Exporter/Datadog, Elasticsearch: Add deadline option.

CHANGELOG.md

@@ -9,6 +9,7 @@ Monitoring client to make RPCs.
 with default value 10 seconds.
 - Use `ZipkinExporterConfiguration` for creating `ZipkinTraceExporter`. Provide a `Deadline` option
 with default value 10 seconds.
+- Provide a `Deadline` option to Datadog and Elasticsearch exporter. Default value is 10 seconds.
 
 ## 0.21.0 - 2019-04-30
 - Add HTTP text format serializer to Tag propagation component.

exporters/trace/datadog/src/main/java/io/opencensus/exporter/trace/datadog/DatadogExporterHandler.java

@@ -16,9 +16,12 @@
 
 package io.opencensus.exporter.trace.datadog;
 
+import com.google.common.util.concurrent.SimpleTimeLimiter;
+import com.google.common.util.concurrent.TimeLimiter;
 import com.google.gson.FieldNamingPolicy;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
+import io.opencensus.common.Duration;
 import io.opencensus.common.Functions;
 import io.opencensus.common.Scope;
 import io.opencensus.common.Timestamp;
@@ -44,7 +47,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
@@ -65,12 +71,14 @@ final class DatadogExporterHandler extends SpanExporter.Handler {
   private final URL agentEndpoint;
   private final String service;
   private final String type;
+  private final Duration deadline;
 
-  DatadogExporterHandler(final String agentEndpoint, final String service, final String type)
+  DatadogExporterHandler(String agentEndpoint, String service, String type, Duration deadline)
       throws MalformedURLException {
     this.agentEndpoint = new URL(agentEndpoint);
     this.service = service;
     this.type = type;
+    this.deadline = deadline;
   }
 
   private static String attributeValueToString(AttributeValue attributeValue) {
@@ -163,7 +171,7 @@ String convertToJson(Collection<SpanData> spanDataList) {
   }
 
   @Override
-  public void export(Collection<SpanData> spanDataList) {
+  public void export(final Collection<SpanData> spanDataList) {
     // Start a new span with explicit 1/10000 sampling probability to avoid the case when user
     // sets the default sampler to always sample and we get the gRPC span of the datadog
     // export call always sampled and go to an infinite loop.
@@ -172,29 +180,44 @@ public void export(Collection<SpanData> spanDataList) {
             .spanBuilder("ExportDatadogTraces")
             .setSampler(probabilitySpampler)
             .startScopedSpan()) {
+      TimeLimiter timeLimiter = SimpleTimeLimiter.create(Executors.newSingleThreadExecutor());
+      timeLimiter.callWithTimeout(
+          new Callable<Void>() {
+            @Override
+            public Void call() throws Exception {
+              doExport(spanDataList);
+              return null;
+            }
+          },
+          deadline.toMillis(),
+          TimeUnit.MILLISECONDS);
+    } catch (Exception e) {
+      handleException(e);
+    }
+  }
 
-      final String data = convertToJson(spanDataList);
-
-      final HttpURLConnection connection = (HttpURLConnection) agentEndpoint.openConnection();
-      connection.setRequestMethod("POST");
-      connection.setRequestProperty("Content-Type", "application/json");
-      connection.setDoOutput(true);
-      OutputStream outputStream = connection.getOutputStream();
-      outputStream.write(data.getBytes(Charset.defaultCharset()));
-      outputStream.flush();
-      outputStream.close();
-      if (connection.getResponseCode() != 200) {
-        tracer
-            .getCurrentSpan()
-            .setStatus(Status.UNKNOWN.withDescription("Response " + connection.getResponseCode()));
-      }
-    } catch (IOException e) {
-      tracer
-          .getCurrentSpan()
-          .setStatus(
-              Status.UNKNOWN.withDescription(
-                  e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage()));
-      // drop span batch
+  private void doExport(Collection<SpanData> spanDataList) throws IOException {
+    final String data = convertToJson(spanDataList);
+
+    final HttpURLConnection connection = (HttpURLConnection) agentEndpoint.openConnection();
+    connection.setRequestMethod("POST");
+    connection.setRequestProperty("Content-Type", "application/json");
+    connection.setDoOutput(true);
+    OutputStream outputStream = connection.getOutputStream();
+    outputStream.write(data.getBytes(Charset.defaultCharset()));
+    outputStream.flush();
+    outputStream.close();
+    if (connection.getResponseCode() != 200) {
+      handleException(new Exception("Response " + connection.getResponseCode()));
     }
   }
+
+  private static void handleException(Exception e) {
+    Status status = e instanceof TimeoutException ? Status.DEADLINE_EXCEEDED : Status.UNKNOWN;
+    tracer
+        .getCurrentSpan()
+        .setStatus(
+            status.withDescription(
+                e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage()));
+  }
 }

exporters/trace/datadog/src/main/java/io/opencensus/exporter/trace/datadog/DatadogTraceConfiguration.java

@@ -17,6 +17,9 @@
 package io.opencensus.exporter.trace.datadog;
 
 import com.google.auto.value.AutoValue;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import io.opencensus.common.Duration;
 import javax.annotation.concurrent.Immutable;
 
 /**
@@ -28,6 +31,9 @@
 @Immutable
 public abstract class DatadogTraceConfiguration {
 
+  @VisibleForTesting static final Duration DEFAULT_DEADLINE = Duration.create(10, 0);
+  @VisibleForTesting static final Duration ZERO = Duration.fromMillis(0);
+
   DatadogTraceConfiguration() {}
 
   /**
@@ -54,14 +60,24 @@ public abstract class DatadogTraceConfiguration {
    */
   public abstract String getType();
 
+  /**
+   * Returns the deadline for exporting to Datadog.
+   *
+   * <p>Default value is 10 seconds.
+   *
+   * @return the export deadline.
+   * @since 0.22
+   */
+  public abstract Duration getDeadline();
+
   /**
    * Return a new {@link Builder}.
    *
    * @return a {@code Builder}
    * @since 0.19
    */
   public static Builder builder() {
-    return new AutoValue_DatadogTraceConfiguration.Builder();
+    return new AutoValue_DatadogTraceConfiguration.Builder().setDeadline(DEFAULT_DEADLINE);
   }
 
   /**
@@ -101,6 +117,28 @@ public abstract static class Builder {
      */
     public abstract Builder setType(String type);
 
-    public abstract DatadogTraceConfiguration build();
+    /**
+     * Sets the deadline for exporting to Datadog.
+     *
+     * @param deadline the export deadline.
+     * @return this
+     * @since 0.22
+     */
+    public abstract Builder setDeadline(Duration deadline);
+
+    abstract Duration getDeadline();
+
+    abstract DatadogTraceConfiguration autoBuild();
+
+    /**
+     * Builds a {@link DatadogTraceConfiguration}.
+     *
+     * @return a {@code DatadogTraceConfiguration}.
+     * @since 0.22
+     */
+    public DatadogTraceConfiguration build() {
+      Preconditions.checkArgument(getDeadline().compareTo(ZERO) > 0, "Deadline must be positive.");
+      return autoBuild();
+    }
   }
 }

exporters/trace/datadog/src/main/java/io/opencensus/exporter/trace/datadog/DatadogTraceExporter.java

@@ -32,10 +32,10 @@
  * <pre>{@code
  * public static void main(String[] args) {
  *   DatadogTraceConfiguration config = DatadogTraceConfiguration.builder()
- * .setAgentEndpoint("http://localhost:8126/v0.3/traces")
- * .setService("myService")
- * .setType("web")
- * .build();
+ *     .setAgentEndpoint("http://localhost:8126/v0.3/traces")
+ *     .setService("myService")
+ *     .setType("web")
+ *     .build();
  * DatadogTraceExporter.createAndRegister(config);
  *   ... // Do work.
  * }
@@ -72,7 +72,7 @@ public static void createAndRegister(DatadogTraceConfiguration configuration)
       String type = configuration.getType();
 
       final DatadogExporterHandler exporterHandler =
-          new DatadogExporterHandler(agentEndpoint, service, type);
+          new DatadogExporterHandler(agentEndpoint, service, type, configuration.getDeadline());
       handler = exporterHandler;
       Tracing.getExportComponent()
           .getSpanExporter()

exporters/trace/datadog/src/test/java/io/opencensus/exporter/trace/datadog/DatadogExporterHandlerTest.java

@@ -17,6 +17,7 @@
 package io.opencensus.exporter.trace.datadog;
 
 import static com.google.common.truth.Truth.assertThat;
+import static io.opencensus.exporter.trace.datadog.DatadogTraceConfiguration.DEFAULT_DEADLINE;
 
 import com.google.common.collect.ImmutableMap;
 import io.opencensus.common.Timestamp;
@@ -56,7 +57,8 @@ public class DatadogExporterHandlerTest {
 
   @Before
   public void setup() throws Exception {
-    this.handler = new DatadogExporterHandler("http://localhost", "service", "web");
+    this.handler =
+        new DatadogExporterHandler("http://localhost", "service", "web", DEFAULT_DEADLINE);
   }
 
   @Test

exporters/trace/elasticsearch/src/main/java/io/opencensus/exporter/trace/elasticsearch/ElasticsearchTraceConfiguration.java

@@ -17,8 +17,10 @@
 package io.opencensus.exporter.trace.elasticsearch;
 
 import com.google.auto.value.AutoValue;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
+import io.opencensus.common.Duration;
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.Immutable;
 
@@ -31,14 +33,17 @@
 @Immutable
 public abstract class ElasticsearchTraceConfiguration {
 
+  @VisibleForTesting static final Duration DEFAULT_DEADLINE = Duration.create(10, 0);
+  @VisibleForTesting static final Duration ZERO = Duration.fromMillis(0);
+
   /**
    * Returns a new {@link Builder}.
    *
    * @return a {@code Builder}.
    * @since 0.20.0
    */
   public static Builder builder() {
-    return new AutoValue_ElasticsearchTraceConfiguration.Builder();
+    return new AutoValue_ElasticsearchTraceConfiguration.Builder().setDeadline(DEFAULT_DEADLINE);
   }
 
   /**
@@ -91,6 +96,16 @@ public static Builder builder() {
    */
   public abstract String getElasticsearchType();
 
+  /**
+   * Returns the deadline for exporting to Elasticsearch.
+   *
+   * <p>Default value is 10 seconds.
+   *
+   * @return the export deadline.
+   * @since 0.22
+   */
+  public abstract Duration getDeadline();
+
   /**
    * Builds a {@link ElasticsearchTraceConfiguration}.
    *
@@ -157,6 +172,15 @@ public abstract static class Builder {
      */
     public abstract Builder setElasticsearchType(String elasticsearchType);
 
+    /**
+     * Sets the deadline for exporting to Elasticsearch.
+     *
+     * @param deadline the export deadline.
+     * @return this
+     * @since 0.22
+     */
+    public abstract Builder setDeadline(Duration deadline);
+
     /**
      * Builder for {@link ElasticsearchTraceConfiguration}.
      *
@@ -174,6 +198,9 @@ public ElasticsearchTraceConfiguration build() {
       Preconditions.checkArgument(
           !Strings.isNullOrEmpty(elasticsearchTraceConfiguration.getElasticsearchIndex()),
           "Invalid Elasticsearch type.");
+      Preconditions.checkArgument(
+          elasticsearchTraceConfiguration.getDeadline().compareTo(ZERO) > 0,
+          "Deadline must be positive.");
       return elasticsearchTraceConfiguration;
     }
   }