[ML] Catch exceptions during pytorch_inference startup (elastic#103873)…

… (elastic#103911)

A couple of test failures (elastic#103808 and elastic#103868) reveal that
an exception thrown while connecting to the pytorch_inference
process can be uncaught and hence cause the whole node to stop.

This change does not fix the underlying problem of failure to
connect to the process that those issues relate to, but it
converts the error from one that crashes a whole node to one
that just fails the affected model deployment.

docs/changelog/103873.yaml

@@ -0,0 +1,5 @@
+pr: 103873
+summary: Catch exceptions during `pytorch_inference` startup
+area: Machine Learning
+type: bug
+issues: []

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/deployment/DeploymentManager.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.common.xcontent.LoggingDeprecationHandler;
 import org.elasticsearch.core.TimeValue;
@@ -200,9 +201,18 @@ public void startDeployment(TrainedModelDeploymentTask task, ActionListener<Trai
  // here, we are being called back on the searching thread, which MAY be a network thread
  // `startAndLoad` creates named pipes, blocking the calling thread, better to execute that in our utility
  // executor.
- executorServiceForDeployment.execute(
- () -> processContext.startAndLoad(modelConfig.getLocation(), modelLoadedListener)
- );
+ executorServiceForDeployment.execute(new AbstractRunnable() {
+
+ @Override
+ public void onFailure(Exception e) {
+ failedDeploymentListener.onFailure(e);
+ }
+
+ @Override
+ protected void doRun() {
+ processContext.startAndLoad(modelConfig.getLocation(), modelLoadedListener);
+ }
+ });
  }, failedDeploymentListener::onFailure)
  );
  } else {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/inference/pytorch/process/NativePyTorchProcessFactory.java

@@ -89,7 +89,7 @@ public NativePyTorchProcess createProcess(
  process.start(executorService);
  } catch (IOException | EsRejectedExecutionException e) {
  String msg = "Failed to connect to pytorch process for job " + task.getDeploymentId();
- logger.error(msg);
+ logger.error(msg, e);
  try {
  IOUtils.close(process);
  } catch (IOException ioe) {