Prevent trace context propagation for self-rescheduling operations to not create everlasting APM transactions

Author: mosche

server/src/main/java/org/elasticsearch/health/node/LocalHealthMonitor.java

@@ -262,7 +262,7 @@ private Monitoring(
          * Schedule the first run of the monitor.
          */
         public void start() {
-            scheduledRun = threadPool.schedule(this, TimeValue.ZERO, executor);
+            scheduleNextRun(TimeValue.ZERO);
         }
 
         /**
@@ -383,8 +383,14 @@ private void scheduleNextRunIfNecessary() {
             if (cancelled) {
                 return;
             }
-            try {
-                scheduledRun = threadPool.schedule(this, interval, executor);
+            scheduleNextRun(interval);
+        }
+
+        private void scheduleNextRun(TimeValue delay) {
+            // Prevent the propagation of the trace context here to not create everlasting APM transactions.
+            // Such a trace context is created when executing a transport action.
+            try (var ignored = threadPool.getThreadContext().newEmptySystemContext()) {
+                scheduledRun = threadPool.schedule(this, delay, executor);
             } catch (final EsRejectedExecutionException e) {
                 logger.debug(() -> format("Scheduled health monitoring was rejected on thread pool [%s]", executor), e);
             }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedRunner.java

@@ -338,7 +338,11 @@ protected void doRun() {
                     }
                     holder.problemTracker.finishReport();
                     if (nextDelayInMsSinceEpoch >= 0) {
-                        doDatafeedRealtime(nextDelayInMsSinceEpoch, jobId, holder);
+                        // Prevent the propagation of the trace context here to not create everlasting APM transactions.
+                        // Such a trace context is created when executing any transport action.
+                        try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
+                            doDatafeedRealtime(nextDelayInMsSinceEpoch, jobId, holder);
+                        }
                     }
                 }
             }, delay, threadPool.executor(MachineLearning.DATAFEED_THREAD_POOL_NAME));