Stop using OpenTelemetry SDK internal AttributesProcessor for pre-aggregated metrics

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/init/AiInstrumenterCustomizerProvider.java

@@ -5,12 +5,15 @@
 
 import com.google.auto.service.AutoService;
 import com.microsoft.applicationinsights.agent.bootstrap.preagg.AiContextCustomizerHolder;
+import com.microsoft.applicationinsights.agent.internal.configuration.Configuration;
+import com.microsoft.applicationinsights.agent.internal.preagg.PreAggregatedMetricsCustomizer;
 import io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizer;
 import io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizerProvider;
 
 /**
  * InstrumenterCustomizerProvider implementation that adds Application Insights context customizer
- * to all instrumenters. This replaces the need for copying the InstrumenterBuilder class.
+ * to all instrumenters, and the pre-aggregated metrics customizer to http and rpc instrumenters.
+ * This replaces the need for copying the InstrumenterBuilder class.
  */
 @AutoService(InstrumenterCustomizerProvider.class)
 public class AiInstrumenterCustomizerProvider implements InstrumenterCustomizerProvider {
@@ -20,5 +23,10 @@ public void customize(InstrumenterCustomizer customizer) {
     // Add the Application Insights context customizer to all instrumenters
     // This was previously done by modifying the InstrumenterBuilder constructor
     customizer.addContextCustomizer(AiContextCustomizerHolder.getInstance());
+
+    Configuration configuration = FirstEntryPoint.getConfiguration();
+    if (configuration != null && configuration.internal.preAggregatedStandardMetrics.enabled) {
+      PreAggregatedMetricsCustomizer.customize(customizer);
+    }
   }
 }

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/init/SecondEntryPoint.java

@@ -39,6 +39,8 @@
 import com.microsoft.applicationinsights.agent.internal.exporter.AgentSpanExporter;
 import com.microsoft.applicationinsights.agent.internal.httpclient.LazyHttpClient;
 import com.microsoft.applicationinsights.agent.internal.legacyheaders.AiLegacyHeaderSpanProcessor;
+import com.microsoft.applicationinsights.agent.internal.preagg.AiViewRegistry;
+import com.microsoft.applicationinsights.agent.internal.preagg.PreAggregatedMetricsCustomizer;
 import com.microsoft.applicationinsights.agent.internal.processors.ExporterWithLogProcessor;
 import com.microsoft.applicationinsights.agent.internal.processors.ExporterWithSpanProcessor;
 import com.microsoft.applicationinsights.agent.internal.processors.LogExporterWithAttributeProcessor;
@@ -65,7 +67,6 @@
 import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
 import io.opentelemetry.sdk.metrics.View;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
-import io.opentelemetry.sdk.metrics.internal.view.AiViewRegistry;
 import io.opentelemetry.sdk.trace.SdkTracerProviderBuilder;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 import io.opentelemetry.sdk.trace.samplers.SamplingDecision;
@@ -276,6 +277,9 @@ public void customize(AutoConfigurationCustomizer autoConfiguration) {
         })
         .addMetricExporterCustomizer((metricExporter, configProperties) -> {
           if (metricExporter instanceof AzureMonitorMetricExporterProvider.MarkerMetricExporter) {
+            // the metric exporters are created when the SDK is built, which is before any
+            // instrumenter is built
+            PreAggregatedMetricsCustomizer.setAzureMonitorMetricExporterEnabled(true);
             return buildMetricExporter(
                 configuration, telemetryClient, metricFilters, configProperties);
           } else {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/preagg/AiViewRegistry.java

@@ -0,0 +1,23 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.applicationinsights.agent.internal.preagg;
+
+import io.opentelemetry.sdk.metrics.InstrumentSelector;
+import io.opentelemetry.sdk.metrics.SdkMeterProviderBuilder;
+import io.opentelemetry.sdk.metrics.View;
+
+// the Application Insights specific attributes are added by PreAggregatedMetricsCustomizer,
+// these views only limit the metric attributes to the ones used by the pre-aggregated metrics
+public class AiViewRegistry {
+
+  public static void registerViews(SdkMeterProviderBuilder builder) {
+    for (MetricView view : MetricView.values()) {
+      builder.registerView(
+          InstrumentSelector.builder().setName(view.getInstrumentName()).build(),
+          View.builder().setAttributeFilter(view.getRetainedAttributeKeys()).build());
+    }
+  }
+
+  private AiViewRegistry() {}
+}

agent/agent-tooling/src/main/java/io/opentelemetry/sdk/metrics/internal/view/MetricView.java → agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/preagg/MetricView.java

@@ -1,8 +1,9 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-package io.opentelemetry.sdk.metrics.internal.view;
+package com.microsoft.applicationinsights.agent.internal.preagg;
 
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.AiSemanticAttributes;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.semconv.HttpAttributes;
 import io.opentelemetry.semconv.ServerAttributes;
@@ -34,12 +35,18 @@ String getInstrumentName() {
     return instrumentName;
   }
 
-  Set<AttributeKey<?>> getAttributeKeys() {
-    return attributeKeys;
-  }
-
-  boolean isCaptureSynthetic() {
-    return captureSynthetic;
+  // the view attribute keys plus the internal attributes added by PreAggregatedMetricsCustomizer
+  Set<String> getRetainedAttributeKeys() {
+    Set<String> keys = new HashSet<>();
+    for (AttributeKey<?> key : attributeKeys) {
+      keys.add(key.getKey());
+    }
+    keys.add(AiSemanticAttributes.INTERNAL_CONNECTION_STRING.getKey());
+    keys.add(AiSemanticAttributes.INTERNAL_ROLE_NAME.getKey());
+    if (captureSynthetic) {
+      keys.add(AiSemanticAttributes.IS_SYNTHETIC.getKey());
+    }
+    return keys;
   }
 
   private static Set<AttributeKey<?>> httpClientDurationAttributeKeys() {

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/preagg/MetricsState.java

@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.applicationinsights.agent.internal.preagg;
+
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.ContextKey;
+import io.opentelemetry.instrumentation.api.incubator.semconv.rpc.RpcClientMetrics;
+import io.opentelemetry.instrumentation.api.incubator.semconv.rpc.RpcServerMetrics;
+import io.opentelemetry.instrumentation.api.semconv.http.HttpClientMetrics;
+import io.opentelemetry.instrumentation.api.semconv.http.HttpServerMetrics;
+import java.lang.reflect.Field;
+import javax.annotation.Nullable;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Detects whether the upstream duration metric will be recorded for an operation.
+ *
+ * <p>Some instrumenters have http or rpc attributes but don't record the duration metrics (e.g. AWS
+ * SDK, RMI). The upstream metrics store their per-operation state in the context in onStart() and
+ * only record the duration in onEnd() when that state is present, so checking for the state tells
+ * whether the pre-aggregated metric will be recorded.
+ *
+ * <p>The state is inherited by nested operations, so the state that was already in the parent
+ * context is captured before the operation listeners run (see {@link #captureInherited}), and only
+ * a state created by the current operation counts.
+ */
+enum MetricsState {
+  HTTP_SERVER(HttpServerMetrics.class, "HTTP_SERVER_METRICS_STATE"),
+  HTTP_CLIENT(HttpClientMetrics.class, "HTTP_CLIENT_REQUEST_METRICS_STATE"),
+  RPC_SERVER(RpcServerMetrics.class, "RPC_SERVER_REQUEST_METRICS_STATE"),
+  RPC_CLIENT(RpcClientMetrics.class, "RPC_CLIENT_REQUEST_METRICS_STATE");
+
+  @SuppressWarnings("ImmutableEnumChecker") // ContextKey is immutable
+  @Nullable
+  private final ContextKey<?> contextKey;
+
+  @SuppressWarnings("ImmutableEnumChecker") // ContextKey is immutable
+  private final ContextKey<Inherited> inheritedContextKey;
+
+  MetricsState(Class<?> metricsClass, String fieldName) {
+    this.contextKey = lookupContextKey(metricsClass, fieldName);
+    this.inheritedContextKey =
+        ContextKey.named("applicationinsights.internal.inherited_metrics_state." + fieldName);
+  }
+
+  boolean isResolved() {
+    return contextKey != null;
+  }
+
+  // needs to be called before the operation listeners' onStart()
+  Context captureInherited(Context parentContext) {
+    if (contextKey == null) {
+      return parentContext;
+    }
+    return parentContext.with(inheritedContextKey, new Inherited(parentContext.get(contextKey)));
+  }
+
+  // identity comparison is intended, the upstream state is a value class
+  @SuppressWarnings("ReferenceEquality")
+  boolean isRecorded(Context context) {
+    if (contextKey == null) {
+      // fall back to assuming the metric will be recorded
+      return true;
+    }
+    Object state = context.get(contextKey);
+    if (state == null) {
+      return false;
+    }
+    Inherited inherited = context.get(inheritedContextKey);
+    // inherited is null when the operation was not started by a customized instrumenter
+    return inherited == null || state != inherited.state;
+  }
+
+  @Nullable
+  static ContextKey<?> lookupContextKey(Class<?> metricsClass, String fieldName) {
+    try {
+      Field field = metricsClass.getDeclaredField(fieldName);
+      field.setAccessible(true);
+      return (ContextKey<?>) field.get(null);
+    } catch (ReflectiveOperationException | RuntimeException e) {
+      // not using a static logger field, since this is called from the enum constructors which run
+      // before the other static fields are initialized
+      LoggerFactory.getLogger(MetricsState.class)
+          .warn(
+              "Unable to find {}.{}, pre-aggregated metrics may be inaccurate",
+              metricsClass.getName(),
+              fieldName,
+              e);
+      return null;
+    }
+  }
+
+  private static final class Inherited {
+    @Nullable private final Object state;
+
+    private Inherited(@Nullable Object state) {
+      this.state = state;
+    }
+  }
+}

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/preagg/PreAggregatedMetricsCustomizer.java

@@ -0,0 +1,157 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.microsoft.applicationinsights.agent.internal.preagg;
+
+import static io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizer.InstrumentationType.HTTP_CLIENT;
+import static io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizer.InstrumentationType.HTTP_SERVER;
+import static io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizer.InstrumentationType.RPC_CLIENT;
+import static io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizer.InstrumentationType.RPC_SERVER;
+
+import com.azure.monitor.opentelemetry.autoconfigure.implementation.AiSemanticAttributes;
+import com.microsoft.applicationinsights.agent.internal.init.AiContextKeys;
+import io.opentelemetry.api.common.Attributes;
+import io.opentelemetry.api.common.AttributesBuilder;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.ContextKey;
+import io.opentelemetry.instrumentation.api.incubator.instrumenter.InstrumenterCustomizer;
+import io.opentelemetry.instrumentation.api.instrumenter.AttributesExtractor;
+import io.opentelemetry.instrumentation.api.instrumenter.ContextCustomizer;
+import io.opentelemetry.instrumentation.api.internal.SemconvStability;
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * Adds the Application Insights specific attributes to the pre-aggregated http and rpc duration
+ * metrics, and marks the span as pre-aggregated so that the exporter doesn't also generate standard
+ * metrics from it.
+ *
+ * <p>The attributes are added in {@link #onEnd}, which runs right before the upstream metrics
+ * record the duration with the same context and attributes.
+ *
+ * <p>Note: this only applies to the agent's instrumenters. Duration metrics recorded by
+ * OpenTelemetry instrumentation libraries bundled in the application (through the OpenTelemetry API
+ * bridge) only go through the views, so their spans are not marked as pre-aggregated.
+ */
+public final class PreAggregatedMetricsCustomizer
+    implements ContextCustomizer<Object>, AttributesExtractor<Object, Object> {
+
+  private static final ContextKey<Boolean> IS_SYNTHETIC =
+      ContextKey.named("applicationinsights.internal.is_synthetic");
+
+  private static volatile boolean azureMonitorMetricExporterEnabled;
+
+  private final List<MetricsState> metricsStates;
+  private final boolean captureSynthetic;
+
+  // the pre-aggregated metrics only reach Application Insights through the Azure Monitor metric
+  // exporter, e.g. with otel.metrics.exporter=none the duration metrics are not collected at all,
+  // so the spans must not be marked as pre-aggregated
+  public static void setAzureMonitorMetricExporterEnabled(boolean enabled) {
+    azureMonitorMetricExporterEnabled = enabled;
+  }
+
+  public static void customize(InstrumenterCustomizer customizer) {
+    if (!azureMonitorMetricExporterEnabled) {
+      return;
+    }
+    customize(customizer, SemconvStability.emitOldRpcSemconv());
+  }
+
+  // visible for testing
+  static void customize(InstrumenterCustomizer customizer, boolean emitOldRpcSemconv) {
+    List<MetricsState> metricsStates = new ArrayList<>();
+    if (customizer.hasType(HTTP_SERVER)) {
+      metricsStates.add(MetricsState.HTTP_SERVER);
+    }
+    if (customizer.hasType(HTTP_CLIENT)) {
+      metricsStates.add(MetricsState.HTTP_CLIENT);
+    }
+    // with stable rpc semconv only, the upstream rpc metrics record rpc.*.call.duration instead of
+    // rpc.*.duration, which are not pre-aggregated standard metrics
+    if (emitOldRpcSemconv && customizer.hasType(RPC_SERVER)) {
+      metricsStates.add(MetricsState.RPC_SERVER);
+    }
+    if (emitOldRpcSemconv && customizer.hasType(RPC_CLIENT)) {
+      metricsStates.add(MetricsState.RPC_CLIENT);
+    }
+    if (metricsStates.isEmpty()) {
+      return;
+    }
+    PreAggregatedMetricsCustomizer instance =
+        new PreAggregatedMetricsCustomizer(
+            metricsStates, metricsStates.contains(MetricsState.HTTP_SERVER));
+    customizer.addContextCustomizer(instance);
+    customizer.addAttributesExtractor(instance);
+  }
+
+  PreAggregatedMetricsCustomizer(List<MetricsState> metricsStates, boolean captureSynthetic) {
+    this.metricsStates = metricsStates;
+    this.captureSynthetic = captureSynthetic;
+  }
+
+  // context customizers run before the operation listeners' onStart(), and receive the start
+  // attributes from all of the attributes extractors, which is needed to get the user agent
+  @Override
+  public Context onStart(Context parentContext, Object request, Attributes startAttributes) {
+    Context context = parentContext;
+    for (MetricsState metricsState : metricsStates) {
+      context = metricsState.captureInherited(context);
+    }
+    if (captureSynthetic) {
+      boolean synthetic = UserAgents.isBot(startAttributes);
+      if (synthetic != Boolean.TRUE.equals(context.get(IS_SYNTHETIC))) {
+        context = context.with(IS_SYNTHETIC, synthetic);
+      }
+    }
+    return context;
+  }
+
+  @Override
+  public void onStart(AttributesBuilder attributes, Context parentContext, Object request) {}
+
+  @Override
+  public void onEnd(
+      AttributesBuilder attributes,
+      Context context,
+      Object request,
+      @Nullable Object response,
+      @Nullable Throwable error) {
+
+    if (!isMetricRecorded(context)) {
+      return;
+    }
+
+    // this is needed for detecting telemetry signals that will trigger pre-aggregated metrics via
+    // auto instrumentations
+    Span.fromContext(context).setAttribute(AiSemanticAttributes.IS_PRE_AGGREGATED, true);
+
+    String connectionString = context.get(AiContextKeys.CONNECTION_STRING);
+    if (connectionString != null) {
+      // support for connectionStringOverrides
+      // and for Classic SDK's setConnectionString()
+      attributes.put(AiSemanticAttributes.INTERNAL_CONNECTION_STRING, connectionString);
+    }
+    String roleName = context.get(AiContextKeys.ROLE_NAME);
+    if (roleName != null) {
+      // support for roleNameOverrides and for Classic SDK's setConnectionString()
+      // and Classic SDK set...
+      attributes.put(AiSemanticAttributes.INTERNAL_ROLE_NAME, roleName);
+    }
+    if (captureSynthetic) {
+      attributes.put(
+          AiSemanticAttributes.IS_SYNTHETIC, Boolean.TRUE.equals(context.get(IS_SYNTHETIC)));
+    }
+  }
+
+  private boolean isMetricRecorded(Context context) {
+    for (MetricsState metricsState : metricsStates) {
+      if (metricsState.isRecorded(context)) {
+        return true;
+      }
+    }
+    return false;
+  }
+}

agent/agent-tooling/src/main/java/io/opentelemetry/sdk/metrics/internal/view/UserAgents.java → agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/preagg/UserAgents.java

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-package io.opentelemetry.sdk.metrics.internal.view;
+package com.microsoft.applicationinsights.agent.internal.preagg;
 
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.semconv.UserAgentAttributes;