[ML] Fix job ID in C++ logs for normalize and memory estimation

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/process/NativeAnalyticsProcessFactory.java

@@ -73,7 +73,7 @@ public NativeAnalyticsProcess createAnalyticsProcess(DataFrameAnalyticsConfig co
         String jobId = config.getId();
         List<Path> filesToDelete = new ArrayList<>();
         ProcessPipes processPipes = new ProcessPipes(env, NAMED_PIPE_HELPER, processConnectTimeout, AnalyticsBuilder.ANALYTICS, jobId,
-                false, true, true, hasState, config.getAnalysis().persistsState());
+            null, false, true, true, hasState, config.getAnalysis().persistsState());
 
         // The extra 2 are for the checksum and the control field
         int numberOfFields = analyticsProcessConfig.cols() + 2;

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/dataframe/process/NativeMemoryUsageEstimationProcessFactory.java

@@ -62,12 +62,11 @@ public NativeMemoryUsageEstimationProcess createAnalyticsProcess(
             ExecutorService executorService,
             Consumer<String> onProcessCrash) {
         List<Path> filesToDelete = new ArrayList<>();
-        // The config ID passed to the process pipes is only used to make the file names unique.  Since memory estimation can be
-        // called many times in quick succession for the same config the config ID alone is not sufficient to guarantee that the
-        // memory estimation process pipe names are unique.  Therefore an increasing counter value is appended to the config ID
-        // to ensure uniqueness between calls.
+        // Since memory estimation can be called many times in quick succession for the same config the config ID alone is not
+        // sufficient to guarantee that the memory estimation process pipe names are unique.  Therefore an increasing counter
+        // value is passed as well as the config ID to ensure uniqueness between calls.
         ProcessPipes processPipes = new ProcessPipes(
-            env, NAMED_PIPE_HELPER, processConnectTimeout, AnalyticsBuilder.ANALYTICS, config.getId() + "_" + counter.incrementAndGet(),
+            env, NAMED_PIPE_HELPER, processConnectTimeout, AnalyticsBuilder.ANALYTICS, config.getId(), counter.incrementAndGet(),
             false, false, true, false, false);
 
         createNativeProcess(config.getId(), analyticsProcessConfig, filesToDelete, processPipes);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/NativeAutodetectProcessFactory.java

@@ -77,7 +77,7 @@ public AutodetectProcess createAutodetectProcess(Job job,
                                                      Consumer<String> onProcessCrash) {
         List<Path> filesToDelete = new ArrayList<>();
         ProcessPipes processPipes = new ProcessPipes(env, NAMED_PIPE_HELPER, processConnectTimeout, AutodetectBuilder.AUTODETECT,
-            job.getId(), false, true, true, params.modelSnapshot() != null,
+            job.getId(), null, false, true, true, params.modelSnapshot() != null,
             AutodetectBuilder.DONT_PERSIST_MODEL_STATE_SETTING.get(settings) == false);
         createNativeProcess(job, params, processPipes, filesToDelete);
         boolean includeTokensField = MachineLearning.CATEGORIZATION_TOKENIZATION_IN_JAVA

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/normalizer/NativeNormalizerProcessFactory.java

@@ -51,11 +51,11 @@ void setProcessConnectTimeout(TimeValue processConnectTimeout) {
     @Override
     public NormalizerProcess createNormalizerProcess(String jobId, String quantilesState, Integer bucketSpan,
                                                      ExecutorService executorService) {
-        // The job ID passed to the process pipes is only used to make the file names unique.  Since normalize can get run many times
-        // in quick succession for the same job the job ID alone is not sufficient to guarantee that the normalizer process pipe names
-        // are unique.  Therefore an increasing counter value is appended to the job ID to ensure uniqueness between calls.
+        // Since normalize can get run many times in quick succession for the same job the job ID alone is not sufficient to
+        // guarantee that the normalizer process pipe names are unique.  Therefore an increasing counter value is passed as
+        // well as the job ID to ensure uniqueness between calls.
         ProcessPipes processPipes = new ProcessPipes(env, NAMED_PIPE_HELPER, processConnectTimeout, NormalizerBuilder.NORMALIZE,
-            jobId + "_" + counter.incrementAndGet(), false, true, true, false, false);
+            jobId, counter.incrementAndGet(), false, true, true, false, false);
         createNativeProcess(jobId, quantilesState, processPipes, bucketSpan);
 
         NativeNormalizerProcess normalizerProcess = new NativeNormalizerProcess(jobId, nativeController, processPipes);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/process/NativeController.java

@@ -69,7 +69,7 @@ public static NativeController makeNativeController(String localNodeName, Enviro
     NativeController(String localNodeName, Environment env, NamedPipeHelper namedPipeHelper, NamedXContentRegistry xContentRegistry)
         throws IOException {
         this.localNodeName = localNodeName;
-        ProcessPipes processPipes = new ProcessPipes(env, namedPipeHelper, CONTROLLER_CONNECT_TIMEOUT, CONTROLLER, null,
+        ProcessPipes processPipes = new ProcessPipes(env, namedPipeHelper, CONTROLLER_CONNECT_TIMEOUT, CONTROLLER, null, null,
                 true, false, true, false, false);
         processPipes.connectLogStream();
         this.cppLogHandler = processPipes.getLogStreamHandler();

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/process/ProcessPipes.java

@@ -76,7 +76,7 @@ public class ProcessPipes {
      *              May be null or empty for processes not associated with a specific job.
      */
     public ProcessPipes(Environment env, NamedPipeHelper namedPipeHelper, Duration timeout, String processName, String jobId,
-                        boolean wantCommandPipe, boolean wantProcessInPipe, boolean wantProcessOutPipe,
+                        Long uniqueId, boolean wantCommandPipe, boolean wantProcessInPipe, boolean wantProcessOutPipe,
                         boolean wantRestorePipe, boolean wantPersistPipe) {
         this.namedPipeHelper = namedPipeHelper;
         this.jobId = jobId;
@@ -91,6 +91,9 @@ public ProcessPipes(Environment env, NamedPipeHelper namedPipeHelper, Duration t
         if (!Strings.isNullOrEmpty(jobId)) {
             prefixBuilder.append(jobId).append('_');
         }
+        if (uniqueId != null) {
+            prefixBuilder.append(uniqueId).append('_');
+        }
         String prefix = prefixBuilder.toString();
         String suffix = String.format(Locale.ROOT, "_%d", JvmInfo.jvmInfo().getPid());
         logPipeName = String.format(Locale.ROOT, "%slog%s", prefix, suffix);

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/process/ProcessPipesTests.java

@@ -62,7 +62,7 @@ public void testProcessPipes() throws Exception {
 
         int timeoutSeconds = randomIntBetween(5, 100);
         ProcessPipes processPipes = new ProcessPipes(env, namedPipeHelper, Duration.ofSeconds(timeoutSeconds), AutodetectBuilder.AUTODETECT,
-            "my_job", false, true, true, true, true);
+            "my_job", null, false, true, true, true, true);
 
         List<String> command = new ArrayList<>();
         processPipes.addArgs(command);
@@ -110,7 +110,7 @@ public void testCloseUnusedPipes_notConnected() {
         Settings settings = Settings.builder().put(Environment.PATH_HOME_SETTING.getKey(), createTempDir().toString()).build();
         Environment env = TestEnvironment.newEnvironment(settings);
 
-        new ProcessPipes(env, namedPipeHelper, Duration.ofSeconds(2), AutodetectBuilder.AUTODETECT, "my_job",
+        new ProcessPipes(env, namedPipeHelper, Duration.ofSeconds(2), AutodetectBuilder.AUTODETECT, "my_job", null,
                 true, true, true, true, true);
     }
 
@@ -138,7 +138,7 @@ public void testCloseOpenedPipesOnError() throws IOException {
         Settings settings = Settings.builder().put(Environment.PATH_HOME_SETTING.getKey(), createTempDir().toString()).build();
         Environment env = TestEnvironment.newEnvironment(settings);
         ProcessPipes processPipes = new ProcessPipes(env, namedPipeHelper, Duration.ofSeconds(2), AutodetectBuilder.AUTODETECT, "my_job",
-                true, true, true, true, true);
+            null, true, true, true, true, true);
 
         processPipes.connectLogStream();
         expectThrows(IOException.class, processPipes::connectOtherStreams);