diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/checkstyle-suppressions.xml b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/checkstyle-suppressions.xml index d54e74099812..5a78a1260acb 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/checkstyle-suppressions.xml +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/checkstyle-suppressions.xml @@ -81,6 +81,7 @@ + @@ -130,6 +131,7 @@ + diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/AzureMonitorExporterBuilder.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/AzureMonitorExporterBuilder.java index 2218794d2e2a..98ca832e997c 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/AzureMonitorExporterBuilder.java +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/AzureMonitorExporterBuilder.java @@ -24,14 +24,18 @@ import com.azure.monitor.opentelemetry.autoconfigure.implementation.configuration.StatsbeatConnectionString; import com.azure.monitor.opentelemetry.autoconfigure.implementation.heartbeat.HeartbeatExporter; import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.ContextTagKeys; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.TelemetryItem; import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryItemExporter; import com.azure.monitor.opentelemetry.autoconfigure.implementation.quickpulse.QuickPulse; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.CustomerSdkStats; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.CustomerSdkStatsTelemetryPipelineListener; import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.Feature; import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.StatsbeatModule; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.AzureMonitorHelper; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.PropertyHelper; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.ResourceParser; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.TempDirs; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.ThreadPoolUtils; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.VersionGenerator; import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; @@ -47,6 +51,9 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; import static java.util.concurrent.TimeUnit.DAYS; @@ -61,6 +68,10 @@ class AzureMonitorExporterBuilder { private static final String STATSBEAT_SHORT_INTERVAL_SECONDS_PROPERTY_NAME = "STATSBEAT_SHORT_INTERVAL_SECONDS_PROPERTY_NAME"; + private static final String SDKSTATS_DISABLED_ENV_VAR = "APPLICATIONINSIGHTS_SDKSTATS_DISABLED"; + private static final String SDKSTATS_EXPORT_INTERVAL_ENV_VAR = "APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL"; + private static final long SDKSTATS_DEFAULT_EXPORT_INTERVAL_SECONDS = 900; // 15 minutes + private static final Map PROPERTIES = CoreUtils.getProperties("azure-monitor-opentelemetry-exporter.properties"); @@ -93,13 +104,21 @@ void initializeIfNot(AzureMonitorAutoConfigureOptions exporterOptions, ConfigPro this.spanDataMapper = createSpanDataMapper(); File tempDir = TempDirs.getApplicationInsightsTempDir(LOGGER, "Telemetry will not be stored to disk and retried on sporadic network failures"); + // Create customer-facing SDKStats if enabled (skip accumulator and listener when disabled) + boolean customerSdkStatsEnabled = isCustomerSdkStatsEnabled(); + CustomerSdkStats customerSdkStats = customerSdkStatsEnabled ? createCustomerSdkStats() : null; + CustomerSdkStatsTelemetryPipelineListener customerSdkStatsListener + = customerSdkStats != null ? new CustomerSdkStatsTelemetryPipelineListener(customerSdkStats) : null; // TODO (heya) change LocalStorageStats.noop() to statsbeatModule.getNonessentialStatsbeat() when we decide to collect non-essential Statsbeat by default. this.builtTelemetryItemExporter = AzureMonitorHelper.createTelemetryItemExporter(httpPipeline, statsbeatModule, - tempDir, LocalStorageStats.noop()); + tempDir, LocalStorageStats.noop(), customerSdkStatsListener); if (LiveMetrics.isEnabled(configProperties)) { this.quickPulse = createQuickPulse(resource); } startStatsbeatModule(statsbeatModule, configProperties, tempDir); // wait till TelemetryItemExporter has been initialized before starting StatsbeatModule + if (customerSdkStatsEnabled) { + startCustomerSdkStats(customerSdkStats, customerSdkStatsListener, resource); + } } private QuickPulse createQuickPulse(Resource resource) { @@ -240,6 +259,56 @@ private void startStatsbeatModule(StatsbeatModule statsbeatModule, ConfigPropert false, initStatsbeatFeatures()); } + private CustomerSdkStats createCustomerSdkStats() { + // Use the raw library version number (e.g. "3.6.0") for the customer-facing stats dimension + String version = PropertyHelper.getSdkVersionNumber(); + return CustomerSdkStats.create(version); + } + + private boolean isCustomerSdkStatsEnabled() { + String disabledValue = configProperties.getString(SDKSTATS_DISABLED_ENV_VAR); + if ("true".equalsIgnoreCase(disabledValue)) { + LOGGER.verbose("Customer SDKStats is disabled via configuration property {}.", SDKSTATS_DISABLED_ENV_VAR); + return false; + } + return true; + } + + private void startCustomerSdkStats(CustomerSdkStats customerSdkStats, + CustomerSdkStatsTelemetryPipelineListener customerSdkStatsListener, Resource resource) { + // Get export interval from configuration or use default + long exportIntervalSeconds + = configProperties.getLong(SDKSTATS_EXPORT_INTERVAL_ENV_VAR, SDKSTATS_DEFAULT_EXPORT_INTERVAL_SECONDS); + if (exportIntervalSeconds <= 0) { + LOGGER.warning("Value for {} must be positive: {}. Using default {} seconds.", + SDKSTATS_EXPORT_INTERVAL_ENV_VAR, exportIntervalSeconds, SDKSTATS_DEFAULT_EXPORT_INTERVAL_SECONDS); + exportIntervalSeconds = SDKSTATS_DEFAULT_EXPORT_INTERVAL_SECONDS; + } + + ConnectionString connectionString = getConnectionString(); + String sdkVersion = VersionGenerator.getSdkVersion(); + String cloudRole = resource.getAttribute(AttributeKey.stringKey("service.name")); + String cloudRoleInstance = resource.getAttribute(AttributeKey.stringKey("service.instance.id")); + + ScheduledExecutorService customerSdkStatsExecutor = Executors + .newSingleThreadScheduledExecutor(ThreadPoolUtils.createDaemonThreadFactory(CustomerSdkStats.class)); + customerSdkStatsListener.setScheduler(customerSdkStatsExecutor); + + LOGGER.verbose("Customer SDKStats scheduler starting with interval {} seconds.", exportIntervalSeconds); + customerSdkStatsExecutor.scheduleWithFixedDelay(() -> { + try { + List items + = customerSdkStats.collectAndReset(connectionString, sdkVersion, cloudRole, cloudRoleInstance); + if (!items.isEmpty()) { + LOGGER.verbose("Customer SDKStats: exporting {} metric items.", items.size()); + builtTelemetryItemExporter.sendWithoutTracking(items); + } + } catch (RuntimeException e) { + LOGGER.warning("Error sending customer SDKStats", e); + } + }, exportIntervalSeconds, exportIntervalSeconds, TimeUnit.SECONDS); + } + private HttpPipeline createStatsbeatHttpPipeline() { if (exporterOptions.httpPipeline != null) { return exporterOptions.httpPipeline; diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryItemExporter.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryItemExporter.java index 0992704183d7..dfb7c18514e7 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryItemExporter.java +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryItemExporter.java @@ -11,6 +11,7 @@ import com.azure.monitor.opentelemetry.autoconfigure.implementation.logging.OperationLogger; import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.ContextTagKeys; import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.TelemetryItem; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.TelemetryBatchMetadata; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.AksResourceAttributes; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.IKeyMasker; import io.opentelemetry.api.common.AttributeKey; @@ -70,6 +71,33 @@ public CompletableResultCode send(List telemetryItems) { return CompletableResultCode.ofAll(resultCodeList); } + /** + * Sends telemetry items without computing customer-facing SDKStats metadata. + * Used for sending customer SDKStats metrics themselves to prevent recursive counting. + * Items sent via this method will NOT trigger CustomerSdkStatsTelemetryPipelineListener + * since the TelemetryPipelineRequest will have empty item count maps. + * + *

This intentionally bypasses {@link #internalSendByBatch} and its AKS + * {@code _OTELRESOURCE_} metric injection, since SDKStats metrics are not + * application telemetry and should not carry OTel resource attributes.

+ */ + public CompletableResultCode sendWithoutTracking(List telemetryItems) { + Map> batches = splitIntoBatches(telemetryItems); + List resultCodeList = new ArrayList<>(); + for (Map.Entry> batch : batches.entrySet()) { + try { + List byteBuffers = serialize(batch.getValue()); + encodeBatchOperationLogger.recordSuccess(); + resultCodeList.add(telemetryPipeline.send(byteBuffers, batch.getKey().connectionString, listener)); + } catch (Throwable t) { + encodeBatchOperationLogger.recordFailure(t.getMessage(), t); + resultCodeList.add(CompletableResultCode.ofFailure()); + } + } + maybeAddToActiveExportResults(resultCodeList); + return CompletableResultCode.ofAll(resultCodeList); + } + // visible for tests Map> splitIntoBatches(List telemetryItems) { @@ -109,6 +137,11 @@ public CompletableResultCode shutdown() { CompletableResultCode internalSendByBatch(TelemetryItemBatchKey telemetryItemBatchKey, List telemetryItems) { List byteBuffers; + + // Compute per-type item counts BEFORE injecting internal metrics (e.g. _OTELRESOURCE_) + // so that only customer telemetry is counted in SDKStats. + TelemetryBatchMetadata batchMetadata = TelemetryBatchMetadata.fromTelemetryItems(telemetryItems); + // Don't send _OTELRESOURCE_ custom metric when OTEL_RESOURCE_ATTRIBUTES env var is empty // Don't send _OTELRESOURCE_ custom metric to Statsbeat yet // Don't Send _OTELRESOURCE_ when the app is running on other env other than AKS @@ -118,6 +151,7 @@ CompletableResultCode internalSendByBatch(TelemetryItemBatchKey telemetryItemBat && AksResourceAttributes.isAks(telemetryItemBatchKey.resource)) { telemetryItems.add(0, createOtelResourceMetric(telemetryItemBatchKey)); } + try { byteBuffers = serialize(telemetryItems); encodeBatchOperationLogger.recordSuccess(); @@ -125,7 +159,7 @@ CompletableResultCode internalSendByBatch(TelemetryItemBatchKey telemetryItemBat encodeBatchOperationLogger.recordFailure(t.getMessage(), t); return CompletableResultCode.ofFailure(); } - return telemetryPipeline.send(byteBuffers, telemetryItemBatchKey.connectionString, listener); + return telemetryPipeline.send(byteBuffers, telemetryItemBatchKey.connectionString, listener, batchMetadata); } // serialize an array of TelemetryItems to an array of byte buffers diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipeline.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipeline.java index 0bd0929e6ce6..8d5b4b7f969f 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipeline.java +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipeline.java @@ -10,6 +10,7 @@ import com.azure.core.util.logging.ClientLogger; import com.azure.core.util.tracing.Tracer; import com.azure.monitor.opentelemetry.autoconfigure.implementation.configuration.ConnectionString; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.TelemetryBatchMetadata; import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.StatusCode; import io.opentelemetry.sdk.common.CompletableResultCode; import reactor.core.publisher.Mono; @@ -45,6 +46,11 @@ public TelemetryPipeline(HttpPipeline pipeline, Runnable statsbeatShutdown) { public CompletableResultCode send(List telemetry, String connectionString, TelemetryPipelineListener listener) { + return send(telemetry, connectionString, listener, TelemetryBatchMetadata.empty()); + } + + public CompletableResultCode send(List telemetry, String connectionString, + TelemetryPipelineListener listener, TelemetryBatchMetadata batchMetadata) { ConnectionString connectionStringObj = ConnectionString.parse(connectionString); @@ -52,7 +58,7 @@ public CompletableResultCode send(List telemetry, String connectionS k -> getFullIngestionUrl(connectionStringObj.getIngestionEndpoint())); TelemetryPipelineRequest request = new TelemetryPipelineRequest(url, connectionString, - connectionStringObj.getInstrumentationKey(), telemetry); + connectionStringObj.getInstrumentationKey(), telemetry, batchMetadata); try { CompletableResultCode result = new CompletableResultCode(); diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineRequest.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineRequest.java index 43ca4257367c..d6f001bd7ab8 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineRequest.java +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineRequest.java @@ -6,6 +6,7 @@ import com.azure.core.http.HttpHeaderName; import com.azure.core.http.HttpMethod; import com.azure.core.http.HttpRequest; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.TelemetryBatchMetadata; import reactor.core.publisher.Flux; import java.net.URL; @@ -20,13 +21,22 @@ public class TelemetryPipelineRequest { private final List byteBuffers; private final int contentLength; + // Customer-facing SDKStats metadata + private final TelemetryBatchMetadata batchMetadata; + TelemetryPipelineRequest(URL url, String connectionString, String instrumentationKey, List byteBuffers) { + this(url, connectionString, instrumentationKey, byteBuffers, TelemetryBatchMetadata.empty()); + } + + public TelemetryPipelineRequest(URL url, String connectionString, String instrumentationKey, + List byteBuffers, TelemetryBatchMetadata batchMetadata) { this.url = url; this.connectionString = connectionString; this.instrumentationKey = instrumentationKey; this.byteBuffers = byteBuffers; contentLength = byteBuffers.stream().mapToInt(ByteBuffer::limit).sum(); + this.batchMetadata = batchMetadata != null ? batchMetadata : TelemetryBatchMetadata.empty(); } public URL getUrl() { @@ -50,6 +60,15 @@ public String getInstrumentationKey() { return instrumentationKey; } + /** + * Returns metadata about the telemetry batch, including item counts by type + * and success/failure breakdowns. Returns empty metadata for batches where + * item counting is not applicable (e.g. statsbeat, disk retries). + */ + public TelemetryBatchMetadata getTelemetryBatchMetadata() { + return batchMetadata; + } + HttpRequest createHttpRequest() { HttpRequest request = new HttpRequest(HttpMethod.POST, url); request.setBody(Flux.fromIterable(byteBuffers)); diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineResponse.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineResponse.java index 142e964c2dd8..2909f195bd90 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineResponse.java +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/pipeline/TelemetryPipelineResponse.java @@ -21,7 +21,7 @@ public class TelemetryPipelineResponse { private final int statusCode; private final String body; - TelemetryPipelineResponse(int statusCode, String body) { + public TelemetryPipelineResponse(int statusCode, String body) { this.statusCode = statusCode; this.body = body; } diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStats.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStats.java new file mode 100644 index 000000000000..a366c3340c54 --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStats.java @@ -0,0 +1,361 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import com.azure.monitor.opentelemetry.autoconfigure.implementation.builders.MetricTelemetryBuilder; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.configuration.ConnectionString; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.ContextTagKeys; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.TelemetryItem; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.FormattedTime; +import reactor.util.annotation.Nullable; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Accumulates customer-facing SDKStats metrics: Item_Success_Count, Item_Dropped_Count, + * Item_Retry_Count. These metrics are sent to the customer's own Application Insights resource. + */ +public class CustomerSdkStats { + + static final String ITEM_SUCCESS_COUNT = "Item_Success_Count"; + static final String ITEM_DROPPED_COUNT = "Item_Dropped_Count"; + static final String ITEM_RETRY_COUNT = "Item_Retry_Count"; + + // Drop code constants + public static final String DROP_CODE_CLIENT_EXCEPTION = "CLIENT_EXCEPTION"; + public static final String DROP_CODE_CLIENT_READONLY = "CLIENT_READONLY"; + public static final String DROP_CODE_CLIENT_PERSISTENCE_CAPACITY = "CLIENT_PERSISTENCE_CAPACITY"; + public static final String DROP_CODE_CLIENT_STORAGE_DISABLED = "CLIENT_STORAGE_DISABLED"; + + // Retry code constants + public static final String RETRY_CODE_CLIENT_EXCEPTION = "CLIENT_EXCEPTION"; + public static final String RETRY_CODE_CLIENT_TIMEOUT = "CLIENT_TIMEOUT"; + + private final ConcurrentHashMap successCounts = new ConcurrentHashMap<>(); + private final ConcurrentHashMap droppedCounts = new ConcurrentHashMap<>(); + private final ConcurrentHashMap retryCounts = new ConcurrentHashMap<>(); + + private final String computeType; + private final String language; + private final String version; + + /** + * Creates a CustomerSdkStats instance using the detected resource provider for compute type. + */ + public static CustomerSdkStats create(String version) { + String computeType = ResourceProvider.initResourceProvider().getValue(); + return new CustomerSdkStats(computeType, "java", version); + } + + // visible for testing + public CustomerSdkStats(String computeType, String language, String version) { + this.computeType = computeType; + this.language = language; + this.version = version; + } + + /** + * Increment success counts for all telemetry types in the batch. + */ + public void incrementSuccessCount(Map itemCountsByType) { + for (Map.Entry entry : itemCountsByType.entrySet()) { + SuccessKey key = new SuccessKey(entry.getKey()); + successCounts.computeIfAbsent(key, k -> new AtomicLong()).addAndGet(entry.getValue()); + } + } + + /** + * Increment dropped counts for all telemetry types in the batch. + * + * @param itemCountsByType the per-type item counts + * @param dropCode the drop code (e.g. "CLIENT_EXCEPTION", "402", etc.) + * @param dropReason the drop reason (e.g. "Exceeded daily quota") + * @param successItemCountsByType success items by type (for telemetry_success dimension on + * REQUEST/DEPENDENCY) + * @param failureItemCountsByType failure items by type + */ + public void incrementDroppedCount(Map itemCountsByType, String dropCode, @Nullable String dropReason, + Map successItemCountsByType, Map failureItemCountsByType) { + for (Map.Entry entry : itemCountsByType.entrySet()) { + String telemetryType = entry.getKey(); + long totalCount = entry.getValue(); + + // For REQUEST and DEPENDENCY, split by telemetry_success + if ("REQUEST".equals(telemetryType) || "DEPENDENCY".equals(telemetryType)) { + long successCount = successItemCountsByType.getOrDefault(telemetryType, 0L); + long failureCount = failureItemCountsByType.getOrDefault(telemetryType, 0L); + // Any items not accounted for in success/failure maps go to the "unknown" bucket + long unaccounted = Math.max(0, totalCount - successCount - failureCount); + + if (successCount > 0) { + DroppedKey key = new DroppedKey(telemetryType, dropCode, dropReason, Boolean.TRUE); + droppedCounts.computeIfAbsent(key, k -> new AtomicLong()).addAndGet(successCount); + } + if (failureCount > 0) { + DroppedKey key = new DroppedKey(telemetryType, dropCode, dropReason, Boolean.FALSE); + droppedCounts.computeIfAbsent(key, k -> new AtomicLong()).addAndGet(failureCount); + } + if (unaccounted > 0) { + // Items where we couldn't determine success/failure + DroppedKey key = new DroppedKey(telemetryType, dropCode, dropReason, null); + droppedCounts.computeIfAbsent(key, k -> new AtomicLong()).addAndGet(unaccounted); + } + } else { + // For non-REQUEST/DEPENDENCY types, telemetry_success is not applicable + DroppedKey key = new DroppedKey(telemetryType, dropCode, dropReason, null); + droppedCounts.computeIfAbsent(key, k -> new AtomicLong()).addAndGet(totalCount); + } + } + } + + /** + * Increment retry counts for all telemetry types in the batch. + */ + public void incrementRetryCount(Map itemCountsByType, String retryCode, + @Nullable String retryReason) { + for (Map.Entry entry : itemCountsByType.entrySet()) { + RetryKey key = new RetryKey(entry.getKey(), retryCode, retryReason); + retryCounts.computeIfAbsent(key, k -> new AtomicLong()).addAndGet(entry.getValue()); + } + } + + /** + * Atomically snapshots and clears all counters, returning a list of TelemetryItem metrics + * to send to the customer's AI resource. + * + * @param connectionString the customer's connection string + * @param sdkVersion the full SDK version string for ai.internal.sdkVersion tag + * @param cloudRole the cloud role name (nullable) + * @param cloudRoleInstance the cloud role instance (nullable) + * @return list of TelemetryItem metrics, empty if no non-zero counters + */ + public List collectAndReset(ConnectionString connectionString, String sdkVersion, + @Nullable String cloudRole, @Nullable String cloudRoleInstance) { + List telemetryItems = new ArrayList<>(); + + // Snapshot and clear success counts + Map successSnapshot = snapshotAndClear(successCounts); + for (Map.Entry entry : successSnapshot.entrySet()) { + long count = entry.getValue(); + if (count == 0) { + continue; + } + SuccessKey key = entry.getKey(); + MetricTelemetryBuilder builder = MetricTelemetryBuilder.create(ITEM_SUCCESS_COUNT, (double) count); + builder.setConnectionString(connectionString); + builder.setTime(FormattedTime.offSetDateTimeFromNow()); + addCommonTags(builder, sdkVersion, cloudRole, cloudRoleInstance); + addCommonProperties(builder); + builder.addProperty("telemetry_type", key.telemetryType); + telemetryItems.add(builder.build()); + } + + // Snapshot and clear dropped counts + Map droppedSnapshot = snapshotAndClear(droppedCounts); + for (Map.Entry entry : droppedSnapshot.entrySet()) { + long count = entry.getValue(); + if (count == 0) { + continue; + } + DroppedKey key = entry.getKey(); + MetricTelemetryBuilder builder = MetricTelemetryBuilder.create(ITEM_DROPPED_COUNT, (double) count); + builder.setConnectionString(connectionString); + builder.setTime(FormattedTime.offSetDateTimeFromNow()); + addCommonTags(builder, sdkVersion, cloudRole, cloudRoleInstance); + addCommonProperties(builder); + builder.addProperty("telemetry_type", key.telemetryType); + builder.addProperty("drop.code", key.dropCode); + if (key.dropReason != null) { + builder.addProperty("drop.reason", key.dropReason); + } + if (key.telemetrySuccess != null) { + builder.addProperty("telemetry_success", key.telemetrySuccess.toString()); + } + telemetryItems.add(builder.build()); + } + + // Snapshot and clear retry counts + Map retrySnapshot = snapshotAndClear(retryCounts); + for (Map.Entry entry : retrySnapshot.entrySet()) { + long count = entry.getValue(); + if (count == 0) { + continue; + } + RetryKey key = entry.getKey(); + MetricTelemetryBuilder builder = MetricTelemetryBuilder.create(ITEM_RETRY_COUNT, (double) count); + builder.setConnectionString(connectionString); + builder.setTime(FormattedTime.offSetDateTimeFromNow()); + addCommonTags(builder, sdkVersion, cloudRole, cloudRoleInstance); + addCommonProperties(builder); + builder.addProperty("telemetry_type", key.telemetryType); + builder.addProperty("retry.code", key.retryCode); + if (key.retryReason != null) { + builder.addProperty("retry.reason", key.retryReason); + } + telemetryItems.add(builder.build()); + } + + return telemetryItems; + } + + private void addCommonTags(MetricTelemetryBuilder builder, String sdkVersion, @Nullable String cloudRole, + @Nullable String cloudRoleInstance) { + builder.addTag(ContextTagKeys.AI_INTERNAL_SDK_VERSION.toString(), sdkVersion); + if (cloudRole != null) { + builder.addTag(ContextTagKeys.AI_CLOUD_ROLE.toString(), cloudRole); + } + if (cloudRoleInstance != null) { + builder.addTag(ContextTagKeys.AI_CLOUD_ROLE_INSTANCE.toString(), cloudRoleInstance); + } + } + + private void addCommonProperties(MetricTelemetryBuilder builder) { + builder.addProperty("computeType", computeType); + builder.addProperty("language", language); + builder.addProperty("version", version); + } + + private static Map snapshotAndClear(ConcurrentHashMap map) { + Map snapshot = new HashMap<>(); + // We iterate and getAndSet(0) to avoid losing concurrent increments + for (Map.Entry entry : map.entrySet()) { + long value = entry.getValue().getAndSet(0); + if (value != 0) { + snapshot.put(entry.getKey(), value); + } + } + // Note: we don't remove zero-valued entries here because doing so with removeIf() + // can race with concurrent increment*() calls, causing those increments to be lost. + // The number of distinct keys is bounded by the set of telemetry types and code/reason + // combinations, so unbounded growth is not a concern in practice. + return snapshot; + } + + // visible for testing + long getSuccessCount(String telemetryType) { + AtomicLong count = successCounts.get(new SuccessKey(telemetryType)); + return count == null ? 0L : count.get(); + } + + // visible for testing + long getDroppedCount(String telemetryType, String dropCode) { + long total = 0; + for (Map.Entry entry : droppedCounts.entrySet()) { + DroppedKey key = entry.getKey(); + if (key.telemetryType.equals(telemetryType) && key.dropCode.equals(dropCode)) { + total += entry.getValue().get(); + } + } + return total; + } + + // visible for testing + long getRetryCount(String telemetryType, String retryCode) { + long total = 0; + for (Map.Entry entry : retryCounts.entrySet()) { + RetryKey key = entry.getKey(); + if (key.telemetryType.equals(telemetryType) && key.retryCode.equals(retryCode)) { + total += entry.getValue().get(); + } + } + return total; + } + + static final class SuccessKey { + final String telemetryType; + + SuccessKey(String telemetryType) { + this.telemetryType = telemetryType; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof SuccessKey)) { + return false; + } + SuccessKey that = (SuccessKey) o; + return telemetryType.equals(that.telemetryType); + } + + @Override + public int hashCode() { + return telemetryType.hashCode(); + } + } + + static final class DroppedKey { + final String telemetryType; + final String dropCode; + final String dropReason; + final Boolean telemetrySuccess; + + DroppedKey(String telemetryType, String dropCode, @Nullable String dropReason, + @Nullable Boolean telemetrySuccess) { + this.telemetryType = telemetryType; + this.dropCode = dropCode; + this.dropReason = dropReason; + this.telemetrySuccess = telemetrySuccess; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof DroppedKey)) { + return false; + } + DroppedKey that = (DroppedKey) o; + return telemetryType.equals(that.telemetryType) + && dropCode.equals(that.dropCode) + && Objects.equals(dropReason, that.dropReason) + && Objects.equals(telemetrySuccess, that.telemetrySuccess); + } + + @Override + public int hashCode() { + return Objects.hash(telemetryType, dropCode, dropReason, telemetrySuccess); + } + } + + static final class RetryKey { + final String telemetryType; + final String retryCode; + final String retryReason; + + RetryKey(String telemetryType, String retryCode, @Nullable String retryReason) { + this.telemetryType = telemetryType; + this.retryCode = retryCode; + this.retryReason = retryReason; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof RetryKey)) { + return false; + } + RetryKey that = (RetryKey) o; + return telemetryType.equals(that.telemetryType) + && retryCode.equals(that.retryCode) + && Objects.equals(retryReason, that.retryReason); + } + + @Override + public int hashCode() { + return Objects.hash(telemetryType, retryCode, retryReason); + } + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsExceptionCategory.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsExceptionCategory.java new file mode 100644 index 000000000000..3199d95ba389 --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsExceptionCategory.java @@ -0,0 +1,110 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import java.net.ConnectException; +import java.net.NoRouteToHostException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.util.concurrent.TimeoutException; + +/** + * Categorizes exceptions into low-cardinality reason strings for customer-facing SDKStats + * drop.reason and retry.reason dimensions. + */ +final class CustomerSdkStatsExceptionCategory { + + static final String TIMEOUT_EXCEPTION = "Timeout exception"; + static final String NETWORK_EXCEPTION = "Network exception"; + static final String STORAGE_EXCEPTION = "Storage exception"; + static final String CLIENT_EXCEPTION = "Client exception"; + + private static final int MAX_CAUSE_CHAIN_DEPTH = 10; + + /** + * Returns a low-cardinality exception category string. + */ + static String categorize(Throwable throwable) { + if (throwable == null) { + return CLIENT_EXCEPTION; + } + return categorizeByType(throwable); + } + + /** + * Returns true if any exception in the cause chain represents a timeout scenario + * (use CLIENT_TIMEOUT retry code), false otherwise (use CLIENT_EXCEPTION retry code). + */ + static boolean containsTimeout(Throwable throwable) { + Throwable current = throwable; + int depth = 0; + while (current != null && depth < MAX_CAUSE_CHAIN_DEPTH) { + if (isTimeoutType(current)) { + return true; + } + current = current.getCause(); + depth++; + } + return false; + } + + private static String categorizeByType(Throwable throwable) { + // Traverse the cause chain to find the most specific category. + // Each type-check method examines only the current node (no nested traversal) + // so that priority ordering (timeout > network > storage) is evaluated + // consistently at each level of the chain. + Throwable current = throwable; + int depth = 0; + while (current != null && depth < MAX_CAUSE_CHAIN_DEPTH) { + if (isTimeoutType(current)) { + return TIMEOUT_EXCEPTION; + } + if (isNetworkType(current)) { + return NETWORK_EXCEPTION; + } + if (isStorageType(current)) { + return STORAGE_EXCEPTION; + } + current = current.getCause(); + depth++; + } + return CLIENT_EXCEPTION; + } + + /** + * Checks whether a single throwable (not its cause chain) is a timeout type. + */ + private static boolean isTimeoutType(Throwable throwable) { + if (throwable instanceof SocketTimeoutException || throwable instanceof TimeoutException) { + return true; + } + // Check class name for netty timeout types without creating a hard dependency + String className = throwable.getClass().getName(); + return className.contains("TimeoutException") + || className.contains("ReadTimeoutException") + || className.contains("WriteTimeoutException"); + } + + private static boolean isNetworkType(Throwable throwable) { + return throwable instanceof UnknownHostException + || throwable instanceof ConnectException + || throwable instanceof NoRouteToHostException; + } + + private static boolean isStorageType(Throwable throwable) { + // Storage-related IOExceptions typically involve file system operations + String message = throwable.getMessage(); + if (message == null) { + return false; + } + String lowerMessage = message.toLowerCase(); + return lowerMessage.contains("no space left") + || lowerMessage.contains("disk full") + || lowerMessage.contains("read-only file system") + || lowerMessage.contains("permission denied"); + } + + private CustomerSdkStatsExceptionCategory() { + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryPipelineListener.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryPipelineListener.java new file mode 100644 index 000000000000..ce517d6c253c --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryPipelineListener.java @@ -0,0 +1,137 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipelineListener; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipelineRequest; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipelineResponse; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.utils.StatusCode; +import io.opentelemetry.sdk.common.CompletableResultCode; + +import java.util.concurrent.ScheduledExecutorService; + +/** + * TelemetryPipelineListener that observes HTTP responses and exceptions from the telemetry + * pipeline, recording item-level success/drop/retry counts into the CustomerSdkStats accumulator. + */ +public class CustomerSdkStatsTelemetryPipelineListener implements TelemetryPipelineListener { + + private final CustomerSdkStats customerSdkStats; + private volatile ScheduledExecutorService scheduler; + + public CustomerSdkStatsTelemetryPipelineListener(CustomerSdkStats customerSdkStats) { + this.customerSdkStats = customerSdkStats; + } + + /** + * Sets the scheduler used for periodic SDKStats export, so it can be shut down + * when the pipeline is shut down. + */ + public void setScheduler(ScheduledExecutorService scheduler) { + this.scheduler = scheduler; + } + + @Override + public void onResponse(TelemetryPipelineRequest request, TelemetryPipelineResponse response) { + TelemetryBatchMetadata batchMetadata = request.getTelemetryBatchMetadata(); + if (batchMetadata.isEmpty()) { + // No item metadata (e.g. from local storage resend) — skip tracking. + // NOTE: This means telemetry retried from disk will not be counted in SDKStats. + // To track those, item-count metadata would need to be persisted alongside the payload. + return; + } + + int statusCode = response.getStatusCode(); + + if (statusCode == 200 || statusCode == 206) { + // 200 = full success; 206 = partial success (some items accepted, some rejected). + // For 206, we count the entire batch as success because: + // - The majority of items were accepted by the ingestion service. + // - The failed items are retried from disk by LocalStorageTelemetryPipelineListener. + // - Disk retries carry empty metadata, so there is no double-counting risk. + // - Splitting counts proportionally per-type would require complex response parsing + // for a rare edge case. + customerSdkStats.incrementSuccessCount(batchMetadata.getItemCountsByType()); + } else if (StatusCode.isRetryable(statusCode)) { + // Retryable status codes: items will be retried via local storage + String retryCode = String.valueOf(statusCode); + String retryReason = getReasonPhraseForStatusCode(statusCode); + customerSdkStats.incrementRetryCount(batchMetadata.getItemCountsByType(), retryCode, retryReason); + } else if (!StatusCode.isRedirect(statusCode)) { + // Non-redirect, non-retryable status codes: items are dropped. + // Redirects are handled transparently by the HTTP client and are not counted. + String dropCode = String.valueOf(statusCode); + String dropReason = getReasonPhraseForStatusCode(statusCode); + customerSdkStats.incrementDroppedCount(batchMetadata.getItemCountsByType(), dropCode, dropReason, + batchMetadata.getSuccessItemCountsByType(), batchMetadata.getFailureItemCountsByType()); + } + } + + @Override + public void onException(TelemetryPipelineRequest request, String errorMessage, Throwable throwable) { + TelemetryBatchMetadata batchMetadata = request.getTelemetryBatchMetadata(); + if (batchMetadata.isEmpty()) { + return; + } + + // Exceptions result in retry via local storage persistence + boolean isTimeout = CustomerSdkStatsExceptionCategory.containsTimeout(throwable); + String retryCode + = isTimeout ? CustomerSdkStats.RETRY_CODE_CLIENT_TIMEOUT : CustomerSdkStats.RETRY_CODE_CLIENT_EXCEPTION; + String retryReason = CustomerSdkStatsExceptionCategory.categorize(throwable); + customerSdkStats.incrementRetryCount(batchMetadata.getItemCountsByType(), retryCode, retryReason); + } + + @Override + public CompletableResultCode shutdown() { + ScheduledExecutorService exec = this.scheduler; + if (exec != null) { + exec.shutdown(); + } + return CompletableResultCode.ofSuccess(); + } + + static String getReasonPhraseForStatusCode(int statusCode) { + switch (statusCode) { + case 400: + return "Bad request"; + + case 401: + return "Unauthorized"; + + case 402: + return "Exceeded daily quota"; + + case 403: + return "Forbidden"; + + case 404: + return "Not found"; + + case 408: + return "Request timeout"; + + case 429: + return "Too many requests"; + + case 439: + return "Exceeded daily quota"; + + case 500: + return "Internal server error"; + + case 502: + return "Bad gateway"; + + case 503: + return "Service unavailable"; + + case 504: + return "Gateway timeout"; + + default: + return "Unknown"; + } + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryType.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryType.java new file mode 100644 index 000000000000..359547766619 --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryType.java @@ -0,0 +1,50 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import reactor.util.annotation.Nullable; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +/** + * Maps TelemetryItem.getName() values to the telemetry_type dimension values + * used in customer-facing SDKStats metrics. + */ +public final class CustomerSdkStatsTelemetryType { + + private static final Map MAPPING; + + static { + Map m = new HashMap<>(); + m.put("Request", "REQUEST"); + m.put("RemoteDependency", "DEPENDENCY"); + m.put("Message", "TRACE"); + m.put("Exception", "EXCEPTION"); + m.put("Metric", "CUSTOM_METRIC"); + m.put("Event", "CUSTOM_EVENT"); + m.put("PageView", "PAGE_VIEW"); + m.put("Availability", "AVAILABILITY"); + MAPPING = Collections.unmodifiableMap(m); + } + + /** + * Maps a TelemetryItem name to its customer-facing telemetry_type value. + * + * @param telemetryItemName the value from TelemetryItem.getName() + * @return the telemetry_type dimension value, or null if the item should be skipped + * (e.g. "Statsbeat" internal metrics) + */ + @Nullable + public static String fromTelemetryItemName(@Nullable String telemetryItemName) { + if (telemetryItemName == null) { + return null; + } + return MAPPING.get(telemetryItemName); + } + + private CustomerSdkStatsTelemetryType() { + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/TelemetryBatchMetadata.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/TelemetryBatchMetadata.java new file mode 100644 index 000000000000..9ff0c377eec8 --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/TelemetryBatchMetadata.java @@ -0,0 +1,118 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.MonitorDomain; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.RemoteDependencyData; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.RequestData; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.TelemetryItem; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Immutable snapshot of per-telemetry-type item counts for a batch of TelemetryItems. + * Used to propagate customer-facing SDKStats metadata through the telemetry pipeline + * without exposing three separate map parameters. + */ +public final class TelemetryBatchMetadata { + + private static final TelemetryBatchMetadata EMPTY + = new TelemetryBatchMetadata(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap()); + + private final Map itemCountsByType; + private final Map successItemCountsByType; + private final Map failureItemCountsByType; + + TelemetryBatchMetadata(Map itemCountsByType, Map successItemCountsByType, + Map failureItemCountsByType) { + this.itemCountsByType = itemCountsByType; + this.successItemCountsByType = successItemCountsByType; + this.failureItemCountsByType = failureItemCountsByType; + } + + /** + * Returns a shared empty instance (no item counts). Used for batches where + * item counting is not applicable (e.g. statsbeat, disk retries, SDKStats metrics). + */ + public static TelemetryBatchMetadata empty() { + return EMPTY; + } + + /** + * Computes per-type item counts from a list of TelemetryItems. + * Internal items (e.g. Statsbeat) are excluded from the counts. + * For REQUEST and DEPENDENCY types, success/failure is split based on isSuccess(). + */ + public static TelemetryBatchMetadata fromTelemetryItems(List telemetryItems) { + Map itemCountsByType = new HashMap<>(); + Map successItemCountsByType = new HashMap<>(); + Map failureItemCountsByType = new HashMap<>(); + + for (TelemetryItem item : telemetryItems) { + String telemetryType = CustomerSdkStatsTelemetryType.fromTelemetryItemName(item.getName()); + if (telemetryType == null) { + // Skip internal items (e.g. Statsbeat) + continue; + } + + itemCountsByType.merge(telemetryType, 1L, Long::sum); + + // Track success/failure for Request and Dependency types + if ("REQUEST".equals(telemetryType) || "DEPENDENCY".equals(telemetryType)) { + MonitorDomain baseData = item.getData() != null ? item.getData().getBaseData() : null; + Boolean success = null; + if (baseData instanceof RequestData) { + success = ((RequestData) baseData).isSuccess(); + } else if (baseData instanceof RemoteDependencyData) { + success = ((RemoteDependencyData) baseData).isSuccess(); + } + if (success != null && success) { + successItemCountsByType.merge(telemetryType, 1L, Long::sum); + } else if (success != null) { + // Only count explicit false as failure; null is treated as unknown + failureItemCountsByType.merge(telemetryType, 1L, Long::sum); + } + } + } + + if (itemCountsByType.isEmpty()) { + return EMPTY; + } + + return new TelemetryBatchMetadata(Collections.unmodifiableMap(itemCountsByType), + Collections.unmodifiableMap(successItemCountsByType), Collections.unmodifiableMap(failureItemCountsByType)); + } + + /** + * Returns item counts by telemetry type (e.g. "REQUEST" -> 200, "DEPENDENCY" -> 300). + * Empty map for batches where item counting is not applicable. + */ + public Map getItemCountsByType() { + return itemCountsByType; + } + + /** + * Returns counts of successful REQUEST/DEPENDENCY items (where isSuccess() == true). + */ + public Map getSuccessItemCountsByType() { + return successItemCountsByType; + } + + /** + * Returns counts of failed REQUEST/DEPENDENCY items (where isSuccess() == false). + */ + public Map getFailureItemCountsByType() { + return failureItemCountsByType; + } + + /** + * Returns true if this metadata has no item counts. + */ + public boolean isEmpty() { + return itemCountsByType.isEmpty(); + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/utils/AzureMonitorHelper.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/utils/AzureMonitorHelper.java index 645a6a56f364..c0df064007cd 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/utils/AzureMonitorHelper.java +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/main/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/utils/AzureMonitorHelper.java @@ -10,29 +10,43 @@ import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryItemExporter; import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipeline; import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipelineListener; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.CustomerSdkStatsTelemetryPipelineListener; import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.StatsbeatModule; import com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat.StatsbeatTelemetryPipelineListener; +import reactor.util.annotation.Nullable; import java.io.File; public final class AzureMonitorHelper { public static TelemetryItemExporter createTelemetryItemExporter(HttpPipeline httpPipeline, - StatsbeatModule statsbeatModule, File tempDir, LocalStorageStats localStorageStats) { + StatsbeatModule statsbeatModule, File tempDir, LocalStorageStats localStorageStats, + @Nullable CustomerSdkStatsTelemetryPipelineListener customerSdkStatsListener) { TelemetryPipeline telemetryPipeline = new TelemetryPipeline(httpPipeline, statsbeatModule::shutdown); + // Listener ordering matters: localStorageListener must come before customerSdkStatsListener + // so that telemetry is persisted to disk (for later retry) before the retry is recorded in + // SDKStats. If the order is reversed, SDKStats would record a retry before the items are + // actually persisted, and a persistence failure would not be reflected. TelemetryPipelineListener telemetryPipelineListener; if (tempDir == null) { - telemetryPipelineListener = new DiagnosticTelemetryPipelineListener( + DiagnosticTelemetryPipelineListener diagnosticListener = new DiagnosticTelemetryPipelineListener( "Sending telemetry to the ingestion service", true, " (telemetry will be lost)"); + telemetryPipelineListener = customerSdkStatsListener != null + ? TelemetryPipelineListener.composite(diagnosticListener, customerSdkStatsListener) + : diagnosticListener; } else { - telemetryPipelineListener = TelemetryPipelineListener.composite( - // suppress warnings on retryable failures, in order to reduce sporadic/annoying - // warnings when storing to disk and retrying shortly afterwards anyways - // will log if that retry from disk fails - new DiagnosticTelemetryPipelineListener("Sending telemetry to the ingestion service", false, ""), - new LocalStorageTelemetryPipelineListener(50, // default to 50MB - TempDirs.getSubDir(tempDir, "telemetry"), telemetryPipeline, localStorageStats, false)); + // suppress warnings on retryable failures, in order to reduce sporadic/annoying + // warnings when storing to disk and retrying shortly afterwards anyways + // will log if that retry from disk fails + DiagnosticTelemetryPipelineListener diagnosticListener + = new DiagnosticTelemetryPipelineListener("Sending telemetry to the ingestion service", false, ""); + LocalStorageTelemetryPipelineListener localStorageListener = new LocalStorageTelemetryPipelineListener(50, // default to 50MB + TempDirs.getSubDir(tempDir, "telemetry"), telemetryPipeline, localStorageStats, false); + telemetryPipelineListener = customerSdkStatsListener != null + ? TelemetryPipelineListener.composite(diagnosticListener, localStorageListener, + customerSdkStatsListener) + : TelemetryPipelineListener.composite(diagnosticListener, localStorageListener); } return new TelemetryItemExporter(telemetryPipeline, telemetryPipelineListener); diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/samples/java/com/azure/monitor/opentelemetry/autoconfigure/SimpleWebAppSample-README.md b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/samples/java/com/azure/monitor/opentelemetry/autoconfigure/SimpleWebAppSample-README.md new file mode 100644 index 000000000000..d9f3aa25abfc --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/samples/java/com/azure/monitor/opentelemetry/autoconfigure/SimpleWebAppSample-README.md @@ -0,0 +1,150 @@ +# SimpleWebAppSample — Manual Execution Guide + +A long-running web app that generates telemetry and validates customer-facing SDKStats metrics (`Item_Success_Count`, `Item_Dropped_Count`, `Item_Retry_Count`). + +--- + +## Prerequisites + +- **Java 8+** (JDK 8, 11, 17, or 21) +- **Maven 3.6+** +- The `azure-monitor-opentelemetry-autoconfigure` module must be compiled (including test sources). + +--- + +## Step-by-Step + +### 1. Set up environment variables + +Ensure `JAVA_HOME` points to your JDK installation and Maven is on your `PATH`. + +**Windows (PowerShell):** +```powershell +$env:JAVA_HOME = "" +$env:PATH = "/bin;$env:JAVA_HOME/bin;$env:PATH" +``` + +**Linux / macOS:** +```bash +export JAVA_HOME= +export PATH=/bin:$JAVA_HOME/bin:$PATH +``` + +### 2. Navigate to the project + +```bash +cd sdk/monitor/azure-monitor-opentelemetry-autoconfigure +``` + +### 3. Compile + +```bash +mvn compile test-compile +``` + +### 4. Choose a test mode + +Set the SDKStats export interval to 60 seconds (instead of the default 900) so results appear sooner: + +**Windows (PowerShell):** +```powershell +$env:APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL = "60" +``` + +**Linux / macOS:** +```bash +export APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL=60 +``` + +Then pick **one** of the following modes: + +#### Success mode (default) + +Telemetry is sent to real Azure Monitor. You will see `Item_Success_Count` in the `customMetrics` table. +Set `APPLICATIONINSIGHTS_CONNECTION_STRING` to your Application Insights connection string. + +```bash +# No TEST_MODE needed — defaults to "success" +``` + +#### Drop mode + +A local mock server on port 9090 returns **400** for all requests. You will see `Item_Dropped_Count` printed in the console. + +**Windows:** `$env:TEST_MODE = "drop"`  |  **Linux/macOS:** `export TEST_MODE=drop` + +#### Retry mode + +A local mock server on port 9090 returns **500** for all requests. You will see `Item_Retry_Count` printed in the console. + +**Windows:** `$env:TEST_MODE = "retry"`  |  **Linux/macOS:** `export TEST_MODE=retry` + +#### Optional: enable Azure SDK logging + +**Windows:** `$env:AZURE_LOG_LEVEL = "informational"`  |  **Linux/macOS:** `export AZURE_LOG_LEVEL=informational` + +### 5. Run the sample + +```bash +mvn exec:java "-Dexec.mainClass=com.azure.monitor.opentelemetry.autoconfigure.SimpleWebAppSample" "-Dexec.classpathScope=test" +``` + +You should see: + +``` +======================================================== + SimpleWebAppSample running on http://localhost:8080 + Mode: SUCCESS — real Azure Monitor → Item_Success_Count +======================================================== +``` + +### 6. Generate telemetry + +Open a **second** terminal and hit the endpoints: + +```bash +curl http://localhost:8080/ +curl http://localhost:8080/dependency +curl http://localhost:8080/error +curl http://localhost:8080/exception +curl http://localhost:8080/load +``` + +| Endpoint | What it does | +|----------------|-------------------------------------------| +| `GET /` | Returns "Hello!" (request span) | +| `GET /dependency` | Makes an outbound HTTP call (dependency span) | +| `GET /error` | Simulates an error span | +| `GET /exception` | Throws & records an exception | +| `GET /load` | Fires 20 mixed child spans for volume | + +### 7. Wait ~60 seconds + +After the export interval elapses, you will see SDKStats metrics: + +- **Success mode**: Query `customMetrics` in Azure Monitor: + ```kusto + customMetrics + | where name startswith "Item_" + | project timestamp, name, value, customDimensions + | order by timestamp desc + ``` +- **Drop/Retry mode**: SDKStats payloads appear in the **first terminal's console output**, tagged with `[SDKStats]`: + ``` + ╔══ MOCK INGESTION #4 [SDKStats] ══════════════════════════════ + ║ POST /v2.1/track → returning 400 + ║ {"name":"Metric","data":{"baseData":{"metrics":[{"name":"Item_Dropped_Count",...}]}}} + ╚═══════════════════════════════════════════════════════════════ + ``` + +### 8. Stop the sample + +Press `Ctrl+C` in the terminal running the sample. + +--- + +## Notes + +- In **drop/retry** modes, both application telemetry and SDKStats metrics go through the same mock pipeline. SDKStats metrics are visible only in the console output, not in Azure Monitor. +- The mock server automatically gunzips incoming payloads and prints each telemetry item on its own line. +- SDKStats dimensions include: `computeType`, `language`, `version`, `telemetry_type`, `telemetry_success`, and mode-specific fields (`drop.code`/`drop.reason` or `retry.code`/`retry.reason`). diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/samples/java/com/azure/monitor/opentelemetry/autoconfigure/SimpleWebAppSample.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/samples/java/com/azure/monitor/opentelemetry/autoconfigure/SimpleWebAppSample.java new file mode 100644 index 000000000000..f39ba06a177e --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/samples/java/com/azure/monitor/opentelemetry/autoconfigure/SimpleWebAppSample.java @@ -0,0 +1,367 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure; + +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.InetSocketAddress; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.zip.GZIPInputStream; + +/** + * A simple long-running web app sample that generates requests, dependencies, traces, and + * exceptions — ideal for observing customer-facing SDKStats metrics in Azure Monitor. + * + *

Test Modes (set via {@code TEST_MODE} environment variable):

+ *
    + *
  • success (default) — normal operation, telemetry reaches ingestion + * → {@code Item_Success_Count} visible in Azure Monitor
  • + *
  • drop — a local mock server returns 400 for all requests + * → {@code Item_Dropped_Count} with drop code 400 (visible in mock server console output)
  • + *
  • retry — a local mock server returns 500 for all requests + * → {@code Item_Retry_Count} with retry code 500 (visible in mock server console output)
  • + *
+ * + *

In drop/retry modes, a mock ingestion server runs on port 9090. Both application telemetry + * and SDKStats metrics go through the same pipeline, so SDKStats metrics also hit the mock. + * The mock server logs all received payloads (gunzipped) so you can inspect the generated + * {@code Item_Dropped_Count} and {@code Item_Retry_Count} TelemetryItems in the console.

+ * + *

Set {@code APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL} to a lower value (e.g. 60) + * so you don't have to wait 15 minutes for the first SDKStats export.

+ * + *

Endpoints:

+ *
    + *
  • {@code GET /} — returns "Hello!"
  • + *
  • {@code GET /dependency} — makes an outbound HTTP call
  • + *
  • {@code GET /error} — simulates an error span
  • + *
  • {@code GET /exception} — throws and records an exception
  • + *
  • {@code GET /load} — fires a batch of 20 mixed requests internally (for quick volume)
  • + *
+ * + *

Usage:

+ *
+ * # Test Item_Success_Count (default — goes to real Azure Monitor):
+ * set APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL=60
+ * mvn compile test-compile exec:java -Dexec.mainClass=...SimpleWebAppSample -Dexec.classpathScope=test
+ *
+ * # Test Item_Dropped_Count (mock server returns 400):
+ * set TEST_MODE=drop
+ * set APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL=60
+ * mvn compile test-compile exec:java -Dexec.mainClass=...SimpleWebAppSample -Dexec.classpathScope=test
+ *
+ * # Test Item_Retry_Count (mock server returns 500):
+ * set TEST_MODE=retry
+ * set APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL=60
+ * mvn compile test-compile exec:java -Dexec.mainClass=...SimpleWebAppSample -Dexec.classpathScope=test
+ * 
+ */ +public class SimpleWebAppSample { + + // ── Connection string for success mode (real Azure Monitor) ───────────── + // Read from APPLICATIONINSIGHTS_CONNECTION_STRING env var; if not set, use a placeholder. + private static final String CONNECTION_STRING_SUCCESS + = System.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING") != null + ? System.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING") + : "InstrumentationKey=" + + ";IngestionEndpoint=https://.in.applicationinsights.azure.com/" + + ";LiveEndpoint=https://.livediagnostics.monitor.azure.com/" + + ";ApplicationId="; + + // ── Connection string for mock server (drop/retry modes) ──────────────── + private static final int MOCK_PORT = 9090; + private static final String CONNECTION_STRING_MOCK + = "InstrumentationKey=00000000-0000-0000-0000-000000000000" + + ";IngestionEndpoint=http://localhost:" + MOCK_PORT + "/" + + ";LiveEndpoint=http://localhost:" + MOCK_PORT + "/"; + + private static final int APP_PORT = 8080; + + private static Tracer tracer; + private static LongCounter requestCounter; + + public static void main(String[] args) throws IOException { + // ── 0. Determine test mode ────────────────────────────────────────── + String testMode = System.getenv("TEST_MODE"); + if (testMode == null || testMode.isEmpty()) { + testMode = "success"; + } + testMode = testMode.toLowerCase(); + + String connectionString; + String modeDescription; + int mockStatusCode = 0; + switch (testMode) { + case "drop": + connectionString = CONNECTION_STRING_MOCK; + modeDescription = "DROP — mock server returns 400 → Item_Dropped_Count"; + mockStatusCode = 400; + break; + case "retry": + connectionString = CONNECTION_STRING_MOCK; + modeDescription = "RETRY — mock server returns 500 → Item_Retry_Count"; + mockStatusCode = 500; + break; + default: + connectionString = CONNECTION_STRING_SUCCESS; + modeDescription = "SUCCESS — real Azure Monitor → Item_Success_Count"; + break; + } + + // ── 1. Start mock ingestion server for drop/retry modes ───────────── + if (mockStatusCode > 0) { + startMockIngestionServer(mockStatusCode); + } + + // ── 2. Configure OpenTelemetry + Azure Monitor ────────────────────── + AutoConfiguredOpenTelemetrySdkBuilder sdkBuilder = AutoConfiguredOpenTelemetrySdk.builder(); + AzureMonitorAutoConfigure.customize(sdkBuilder, connectionString); + OpenTelemetry openTelemetry = sdkBuilder.build().getOpenTelemetrySdk(); + + tracer = openTelemetry.getTracer("SimpleWebApp"); + Meter meter = openTelemetry.meterBuilder("SimpleWebApp").build(); + requestCounter = meter.counterBuilder("http.server.request_count").build(); + + // ── 3. Start the web app HTTP server ──────────────────────────────── + HttpServer server = HttpServer.create(new InetSocketAddress(APP_PORT), 0); + server.createContext("/", SimpleWebAppSample::handleRoot); + server.createContext("/dependency", SimpleWebAppSample::handleDependency); + server.createContext("/error", SimpleWebAppSample::handleError); + server.createContext("/exception", SimpleWebAppSample::handleException); + server.createContext("/load", SimpleWebAppSample::handleLoad); + server.start(); + + System.out.println("========================================================"); + System.out.println(" SimpleWebAppSample running on http://localhost:" + APP_PORT); + System.out.println(" Mode: " + modeDescription); + System.out.println("========================================================"); + System.out.println(" Endpoints:"); + System.out.println(" GET / — hello"); + System.out.println(" GET /dependency — outbound HTTP call"); + System.out.println(" GET /error — simulated error span"); + System.out.println(" GET /exception — throws & records exception"); + System.out.println(" GET /load — fires 20 mixed requests"); + System.out.println(); + System.out.println(" Test modes (set TEST_MODE env var):"); + System.out.println(" success (default) — Item_Success_Count (real Azure Monitor)"); + System.out.println(" drop — Item_Dropped_Count (mock → 400)"); + System.out.println(" retry — Item_Retry_Count (mock → 500)"); + System.out.println(); + System.out.println(" Tip: set APPLICATIONINSIGHTS_SDKSTATS_EXPORT_INTERVAL=60"); + System.out.println(" to see customer SDKStats sooner."); + System.out.println("========================================================"); + } + + // ── Mock Ingestion Server ─────────────────────────────────────────────── + + /** + * Starts a local HTTP server on {@link #MOCK_PORT} that simulates ingestion failures. + * It handles POST /v2.1/track by returning the given status code and logging the + * gunzipped request payload so you can inspect SDKStats TelemetryItems in the console. + */ + @SuppressWarnings("try") + private static void startMockIngestionServer(int statusCode) throws IOException { + AtomicInteger requestCount = new AtomicInteger(); + HttpServer mock = HttpServer.create(new InetSocketAddress(MOCK_PORT), 0); + mock.createContext("/", exchange -> { + int reqNum = requestCount.incrementAndGet(); + String method = exchange.getRequestMethod(); + String path = exchange.getRequestURI().getPath(); + + // Read and gunzip the request body + String body = "(empty)"; + try { + byte[] rawBytes = readAllBytes(exchange.getRequestBody()); + if (rawBytes.length > 0) { + String contentEncoding = exchange.getRequestHeaders().getFirst("Content-Encoding"); + if ("gzip".equalsIgnoreCase(contentEncoding)) { + body = gunzip(rawBytes); + } else { + body = new String(rawBytes, StandardCharsets.UTF_8); + } + } + } catch (Exception e) { + body = "(failed to read body: " + e.getMessage() + ")"; + } + + // Log the request to console + boolean isSdkStats = body.contains("Item_Success_Count") + || body.contains("Item_Dropped_Count") + || body.contains("Item_Retry_Count"); + String tag = isSdkStats ? " [SDKStats]" : ""; + System.out.println(); + System.out.println("╔══ MOCK INGESTION #" + reqNum + tag + " ══════════════════════════════"); + System.out.println("║ " + method + " " + path + " → returning " + statusCode); + // Print each telemetry item on its own line for readability + for (String line : body.split("\n")) { + System.out.println("║ " + line.trim()); + } + System.out.println("╚═══════════════════════════════════════════════════════"); + + // Return the configured error status with a minimal JSON body + String responseBody = "{\"itemsReceived\":0,\"itemsAccepted\":0,\"errors\":[]}"; + byte[] responseBytes = responseBody.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(statusCode, responseBytes.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(responseBytes); + } + }); + mock.start(); + System.out.println("[Mock] Ingestion server started on http://localhost:" + MOCK_PORT + + " — returning " + statusCode + " for all requests"); + } + + private static byte[] readAllBytes(InputStream is) throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + byte[] tmp = new byte[4096]; + int n; + while ((n = is.read(tmp)) != -1) { + buffer.write(tmp, 0, n); + } + return buffer.toByteArray(); + } + + private static String gunzip(byte[] compressed) throws IOException { + try (GZIPInputStream gis = new GZIPInputStream(new java.io.ByteArrayInputStream(compressed))) { + return new String(readAllBytes(gis), StandardCharsets.UTF_8); + } + } + + // ── Handlers ──────────────────────────────────────────────────────────── + + @SuppressWarnings("try") + private static void handleRoot(HttpExchange exchange) throws IOException { + Span span = tracer.spanBuilder("GET /").setSpanKind(SpanKind.SERVER).startSpan(); + try (Scope ignored = span.makeCurrent()) { + requestCounter.add(1, Attributes.of(AttributeKey.stringKey("path"), "/")); + respond(exchange, 200, "Hello from SimpleWebAppSample!"); + } finally { + span.end(); + } + } + + @SuppressWarnings("try") + private static void handleDependency(HttpExchange exchange) throws IOException { + Span span = tracer.spanBuilder("GET /dependency").setSpanKind(SpanKind.SERVER).startSpan(); + try (Scope ignored = span.makeCurrent()) { + requestCounter.add(1, Attributes.of(AttributeKey.stringKey("path"), "/dependency")); + + // Simulate an outbound dependency call + Span depSpan = tracer.spanBuilder("HTTP GET httpbin.org").setSpanKind(SpanKind.CLIENT).startSpan(); + try (Scope depScope = depSpan.makeCurrent()) { + URL url = new URL("https://httpbin.org/get"); + HttpURLConnection conn = (HttpURLConnection) url.openConnection(); + conn.setConnectTimeout(5000); + conn.setReadTimeout(5000); + int status = conn.getResponseCode(); + depSpan.setAttribute("http.status_code", status); + conn.disconnect(); + respond(exchange, 200, "Dependency call returned " + status); + } catch (Exception e) { + depSpan.recordException(e); + depSpan.setStatus(StatusCode.ERROR, e.getMessage()); + respond(exchange, 502, "Dependency call failed: " + e.getMessage()); + } finally { + depSpan.end(); + } + } finally { + span.end(); + } + } + + @SuppressWarnings("try") + private static void handleError(HttpExchange exchange) throws IOException { + Span span = tracer.spanBuilder("GET /error").setSpanKind(SpanKind.SERVER).startSpan(); + try (Scope ignored = span.makeCurrent()) { + requestCounter.add(1, Attributes.of(AttributeKey.stringKey("path"), "/error")); + span.setStatus(StatusCode.ERROR, "simulated server error"); + respond(exchange, 500, "Simulated 500 error"); + } finally { + span.end(); + } + } + + @SuppressWarnings("try") + private static void handleException(HttpExchange exchange) throws IOException { + Span span = tracer.spanBuilder("GET /exception").setSpanKind(SpanKind.SERVER).startSpan(); + try (Scope ignored = span.makeCurrent()) { + requestCounter.add(1, Attributes.of(AttributeKey.stringKey("path"), "/exception")); + try { + throw new RuntimeException("Something went wrong!"); + } catch (RuntimeException e) { + span.recordException(e); + span.setStatus(StatusCode.ERROR, e.getMessage()); + respond(exchange, 500, "Exception recorded: " + e.getMessage()); + } + } finally { + span.end(); + } + } + + @SuppressWarnings("try") + private static void handleLoad(HttpExchange exchange) throws IOException { + Span span = tracer.spanBuilder("GET /load").setSpanKind(SpanKind.SERVER).startSpan(); + try (Scope ignored = span.makeCurrent()) { + // Generate a batch of child spans to create volume + int count = 20; + for (int i = 0; i < count; i++) { + String op = randomOp(); + Span child = tracer.spanBuilder("load-" + op).setSpanKind(SpanKind.INTERNAL).startSpan(); + try (Scope childScope = child.makeCurrent()) { + // simulate some work + Thread.sleep(ThreadLocalRandom.current().nextInt(5, 50)); + if ("error".equals(op)) { + child.setStatus(StatusCode.ERROR, "load error"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + child.end(); + } + } + requestCounter.add(count, Attributes.of(AttributeKey.stringKey("path"), "/load")); + respond(exchange, 200, "Generated " + count + " spans"); + } finally { + span.end(); + } + } + + // ── Helpers ───────────────────────────────────────────────────────────── + + private static void respond(HttpExchange exchange, int status, String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "text/plain; charset=UTF-8"); + exchange.sendResponseHeaders(status, bytes.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(bytes); + } + } + + private static String randomOp() { + String[] ops = {"ok", "ok", "ok", "error", "dep"}; + return ops[ThreadLocalRandom.current().nextInt(ops.length)]; + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsExceptionCategoryTest.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsExceptionCategoryTest.java new file mode 100644 index 000000000000..dfa48c585f87 --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsExceptionCategoryTest.java @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.net.ConnectException; +import java.net.SocketTimeoutException; +import java.net.UnknownHostException; +import java.util.concurrent.TimeoutException; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CustomerSdkStatsExceptionCategoryTest { + + @Test + public void testSocketTimeoutException() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(new SocketTimeoutException("Read timed out"))) + .isEqualTo("Timeout exception"); + assertThat(CustomerSdkStatsExceptionCategory.containsTimeout(new SocketTimeoutException("Read timed out"))) + .isTrue(); + } + + @Test + public void testTimeoutException() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(new TimeoutException("timeout"))) + .isEqualTo("Timeout exception"); + assertThat(CustomerSdkStatsExceptionCategory.containsTimeout(new TimeoutException("timeout"))).isTrue(); + } + + @Test + public void testUnknownHostException() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(new UnknownHostException("host.example.com"))) + .isEqualTo("Network exception"); + assertThat(CustomerSdkStatsExceptionCategory.containsTimeout(new UnknownHostException("host.example.com"))) + .isFalse(); + } + + @Test + public void testConnectException() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(new ConnectException("Connection refused"))) + .isEqualTo("Network exception"); + } + + @Test + public void testIOException() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(new IOException("Disk full"))) + .isEqualTo("Storage exception"); + } + + @Test + public void testGenericRuntimeException() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(new RuntimeException("Something went wrong"))) + .isEqualTo("Client exception"); + assertThat(CustomerSdkStatsExceptionCategory.containsTimeout(new RuntimeException("Something went wrong"))) + .isFalse(); + } + + @Test + public void testNullThrowable() { + assertThat(CustomerSdkStatsExceptionCategory.categorize(null)).isEqualTo("Client exception"); + assertThat(CustomerSdkStatsExceptionCategory.containsTimeout(null)).isFalse(); + } + + @Test + public void testWrappedSocketTimeoutException() { + // A SocketTimeoutException wrapped in a RuntimeException should still be detected via cause + RuntimeException wrapper = new RuntimeException("wrapper", new SocketTimeoutException("Read timed out")); + assertThat(CustomerSdkStatsExceptionCategory.categorize(wrapper)).isEqualTo("Timeout exception"); + assertThat(CustomerSdkStatsExceptionCategory.containsTimeout(wrapper)).isTrue(); + } + + @Test + public void testWrappedUnknownHostException() { + RuntimeException wrapper = new RuntimeException("wrapper", new UnknownHostException("host.example.com")); + assertThat(CustomerSdkStatsExceptionCategory.categorize(wrapper)).isEqualTo("Network exception"); + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryPipelineListenerTest.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryPipelineListenerTest.java new file mode 100644 index 000000000000..1c99704392ac --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryPipelineListenerTest.java @@ -0,0 +1,175 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipelineRequest; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.pipeline.TelemetryPipelineResponse; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.net.MalformedURLException; +import java.net.SocketTimeoutException; +import java.net.URL; +import java.net.UnknownHostException; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CustomerSdkStatsTelemetryPipelineListenerTest { + + private CustomerSdkStats customerSdkStats; + private CustomerSdkStatsTelemetryPipelineListener listener; + + @BeforeEach + public void init() { + customerSdkStats = new CustomerSdkStats("unknown", "java", "3.5.1"); + listener = new CustomerSdkStatsTelemetryPipelineListener(customerSdkStats); + } + + @Test + public void testSuccessResponse() { + Map itemCounts = new HashMap<>(); + itemCounts.put("REQUEST", 5L); + itemCounts.put("DEPENDENCY", 3L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + TelemetryPipelineResponse response = new TelemetryPipelineResponse(200, "OK"); + + listener.onResponse(request, response); + + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(5); + assertThat(customerSdkStats.getSuccessCount("DEPENDENCY")).isEqualTo(3); + } + + @Test + public void testPartialSuccessResponse206() { + Map itemCounts = new HashMap<>(); + itemCounts.put("REQUEST", 8L); + itemCounts.put("TRACE", 2L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + // 206 = partial success: some items accepted, some rejected + TelemetryPipelineResponse response + = new TelemetryPipelineResponse(206, "{\"itemsReceived\":10,\"itemsAccepted\":7,\"errors\":[]}"); + + listener.onResponse(request, response); + + // Entire batch counted as success (failed items retried from disk with empty metadata) + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(8); + assertThat(customerSdkStats.getSuccessCount("TRACE")).isEqualTo(2); + assertThat(customerSdkStats.getDroppedCount("REQUEST", "206")).isEqualTo(0); + } + + @Test + public void testRetryableStatusCode429() { + Map itemCounts = Collections.singletonMap("TRACE", 10L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + TelemetryPipelineResponse response = new TelemetryPipelineResponse(429, "Too Many Requests"); + + listener.onResponse(request, response); + + assertThat(customerSdkStats.getRetryCount("TRACE", "429")).isEqualTo(10); + } + + @Test + public void testRetryableStatusCode500() { + Map itemCounts = Collections.singletonMap("DEPENDENCY", 5L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + TelemetryPipelineResponse response = new TelemetryPipelineResponse(500, "Internal Server Error"); + + listener.onResponse(request, response); + + assertThat(customerSdkStats.getRetryCount("DEPENDENCY", "500")).isEqualTo(5); + } + + @Test + public void testNonRetryableStatusCode402() { + Map itemCounts = Collections.singletonMap("CUSTOM_METRIC", 20L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + TelemetryPipelineResponse response = new TelemetryPipelineResponse(402, "Payment Required"); + + listener.onResponse(request, response); + + assertThat(customerSdkStats.getDroppedCount("CUSTOM_METRIC", "402")).isEqualTo(20); + } + + @Test + public void testTimeoutException() { + Map itemCounts = Collections.singletonMap("REQUEST", 7L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + + listener.onException(request, "Read timed out", new SocketTimeoutException("Read timed out")); + + assertThat(customerSdkStats.getRetryCount("REQUEST", "CLIENT_TIMEOUT")).isEqualTo(7); + } + + @Test + public void testNetworkException() { + Map itemCounts = Collections.singletonMap("DEPENDENCY", 12L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + + listener.onException(request, "Unknown host", new UnknownHostException("host.example.com")); + + assertThat(customerSdkStats.getRetryCount("DEPENDENCY", "CLIENT_EXCEPTION")).isEqualTo(12); + } + + @Test + public void testEmptyItemCountsSkipped() { + TelemetryPipelineRequest request = createRequest(Collections.emptyMap()); + TelemetryPipelineResponse response = new TelemetryPipelineResponse(200, "OK"); + + listener.onResponse(request, response); + + // Nothing should be tracked + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(0); + } + + @Test + public void testRedirectSkipped() { + Map itemCounts = Collections.singletonMap("REQUEST", 5L); + + TelemetryPipelineRequest request = createRequest(itemCounts); + TelemetryPipelineResponse response = new TelemetryPipelineResponse(307, "Temporary Redirect"); + + listener.onResponse(request, response); + + // Redirects should not be tracked + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(0); + assertThat(customerSdkStats.getRetryCount("REQUEST", "307")).isEqualTo(0); + assertThat(customerSdkStats.getDroppedCount("REQUEST", "307")).isEqualTo(0); + } + + @Test + public void testReasonPhraseForStatusCode() { + assertThat(CustomerSdkStatsTelemetryPipelineListener.getReasonPhraseForStatusCode(402)) + .isEqualTo("Exceeded daily quota"); + assertThat(CustomerSdkStatsTelemetryPipelineListener.getReasonPhraseForStatusCode(429)) + .isEqualTo("Too many requests"); + assertThat(CustomerSdkStatsTelemetryPipelineListener.getReasonPhraseForStatusCode(500)) + .isEqualTo("Internal server error"); + assertThat(CustomerSdkStatsTelemetryPipelineListener.getReasonPhraseForStatusCode(999)).isEqualTo("Unknown"); + } + + private static TelemetryPipelineRequest createRequest(Map itemCountsByType) { + List byteBuffers = Collections.singletonList(ByteBuffer.allocate(0)); + try { + URL url = new URL("https://dc.services.visualstudio.com/v2.1/track"); + TelemetryBatchMetadata batchMetadata + = new TelemetryBatchMetadata(itemCountsByType, Collections.emptyMap(), Collections.emptyMap()); + return new TelemetryPipelineRequest(url, "InstrumentationKey=00000000-0000-0000-0000-0FEEDDADBEEF", + "00000000-0000-0000-0000-0FEEDDADBEEF", byteBuffers, batchMetadata); + } catch (MalformedURLException e) { + throw new RuntimeException(e); + } + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryTypeTest.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryTypeTest.java new file mode 100644 index 000000000000..488012fed406 --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTelemetryTypeTest.java @@ -0,0 +1,61 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CustomerSdkStatsTelemetryTypeTest { + + @Test + public void testRequestMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Request")).isEqualTo("REQUEST"); + } + + @Test + public void testDependencyMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("RemoteDependency")).isEqualTo("DEPENDENCY"); + } + + @Test + public void testTraceMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Message")).isEqualTo("TRACE"); + } + + @Test + public void testExceptionMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Exception")).isEqualTo("EXCEPTION"); + } + + @Test + public void testMetricMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Metric")).isEqualTo("CUSTOM_METRIC"); + } + + @Test + public void testEventMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Event")).isEqualTo("CUSTOM_EVENT"); + } + + @Test + public void testPageViewMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("PageView")).isEqualTo("PAGE_VIEW"); + } + + @Test + public void testAvailabilityMapping() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Availability")).isEqualTo("AVAILABILITY"); + } + + @Test + public void testStatsbeatReturnsNull() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("Statsbeat")).isNull(); + } + + @Test + public void testUnknownTypeReturnsNull() { + assertThat(CustomerSdkStatsTelemetryType.fromTelemetryItemName("UnknownType")).isNull(); + } +} diff --git a/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTest.java b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTest.java new file mode 100644 index 000000000000..767a5ed565dc --- /dev/null +++ b/sdk/monitor/azure-monitor-opentelemetry-autoconfigure/src/test/java/com/azure/monitor/opentelemetry/autoconfigure/implementation/statsbeat/CustomerSdkStatsTest.java @@ -0,0 +1,170 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.monitor.opentelemetry.autoconfigure.implementation.statsbeat; + +import com.azure.monitor.opentelemetry.autoconfigure.implementation.configuration.ConnectionString; +import com.azure.monitor.opentelemetry.autoconfigure.implementation.models.TelemetryItem; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +public class CustomerSdkStatsTest { + + private CustomerSdkStats customerSdkStats; + private static final ConnectionString CONNECTION_STRING + = ConnectionString.parse("InstrumentationKey=00000000-0000-0000-0000-0FEEDDADBEEF"); + private static final String SDK_VERSION = "java:3.5.1"; + private static final String CLOUD_ROLE = "TestRole"; + private static final String CLOUD_ROLE_INSTANCE = "TestInstance"; + + @BeforeEach + public void init() { + customerSdkStats = new CustomerSdkStats("unknown", "java", "3.5.1"); + } + + @Test + public void testIncrementSuccessCount() { + Map itemCountsByType = new HashMap<>(); + itemCountsByType.put("REQUEST", 5L); + itemCountsByType.put("DEPENDENCY", 3L); + + customerSdkStats.incrementSuccessCount(itemCountsByType); + + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(5); + assertThat(customerSdkStats.getSuccessCount("DEPENDENCY")).isEqualTo(3); + assertThat(customerSdkStats.getSuccessCount("TRACE")).isEqualTo(0); + } + + @Test + public void testIncrementSuccessCountMultipleTimes() { + Map batch1 = Collections.singletonMap("REQUEST", 3L); + Map batch2 = Collections.singletonMap("REQUEST", 7L); + + customerSdkStats.incrementSuccessCount(batch1); + customerSdkStats.incrementSuccessCount(batch2); + + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(10); + } + + @Test + public void testIncrementDroppedCount() { + Map itemCountsByType = new HashMap<>(); + itemCountsByType.put("TRACE", 10L); + Map successItems = Collections.emptyMap(); + Map failureItems = Collections.emptyMap(); + + customerSdkStats.incrementDroppedCount(itemCountsByType, "402", "Exceeded daily quota", successItems, + failureItems); + + assertThat(customerSdkStats.getDroppedCount("TRACE", "402")).isEqualTo(10); + } + + @Test + public void testIncrementDroppedCountWithTelemetrySuccess() { + Map itemCountsByType = new HashMap<>(); + itemCountsByType.put("REQUEST", 8L); + Map successItems = Collections.singletonMap("REQUEST", 5L); + Map failureItems = Collections.singletonMap("REQUEST", 3L); + + customerSdkStats.incrementDroppedCount(itemCountsByType, "402", "Exceeded daily quota", successItems, + failureItems); + + // Total dropped for REQUEST with code 402 should be 8 (5 success + 3 failure) + assertThat(customerSdkStats.getDroppedCount("REQUEST", "402")).isEqualTo(8); + } + + @Test + public void testIncrementRetryCount() { + Map itemCountsByType = new HashMap<>(); + itemCountsByType.put("DEPENDENCY", 15L); + itemCountsByType.put("REQUEST", 5L); + + customerSdkStats.incrementRetryCount(itemCountsByType, "429", "Too many requests"); + + assertThat(customerSdkStats.getRetryCount("DEPENDENCY", "429")).isEqualTo(15); + assertThat(customerSdkStats.getRetryCount("REQUEST", "429")).isEqualTo(5); + } + + @Test + public void testIncrementRetryCountClientTimeout() { + Map itemCountsByType = Collections.singletonMap("TRACE", 20L); + + customerSdkStats.incrementRetryCount(itemCountsByType, CustomerSdkStats.RETRY_CODE_CLIENT_TIMEOUT, + "Timeout exception"); + + assertThat(customerSdkStats.getRetryCount("TRACE", CustomerSdkStats.RETRY_CODE_CLIENT_TIMEOUT)).isEqualTo(20); + } + + @Test + public void testCollectAndResetReturnsCorrectMetrics() { + // Add some data + customerSdkStats.incrementSuccessCount(Collections.singletonMap("REQUEST", 10L)); + customerSdkStats.incrementDroppedCount(Collections.singletonMap("DEPENDENCY", 5L), "402", + "Exceeded daily quota", Collections.emptyMap(), Collections.emptyMap()); + customerSdkStats.incrementRetryCount(Collections.singletonMap("TRACE", 3L), "429", "Too many requests"); + + List items + = customerSdkStats.collectAndReset(CONNECTION_STRING, SDK_VERSION, CLOUD_ROLE, CLOUD_ROLE_INSTANCE); + + // Should have 3 metric items + assertThat(items).hasSize(3); + + // Verify that counters are cleared + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(0); + assertThat(customerSdkStats.getDroppedCount("DEPENDENCY", "402")).isEqualTo(0); + assertThat(customerSdkStats.getRetryCount("TRACE", "429")).isEqualTo(0); + } + + @Test + public void testCollectAndResetEmptyReturnsEmptyList() { + List items + = customerSdkStats.collectAndReset(CONNECTION_STRING, SDK_VERSION, CLOUD_ROLE, CLOUD_ROLE_INSTANCE); + + assertThat(items).isEmpty(); + } + + @Test + public void testCollectAndResetClearsCounters() { + customerSdkStats.incrementSuccessCount(Collections.singletonMap("REQUEST", 100L)); + + // First collect + List items1 + = customerSdkStats.collectAndReset(CONNECTION_STRING, SDK_VERSION, CLOUD_ROLE, CLOUD_ROLE_INSTANCE); + assertThat(items1).hasSize(1); + + // Second collect should be empty + List items2 + = customerSdkStats.collectAndReset(CONNECTION_STRING, SDK_VERSION, CLOUD_ROLE, CLOUD_ROLE_INSTANCE); + assertThat(items2).isEmpty(); + } + + @Test + public void testConcurrentIncrements() throws InterruptedException { + int threads = 10; + int incrementsPerThread = 1000; + ExecutorService executor = Executors.newFixedThreadPool(threads); + + for (int i = 0; i < threads; i++) { + executor.submit(() -> { + for (int j = 0; j < incrementsPerThread; j++) { + customerSdkStats.incrementSuccessCount(Collections.singletonMap("REQUEST", 1L)); + } + }); + } + + executor.shutdown(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + + assertThat(customerSdkStats.getSuccessCount("REQUEST")).isEqualTo(threads * incrementsPerThread); + } +}