Replace randomKey with stable trace_processor_host_id and split metric timestamps (#6672)

- Remove randomKey UUID label that caused cardinality explosion (new
  throwaway time series per metric emission)
- Add stable trace_processor_host_id label (hostname) to prevent
  cross-host collisions in AMP without creating new series each window
- Split timestamp granularity by metric type:
  - Sum metrics (request/error/fault): seconds-truncated timestamps
    for stable time series with minimal late-span collision risk
  - Histogram metrics (latency): minutes-truncated timestamps to
    aggregate more samples for richer bucket distributions

Signed-off-by: Vamsi Manohar <reddyvam@amazon.com>

Author: vamsimanohar

data-prepper-plugins/otel-apm-service-map-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/otel_apm_service_map/OTelApmServiceMapProcessor.java

@@ -45,9 +45,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.net.InetAddress;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -71,6 +73,7 @@ public class OTelApmServiceMapProcessor extends AbstractProcessor<Record<Event>,
 
     private static final String SPANS_DB_SIZE = "spansDbSize";
     private static final String SPANS_DB_COUNT = "spansDbCount";
+    private static final String HOST_ID = resolveHostId();
 
     private static final Logger LOG = LoggerFactory.getLogger(OTelApmServiceMapProcessor.class);
     private static final String EVENT_TYPE_OTEL_APM_SERVICE_MAP = "SERVICE_MAP";
@@ -375,7 +378,8 @@ private Collection<Record<Event>> processCurrentWindowSpans() {
 
         final EphemeralSpanDecorations ephemeralDecorations = new EphemeralSpanDecorations();
 
-        final Map<MetricKey, MetricAggregationState> metricsStateByKey = new HashMap<>();
+        final Map<MetricKey, MetricAggregationState> sumStateByKey = new HashMap<>();
+        final Map<MetricKey, MetricAggregationState> histogramStateByKey = new HashMap<>();
 
         final Map<String, Collection<SpanStateData>> previousSpansByTraceId = buildSpansByTraceIdMap(previousWindow);
         final Map<String, Collection<SpanStateData>> currentSpansByTraceId = buildSpansByTraceIdMap(currentWindow);
@@ -388,11 +392,11 @@ private Collection<Record<Event>> processCurrentWindowSpans() {
             if (!traceData.getProcessingSpans().isEmpty()) {
                 decorateSpansInTraceWithEphemeralStorage(traceData);
 
-                apmEvents.addAll(generateNodeOperationDetailEvents(traceData, currentTime, metricsStateByKey));
+                apmEvents.addAll(generateNodeOperationDetailEvents(traceData, currentTime, sumStateByKey, histogramStateByKey));
             }
         }
 
-        final List<JacksonMetric> metrics = ApmServiceMapMetricsUtil.createMetricsFromAggregatedState(metricsStateByKey);
+        final List<JacksonMetric> metrics = ApmServiceMapMetricsUtil.createMetricsFromAggregatedState(sumStateByKey, histogramStateByKey);
         metrics.sort(Comparator.comparing(JacksonMetric::getTime));
 
         final List<Record<Event>> apmEventsSorted = new ArrayList<>();
@@ -444,13 +448,17 @@ private Map<String, String> extractGroupByAttributes(final Span span) {
     }
 
     /**
-     * Get anchor timestamp from span's endTime, truncated to minute boundary
+     * Get anchor timestamp from span's endTime, truncated to the specified unit.
+     * Sum metrics use SECONDS to minimize collisions while keeping stable time series.
+     * Histogram metrics use MINUTES to aggregate more samples for richer bucket distributions.
      *
      * @param spanStateData The span to extract timestamp from
      * @param fallbackTime Current system time to use if span endTime is null
-     * @return Instant truncated to the lower 1-minute boundary
+     * @param truncationUnit The ChronoUnit to truncate the timestamp to
+     * @return Instant truncated to the specified boundary
      */
-    private Instant getAnchorTimestampFromSpan(final SpanStateData spanStateData, final Instant fallbackTime) {
+    private Instant getAnchorTimestampFromSpan(final SpanStateData spanStateData, final Instant fallbackTime,
+                                               final ChronoUnit truncationUnit) {
         Instant timestamp = fallbackTime; // Default to current system time
 
         final String endTime = spanStateData.getEndTime();
@@ -463,7 +471,25 @@ private Instant getAnchorTimestampFromSpan(final SpanStateData spanStateData, fi
                      endTime, e.getMessage());
         }
 
-        return timestamp.truncatedTo(java.time.temporal.ChronoUnit.MINUTES);
+        return timestamp.truncatedTo(truncationUnit);
+    }
+
+    /**
+     * Resolve a stable host identifier for this Data Prepper instance.
+     * Uses a truncated SHA-256 hash of the hostname to ensure uniqueness
+     * without revealing the actual hostname in emitted metrics.
+     * Falls back to a random UUID if hostname resolution fails.
+     */
+    private static String resolveHostId() {
+        try {
+            final String hostname = InetAddress.getLocalHost().getHostName();
+            final java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
+            final byte[] hash = digest.digest(hostname.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+            return Hex.encodeHexString(hash).substring(0, 16);
+        } catch (Exception e) {
+            LOG.warn("Failed to resolve hostname for trace_processor_host_id, using random UUID: {}", e.getMessage());
+            return java.util.UUID.randomUUID().toString();
+        }
     }
 
     /**
@@ -720,7 +746,8 @@ private void decorateServerSpansSecondPassWithEphemeralStorage(final ThreeWindow
      */
     private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeWindowTraceDataWithDecorations traceData,
                                                                         final Instant currentTime,
-                                                                        final Map<MetricKey, MetricAggregationState> metricsStateByKey) {
+                                                                        final Map<MetricKey, MetricAggregationState> sumStateByKey,
+                                                                        final Map<MetricKey, MetricAggregationState> histogramStateByKey) {
         final Collection<Record<Event>> events = new HashSet<>();
 
         // Step 1: CLIENT spans — primary emission path
@@ -746,10 +773,11 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                             : null;
                     final Operation targetOp = new Operation(decoration.getRemoteOperation());
 
-                    final Instant anchorTimestamp = getAnchorTimestampFromSpan(clientSpan, currentTime);
+                    final Instant sumAnchor = getAnchorTimestampFromSpan(clientSpan, currentTime, ChronoUnit.SECONDS);
+                    final Instant histAnchor = getAnchorTimestampFromSpan(clientSpan, currentTime, ChronoUnit.MINUTES);
 
                     final NodeOperationDetail nodeOperationDetail = new NodeOperationDetail(
-                            sourceNode, targetNode, sourceOp, targetOp, anchorTimestamp);
+                            sourceNode, targetNode, sourceOp, targetOp, sumAnchor);
 
                     final EventMetadata eventMetadata = new DefaultEventMetadata.Builder()
                             .withEventType(EVENT_TYPE_OTEL_APM_SERVICE_MAP).build();
@@ -763,7 +791,8 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
 
                     if (decoration.getParentServerOperationName() != null) {
                         ApmServiceMapMetricsUtil.generateMetricsForClientSpan(
-                                clientSpan, decoration, currentTime, metricsStateByKey, anchorTimestamp);
+                                clientSpan, decoration, currentTime, sumStateByKey, histogramStateByKey,
+                                sumAnchor, histAnchor, HOST_ID);
                     }
                 }
             }
@@ -772,9 +801,11 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
         // Step 2: SERVER spans — metrics for all, leaf NodeOperationDetail for those with no CLIENT descendants
         for (SpanStateData serverSpan : traceData.getProcessingSpans()) {
             if (SPAN_KIND_SERVER.equals(serverSpan.getSpanKind())) {
-                final Instant anchorTimestamp = getAnchorTimestampFromSpan(serverSpan, currentTime);
+                final Instant sumAnchor = getAnchorTimestampFromSpan(serverSpan, currentTime, ChronoUnit.SECONDS);
+                final Instant histAnchor = getAnchorTimestampFromSpan(serverSpan, currentTime, ChronoUnit.MINUTES);
                 ApmServiceMapMetricsUtil.generateMetricsForServerSpan(
-                        serverSpan, currentTime, metricsStateByKey, anchorTimestamp);
+                        serverSpan, currentTime, sumStateByKey, histogramStateByKey,
+                        sumAnchor, histAnchor, HOST_ID);
 
                 final ServerSpanDecoration decoration = traceData.getDecorations().getServerDecoration(serverSpan.getSpanId());
 
@@ -788,7 +819,7 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                     final Operation sourceOp = new Operation(serverSpan.getOperationName());
 
                     final NodeOperationDetail nodeOperationDetail = new NodeOperationDetail(
-                            sourceNode, null, sourceOp, null, anchorTimestamp);
+                            sourceNode, null, sourceOp, null, sumAnchor);
 
                     final EventMetadata eventMetadata = new DefaultEventMetadata.Builder()
                             .withEventType(EVENT_TYPE_OTEL_APM_SERVICE_MAP).build();