Add configurable metric_timestamp_source and shared HostContext for APM service map processor

- Add MetricTimestampSource enum (arrival_time / span_end_time) with
  @JsonCreator for config deserialization
- Add metric_timestamp_source config field defaulting to arrival_time,
  which uses processing time at window evaluation to avoid late-span
  data loss in Prometheus/AMP
- Split metric aggregation: Sum metrics use seconds-truncated timestamps,
  Histogram metrics use minutes-truncated timestamps
- Create shared HostContext utility in data-prepper-api for consistent
  hostname resolution across components
- Replace randomKey UUID with stable service_map_processor_host_id label
  derived from SHA-256 hash of hostname via HostContext
- Add putCommonLabels() helper and HOST_ID_LABEL constant in
  ApmServiceMapMetricsUtil for consistent label construction
- NodeOperationDetail events use minutes-truncated timestamps to
  preserve pre-existing behavior

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

Author: vamsimanohar

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/host/HostContext.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.model.host;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+
+/**
+ * Provides the hostname of the current Data Prepper instance.
+ * This is intended as a shared utility so that hostname resolution
+ * is consistent across all components (processors, source coordinators, etc.).
+ */
+public class HostContext {
+
+    private static final Logger LOG = LoggerFactory.getLogger(HostContext.class);
+    private static final String UNKNOWN_HOST = "unknown";
+    private static final String HOSTNAME = resolveHostname();
+
+    static String resolveHostname() {
+        try {
+            return InetAddress.getLocalHost().getHostName();
+        } catch (final Exception e) {
+            LOG.warn("Failed to resolve hostname, using '{}': {}", UNKNOWN_HOST, e.getMessage());
+            return UNKNOWN_HOST;
+        }
+    }
+
+    /**
+     * Returns the hostname of the current Data Prepper host.
+     *
+     * @return the hostname
+     */
+    public static String getHostname() {
+        return HOSTNAME;
+    }
+}

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/host/HostContextTest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.model.host;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.emptyString;
+import static org.mockito.Mockito.mockStatic;
+
+class HostContextTest {
+
+    @Test
+    void getHostname_returns_non_null_non_empty_value() {
+        final String hostname = HostContext.getHostname();
+        assertThat(hostname, notNullValue());
+        assertThat(hostname, not(emptyString()));
+    }
+
+    @Test
+    void getHostname_returns_consistent_value() {
+        final String first = HostContext.getHostname();
+        final String second = HostContext.getHostname();
+        assertThat(first, equalTo(second));
+    }
+
+    @Test
+    void getHostname_matches_InetAddress_hostname() throws UnknownHostException {
+        final String expected = InetAddress.getLocalHost().getHostName();
+        assertThat(HostContext.getHostname(), equalTo(expected));
+    }
+
+    @Test
+    void resolveHostname_returns_valid_hostname() throws UnknownHostException {
+        final String hostname = HostContext.resolveHostname();
+        assertThat(hostname, equalTo(InetAddress.getLocalHost().getHostName()));
+    }
+
+    @Test
+    void resolveHostname_returns_unknown_when_hostname_cannot_be_resolved() {
+        try (final MockedStatic<InetAddress> inetAddressMock = mockStatic(InetAddress.class)) {
+            inetAddressMock.when(InetAddress::getLocalHost)
+                    .thenThrow(new UnknownHostException("test exception"));
+
+            assertThat(HostContext.resolveHostname(), equalTo("unknown"));
+        }
+    }
+
+    @Test
+    void constructor_can_be_created() {
+        final HostContext hostContext = new HostContext();
+        assertThat(hostContext, notNullValue());
+    }
+}

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ */
+
+package org.opensearch.dataprepper.plugins.processor.otel_apm_service_map;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public enum MetricTimestampSource {
+    ARRIVAL_TIME("arrival_time"),
+    SPAN_END_TIME("span_end_time");
+
+    private static final Map<String, MetricTimestampSource> OPTIONS_MAP = Arrays.stream(MetricTimestampSource.values())
+            .collect(Collectors.toMap(
+                    value -> value.option,
+                    value -> value
+            ));
+
+    private final String option;
+
+    MetricTimestampSource(final String option) {
+        this.option = option;
+    }
+
+    public String getOption() {
+        return option;
+    }
+
+    @JsonCreator
+    public static MetricTimestampSource fromOptionValue(final String option) {
+        return OPTIONS_MAP.get(option);
+    }
+}

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

@@ -45,7 +45,10 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
-import java.net.InetAddress;
+import org.opensearch.dataprepper.model.host.HostContext;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
 import java.time.Clock;
 import java.time.Duration;
 import java.time.Instant;
@@ -73,7 +76,6 @@ 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";
@@ -95,7 +97,9 @@ public class OTelApmServiceMapProcessor extends AbstractProcessor<Record<Event>,
     private static Clock clock;
 
     private final int thisProcessorId;
+    private final String hostId;
     private final List<String> groupByAttributes;
+    private final MetricTimestampSource metricTimestampSource;
     private final EventFactory eventFactory;
 
     @DataPrepperPluginConstructor
@@ -110,7 +114,8 @@ public OTelApmServiceMapProcessor(
                 pipelineDescription.getNumberOfProcessWorkers(),
                 eventFactory,
                 pluginMetrics,
-                config.getGroupByAttributes());
+                config.getGroupByAttributes(),
+                config.getMetricTimestampSource());
     }
 
     OTelApmServiceMapProcessor(final Duration windowDuration,
@@ -119,7 +124,8 @@ public OTelApmServiceMapProcessor(
                                final int processWorkers,
                                final EventFactory eventFactory,
                                final PluginMetrics pluginMetrics) {
-        this(windowDuration, databasePath, clock, processWorkers, eventFactory, pluginMetrics, Collections.emptyList());
+        this(windowDuration, databasePath, clock, processWorkers, eventFactory, pluginMetrics,
+                Collections.emptyList(), MetricTimestampSource.SPAN_END_TIME);
     }
 
     OTelApmServiceMapProcessor(final Duration windowDuration,
@@ -129,9 +135,23 @@ public OTelApmServiceMapProcessor(
                                final EventFactory eventFactory,
                                final PluginMetrics pluginMetrics,
                                final List<String> groupByAttributes) {
+        this(windowDuration, databasePath, clock, processWorkers, eventFactory, pluginMetrics,
+                groupByAttributes, MetricTimestampSource.SPAN_END_TIME);
+    }
+
+    OTelApmServiceMapProcessor(final Duration windowDuration,
+                               final File databasePath,
+                               final Clock clock,
+                               final int processWorkers,
+                               final EventFactory eventFactory,
+                               final PluginMetrics pluginMetrics,
+                               final List<String> groupByAttributes,
+                               final MetricTimestampSource metricTimestampSource) {
         super(pluginMetrics);
 
+        this.hostId = resolveHostId();
         this.groupByAttributes = groupByAttributes != null ? Collections.unmodifiableList(groupByAttributes) : Collections.emptyList();
+        this.metricTimestampSource = metricTimestampSource != null ? metricTimestampSource : MetricTimestampSource.ARRIVAL_TIME;
 
         this.eventFactory = eventFactory;
         OTelApmServiceMapProcessor.clock = clock;
@@ -448,19 +468,23 @@ private Map<String, String> extractGroupByAttributes(final Span span) {
     }
 
     /**
-     * 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.
+     * Get anchor timestamp for metrics, truncated to the specified unit.
+     * When metric_timestamp_source is ARRIVAL_TIME, uses fallbackTime (clock.instant()).
+     * When metric_timestamp_source is SPAN_END_TIME, uses the span's endTime field.
      *
      * @param spanStateData The span to extract timestamp from
-     * @param fallbackTime Current system time to use if span endTime is null
+     * @param fallbackTime Current system time to use as arrival time or if span endTime is null
      * @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,
                                                final ChronoUnit truncationUnit) {
-        Instant timestamp = fallbackTime; // Default to current system time
+        if (metricTimestampSource == MetricTimestampSource.ARRIVAL_TIME) {
+            return fallbackTime.truncatedTo(truncationUnit);
+        }
 
+        // SPAN_END_TIME mode: parse span's endTime, fall back to system time
+        Instant timestamp = fallbackTime;
         final String endTime = spanStateData.getEndTime();
         try {
             if (endTime != null && !endTime.isEmpty()) {
@@ -476,19 +500,17 @@ private Instant getAnchorTimestampFromSpan(final SpanStateData spanStateData, fi
 
     /**
      * 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.
+     * Uses a truncated SHA-256 hash of the hostname (from {@link HostContext})
+     * to ensure uniqueness without revealing the actual hostname in emitted metrics.
      */
-    private static String resolveHostId() {
+    private 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));
+            final String hostname = HostContext.getHostname();
+            final MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            final byte[] hash = digest.digest(hostname.getBytes(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();
+        } catch (final java.security.NoSuchAlgorithmException e) {
+            throw new RuntimeException("SHA-256 algorithm not available", e);
         }
     }
 
@@ -777,7 +799,7 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                     final Instant histAnchor = getAnchorTimestampFromSpan(clientSpan, currentTime, ChronoUnit.MINUTES);
 
                     final NodeOperationDetail nodeOperationDetail = new NodeOperationDetail(
-                            sourceNode, targetNode, sourceOp, targetOp, sumAnchor);
+                            sourceNode, targetNode, sourceOp, targetOp, histAnchor);
 
                     final EventMetadata eventMetadata = new DefaultEventMetadata.Builder()
                             .withEventType(EVENT_TYPE_OTEL_APM_SERVICE_MAP).build();
@@ -792,7 +814,7 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                     if (decoration.getParentServerOperationName() != null) {
                         ApmServiceMapMetricsUtil.generateMetricsForClientSpan(
                                 clientSpan, decoration, currentTime, sumStateByKey, histogramStateByKey,
-                                sumAnchor, histAnchor, HOST_ID);
+                                sumAnchor, histAnchor, hostId);
                     }
                 }
             }
@@ -805,7 +827,7 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                 final Instant histAnchor = getAnchorTimestampFromSpan(serverSpan, currentTime, ChronoUnit.MINUTES);
                 ApmServiceMapMetricsUtil.generateMetricsForServerSpan(
                         serverSpan, currentTime, sumStateByKey, histogramStateByKey,
-                        sumAnchor, histAnchor, HOST_ID);
+                        sumAnchor, histAnchor, hostId);
 
                 final ServerSpanDecoration decoration = traceData.getDecorations().getServerDecoration(serverSpan.getSpanId());
 
@@ -819,7 +841,7 @@ private Collection<Record<Event>> generateNodeOperationDetailEvents(final ThreeW
                     final Operation sourceOp = new Operation(serverSpan.getOperationName());
 
                     final NodeOperationDetail nodeOperationDetail = new NodeOperationDetail(
-                            sourceNode, null, sourceOp, null, sumAnchor);
+                            sourceNode, null, sourceOp, null, histAnchor);
 
                     final EventMetadata eventMetadata = new DefaultEventMetadata.Builder()
                             .withEventType(EVENT_TYPE_OTEL_APM_SERVICE_MAP).build();

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

@@ -43,6 +43,11 @@ public class OTelApmServiceMapProcessorConfig {
             "when present on the span's resource attributes. Only applied to primary Service objects, not dependency services.")
     private List<String> groupByAttributes = Collections.emptyList();
 
+    @JsonProperty("metric_timestamp_source")
+    @JsonPropertyDescription("The timestamp source for emitted metrics. 'arrival_time' (default) uses processing time " +
+            "at window evaluation, avoiding late-span data loss in Prometheus/AMP. 'span_end_time' uses the span's endTime field.")
+    private MetricTimestampSource metricTimestampSource = MetricTimestampSource.ARRIVAL_TIME;
+
     public Duration getWindowDuration() {
         return windowDuration;
     }
@@ -54,4 +59,8 @@ public String getDbPath() {
     public List<String> getGroupByAttributes() {
         return groupByAttributes != null ? Collections.unmodifiableList(groupByAttributes) : Collections.emptyList();
     }
+
+    public MetricTimestampSource getMetricTimestampSource() {
+        return metricTimestampSource;
+    }
 }