Exporter/Trace: Refactor to use TimeLimitedHandler. (#1890)

Author: songy23

buildscripts/import-control.xml

@@ -224,6 +224,7 @@ General guidelines on imports:
     </subpackage>
     <subpackage name="trace">
       <allow pkg="io.opencensus.trace"/>
+      <allow pkg="io.opencensus.exporter.trace.util"/>
       <subpackage name="instana">
         <allow pkg="io.opencensus.exporter.trace.instana"/>
       </subpackage>

exporters/trace/datadog/build.gradle

@@ -9,6 +9,7 @@ dependencies {
     compileOnly libraries.findbugs_annotations
 
     compile project(':opencensus-api'),
+        project(':opencensus-exporter-trace-util'),
         libraries.guava,
         libraries.gson, 
         libraries.auto_value

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

@@ -16,26 +16,19 @@
 
 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;
+import io.opencensus.exporter.trace.util.TimeLimitedHandler;
 import io.opencensus.trace.AttributeValue;
-import io.opencensus.trace.Sampler;
 import io.opencensus.trace.SpanContext;
 import io.opencensus.trace.SpanId;
 import io.opencensus.trace.Status;
-import io.opencensus.trace.Tracer;
 import io.opencensus.trace.Tracing;
 import io.opencensus.trace.export.SpanData;
-import io.opencensus.trace.export.SpanExporter;
-import io.opencensus.trace.samplers.Samplers;
-import java.io.IOException;
 import java.io.OutputStream;
 import java.net.HttpURLConnection;
 import java.net.MalformedURLException;
@@ -47,10 +40,7 @@
 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;
 
@@ -59,10 +49,9 @@
   "AndroidJdkLibsChecker",
   "Java7ApiChecker"
 })
-final class DatadogExporterHandler extends SpanExporter.Handler {
+final class DatadogExporterHandler extends TimeLimitedHandler {
 
-  private static final Tracer tracer = Tracing.getTracer();
-  private static final Sampler probabilitySpampler = Samplers.probabilitySampler(0.0001);
+  private static final String EXPORT_SPAN_NAME = "ExportDatadogTraces";
   private static final Gson gson =
       new GsonBuilder()
           .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
@@ -71,14 +60,13 @@ final class DatadogExporterHandler extends SpanExporter.Handler {
   private final URL agentEndpoint;
   private final String service;
   private final String type;
-  private final Duration deadline;
 
   DatadogExporterHandler(String agentEndpoint, String service, String type, Duration deadline)
       throws MalformedURLException {
+    super(deadline, EXPORT_SPAN_NAME);
     this.agentEndpoint = new URL(agentEndpoint);
     this.service = service;
     this.type = type;
-    this.deadline = deadline;
   }
 
   private static String attributeValueToString(AttributeValue attributeValue) {
@@ -171,32 +159,7 @@ String convertToJson(Collection<SpanData> spanDataList) {
   }
 
   @Override
-  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.
-    try (Scope ss =
-        tracer
-            .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);
-    }
-  }
-
-  private void doExport(Collection<SpanData> spanDataList) throws IOException {
+  public void timeLimitedExport(Collection<SpanData> spanDataList) throws Exception {
     final String data = convertToJson(spanDataList);
 
     final HttpURLConnection connection = (HttpURLConnection) agentEndpoint.openConnection();
@@ -208,16 +171,7 @@ private void doExport(Collection<SpanData> spanDataList) throws IOException {
     outputStream.flush();
     outputStream.close();
     if (connection.getResponseCode() != 200) {
-      handleException(new Exception("Response " + connection.getResponseCode()));
+      throw 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/elasticsearch/build.gradle

@@ -11,6 +11,7 @@ dependencies {
     compileOnly libraries.auto_value
 
     compile project(':opencensus-api'),
+            project(':opencensus-exporter-trace-util'),
             libraries.guava
 
     testCompile project(':opencensus-api')

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

@@ -17,17 +17,9 @@
 package io.opencensus.exporter.trace.elasticsearch;
 
 import com.google.common.io.BaseEncoding;
-import com.google.common.util.concurrent.SimpleTimeLimiter;
-import com.google.common.util.concurrent.TimeLimiter;
-import io.opencensus.common.Duration;
-import io.opencensus.common.Scope;
-import io.opencensus.trace.Sampler;
-import io.opencensus.trace.Status;
-import io.opencensus.trace.Tracer;
-import io.opencensus.trace.Tracing;
+import io.opencensus.exporter.trace.util.TimeLimitedHandler;
 import io.opencensus.trace.export.SpanData;
-import io.opencensus.trace.export.SpanExporter;
-import io.opencensus.trace.samplers.Samplers;
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -37,34 +29,28 @@
 import java.nio.charset.Charset;
 import java.util.Collection;
 import java.util.List;
-import java.util.concurrent.Callable;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
 
-final class ElasticsearchTraceHandler extends SpanExporter.Handler {
+final class ElasticsearchTraceHandler extends TimeLimitedHandler {
 
   private final ElasticsearchTraceConfiguration elasticsearchTraceConfiguration;
   private final String appName;
   private final URL indexUrl;
-  private final Duration deadline;
   private static final String CONTENT_TYPE = "application/json";
   private static final String REQUEST_METHOD = "POST";
   private static final int CONNECTION_TIMEOUT_MILLISECONDS = 6000;
-  private static final Tracer tracer = Tracing.getTracer();
-  private static final Sampler probabilitySampler = Samplers.probabilitySampler(0.0001);
+  private static final String EXPORT_SPAN_NAME = "ExportElasticsearchTraces";
 
   ElasticsearchTraceHandler(ElasticsearchTraceConfiguration elasticsearchTraceConfiguration)
       throws MalformedURLException {
-
+    super(elasticsearchTraceConfiguration.getDeadline(), EXPORT_SPAN_NAME);
     this.elasticsearchTraceConfiguration = elasticsearchTraceConfiguration;
     StringBuilder sb = new StringBuilder();
     sb.append(elasticsearchTraceConfiguration.getElasticsearchUrl()).append("/");
     sb.append(elasticsearchTraceConfiguration.getElasticsearchIndex()).append("/");
     sb.append(elasticsearchTraceConfiguration.getElasticsearchType()).append("/");
     indexUrl = new URL(sb.toString());
     appName = elasticsearchTraceConfiguration.getAppName();
-    deadline = elasticsearchTraceConfiguration.getDeadline();
   }
 
   /**
@@ -73,33 +59,7 @@ final class ElasticsearchTraceHandler extends SpanExporter.Handler {
    * @param spanDataList Collection of {@code SpanData} to be exported.
    */
   @Override
-  public void export(final Collection<SpanData> spanDataList) {
-    Scope scope =
-        tracer
-            .spanBuilder("ExportElasticsearchTraces")
-            .setSampler(probabilitySampler)
-            .setRecordEvents(true)
-            .startScopedSpan();
-    try {
-      TimeLimiter timeLimiter = SimpleTimeLimiter.create(Executors.newSingleThreadExecutor());
-      timeLimiter.callWithTimeout(
-          new Callable<Void>() {
-            @Override
-            public Void call() {
-              doExport(spanDataList);
-              return null;
-            }
-          },
-          deadline.toMillis(),
-          TimeUnit.MILLISECONDS);
-    } catch (Exception e) {
-      handleException(e);
-    } finally {
-      scope.close();
-    }
-  }
-
-  private void doExport(Collection<SpanData> spanDataList) {
+  public void timeLimitedExport(Collection<SpanData> spanDataList) throws Exception {
     List<String> jsonList = JsonConversionUtils.convertToJson(appName, spanDataList);
     if (jsonList.isEmpty()) {
       return;
@@ -130,36 +90,23 @@ private void doExport(Collection<SpanData> spanDataList) {
         outputStream.flush();
         inputStream = connection.getInputStream();
         if (connection.getResponseCode() != 200) {
-          handleException(new Exception("Response " + connection.getResponseCode()));
+          throw new Exception("Response " + connection.getResponseCode());
         }
-      } catch (IOException e) {
-        handleException(e);
-        // dropping span batch
       } finally {
-        if (inputStream != null) {
-          try {
-            inputStream.close();
-          } catch (IOException e) {
-            // ignore
-          }
-        }
-        if (outputStream != null) {
-          try {
-            outputStream.close();
-          } catch (IOException e) {
-            // ignore
-          }
-        }
+        closeStream(inputStream);
+        closeStream(outputStream);
       }
     }
   }
 
-  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()));
+  // Closes an input or output stream and ignores potential IOException.
+  private static void closeStream(@Nullable Closeable stream) {
+    if (stream != null) {
+      try {
+        stream.close();
+      } catch (IOException e) {
+        // ignore
+      }
+    }
   }
 }

exporters/trace/instana/build.gradle

@@ -7,6 +7,7 @@ description = 'OpenCensus Trace Instana Exporter'
 
 dependencies {
     compile project(':opencensus-api'),
+            project(':opencensus-exporter-trace-util'),
             libraries.guava
 
     signature "org.codehaus.mojo.signature:java17:1.0@signature"

exporters/trace/jaeger/build.gradle

@@ -9,6 +9,7 @@ dependencies {
     compileOnly libraries.auto_value
 
     compile project(':opencensus-api'),
+            project(':opencensus-exporter-trace-util'),
             libraries.guava
 
     compile(libraries.jaeger_reporter) {