Add a node id to all spans

CreateSnapshot/src/main/java/com/rfs/CreateSnapshot.java

@@ -7,6 +7,7 @@
 import org.opensearch.migrations.tracing.ActiveContextTrackerByActivityType;
 import org.opensearch.migrations.tracing.CompositeContextTracker;
 import org.opensearch.migrations.tracing.RootOtelContext;
+import org.opensearch.migrations.utils.ProcessHelpers;
 
 import com.beust.jcommander.JCommander;
 import com.beust.jcommander.Parameter;
@@ -83,7 +84,8 @@
         }
 
         var rootContext = new RootSnapshotContext(
-            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(arguments.otelCollectorEndpoint, "rfs"),
+            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(arguments.otelCollectorEndpoint,
+                RootSnapshotContext.SCOPE_NAME, ProcessHelpers.getNodeInstanceName()),
             new CompositeContextTracker(new ActiveContextTracker(), new ActiveContextTrackerByActivityType())
         );
 

DocumentsFromSnapshotMigration/src/main/java/com/rfs/RfsMigrateDocuments.java

@@ -6,7 +6,6 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.util.List;
-import java.util.UUID;
 import java.util.function.Function;
 
 import org.opensearch.migrations.Version;
@@ -17,6 +16,7 @@
 import org.opensearch.migrations.tracing.ActiveContextTrackerByActivityType;
 import org.opensearch.migrations.tracing.CompositeContextTracker;
 import org.opensearch.migrations.tracing.RootOtelContext;
+import org.opensearch.migrations.utils.ProcessHelpers;
 
 import com.beust.jcommander.IStringConverter;
 import com.beust.jcommander.JCommander;
@@ -41,7 +41,6 @@
 import com.rfs.common.http.ConnectionContext;
 import com.rfs.models.IndexMetadata;
 import com.rfs.models.ShardMetadata;
-import com.rfs.tracing.RootWorkCoordinationContext;
 import com.rfs.worker.DocumentsRunner;
 import com.rfs.worker.ShardWorkPreparer;
 import lombok.extern.slf4j.Slf4j;
@@ -168,6 +167,10 @@
     }
 
     public static void main(String[] args) throws Exception {
+        System.err.println("Got args: " + String.join("; ", args));
+        var workerId = ProcessHelpers.getNodeInstanceName();
+        log.info("Starting RfsMigrateDocuments with workerId =" + workerId);
+
         Args arguments = new Args();
         JCommander jCommander = JCommander.newBuilder().addObject(arguments).build();
         jCommander.parse(args);
@@ -179,7 +182,7 @@
 
         validateArgs(arguments);
 
-        var rootDocumentContext = makeRootContext(arguments);
+        var rootDocumentContext = makeRootContext(arguments, workerId);
         var luceneDirPath = Paths.get(arguments.luceneDir);
         var snapshotLocalDirPath = arguments.snapshotLocalDir != null ? Paths.get(arguments.snapshotLocalDir) : null;
 
@@ -188,16 +191,13 @@
             System.exit(PROCESS_TIMED_OUT);
         }, Clock.systemUTC())) {
             ConnectionContext connectionContext = arguments.targetArgs.toConnectionContext();
-            final var workerId = UUID.randomUUID().toString();
             var workCoordinator = new OpenSearchWorkCoordinator(
                 new CoordinateWorkHttpClient(connectionContext),
                 TOLERABLE_CLIENT_SERVER_CLOCK_DIFFERENCE_SECONDS,
                 workerId
             );
             MDC.put(LOGGING_MDC_WORKER_ID, workerId); // I don't see a need to clean this up since we're in main
             TryHandlePhaseFailure.executeWithTryCatch(() -> {
-                log.info("Running RfsMigrateDocuments with workerId = " + workerId);
-
                 OpenSearchClient targetClient = new OpenSearchClient(connectionContext);
                 DocumentReindexer reindexer = new DocumentReindexer(targetClient,
                     arguments.numDocsPerBulkRequest,
@@ -243,17 +243,17 @@
         }
     }
 
-    private static RootDocumentMigrationContext makeRootContext(Args arguments) {
+    private static RootDocumentMigrationContext makeRootContext(Args arguments, String workerId) {
         var compositeContextTracker = new CompositeContextTracker(
             new ActiveContextTracker(),
             new ActiveContextTrackerByActivityType()
         );
         var otelSdk = RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(
             arguments.otelCollectorEndpoint,
-            "docMigration"
+            RootDocumentMigrationContext.SCOPE_NAME,
+            workerId
         );
-        var workContext = new RootWorkCoordinationContext(otelSdk, compositeContextTracker);
-        return new RootDocumentMigrationContext(otelSdk, compositeContextTracker, workContext);
+        return new RootDocumentMigrationContext(otelSdk, compositeContextTracker);
     }
 
     public static DocumentsRunner.CompletionStatus run(

DocumentsFromSnapshotMigration/src/test/java/com/rfs/EndToEndTest.java

@@ -13,7 +13,6 @@
 import org.opensearch.migrations.metadata.tracing.MetadataMigrationTestContext;
 import org.opensearch.migrations.reindexer.tracing.DocumentMigrationTestContext;
 import org.opensearch.migrations.snapshot.creation.tracing.SnapshotTestContext;
-import org.opensearch.migrations.workcoordination.tracing.WorkCoordinationTestContext;
 
 import com.rfs.common.FileSystemRepo;
 import com.rfs.common.FileSystemSnapshotCreator;
@@ -37,8 +36,7 @@ public class EndToEndTest extends SourceTestBase {
     public void migrateFrom_ES_v6_8(final SearchClusterContainer.ContainerVersion targetVersion) throws Exception {
         final var snapshotContext = SnapshotTestContext.factory().noOtelTracking();
         final var metadataContext = MetadataMigrationTestContext.factory().noOtelTracking();
-        final var workCoordinationContext = WorkCoordinationTestContext.factory().noOtelTracking();
-        final var testDocMigrationContext = DocumentMigrationTestContext.factory(workCoordinationContext).noOtelTracking();
+        final var testDocMigrationContext = DocumentMigrationTestContext.factory().noOtelTracking();
 
         try (
             final var sourceCluster = new SearchClusterContainer(SearchClusterContainer.ES_V6_8_23);
@@ -163,8 +161,7 @@ private void migrateFrom_ES_v7_X(
     ) {
         final var snapshotContext = SnapshotTestContext.factory().noOtelTracking();
         final var metadataContext = MetadataMigrationTestContext.factory().noOtelTracking();
-        final var workCoordinationContext = WorkCoordinationTestContext.factory().noOtelTracking();
-        final var testDocMigrationContext = DocumentMigrationTestContext.factory(workCoordinationContext).noOtelTracking();
+        final var testDocMigrationContext = DocumentMigrationTestContext.factory().noOtelTracking();
 
         try {
 

DocumentsFromSnapshotMigration/src/test/java/com/rfs/ParallelDocumentMigrationsTest.java

@@ -23,7 +23,6 @@
 import org.opensearch.migrations.metadata.tracing.MetadataMigrationTestContext;
 import org.opensearch.migrations.reindexer.tracing.DocumentMigrationTestContext;
 import org.opensearch.migrations.snapshot.creation.tracing.SnapshotTestContext;
-import org.opensearch.migrations.workcoordination.tracing.WorkCoordinationTestContext;
 
 import com.rfs.common.FileSystemRepo;
 import com.rfs.common.FileSystemSnapshotCreator;
@@ -45,7 +44,7 @@ public class ParallelDocumentMigrationsTest extends SourceTestBase {
 
     public static Stream<Arguments> makeDocumentMigrationArgs() {
         List<Object[]> sourceImageArgs = SOURCE_IMAGES.stream()
-            .map(name -> makeParamsForBase(name))
+            .map(SourceTestBase::makeParamsForBase)
             .collect(Collectors.toList());
         var targetImageNames = TARGET_IMAGES.stream()
             .collect(Collectors.toList());
@@ -82,8 +81,7 @@ public void testDocumentMigration(
         var executorService = Executors.newFixedThreadPool(numWorkers);
         final var testSnapshotContext = SnapshotTestContext.factory().noOtelTracking();
         final var testMetadataMigrationContext = MetadataMigrationTestContext.factory().noOtelTracking();
-        final var workCoordinationContext = WorkCoordinationTestContext.factory().withAllTracking();
-        final var testDocMigrationContext = DocumentMigrationTestContext.factory(workCoordinationContext)
+        final var testDocMigrationContext = DocumentMigrationTestContext.factory()
             .withAllTracking();
 
         try (
@@ -197,7 +195,7 @@ public void testDocumentMigration(
     }
 
     private void verifyWorkMetrics(DocumentMigrationTestContext rootContext, int numWorkers, int numRuns) {
-        var workMetrics = rootContext.getWorkCoordinationContext().inMemoryInstrumentationBundle.getFinishedMetrics();
+        var workMetrics = rootContext.inMemoryInstrumentationBundle.getFinishedMetrics();
         var migrationMetrics = rootContext.inMemoryInstrumentationBundle.getFinishedMetrics();
 
         verifyCoordinatorBehavior(workMetrics, numRuns);

DocumentsFromSnapshotMigration/src/test/java/com/rfs/ProcessLifecycleTest.java

@@ -61,7 +61,7 @@ enum FailHow {
         // The Document Migration process will throw an exception immediately, which will cause an exit.
         "AT_STARTUP, 1",
         // This test is dependent upon the max lease duration that is passed to the command line. It's set
-        // to such a short value (1s), that no document migration will exit in that amount of time. For good
+        // to such a short value (1s) that no document migration will exit in that amount of time. For good
         // measure though, the toxiproxy also adds latency to the requests to make it impossible for the
         // migration to complete w/in that 1s.
         "WITH_DELAYS, 2" })

MetadataMigration/src/main/java/org/opensearch/migrations/MetadataMigration.java

@@ -8,6 +8,7 @@
 import org.opensearch.migrations.tracing.ActiveContextTrackerByActivityType;
 import org.opensearch.migrations.tracing.CompositeContextTracker;
 import org.opensearch.migrations.tracing.RootOtelContext;
+import org.opensearch.migrations.utils.ProcessHelpers;
 
 import com.beust.jcommander.JCommander;
 import lombok.extern.slf4j.Slf4j;
@@ -26,7 +27,8 @@
         }
 
         var context = new RootMetadataMigrationContext(
-            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(arguments.otelCollectorEndpoint, "metadata"),
+            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(arguments.otelCollectorEndpoint, "metadata",
+                ProcessHelpers.getNodeInstanceName()),
             new CompositeContextTracker(new ActiveContextTracker(), new ActiveContextTrackerByActivityType())
         );
 

RFS/src/main/java/com/rfs/tracing/RootWorkCoordinationContext.java

@@ -9,6 +9,8 @@
 public class RootWorkCoordinationContext extends RootOtelContext {
     private static final String SCOPE_NAME = "workCoordination";
 
+    public final RootOtelContext enclosingScope;
+
     public final WorkCoordinationContexts.InitializeCoordinatorStateContext.MetricInstruments initializeMetrics;
     public final WorkCoordinationContexts.CreateUnassignedWorkItemContext.MetricInstruments createUnassignedWorkMetrics;
     public final WorkCoordinationContexts.Refresh.MetricInstruments refreshMetrics;
@@ -18,7 +20,15 @@ public class RootWorkCoordinationContext extends RootOtelContext {
     public final WorkCoordinationContexts.AcquireNextWorkItemContext.MetricInstruments acquireNextWorkMetrics;
 
     public RootWorkCoordinationContext(OpenTelemetry sdk, IContextTracker contextTracker) {
+        this(sdk, contextTracker, null);
+    }
+
+    public RootWorkCoordinationContext(OpenTelemetry sdk,
+                                       IContextTracker contextTracker,
+                                       RootOtelContext enclosingScope)
+    {
         super(SCOPE_NAME, contextTracker, sdk);
+        this.enclosingScope = enclosingScope;
         var meter = this.getMeterProvider().get(SCOPE_NAME);
 
         initializeMetrics = WorkCoordinationContexts.InitializeCoordinatorStateContext.makeMetrics(meter);

RFS/src/main/java/org/opensearch/migrations/reindexer/tracing/RootDocumentMigrationContext.java

@@ -19,12 +19,11 @@ public class RootDocumentMigrationContext extends BaseRootRfsContext implements
 
     public RootDocumentMigrationContext(
         OpenTelemetry sdk,
-        IContextTracker contextTracker,
-        RootWorkCoordinationContext workCoordinationContext
+        IContextTracker contextTracker
     ) {
         super(SCOPE_NAME, sdk, contextTracker);
         var meter = this.getMeterProvider().get(SCOPE_NAME);
-        this.workCoordinationContext = workCoordinationContext;
+        workCoordinationContext = new RootWorkCoordinationContext(sdk, contextTracker, this);
         documentReindexInstruments = DocumentMigrationContexts.DocumentReindexContext.makeMetrics(meter);
         shardSetupMetrics = DocumentMigrationContexts.ShardSetupAttemptContext.makeMetrics(meter);
         addShardWorkItemMetrics = DocumentMigrationContexts.AddShardWorkItemContext.makeMetrics(meter);

RFS/src/test/java/com/rfs/integration/SnapshotStateTest.java

@@ -10,7 +10,6 @@
 import org.junit.jupiter.api.io.TempDir;
 
 import org.opensearch.migrations.reindexer.tracing.DocumentMigrationTestContext;
-import org.opensearch.migrations.workcoordination.tracing.WorkCoordinationTestContext;
 
 import com.rfs.common.DocumentReindexer.BulkDocSection;
 import com.rfs.common.OpenSearchClient;
@@ -67,8 +66,7 @@ public void tearDown() throws Exception {
     @Test
     public void SingleSnapshot_SingleDocument() throws Exception {
         // Setup
-        final var workCoordinationContext = WorkCoordinationTestContext.factory().noOtelTracking();
-        final var testContext = DocumentMigrationTestContext.factory(workCoordinationContext).noOtelTracking();
+        final var testContext = DocumentMigrationTestContext.factory().noOtelTracking();
         final var indexName = "my-index";
         final var document1Id = "doc1";
         final var document1Body = "   \n {\n\"fo$o\":\"bar\"\n}\t \n"; // Verify that we trim and remove newlines
@@ -106,8 +104,7 @@ public void SingleSnapshot_SingleDocument() throws Exception {
     @Test
     public void SingleSnapshot_SingleDocument_Then_DeletedDocument() throws Exception {
         // Setup
-        final var workCoordinationContext = WorkCoordinationTestContext.factory().noOtelTracking();
-        final var testContext = DocumentMigrationTestContext.factory(workCoordinationContext).noOtelTracking();
+        final var testContext = DocumentMigrationTestContext.factory().noOtelTracking();
         final var indexName = "my-index-with-deleted-item";
         final var document1Id = "doc1-going-to-be-deleted";
         final var document1Body = "{\"foo\":\"bar\"}";
@@ -140,8 +137,7 @@ public void SingleSnapshot_SingleDocument_Then_DeletedDocument() throws Exceptio
     @Test
     public void SingleSnapshot_SingleDocument_Then_UpdateDocument() throws Exception {
         // Setup
-        final var workCoordinationContext = WorkCoordinationTestContext.factory().noOtelTracking();
-        final var testContext = DocumentMigrationTestContext.factory(workCoordinationContext).noOtelTracking();
+        final var testContext = DocumentMigrationTestContext.factory().noOtelTracking();
         final var indexName = "my-index-with-updated-item";
         final var document1Id = "doc1-going-to-be-updated";
         final var document1BodyOriginal = "{\"foo\":\"bar\"}";

RFS/src/testFixtures/java/org/opensearch/migrations/reindexer/tracing/DocumentMigrationTestContext.java

@@ -2,7 +2,6 @@
 
 import org.opensearch.migrations.tracing.IContextTracker;
 import org.opensearch.migrations.tracing.InMemoryInstrumentationBundle;
-import org.opensearch.migrations.workcoordination.tracing.WorkCoordinationTestContext;
 
 import com.rfs.framework.tracing.TrackingTestContextFactory;
 import com.rfs.tracing.IRfsContexts;
@@ -11,26 +10,16 @@
 public class DocumentMigrationTestContext extends RootDocumentMigrationContext {
     public final InMemoryInstrumentationBundle inMemoryInstrumentationBundle;
 
-    @Override
-    public WorkCoordinationTestContext getWorkCoordinationContext() {
-        return (WorkCoordinationTestContext) super.getWorkCoordinationContext();
-    }
-
     public DocumentMigrationTestContext(
         InMemoryInstrumentationBundle inMemoryInstrumentationBundle,
-        IContextTracker contextTracker,
-        WorkCoordinationTestContext workCoordinationContext
+        IContextTracker contextTracker
     ) {
-        super(inMemoryInstrumentationBundle.openTelemetrySdk, contextTracker, workCoordinationContext);
+        super(inMemoryInstrumentationBundle.openTelemetrySdk, contextTracker);
         this.inMemoryInstrumentationBundle = inMemoryInstrumentationBundle;
     }
 
-    public static TrackingTestContextFactory<DocumentMigrationTestContext> factory(
-        WorkCoordinationTestContext rootWorkCoordinationContext
-    ) {
-        return new TrackingTestContextFactory<>(
-            (a, b) -> new DocumentMigrationTestContext(a, b, rootWorkCoordinationContext)
-        );
+    public static TrackingTestContextFactory<DocumentMigrationTestContext> factory() {
+        return new TrackingTestContextFactory<>(DocumentMigrationTestContext::new);
     }
 
     public IRfsContexts.IRequestContext createUnboundRequestContext() {

TrafficCapture/trafficCaptureProxyServer/src/main/java/org/opensearch/migrations/trafficcapture/proxyserver/CaptureProxy.java

@@ -40,6 +40,7 @@
 import org.opensearch.migrations.trafficcapture.netty.HeaderValueFilteringCapturePredicate;
 import org.opensearch.migrations.trafficcapture.proxyserver.netty.BacksideConnectionPool;
 import org.opensearch.migrations.trafficcapture.proxyserver.netty.NettyScanningHttpProxy;
+import org.opensearch.migrations.utils.ProcessHelpers;
 import org.opensearch.security.ssl.DefaultSecurityKeyStore;
 import org.opensearch.security.ssl.util.SSLConfigConstants;
 
@@ -306,14 +307,15 @@ protected static Map<String, String> convertPairListToMap(List<String> list) {
     }
 
     public static void main(String[] args) throws InterruptedException, IOException {
-        System.err.println("Starting Capture Proxy");
         System.err.println("Got args: " + String.join("; ", args));
+        log.info("Starting Capture Proxy on " + ProcessHelpers.getNodeInstanceName());
 
         var params = parseArgs(args);
         var backsideUri = convertStringToUri(params.backsideUriString);
 
         var rootContext = new RootCaptureContext(
-            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(params.otelCollectorEndpoint, "capture"),
+            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(params.otelCollectorEndpoint, "capture",
+                ProcessHelpers.getNodeInstanceName()),
             new CompositeContextTracker(new ActiveContextTracker(), new ActiveContextTrackerByActivityType())
         );
 

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/TrafficReplayer.java

@@ -27,6 +27,7 @@
 import org.opensearch.migrations.transform.RemovingAuthTransformerFactory;
 import org.opensearch.migrations.transform.SigV4AuthTransformerFactory;
 import org.opensearch.migrations.transform.StaticAuthTransformerFactory;
+import org.opensearch.migrations.utils.ProcessHelpers;
 
 import com.beust.jcommander.JCommander;
 import com.beust.jcommander.Parameter;
@@ -221,11 +222,13 @@
     }
 
     public static void main(String[] args) throws Exception {
+        System.err.println("Got args: " + String.join("; ", args));
+        final var workerId = ProcessHelpers.getNodeInstanceName();
+        log.info("Starting Traffic Replayer with id=" + workerId);
+
         var activeContextLogger = LoggerFactory.getLogger(ALL_ACTIVE_CONTEXTS_MONITOR_LOGGER);
         var params = parseArgs(args);
         URI uri;
-        System.err.println("Starting Traffic Replayer");
-        System.err.println("Got args: " + String.join("; ", args));
         try {
             uri = new URI(params.targetUriString);
         } catch (Exception e) {
@@ -258,7 +261,9 @@
         );
         var contextTrackers = new CompositeContextTracker(globalContextTracker, perContextTracker);
         var topContext = new RootReplayerContext(
-            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(params.otelCollectorEndpoint, "replay"),
+            RootOtelContext.initializeOpenTelemetryWithCollectorOrAsNoop(params.otelCollectorEndpoint,
+                "replay",
+                ProcessHelpers.getNodeInstanceName()),
             contextTrackers
         );