Rate limited sampling

Also changed mapping of unparented INTERNAL spans from Dependencies to
REquests.

Author: trask

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/configuration/Configuration.java

@@ -74,7 +74,8 @@ public void validate() {
     preview.validate();
   }
 
-  // TODO (trask) investigate options for mapping lowercase values to otel enum directly
+  // used for now to still support sampling overrides v1
+  @Deprecated
   public enum SpanKind {
     @JsonProperty("server")
     SERVER(io.opentelemetry.api.trace.SpanKind.SERVER),
@@ -151,11 +152,24 @@ public static class Role {
 
   public static class Sampling {
 
-    public float percentage = 100;
+    // simple story: fixed percentage of all telemetry (traceId-based if traceId is available)
+    @Nullable public Float percentage;
+
+    // when this is set, percentage above is ignored, when neither is set, 5 req/sec is the default
+    @Nullable public Double targetRequestsPerSecondLimit;
+
+    // TODO (trask) print sampling percentages when it changes, make sure that with small adaption
+    // it never goes tooo low in which case consistent sampling across nodes could be compromised
   }
 
   public static class SamplingPreview {
 
+    // used for now to still support sampling overrides v1
+    @Deprecated public List<SamplingOverrideV1> overrides = new ArrayList<>();
+
+    @JsonProperty("overrides-v2")
+    public SamplingOverridesV2Configuration newOverrides = new SamplingOverridesV2Configuration();
+
     // this is not the default for now at least, because
     //
     // parent not-sampled -> child not-sampled (always, to avoid broken traces)
@@ -177,8 +191,15 @@ public static class SamplingPreview {
     //
     // IMPORTANT if changing this default, we need to keep it at least on Azure Functions
     public boolean parentBased;
+  }
+
+  public static class SamplingOverridesV2Configuration {
 
-    public List<SamplingOverride> overrides = new ArrayList<>();
+    public List<SamplingOverrideV2> requests = new ArrayList<>();
+    public List<SamplingOverrideV2> dependencies = new ArrayList<>();
+    public List<SamplingOverrideV2> logs = new ArrayList<>();
+    public List<SamplingOverrideV2> exceptions = new ArrayList<>();
+    // FUTURE: public List<SamplingOverride> customEvents = new ArrayList<>();
   }
 
   public static class JmxMetric {
@@ -352,9 +373,21 @@ public static class PreviewConfiguration {
         new HashSet<>(asList("b3", "b3multi"));
 
     public void validate() {
-      for (Configuration.SamplingOverride samplingOverride : sampling.overrides) {
+      for (SamplingOverrideV1 samplingOverride : sampling.overrides) {
         samplingOverride.validate();
       }
+      for (SamplingOverrideV2 samplingOverride : sampling.newOverrides.requests) {
+        samplingOverride.validate(false);
+      }
+      for (SamplingOverrideV2 samplingOverride : sampling.newOverrides.dependencies) {
+        samplingOverride.validate(true);
+      }
+      for (SamplingOverrideV2 samplingOverride : sampling.newOverrides.logs) {
+        samplingOverride.validate(true);
+      }
+      for (SamplingOverrideV2 samplingOverride : sampling.newOverrides.exceptions) {
+        samplingOverride.validate(true);
+      }
       for (Configuration.InstrumentationKeyOverride instrumentationKeyOverride :
           instrumentationKeyOverrides) {
         instrumentationKeyOverride.validate();
@@ -577,7 +610,9 @@ private static boolean isRuntimeAttached() {
     return Boolean.getBoolean("applicationinsights.internal.runtime.attached");
   }
 
-  public static class SamplingOverride {
+  // used for now to still support sampling overrides v1
+  @Deprecated
+  public static class SamplingOverrideV1 {
     // TODO (trask) consider making this required when moving out of preview
     @Nullable public SpanKind spanKind;
     // not using include/exclude, because you can still get exclude with this by adding a second
@@ -612,6 +647,52 @@ public void validate() {
     }
   }
 
+  public static class SamplingOverrideV2 {
+    // not using include/exclude, because you can still get exclude with this by adding a second
+    // (exclude) override above it (since only the first matching override is used)
+    public List<SamplingOverrideAttribute> attributes = new ArrayList<>();
+    @Nullable public Double targetPerSecondLimit;
+    @Nullable public Float fixedPercentage;
+
+    @Nullable public Boolean standalone;
+
+    public void validate(boolean allowStandalone) {
+      if (targetPerSecondLimit == null && fixedPercentage == null) {
+        // TODO add doc and go link, similar to telemetry processors
+        throw new FriendlyException(
+            "A sampling override configuration is missing one of \"targetPerSecondLimit\" or"
+                + " \"fixedPercentage\".",
+            "Please provide one of \"targetPerSecondLimit\" or \"fixedPercentage\" for the sampling"
+                + " override configuration.");
+      }
+      if (targetPerSecondLimit != null && fixedPercentage != null) {
+        // TODO add doc and go link, similar to telemetry processors
+        throw new FriendlyException(
+            "A sampling override configuration has both a \"targetPerSecondLimit\" and a"
+                + " \"fixedPercentage\".",
+            "Please provide only one of \"targetPerSecondLimit\" or \"fixedPercentage\" for the"
+                + " sampling override configuration.");
+      }
+      if (fixedPercentage != null && (fixedPercentage < 0 || fixedPercentage > 100)) {
+        // TODO add doc and go link, similar to telemetry processors
+        throw new FriendlyException(
+            "A sampling override configuration has a \"fixedPercentage\" that is not between 0 and"
+                + " 100.",
+            "Please provide a \"fixedPercentage\" that is between 0 and 100 for the sampling"
+                + " override configuration.");
+      }
+      if (standalone != null && !allowStandalone) {
+        // TODO add doc and go link, similar to telemetry processors
+        throw new FriendlyException(
+            "A sampling \"requestOverride\" configuration has specified \"standalone\".",
+            "Please remove \"standalone\" from the \"requestOverride\" configuration.");
+      }
+      for (SamplingOverrideAttribute attribute : attributes) {
+        attribute.validate();
+      }
+    }
+  }
+
   public static class SamplingOverrideAttribute {
     public String key;
     @Nullable public String value;

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/configuration/ConfigurationBuilder.java

@@ -32,7 +32,7 @@
 import com.microsoft.applicationinsights.agent.bootstrap.diagnostics.DiagnosticsHelper;
 import com.microsoft.applicationinsights.agent.internal.common.FriendlyException;
 import com.microsoft.applicationinsights.agent.internal.configuration.Configuration.JmxMetric;
-import com.microsoft.applicationinsights.agent.internal.configuration.Configuration.SamplingOverride;
+import com.microsoft.applicationinsights.agent.internal.configuration.Configuration.SamplingOverrideV1;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URL;
@@ -181,19 +181,35 @@ public static Configuration create(Path agentJarPath, @Nullable RpConfiguration
               + " and it is now enabled by default,"
               + " so no need to enable it under preview configuration");
     }
+    if (!config.preview.sampling.overrides.isEmpty()) {
+      configurationLogger.warn(
+          "\"preview\": { \"sampling\": { \"overrides\": ... } } has been deprecated,"
+              + " and support for it will be removed in a future release, please transition to"
+              + " \"preview\": { \"sampling\": { \"requestOverrides\": ... } } and"
+              + " \"preview\": { \"sampling\": { \"dependencyOverrides\": ... } }");
+    }
+
     logWarningIfUsingInternalAttributes(config);
 
     overlayFromEnv(config, agentJarPath.getParent());
-    config.sampling.percentage = roundToNearest(config.sampling.percentage, true);
-    for (SamplingOverride override : config.preview.sampling.overrides) {
-      override.percentage = roundToNearest(override.percentage, true);
+    if (config.sampling.percentage != null) {
+      config.sampling.percentage = roundToNearest(config.sampling.percentage, true);
     }
+    roundFixedPercentageToNearest(config.preview.sampling.newOverrides.requests);
+    roundFixedPercentageToNearest(config.preview.sampling.newOverrides.dependencies);
+    roundFixedPercentageToNearest(config.preview.sampling.newOverrides.logs);
+    roundFixedPercentageToNearest(config.preview.sampling.newOverrides.exceptions);
     // rp configuration should always be last (so it takes precedence)
     // currently applicationinsights-rp.json is only used by Azure Spring Cloud
     if (rpConfiguration != null) {
       overlayFromEnv(rpConfiguration);
       overlayRpConfiguration(config, rpConfiguration);
     }
+    // only fall back to default sampling configuration after all overlays have been performed
+    if (config.sampling.targetRequestsPerSecondLimit == null
+        && config.sampling.percentage == null) {
+      config.sampling.targetRequestsPerSecondLimit = 5.0;
+    }
     // only set role instance to host name as a last resort
     if (config.role.instance == null) {
       String hostname = HostName.get();
@@ -202,6 +218,15 @@ public static Configuration create(Path agentJarPath, @Nullable RpConfiguration
     return config;
   }
 
+  private static void roundFixedPercentageToNearest(
+      List<Configuration.SamplingOverrideV2> overrides) {
+    for (Configuration.SamplingOverrideV2 override : overrides) {
+      if (override.fixedPercentage != null) {
+        override.fixedPercentage = roundToNearest(override.fixedPercentage, true);
+      }
+    }
+  }
+
   private static void logWarningIfUsingInternalAttributes(Configuration config) {
     for (Configuration.ProcessorConfig processor : config.preview.processors) {
       if (processor.include != null) {
@@ -219,7 +244,7 @@ private static void logWarningIfUsingInternalAttributes(Configuration config) {
         }
       }
     }
-    for (SamplingOverride override : config.preview.sampling.overrides) {
+    for (SamplingOverrideV1 override : config.preview.sampling.overrides) {
       for (Configuration.SamplingOverrideAttribute attribute : override.attributes) {
         logWarningIfUsingInternalAttributes(attribute.key);
       }
@@ -466,10 +491,9 @@ static void overlayFromEnv(Configuration config, Path baseDir) throws IOExceptio
             APPLICATIONINSIGHTS_SELF_DIAGNOSTICS_FILE_PATH, config.selfDiagnostics.file.path);
 
     config.preview.metricIntervalSeconds =
-        (int)
-            overlayWithEnvVar(
-                APPLICATIONINSIGHTS_PREVIEW_METRIC_INTERVAL_SECONDS,
-                config.preview.metricIntervalSeconds);
+        overlayWithEnvVar(
+            APPLICATIONINSIGHTS_PREVIEW_METRIC_INTERVAL_SECONDS,
+            config.preview.metricIntervalSeconds);
 
     config.preview.instrumentation.springIntegration.enabled =
         overlayWithEnvVar(
@@ -576,6 +600,8 @@ static void overlayRpConfiguration(Configuration config, RpConfiguration rpConfi
     }
     if (rpConfiguration.sampling != null) {
       config.sampling.percentage = rpConfiguration.sampling.percentage;
+      config.sampling.targetRequestsPerSecondLimit =
+          rpConfiguration.sampling.targetRequestsPerSecondLimit;
     }
     if (isTrimEmpty(config.role.name)) {
       // only use rp configuration role name as a fallback, similar to WEBSITE_SITE_NAME
@@ -626,7 +652,8 @@ public static String overlayWithEnvVar(String name, String defaultValue) {
     return defaultValue;
   }
 
-  static float overlayWithEnvVar(String name, float defaultValue) {
+  @Nullable
+  static Float overlayWithEnvVar(String name, @Nullable Float defaultValue) {
     String value = getEnvVar(name);
     if (value != null) {
       configurationLogger.debug("using environment variable: {}", name);
@@ -637,6 +664,17 @@ static float overlayWithEnvVar(String name, float defaultValue) {
     return defaultValue;
   }
 
+  static int overlayWithEnvVar(String name, int defaultValue) {
+    String value = getEnvVar(name);
+    if (value != null) {
+      configurationLogger.debug("using environment variable: {}", name);
+      // intentionally allowing NumberFormatException to bubble up as invalid configuration and
+      // prevent agent from starting
+      return Integer.parseInt(value);
+    }
+    return defaultValue;
+  }
+
   static boolean overlayWithEnvVar(String name, boolean defaultValue) {
     String value = getEnvVar(name);
     if (value != null) {
@@ -802,7 +840,6 @@ public static float roundToNearest(float samplingPercentage) {
     return roundToNearest(samplingPercentage, false);
   }
 
-  // visible for testing
   private static float roundToNearest(float samplingPercentage, boolean doNotLogWarnMessages) {
     if (samplingPercentage == 0) {
       return 0;

agent/agent-tooling/src/main/java/com/microsoft/applicationinsights/agent/internal/exporter/AgentLogExporter.java

@@ -28,15 +28,22 @@
 import com.azure.monitor.opentelemetry.exporter.implementation.logging.OperationLogger;
 import com.azure.monitor.opentelemetry.exporter.implementation.models.TelemetryItem;
 import com.azure.monitor.opentelemetry.exporter.implementation.quickpulse.QuickPulse;
+import com.microsoft.applicationinsights.agent.internal.configuration.Configuration;
+import com.microsoft.applicationinsights.agent.internal.sampling.AiSampler;
+import com.microsoft.applicationinsights.agent.internal.sampling.SamplingOverridesV2;
 import com.microsoft.applicationinsights.agent.internal.telemetry.BatchItemProcessor;
 import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryClient;
 import com.microsoft.applicationinsights.agent.internal.telemetry.TelemetryObservers;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import io.opentelemetry.api.trace.SpanContext;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.logs.data.LogData;
 import io.opentelemetry.sdk.logs.data.Severity;
 import io.opentelemetry.sdk.logs.export.LogExporter;
+import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
 import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Consumer;
 import javax.annotation.Nullable;
 import org.slf4j.Logger;
@@ -52,15 +59,21 @@ public class AgentLogExporter implements LogExporter {
   // TODO (trask) could implement this in a filtering LogExporter instead
   private volatile Severity threshold;
 
+  private final SamplingOverridesV2 logSamplingOverrides;
+  private final SamplingOverridesV2 exceptionSamplingOverrides;
   private final LogDataMapper mapper;
   private final Consumer<TelemetryItem> telemetryItemConsumer;
 
   public AgentLogExporter(
       Severity threshold,
+      List<Configuration.SamplingOverrideV2> logSamplingOverrides,
+      List<Configuration.SamplingOverrideV2> exceptionSamplingOverrides,
       LogDataMapper mapper,
       @Nullable QuickPulse quickPulse,
       BatchItemProcessor batchItemProcessor) {
     this.threshold = threshold;
+    this.logSamplingOverrides = new SamplingOverridesV2(logSamplingOverrides);
+    this.exceptionSamplingOverrides = new SamplingOverridesV2(exceptionSamplingOverrides);
     this.mapper = mapper;
     telemetryItemConsumer =
         telemetryItem -> {
@@ -86,18 +99,44 @@ public CompletableResultCode export(Collection<LogData> logs) {
       return CompletableResultCode.ofFailure();
     }
     for (LogData log : logs) {
-      SpanContext spanContext = log.getSpanContext();
-      if (spanContext.isValid() && !spanContext.getTraceFlags().isSampled()) {
-        continue;
-      }
       logger.debug("exporting log: {}", log);
       try {
         int severity = log.getSeverity().getSeverityNumber();
         int threshold = this.threshold.getSeverityNumber();
         if (severity < threshold) {
           continue;
         }
-        mapper.map(log, telemetryItemConsumer);
+
+        String stack = log.getAttributes().get(SemanticAttributes.EXCEPTION_STACKTRACE);
+
+        SamplingOverridesV2 samplingOverrides =
+            stack != null ? exceptionSamplingOverrides : logSamplingOverrides;
+
+        Float samplingPercentage = samplingOverrides.getOverridePercentage(log.getAttributes());
+
+        SpanContext spanContext = log.getSpanContext();
+
+        if (samplingPercentage != null && !shouldSample(spanContext, samplingPercentage)) {
+          continue;
+        }
+
+        if (samplingPercentage == null
+            && spanContext.isValid()
+            && !spanContext.getTraceFlags().isSampled()) {
+          // if there is no sampling override, and the log is part of an unsampled trace, then don't
+          // capture it
+          continue;
+        }
+
+        Long itemCount = null;
+        if (samplingPercentage != null) {
+          // samplingPercentage cannot be 0 here
+          itemCount = Math.round(100.0 / samplingPercentage);
+        }
+
+        TelemetryItem telemetryItem = mapper.map(log, stack, itemCount);
+        telemetryItemConsumer.accept(telemetryItem);
+
         exportingLogLogger.recordSuccess();
       } catch (Throwable t) {
         exportingLogLogger.recordFailure(t.getMessage(), t, EXPORTER_MAPPING_ERROR);
@@ -116,4 +155,22 @@ public CompletableResultCode flush() {
   public CompletableResultCode shutdown() {
     return CompletableResultCode.ofSuccess();
   }
+
+  @SuppressFBWarnings(
+      value = "SECPR", // Predictable pseudorandom number generator
+      justification = "Predictable random is ok for sampling decision")
+  private static boolean shouldSample(SpanContext spanContext, float percentage) {
+    if (percentage == 100) {
+      // optimization, no need to calculate score
+      return true;
+    }
+    if (percentage == 0) {
+      // optimization, no need to calculate score
+      return false;
+    }
+    if (spanContext.isValid()) {
+      return AiSampler.shouldRecordAndSample(spanContext.getTraceId(), percentage);
+    }
+    return ThreadLocalRandom.current().nextDouble() < percentage / 100;
+  }
 }