open-telemetry · trask · Jul 11, 2024 · May 28, 2024 · May 28, 2024 · Jun 6, 2024
@@ -70,3 +70,7 @@ components:
     - jackshirazi
     - jonaskunz
     - sylvainjuge
+  inferred-spans:
+    - jackshirazi
+    - jonaskunz
+    - sylvainjuge
@@ -57,7 +57,12 @@ val DEPENDENCIES = listOf(
   "org.junit-pioneer:junit-pioneer:1.9.1",
   "org.skyscreamer:jsonassert:1.5.1",
   "org.apache.kafka:kafka-clients:3.7.0",
-  "org.testcontainers:kafka:1.19.8"
+  "org.testcontainers:kafka:1.19.8",
+  "com.lmax:disruptor:3.4.4",
+  "org.jctools:jctools-core:4.0.3",
+  "tools.profiler:async-profiler:3.0",
+  "com.blogspot.mydailyjava:weak-lock-free:0.18",
+  "org.agrona:agrona:1.21.2"
 )
 
 javaPlatform {

@@ -0,0 +1,45 @@
+import net.ltgt.gradle.errorprone.errorprone
+
+plugins {
+  id("otel.java-conventions")
+  id("otel.publish-conventions")
+}
+
+description = "OpenTelemetry Java profiling based inferred spans module"
+otelJava.moduleName.set("io.opentelemetry.contrib.inferredspans")
+
+dependencies {
+  annotationProcessor("com.google.auto.service:auto-service")
+  compileOnly("com.google.auto.service:auto-service-annotations")
+  compileOnly("io.opentelemetry:opentelemetry-sdk")
+  compileOnly("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi")
+  implementation("com.lmax:disruptor")
+  implementation("org.jctools:jctools-core")
+  implementation("tools.profiler:async-profiler")
+  implementation("com.blogspot.mydailyjava:weak-lock-free")
+  implementation("org.agrona:agrona")
+
+  testAnnotationProcessor("com.google.auto.service:auto-service")
+  testCompileOnly("com.google.auto.service:auto-service-annotations")
+  testImplementation("io.opentelemetry.semconv:opentelemetry-semconv-incubating:1.25.0-alpha")
+  testImplementation("io.opentelemetry:opentelemetry-sdk")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-extension-autoconfigure")
+  testImplementation("io.opentelemetry:opentelemetry-sdk-testing")
+  testImplementation("io.opentelemetry:opentelemetry-api-incubator")
+  testImplementation("io.opentelemetry:opentelemetry-exporter-logging")
+}
+
+tasks {
+  withType<JavaCompile>().configureEach {
+    with(options) {
+      errorprone {
+        // This code uses nullable reference in many places due to performance
+        // and makes assumptions of when these references are non-null
+        // In the code we express those assumptions as assertions
+        // instead of Object.requireNonNull because the NPEs raised by actual
+        // null dereferencing are more helpful than the ones raised by Object.requireNonNull
+        option("NullAway:AssertsEnabled", "true")
+      }
+    }
+  }
+}
@@ -0,0 +1,113 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.inferredspans;
+
+import com.google.auto.service.AutoService;
+import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer;
+import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+@AutoService(AutoConfigurationCustomizerProvider.class)
+public class InferredSpansAutoConfig implements AutoConfigurationCustomizerProvider {
+
+  private static final Logger log = Logger.getLogger(InferredSpansAutoConfig.class.getName());
+
+  static final String ENABLED_OPTION = "otel.inferred.spans.enabled";
+  static final String LOGGING_OPTION = "otel.inferred.spans.logging.enabled";
+  static final String DIAGNOSTIC_FILES_OPTION = "otel.inferred.spans.backup.diagnostic.files";
+  static final String SAFEMODE_OPTION = "otel.inferred.spans.safe.mode";
+  static final String POSTPROCESSING_OPTION = "otel.inferred.spans.post.processing.enabled";
+  static final String SAMPLING_INTERVAL_OPTION = "otel.inferred.spans.sampling.interval";
+  static final String MIN_DURATION_OPTION = "otel.inferred.spans.min.duration";
+  static final String INCLUDED_CLASSES_OPTION = "otel.inferred.spans.included.classes";
+  static final String EXCLUDED_CLASSES_OPTION = "otel.inferred.spans.excluded.classes";
+  static final String INTERVAL_OPTION = "otel.inferred.spans.interval";
+  static final String DURATION_OPTION = "otel.inferred.spans.duration";
+  static final String LIB_DIRECTORY_OPTION = "otel.inferred.spans.lib.directory";
+
+  @Override
+  public void customize(AutoConfigurationCustomizer config) {
+    config.addTracerProviderCustomizer(
+        (providerBuilder, properties) -> {
+          if (properties.getBoolean(ENABLED_OPTION, false)) {
+            InferredSpansProcessorBuilder builder = InferredSpansProcessor.builder();
+
+            PropertiesApplier applier = new PropertiesApplier(properties);
+
+            applier.applyBool(LOGGING_OPTION, builder::profilerLoggingEnabled);
+            applier.applyBool(DIAGNOSTIC_FILES_OPTION, builder::backupDiagnosticFiles);
+            applier.applyInt(SAFEMODE_OPTION, builder::asyncProfilerSafeMode);
+            applier.applyBool(POSTPROCESSING_OPTION, builder::postProcessingEnabled);
+            applier.applyDuration(SAMPLING_INTERVAL_OPTION, builder::samplingInterval);
+            applier.applyDuration(MIN_DURATION_OPTION, builder::inferredSpansMinDuration);
+            applier.applyWildcards(INCLUDED_CLASSES_OPTION, builder::includedClasses);
+            applier.applyWildcards(EXCLUDED_CLASSES_OPTION, builder::excludedClasses);
+            applier.applyDuration(INTERVAL_OPTION, builder::profilerInterval);
+            applier.applyDuration(DURATION_OPTION, builder::profilingDuration);
+            applier.applyString(LIB_DIRECTORY_OPTION, builder::profilerLibDirectory);
+
+            providerBuilder.addSpanProcessor(builder.build());
+          } else {
+            log.finest(
+                "Not enabling inferred spans processor because " + ENABLED_OPTION + " is not set");
+          }
+          return providerBuilder;
+        });
+  }
+
+  private static class PropertiesApplier {
+
+    private final ConfigProperties properties;
+
+    public PropertiesApplier(ConfigProperties properties) {
+      this.properties = properties;
+    }
+
+    public void applyBool(String configKey, Consumer<Boolean> funcToApply) {
+      applyValue(properties.getBoolean(configKey), funcToApply);
+    }
+
+    public void applyInt(String configKey, Consumer<Integer> funcToApply) {
+      applyValue(properties.getInt(configKey), funcToApply);
+    }
+
+    public void applyDuration(String configKey, Consumer<Duration> funcToApply) {
+      applyValue(properties.getDuration(configKey), funcToApply);
+    }
+
+    public void applyString(String configKey, Consumer<String> funcToApply) {
+      applyValue(properties.getString(configKey), funcToApply);
+    }
+
+    public void applyWildcards(
+        String configKey, Consumer<? super List<WildcardMatcher>> funcToApply) {
+      String wildcardListString = properties.getString(configKey);
+      if (wildcardListString != null && !wildcardListString.isEmpty()) {
+        List<WildcardMatcher> values =
+            Arrays.stream(wildcardListString.split(","))
+                .filter(str -> !str.isEmpty())
+                .map(WildcardMatcher::valueOf)
+                .collect(Collectors.toList());
+        if (!values.isEmpty()) {
+          funcToApply.accept(values);
+        }
+      }
+    }
+
+    private static <T> void applyValue(@Nullable T value, Consumer<T> funcToApply) {
+      if (value != null) {
+        funcToApply.accept(value);
+      }
+    }
+  }
+}
@@ -0,0 +1,143 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.inferredspans;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.api.trace.TracerProvider;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.contrib.inferredspans.internal.InferredSpansConfiguration;
+import io.opentelemetry.contrib.inferredspans.internal.SamplingProfiler;
+import io.opentelemetry.contrib.inferredspans.internal.SpanAnchoredClock;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.ReadWriteSpan;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.SpanProcessor;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.Nullable;
+
+public class InferredSpansProcessor implements SpanProcessor {
+
+  private static final Logger logger = Logger.getLogger(InferredSpansProcessor.class.getName());
+
+  public static final String TRACER_NAME = "inferred-spans";
+
+  public static final String TRACER_VERSION = readInferredSpansVersion();
+
+  // Visible for testing
+  final SamplingProfiler profiler;
+
+  private Supplier<TracerProvider> tracerProvider = GlobalOpenTelemetry::getTracerProvider;
+
+  @Nullable private volatile Tracer tracer;
+
+  InferredSpansProcessor(
+      InferredSpansConfiguration config,
+      SpanAnchoredClock clock,
+      boolean startScheduledProfiling,
+      @Nullable File activationEventsFile,
+      @Nullable File jfrFile) {
+    profiler = new SamplingProfiler(config, clock, this::getTracer, activationEventsFile, jfrFile);
+    if (startScheduledProfiling) {
+      profiler.start();
+    }
+  }
+
+  public static InferredSpansProcessorBuilder builder() {
+    return new InferredSpansProcessorBuilder();
+  }
+
+  /**
+   * Allows customization of the TraceProvider to use. If not set, a TraceProvider from {@link
+   * GlobalOpenTelemetry} will be used.
+   *
+   * @param provider the provider to use. Null means that {@link GlobalOpenTelemetry} will be used
+   *     lazily.
+   */
+  public synchronized void setTracerProvider(TracerProvider provider) {
+    if (provider == null) {
+      this.tracerProvider = GlobalOpenTelemetry::getTracerProvider;
+    } else {
+      this.tracerProvider = () -> provider;
+    }
+  }
+
+  @Override
+  public void onStart(Context parentContext, ReadWriteSpan span) {
+    profiler.getClock().onSpanStart(span, parentContext);
+  }
+
+  @Override
+  public boolean isStartRequired() {
+    return true;
+  }
+
+  @Override
+  public void onEnd(ReadableSpan span) {}
+
+  @Override
+  public boolean isEndRequired() {
+    return false;
+  }
+
+  @Override
+  @SuppressWarnings({"FutureReturnValueIgnored", "InterruptedExceptionSwallowed"})
+  public CompletableResultCode shutdown() {
+    CompletableResultCode result = new CompletableResultCode();
+    logger.fine("Stopping Inferred Spans Processor");
+    ThreadFactory threadFactory =
+        r -> {
+          Thread thread = new Thread(r);
+          thread.setDaemon(false);
+          thread.setName("otel-inferred-spans-shutdown");
+          return thread;
+        };
+    Executors.newSingleThreadExecutor(threadFactory)
+        .submit(
+            () -> {
+              try {
+                profiler.stop();
+                result.succeed();
+              } catch (Throwable e) {
+                logger.log(Level.SEVERE, "Failed to stop Inferred Spans Processor", e);
+                result.fail();
+              }
+            });
+    return result;
+  }
+
+  private Tracer getTracer() {
+    if (tracer == null) {
+      synchronized (this) {
+        if (tracer == null) {
+          tracer = tracerProvider.get().get(TRACER_NAME, TRACER_VERSION);
+        }
+      }
+    }
+    return tracer;
+  }
+
+  private static String readInferredSpansVersion() {
+    try (InputStream is = InferredSpansProcessor.class.getResourceAsStream("version.properties")) {
+      Properties properties = new Properties();
+      properties.load(is);
+      String version = (String) properties.get("contrib.version");
+      Objects.requireNonNull(version);
+      return version;
+    } catch (IOException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+}