package de.jeha.s3pt;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.Protocol;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import de.jeha.s3pt.operations.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author jenshadlich@googlemail.com
 */
public class S3PerformanceTest implements Runnable {

    private static final Logger LOG = LoggerFactory.getLogger(S3PerformanceTest.class);

    private final String accessKey;
    private final String secretKey;
    private final String endpointUrl;
    private final String bucketName;
    private final Operation operation;
    private final int threads;
    private final int n;
    private final int size;
    private final boolean useHttp;
    private final boolean useGzip;
    private final boolean useOldS3Signer;
    private final boolean useKeepAlive;

    /**
     * @param accessKey
     * @param secretKey
     * @param endpointUrl
     * @param bucketName
     * @param operation
     * @param threads
     * @param n
     * @param size
     * @param useHttp
     * @param useGzip
     * @param useOldS3Signer
     * @param useKeepAlive
     */
    public S3PerformanceTest(String accessKey, String secretKey, String endpointUrl, String bucketName,
                             Operation operation, int threads, int n, int size, boolean useHttp, boolean useGzip,
                             boolean useOldS3Signer, boolean useKeepAlive) {
        this.accessKey = accessKey;
        this.secretKey = secretKey;
        this.endpointUrl = endpointUrl;
        this.bucketName = bucketName;
        this.operation = operation;
        this.threads = threads;
        this.n = n;
        this.size = size;
        this.useHttp = useHttp;
        this.useGzip = useGzip;
        this.useOldS3Signer = useOldS3Signer;
        this.useKeepAlive = useKeepAlive;
    }

    @Override
    public void run() {
        AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretKey);

        ClientConfiguration clientConfig = new ClientConfiguration()
                .withProtocol(useHttp ? Protocol.HTTP : Protocol.HTTPS)
                .withUserAgent("s3pt")
                .withGzip(useGzip)
                .withTcpKeepAlive(useKeepAlive);

        if (useOldS3Signer) {
            clientConfig.setSignerOverride("S3Signer");
        }

        AmazonS3 s3Client = new AmazonS3Client(credentials, clientConfig);
        s3Client.setEndpoint(endpointUrl);

        ExecutorService executorService = Executors.newFixedThreadPool(threads);

        List<Callable<OperationResult>> operations = new ArrayList<>();
        if (operation.isMultiThreaded()) {
            for (int i = 0; i < threads; i++) {
                operations.add(createOperation(operation, s3Client));
            }
        } else {
            if (threads > 1) {
                LOG.warn("operation {} does not support multiple threads, use single thread", operation);
            }
            operations.add(createOperation(operation, s3Client));
        }

        try {
            List<Future<OperationResult>> futureResults = executorService.invokeAll(operations);

            List<OperationResult> results = new ArrayList<>();
            for (Future<OperationResult> result : futureResults) {
                results.add(result.get());
            }

            printResults(results);

        } catch (InterruptedException | ExecutionException e) {
            LOG.error("An error occurred", e);
        }

        executorService.shutdown();

        LOG.info("Done");
    }

    private AbstractOperation createOperation(Operation operation, AmazonS3 s3Client) {
        switch (operation) {
            case CLEAR_BUCKET:
                return new ClearBucket(s3Client, bucketName, n);
            case CREATE_BUCKET:
                return new CreateBucket(s3Client, bucketName);
            case RANDOM_READ:
                return new RandomRead(s3Client, bucketName, n);
            case UPLOAD:
                return new Upload(s3Client, bucketName, n, size);
            case UPLOAD_AND_READ:
                return new UploadAndRead(s3Client, bucketName, n, size);
            default:
                throw new UnsupportedOperationException("Unknown operation: " + operation);
        }
    }

    private void printResults(List<OperationResult> results) {
        int min = (int) results.stream().mapToDouble(x -> x.getStats().getMin()).average().orElse(0.0);
        int max = (int) results.stream().mapToDouble(x -> x.getStats().getMax()).average().orElse(0.0);
        int avg = (int) results.stream().mapToDouble(x -> x.getStats().getGeometricMean()).average().orElse(0.0);
        int p50 = (int) results.stream().mapToDouble(x -> x.getStats().getPercentile(50)).average().orElse(0.0);
        int p75 = (int) results.stream().mapToDouble(x -> x.getStats().getPercentile(75)).average().orElse(0.0);
        int p95 = (int) results.stream().mapToDouble(x -> x.getStats().getPercentile(95)).average().orElse(0.0);
        int p98 = (int) results.stream().mapToDouble(x -> x.getStats().getPercentile(98)).average().orElse(0.0);
        int p99 = (int) results.stream().mapToDouble(x -> x.getStats().getPercentile(99)).average().orElse(0.0);
        double ops = results.stream().mapToDouble(x -> x.getStats().getN() / x.getStats().getSum() * 1000).sum();

        LOG.info("Operation statistics:");
        LOG.info("min = {} ms", min);
        LOG.info("max = {} ms", max);
        LOG.info("avg = {} ms", avg);
        LOG.info("p50 = {} ms", p50);
        LOG.info("p75 = {} ms", p75);
        LOG.info("p95 = {} ms", p95);
        LOG.info("p98 = {} ms", p98);
        LOG.info("p99 = {} ms", p99);
        LOG.info("throughput = {} operations/s ({} threads)", (int) ops, threads);
    }

}

@@ -1088,16 +1088,32 @@
 ccessKey
+      access key
 %0A     * 
@@ -1128,16 +1128,32 @@
 ecretKey
+      secret key
 %0A     * 
@@ -1170,16 +1170,57 @@
 pointUrl
+    endpoint url, e.g. 's3.amazonaws.com'
 %0A     * 
@@ -1236,16 +1236,35 @@
 cketName
+     name of bucket
 %0A     * 
@@ -1279,16 +1279,31 @@
 peration
+      operation
 %0A     * 
@@ -1320,116 +1320,355 @@
 eads
-%0A     * @param n%0A     * @param size%0A     * @param useHttp%0A     * @param useGzip%0A     * @param useOldS3Signer
+        number of threads%0A     * @param n              number of operations%0A     * @param size           size (if applicable), e.g. for UPLOAD operation%0A     * @param useHttp        switch to HTTP when%0A     * @param useGzip        enable GZIP compression%0A     * @param useOldS3Signer use %22old%22 S3Signer for endpoints that do not support v4 signing
 %0A   
@@ -1690,16 +1690,37 @@
 eepAlive
+   use TCP keep alive
 %0A     */

update javadoc