First pass at adding asynchronous flush support

Author: jaredcnance

build.gradle

@@ -14,117 +14,118 @@
 
 
 plugins {
-	id 'java-library'
-	id 'com.diffplug.spotless' version '5.8.2'
-	id 'maven-publish'
-	id 'signing'
+    id 'java-library'
+    id 'com.diffplug.spotless' version '5.8.2'
+    id 'maven-publish'
+    id 'signing'
 }
 
 group "software.amazon.cloudwatchlogs"
 
 allprojects {
-	compileJava {
-		sourceCompatibility = '1.8'
-		targetCompatibility = '1.8'
-	}
+    compileJava {
+        sourceCompatibility = '1.8'
+        targetCompatibility = '1.8'
+    }
 
-	version = '1.0.4'
+    version = '1.0.4'
 }
 
 java {
-	withJavadocJar()
-	withSourcesJar()
+    withJavadocJar()
+    withSourcesJar()
 }
 
 repositories {
-	jcenter()
-	mavenCentral()
+    jcenter()
+    mavenCentral()
 }
 
 sourceSets {
-	integrationTest {
-		java {
-			compileClasspath += main.output + test.output
-			runtimeClasspath += main.output + test.output
-			srcDir file('src/integration-test/java')
-		}
-		resources.srcDir file('src/integration-test/resources')
-	}
+    integrationTest {
+        java {
+            compileClasspath += main.output + test.output
+            runtimeClasspath += main.output + test.output
+            srcDir file('src/integration-test/java')
+        }
+        resources.srcDir file('src/integration-test/resources')
+    }
 }
 
 configurations {
-	integrationTestImplementation.extendsFrom testImplementation
+    integrationTestImplementation.extendsFrom testImplementation
 }
 
 dependencies {
-	annotationProcessor 'org.projectlombok:lombok:1.18.12'
-
-	compileOnly 'org.projectlombok:lombok:1.18.12'
-	implementation 'com.fasterxml.jackson.core:jackson-core:2.11.1'
-	implementation 'com.fasterxml.jackson.core:jackson-annotations:2.11.1'
-	implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.11.1'
-	implementation 'org.slf4j:slf4j-api:1.7.30'
-
-	// Use JUnit test framework
-	testImplementation 'software.amazon.awssdk:cloudwatch:2.13.54'
-	testImplementation 'junit:junit:4.13'
-	testImplementation 'org.apache.commons:commons-lang3:3.10'
-	testImplementation "org.mockito:mockito-core:2.+"
-	testImplementation "org.powermock:powermock-module-junit4:2.0.2"
-	testImplementation "org.powermock:powermock-api-mockito2:2.0.2"
-	testImplementation "com.github.javafaker:javafaker:1.0.2"
-	testImplementation "com.github.tomakehurst:wiremock-jre8:2.27.0"
-	testImplementation 'software.amazon.awssdk:cloudwatch:2.13.54'
-	testCompileOnly 'org.projectlombok:lombok:1.18.12'
-	testAnnotationProcessor 'org.projectlombok:lombok:1.18.12'
+    annotationProcessor 'org.projectlombok:lombok:1.18.12'
+
+    compileOnly 'org.projectlombok:lombok:1.18.12'
+    implementation 'com.fasterxml.jackson.core:jackson-core:2.11.1'
+    implementation 'com.fasterxml.jackson.core:jackson-annotations:2.11.1'
+    implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.11.1'
+    implementation 'org.slf4j:slf4j-api:1.7.30'
+    implementation 'org.javatuples:javatuples:1.2'
+
+    // Use JUnit test framework
+    testImplementation 'software.amazon.awssdk:cloudwatch:2.13.54'
+    testImplementation 'junit:junit:4.13'
+    testImplementation 'org.apache.commons:commons-lang3:3.10'
+    testImplementation "org.mockito:mockito-core:2.+"
+    testImplementation "org.powermock:powermock-module-junit4:2.0.2"
+    testImplementation "org.powermock:powermock-api-mockito2:2.0.2"
+    testImplementation "com.github.javafaker:javafaker:1.0.2"
+    testImplementation "com.github.tomakehurst:wiremock-jre8:2.27.0"
+    testImplementation 'software.amazon.awssdk:cloudwatch:2.13.54'
+    testCompileOnly 'org.projectlombok:lombok:1.18.12'
+    testAnnotationProcessor 'org.projectlombok:lombok:1.18.12'
 }
 
 spotless {
-	format 'misc', {
-		target '*.gradle', '*.md', '.gitignore'
+    format 'misc', {
+        target '*.gradle', '*.md', '.gitignore'
 
-		trimTrailingWhitespace()
-		indentWithTabs()
-		endWithNewline()
-	}
+        trimTrailingWhitespace()
+        indentWithTabs()
+        endWithNewline()
+    }
 
-	java {
-		importOrder()
-		googleJavaFormat('1.7').aosp()
-		removeUnusedImports()
+    java {
+        importOrder()
+        googleJavaFormat('1.7').aosp()
+        removeUnusedImports()
 
-	}
+    }
 }
 
 test {
-	outputs.upToDateWhen {false}
+    outputs.upToDateWhen {false}
 }
 
 jar {
-	manifest {
-		attributes 'Implementation-Version': archiveVersion.get()
-	}
+    manifest {
+        attributes 'Implementation-Version': archiveVersion.get()
+    }
 }
 
 
 task integrationTest(type: Test) {
-	testClassesDirs = sourceSets.integrationTest.output.classesDirs
-	classpath = sourceSets.integrationTest.runtimeClasspath
-	testLogging.showStandardStreams = true
+    testClassesDirs = sourceSets.integrationTest.output.classesDirs
+    classpath = sourceSets.integrationTest.runtimeClasspath
+    testLogging.showStandardStreams = true
 }
 
 task integ(type: Exec) {
-	commandLine './bin/run-integ-tests.sh'
+    commandLine './bin/run-integ-tests.sh'
 }
 
 
 tasks.withType(JavaCompile) {
-	options.compilerArgs << "-Xlint:unchecked" << "-Xlint:deprecation"
+    options.compilerArgs << "-Xlint:unchecked" << "-Xlint:deprecation"
 }
 
 tasks.named('wrapper') {
-	gradleVersion = '6.5.1'
-	distributionType = Wrapper.DistributionType.ALL
+    gradleVersion = '6.5.1'
+    distributionType = Wrapper.DistributionType.ALL
 }
 
 def repoUrl = "https://aws.oss.sonatype.org/service/local/staging/deploy/maven2/"
@@ -134,66 +135,66 @@ def buildNumber = System.getenv('CODEBUILD_BUILD_NUMBER')
 
 
 publishing {
-	publications {
-		mavenJava(MavenPublication) {
-			ext.get_version = {
-				if (buildNumber == null) {
-					return version + ".dev"
-				} else {
-					return version + "." + buildNumber
-				}
-			}
-			//TODO: add the build version properly
-			//version = get_version()
-			from components.java
-			versionMapping {
-				usage('java-api') {
-					fromResolutionOf('runtimeClasspath')
-				}
-				usage('java-runtime') {
-					fromResolutionResult()
-				}
-			}
-			pom {
-				name = 'aws-embedded-metrics'
-				description = 'aws-embedded-metrics for java'
-				url = 'https://github.com/awslabs/aws-embedded-metrics-java'
-				scm {
-					url = 'https://github.com/awslabs/aws-embedded-metrics-java'
-					connection = 'https://github.com/awslabs/aws-embedded-metrics-java'
-					developerConnection = 'https://github.com/awslabs/aws-embedded-metrics-java'
-				}
-				licenses {
-					license {
-						name = 'The Apache License, Version 2.0'
-						url = 'https://github.com/awslabs/aws-embedded-metrics-java/blob/master/LICENSE'
-					}
-				}
-				developers {
-					developer {
-						id = 'aws_emf'
-						name = 'AWS CloudWatch'
-						email = 'logs-emf@amazon.com'
-					}
-				}
-			}
-		}
-	}
-	repositories {
-		maven {
-			name = "mavenRepo"
-			url = repoUrl
-			credentials {
-				username = repoUserName
-				password = repoPassword
-			}
-		}
-	}
+    publications {
+        mavenJava(MavenPublication) {
+            ext.get_version = {
+                if (buildNumber == null) {
+                    return version + ".dev"
+                } else {
+                    return version + "." + buildNumber
+                }
+            }
+            //TODO: add the build version properly
+            //version = get_version()
+            from components.java
+            versionMapping {
+                usage('java-api') {
+                    fromResolutionOf('runtimeClasspath')
+                }
+                usage('java-runtime') {
+                    fromResolutionResult()
+                }
+            }
+            pom {
+                name = 'aws-embedded-metrics'
+                description = 'aws-embedded-metrics for java'
+                url = 'https://github.com/awslabs/aws-embedded-metrics-java'
+                scm {
+                    url = 'https://github.com/awslabs/aws-embedded-metrics-java'
+                    connection = 'https://github.com/awslabs/aws-embedded-metrics-java'
+                    developerConnection = 'https://github.com/awslabs/aws-embedded-metrics-java'
+                }
+                licenses {
+                    license {
+                        name = 'The Apache License, Version 2.0'
+                        url = 'https://github.com/awslabs/aws-embedded-metrics-java/blob/master/LICENSE'
+                    }
+                }
+                developers {
+                    developer {
+                        id = 'aws_emf'
+                        name = 'AWS CloudWatch'
+                        email = 'logs-emf@amazon.com'
+                    }
+                }
+            }
+        }
+    }
+    repositories {
+        maven {
+            name = "mavenRepo"
+            url = repoUrl
+            credentials {
+                username = repoUserName
+                password = repoPassword
+            }
+        }
+    }
 }
 
 signing {
-	def signingKey = findProperty("signingKey")
-	def signingPassword = findProperty("signingPassword")
-	useInMemoryPgpKeys(signingKey, signingPassword)
-	sign publishing.publications.mavenJava
+    def signingKey = findProperty("signingKey")
+    def signingPassword = findProperty("signingPassword")
+    useInMemoryPgpKeys(signingKey, signingPassword)
+    sign publishing.publications.mavenJava
 }

examples/agent/src/main/java/agent/App.java

@@ -1,16 +1,22 @@
 package agent;
 
+import software.amazon.cloudwatchlogs.emf.config.EnvironmentConfigurationProvider;
+import software.amazon.cloudwatchlogs.emf.environment.DefaultEnvironment;
 import software.amazon.cloudwatchlogs.emf.logger.MetricsLogger;
 import software.amazon.cloudwatchlogs.emf.model.DimensionSet;
 import software.amazon.cloudwatchlogs.emf.model.Unit;
 
+import java.util.concurrent.TimeUnit;
+
 public class App {
 
     public static void main(String[] args) {
-        MetricsLogger logger = new MetricsLogger();
+        DefaultEnvironment environment = new DefaultEnvironment(EnvironmentConfigurationProvider.getConfig());
+        MetricsLogger logger = new MetricsLogger(environment);
         logger.putDimensions(DimensionSet.of("Operation", "Agent"));
         logger.putMetric("ExampleMetric", 100, Unit.MILLISECONDS);
         logger.putProperty("RequestId", "422b1569-16f6-4a03-b8f0-fe3fd9b100f8");
         logger.flush();
+        environment.getSink().shutdown().orTimeout(1000L, TimeUnit.MILLISECONDS);
     }
 }

src/main/java/software/amazon/cloudwatchlogs/emf/Constants.java

@@ -24,4 +24,10 @@ public class Constants {
     public static final int MAX_METRICS_PER_EVENT = 100;
 
     public static final int MAX_DATAPOINTS_PER_METRIC = 100;
+
+    public static final int DEFAULT_ASYNC_BUFFER_SIZE = 100;
+
+    public static final int MAX_ATTEMPTS_PER_MESSAGE = 10;
+
+    public static final int MAX_BACKOFF_MILLIS = 2000;
 }

src/main/java/software/amazon/cloudwatchlogs/emf/config/Configuration.java

@@ -18,8 +18,10 @@
 
 import java.util.Optional;
 import lombok.AllArgsConstructor;
+import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
+import software.amazon.cloudwatchlogs.emf.Constants;
 import software.amazon.cloudwatchlogs.emf.environment.Environments;
 import software.amazon.cloudwatchlogs.emf.util.StringUtils;
 
@@ -54,6 +56,11 @@ public class Configuration {
      */
     @Setter Environments environmentOverride;
 
+    /**
+     * Queue length for asynchronous sinks.
+     */
+    @Setter @Getter int asyncBufferSize = Constants.DEFAULT_ASYNC_BUFFER_SIZE;
+
     public Optional<String> getServiceName() {
         return getStringOptional(serviceName);
     }
@@ -85,6 +92,6 @@ private Optional<String> getStringOptional(String value) {
         if (StringUtils.isNullOrEmpty(value)) {
             return Optional.empty();
         }
-        return Optional.ofNullable(value);
+        return Optional.of(value);
     }
 }

src/main/java/software/amazon/cloudwatchlogs/emf/config/ConfigurationKeys.java

@@ -27,4 +27,5 @@ public class ConfigurationKeys {
     public static final String LOG_STREAM_NAME = "LOG_STREAM_NAME";
     public static final String AGENT_ENDPOINT = "AGENT_ENDPOINT";
     public static final String ENVIRONMENT_OVERRIDE = "ENVIRONMENT";
+    public static final String ASYNC_BUFFER_SIZE = "ASYNC_BUFFER_SIZE";
 }