[ML] Typed routing behind the established controller protocol

.buildkite/scripts/steps/run_tests.sh

@@ -49,12 +49,21 @@ TEST_OUTCOME=0
 
 if [[ "$HARDWARE_ARCH" = aarch64 && -z "${CPP_CROSS_COMPILE:-}" && "$(uname)" = Linux ]]; then
     # --- Linux aarch64: run tests inside Docker container from base image ---
+    # aarch64 Buildkite k8s pods are the only runners here with userns
+    # capability (mount("proc", ...) succeeds), so this is the only branch
+    # that can exercise ML_SANDBOX2_REQUIRE=enforced - and it runs only that
+    # mode: aarch64 is pinned to enforced, x86_64 stays fail-closed. A
+    # second fail_closed pass on this same host/kernel would assert the
+    # absence of the very userns capability the enforced pass just proved
+    # present, so exactly one of the two could ever pass.
+    export ML_SANDBOX2_REQUIRE=enforced
+
     BASE_IMAGE="docker.elastic.co/ml-dev/ml-linux-aarch64-native-build:17"
 
 . ./dev-tools/docker/prefetch_docker_image.sh
     prefetch_docker_image "$BASE_IMAGE"
 
-    echo "--- Running tests (Docker)"
+    echo "--- Running tests (Docker, ML_SANDBOX2_REQUIRE=${ML_SANDBOX2_REQUIRE})"
     docker run --rm \
         -v "$(pwd)/${BUILD_DIR}:/ml-cpp/${BUILD_DIR}" \
         -v "$(pwd)/build:/ml-cpp/build" \
@@ -64,6 +73,7 @@ if [[ "$HARDWARE_ARCH" = aarch64 && -z "${CPP_CROSS_COMPILE:-}" && "$(uname)" =
         -v "$(pwd)/set_env.sh:/ml-cpp/set_env.sh:ro" \
         -v "$(pwd)/gradle.properties:/ml-cpp/gradle.properties:ro" \
         -e BOOST_TEST_OUTPUT_FORMAT_FLAGS="${BOOST_TEST_OUTPUT_FORMAT_FLAGS:-}" \
+        -e ML_SANDBOX2_REQUIRE="${ML_SANDBOX2_REQUIRE}" \
         ${TEST_TIMEOUT:+-e TEST_TIMEOUT="${TEST_TIMEOUT}"} \
         -w /ml-cpp \
         $BASE_IMAGE bash -c '
@@ -87,6 +97,15 @@ if [[ "$HARDWARE_ARCH" = aarch64 && -z "${CPP_CROSS_COMPILE:-}" && "$(uname)" =
 
 else
     # --- Linux x86_64 / macOS: run tests directly ---
+    # x86_64 Buildkite k8s pods get EPERM on mount("proc", ...) - there is no
+    # userns-capable x86_64 CI runner today, so this is an accepted gap in
+    # enforced-mode coverage on that architecture. Only fail_closed runs
+    # here; do not add an enforced pass to this branch. This
+    # also covers aarch64 cross-compile builds, which fall through to this
+    # same branch via the "-z ${CPP_CROSS_COMPILE:-}" condition above, so
+    # they get fail_closed coverage too rather than being skipped entirely.
+    export ML_SANDBOX2_REQUIRE=fail_closed
+
     . ./set_env.sh
 
     find ${BUILD_DIR}/test -name "ml_test_*" -type f -exec chmod +x {} \;
@@ -101,7 +120,7 @@ else
         export DYLD_LIBRARY_PATH="${LIB_DIRS}${DYLD_LIBRARY_PATH:+:$DYLD_LIBRARY_PATH}"
     fi
 
-    echo "--- Running tests"
+    echo "--- Running tests (ML_SANDBOX2_REQUIRE=${ML_SANDBOX2_REQUIRE})"
     cmake \
         -DSOURCE_DIR="$(pwd)" \
         -DBUILD_DIR="$(pwd)/${BUILD_DIR}" \

3rd_party/3rd_party.cmake

@@ -170,6 +170,21 @@ function(install_libs _target _source_dir _prefix _postfix)
   message(STATUS "_target=${_target} _source_dir=${_source_dir} _prefix=${_prefix} _postfix=${_postfix} LIBRARIES=${LIBRARIES}")
 
 
+  # Each requested library must be found in its own right. A coarse
+  # "did the source directory contain anything matching *${_prefix}*${_postfix}?"
+  # guard is not enough: an unrelated library that happens to share the prefix
+  # and suffix (e.g. libmkl_scalapack_lp64.so.2 when every library actually
+  # requested has moved to .so.3) satisfies it, and every individual library is
+  # then skipped silently. That ships a distribution whose binaries cannot
+  # resolve their NEEDED libraries, and the only symptom is the dynamic loader
+  # killing the process with exit code 127 before it can log anything.
+  foreach(LIBRARY ${LIBRARIES})
+    file(GLOB _CHECK_LIBS ${_source_dir}/*${_prefix}${LIBRARY}*${_postfix})
+    if(NOT _CHECK_LIBS)
+      message(FATAL_ERROR "${_target}: no library matching '${_prefix}${LIBRARY}*${_postfix}' found in ${_source_dir}")
+    endif()
+  endforeach()
+
   file(GLOB _LIBS ${_source_dir}/*${_prefix}*${_postfix})
 
   if(_LIBS)
@@ -224,22 +239,47 @@ install_libs("zlib" ${ZLIB_LOCATION} "" "${ZLIB_EXTENSION}" "zlib")
 install_libs("Torch libraries" ${TORCH_LOCATION} "" "${TORCH_EXTENSION}" "${TORCH_LIBRARIES}")
 install_libs("Intel MKL libraries" ${MKL_LOCATION} "${MKL_PREFIX}" "${MKL_EXTENSION}" "${MKL_LIBRARIES}")
 
-# On Linux, replace the RPATH for 3rd party libraries that already have one.
+# On Linux, set the RPATH of every bundled 3rd party library to $ORIGIN.
 # (Only Linux targets will have a location for the gcc runtime library.)
+#
+# These libraries are all installed flat into the same directory, so $ORIGIN lets
+# each one find its siblings at runtime. This must be done unconditionally rather
+# than only for libraries that already declare an RPATH: some prebuilt libraries
+# (notably Boost, depending on how it was built) ship with no RPATH at all, and
+# because DT_RUNPATH is not inherited transitively, a library with a sibling
+# dependency (e.g. libboost_log -> libboost_atomic) fails to load at runtime even
+# though the dependency sits right beside it. The native controller has its
+# environment cleared by Elasticsearch's Spawner, so RPATH is the sole resolution
+# mechanism - there is no LD_LIBRARY_PATH fallback.
+#
+# The one exception is Intel MKL, which must be left exactly as Intel ships it.
+# Rewriting its RPATH makes pytorch_inference die with SIGSEGV during the first
+# inference of real models (ELSER, E5) on hosts with glibc 2.34 (Amazon Linux
+# 2023, the ES integration test agents), while tiny test models and newer glibc
+# versions are unaffected. The libraries also do not need it: libmkl_core,
+# libmkl_intel_lp64 and libmkl_gnu_thread are NEEDED by libtorch_cpu, which already
+# has an $ORIGIN RPATH, and the CPU-specific kernels MKL dlopen()s later only NEED
+# libmkl_core, which is resolved by SONAME because it is already loaded.
 if (GCC_RT_LOCATION)
   execute_process(COMMAND find . -type f COMMAND egrep -v "^core|-debug$|libMl" COMMAND xargs COMMAND sed -e "s/ /;/g" OUTPUT_VARIABLE FOUND_LIBRARIES WORKING_DIRECTORY "${INSTALL_DIR}" OUTPUT_STRIP_TRAILING_WHITESPACE)
   foreach(LIBRARY ${FOUND_LIBRARIES})
-    execute_process(COMMAND patchelf --print-rpath ${LIBRARY} COMMAND grep lib OUTPUT_VARIABLE RPATH_VAR ERROR_VARIABLE RPATH_ERR WORKING_DIRECTORY "${INSTALL_DIR}" OUTPUT_STRIP_TRAILING_WHITESPACE)
-    if(RPATH_VAR)
-      message(STATUS "Attempting to overwrite existing RPATH ${RPATH_VAR} in ${LIBRARY}")
-      execute_process(COMMAND patchelf --force-rpath --set-rpath "$ORIGIN" ${LIBRARY} OUTPUT_VARIABLE SET_RPATH_OUT ERROR_VARIABLE SET_RPATH_ERR WORKING_DIRECTORY "${INSTALL_DIR}" OUTPUT_STRIP_TRAILING_WHITESPACE)
+    get_filename_component(LIBRARY_NAME ${LIBRARY} NAME)
+    if(LIBRARY_NAME MATCHES "^libmkl_")
+      message(STATUS "Leaving RPATH of Intel MKL library ${LIBRARY} unchanged")
+      continue()
+    endif()
+    # Only ELF objects can carry an RPATH. patchelf --print-rpath exits non-zero
+    # on anything else, so use it to skip non-ELF files without failing the build.
+    execute_process(COMMAND patchelf --print-rpath ${LIBRARY} RESULT_VARIABLE IS_ELF_RESULT OUTPUT_QUIET ERROR_QUIET WORKING_DIRECTORY "${INSTALL_DIR}")
+    if(IS_ELF_RESULT EQUAL 0)
+      execute_process(COMMAND patchelf --force-rpath --set-rpath "$ORIGIN" ${LIBRARY} ERROR_VARIABLE SET_RPATH_ERR WORKING_DIRECTORY "${INSTALL_DIR}" OUTPUT_STRIP_TRAILING_WHITESPACE)
       if(SET_RPATH_ERR)
         message(FATAL_ERROR "Error setting RPATH in ${LIBRARY}: ${SET_RPATH_ERR}")
       else()
-        message(STATUS "Set RPATH in ${LIBRARY}")
+        message(STATUS "Set RPATH to $ORIGIN in ${LIBRARY}")
       endif()
     else()
-      message(STATUS "Did not set RPATH in ${LIBRARY}")
+      message(STATUS "Skipping non-ELF file ${LIBRARY}")
     endif()
   endforeach()
 endif()

3rd_party/controller-protocol.version

@@ -0,0 +1 @@
+controller-protocol-version=2

bin/controller/CCommandProcessor.cc

@@ -15,11 +15,26 @@
 #include <core/CStringUtils.h>
 
 #include <algorithm>
+#include <cstdlib>
 #include <istream>
+#include <string>
 
 namespace {
 const std::string TAB(1, '\t');
 const std::string EMPTY_STRING;
+//! Operator kill-switch: forces the legacy route for the configured
+//! sandboxed process path. Mutually exclusive with REQUIRE_SANDBOX_TOKEN -
+//! a start command naming both is ambiguous about its own route and is
+//! rejected outright, never resolved by precedence.
+const std::string DISABLE_SANDBOX_TOKEN{"--disableSandbox"};
+
+//! Operator opt-in: forces the Sandbox2 route (E_Sandbox2, no automatic
+//! legacy fallback) for the configured sandboxed process path. Symmetric
+//! counterpart to DISABLE_SANDBOX_TOKEN - together these are the only two
+//! controller-control tokens the command wire format defines; any other
+//! unrecognised "--" prefixed token is passed through to the spawned
+//! process unchanged.
+const std::string REQUIRE_SANDBOX_TOKEN{"--requireSandbox"};
 }
 
 namespace ml {
@@ -30,8 +45,9 @@ const std::string CCommandProcessor::START{"start"};
 const std::string CCommandProcessor::KILL{"kill"};
 
 CCommandProcessor::CCommandProcessor(const TStrVec& permittedProcessPaths,
+                                     const TStrVec& sandboxedProcessPaths,
                                      std::ostream& responseStream)
-    : m_Spawner{permittedProcessPaths}, m_ResponseWriter{responseStream} {
+    : m_Spawner{permittedProcessPaths, sandboxedProcessPaths}, m_ResponseWriter{responseStream} {
 }
 
 void CCommandProcessor::processCommands(std::istream& commandStream) {
@@ -92,7 +108,131 @@ bool CCommandProcessor::handleStart(std::uint32_t id, TStrVec tokens) {
     std::string processPath{std::move(tokens[0])};
     tokens.erase(tokens.begin());
 
-    if (m_Spawner.spawn(processPath, tokens) == false) {
+    // Scan for both routing tokens before any spawn decision is made.
+    // Never "last one wins"/"first one wins" on duplicates of either token -
+    // count them all and reject outright if either appears more than once.
+    std::size_t disableSandboxCount{0};
+    TStrVec::iterator firstDisableSandbox{tokens.end()};
+    std::size_t requireSandboxCount{0};
+    TStrVec::iterator firstRequireSandbox{tokens.end()};
+    for (auto iter = tokens.begin(); iter != tokens.end(); ++iter) {
+        if (*iter == DISABLE_SANDBOX_TOKEN) {
+            if (disableSandboxCount == 0) {
+                firstDisableSandbox = iter;
+            }
+            ++disableSandboxCount;
+        } else if (*iter == REQUIRE_SANDBOX_TOKEN) {
+            if (requireSandboxCount == 0) {
+                firstRequireSandbox = iter;
+            }
+            ++requireSandboxCount;
+        }
+    }
+
+    if (disableSandboxCount >= 2) {
+        std::string error{"Rejecting command: '" + DISABLE_SANDBOX_TOKEN + "' specified " +
+                          core::CStringUtils::typeToString(disableSandboxCount) +
+                          " times for process '" + processPath + '\''};
+        LOG_ERROR(<< error << " in command with ID " << id);
+        m_ResponseWriter.writeResponse(id, false, error);
+        return false;
+    }
+
+    if (requireSandboxCount >= 2) {
+        std::string error{"Rejecting command: '" + REQUIRE_SANDBOX_TOKEN + "' specified " +
+                          core::CStringUtils::typeToString(requireSandboxCount) +
+                          " times for process '" + processPath + '\''};
+        LOG_ERROR(<< error << " in command with ID " << id);
+        m_ResponseWriter.writeResponse(id, false, error);
+        return false;
+    }
+
+    if (disableSandboxCount == 1 && requireSandboxCount == 1) {
+        std::string error{"Rejecting command: '" + DISABLE_SANDBOX_TOKEN +
+                          "' and '" + REQUIRE_SANDBOX_TOKEN +
+                          "' are mutually exclusive, both specified for process '" +
+                          processPath + '\''};
+        LOG_ERROR(<< error << " in command with ID " << id);
+        m_ResponseWriter.writeResponse(id, false, error);
+        return false;
+    }
+
+    // One shared predicate with the router (which uses the same call to gate
+    // dispatch and sandbox2_launch-signal emission), never a second std::find over a
+    // second copy of the list.
+    const bool isConfiguredSandboxedPath{m_Spawner.isSandboxedProcessPath(processPath)};
+
+    CProcessSpawnerRouter::ERoute route{CProcessSpawnerRouter::ERoute::E_Sandbox2};
+    // Provenance of a legacy route, recorded at the one place it is known so
+    // the router's sandbox2_launch signal can report it as "legacy_reason". Stays
+    // E_NotLegacy for every E_Sandbox2 route, where the field is omitted.
+    CProcessSpawnerRouter::ELegacyReason legacyReason{
+        CProcessSpawnerRouter::ELegacyReason::E_NotLegacy};
+    if (requireSandboxCount == 1) {
+        if (isConfiguredSandboxedPath == false) {
+            std::string error{"Rejecting command: '" + REQUIRE_SANDBOX_TOKEN +
+                              "' is only valid for the configured sandboxed process, "
+                              "not '" +
+                              processPath + '\''};
+            LOG_ERROR(<< error << " in command with ID " << id);
+            m_ResponseWriter.writeResponse(id, false, error);
+            return false;
+        }
+
+        // Operator opt-in validated against this exact processPath: strip
+        // it before it reaches the spawner. Route is already E_Sandbox2
+        // (the default above), so nothing else changes here beyond
+        // stripping and logging the decision at the one place its
+        // provenance is known.
+        LOG_INFO(<< "Routing '" << processPath << "' to Sandbox2: operator opt-in "
+                 << REQUIRE_SANDBOX_TOKEN << " in command with ID " << id);
+        tokens.erase(firstRequireSandbox);
+    } else if (disableSandboxCount == 1) {
+        if (isConfiguredSandboxedPath == false) {
+            std::string error{"Rejecting command: '" + DISABLE_SANDBOX_TOKEN +
+                              "' is only valid for the configured sandboxed process, "
+                              "not '" +
+                              processPath + '\''};
+            LOG_ERROR(<< error << " in command with ID " << id);
+            m_ResponseWriter.writeResponse(id, false, error);
+            return false;
+        }
+
+        // Operator kill-switch validated against this exact processPath:
+        // strip it before it reaches the spawner and route to legacy. This
+        // is the one place the route's operator provenance is known, so it
+        // is logged here rather than in the router, which only ever sees an
+        // already-decided route.
+        LOG_INFO(<< "Routing '" << processPath << "' to the legacy path: operator kill switch "
+                 << DISABLE_SANDBOX_TOKEN << " in command with ID " << id);
+        route = CProcessSpawnerRouter::ERoute::E_Legacy;
+        legacyReason = CProcessSpawnerRouter::ELegacyReason::E_KillSwitch;
+        tokens.erase(firstDisableSandbox);
+    } else {
+        // No token at all: the route is only a decision at all for a
+        // configured sandboxed process path (every other permitted process
+        // dispatches to the legacy spawner either way, and must not be
+        // described as an explicitly-selected legacy route in the log).
+        //
+        // Permanent behaviour, not a rollout seam: a caller that sends
+        // neither token always takes the legacy route - byte-for-byte the
+        // pre-typed-routing behaviour on every platform, including builds
+        // with no Sandbox2 support at all. Elasticsearch is expected to
+        // always send exactly one of the two tokens on every start command
+        // for a sandboxed-eligible process, so this branch exists for
+        // non-ES callers (support/debug scripts, direct controller
+        // invocation) and the test harness.
+        if (isConfiguredSandboxedPath) {
+            route = CProcessSpawnerRouter::ERoute::E_Legacy;
+            legacyReason = CProcessSpawnerRouter::ELegacyReason::E_NoTokenDefault;
+            LOG_DEBUG(<< "Routing '" << processPath << "' to the legacy path: neither "
+                      << DISABLE_SANDBOX_TOKEN << " nor "
+                      << REQUIRE_SANDBOX_TOKEN << " token was present");
+        }
+    }
+
+    core::CProcess::TPid childPid{0};
+    if (m_Spawner.spawn(route, processPath, tokens, childPid, legacyReason) == false) {
         std::string error{"Failed to start process '" + processPath + '\''};
         LOG_ERROR(<< error << " in command with ID " << id);
         m_ResponseWriter.writeResponse(id, false, error);

bin/controller/CCommandProcessor.h

@@ -11,8 +11,7 @@
 #ifndef INCLUDED_ml_controller_CCommandProcessor_h
 #define INCLUDED_ml_controller_CCommandProcessor_h
 
-#include <core/CDetachedProcessSpawner.h>
-
+#include "CProcessSpawnerRouter.h"
 #include "CResponseJsonWriter.h"
 
 #include <cstdint>
@@ -63,7 +62,16 @@ class CCommandProcessor {
     static const std::string KILL;
 
 public:
-    CCommandProcessor(const TStrVec& permittedProcessPaths, std::ostream& responseStream);
+    //! \param permittedProcessPaths Processes that may be started/killed.
+    //! \param sandboxedProcessPaths Subset of \p permittedProcessPaths for
+    //!        which the operator kill-switch token (\c --disableSandbox) is
+    //!        meaningful. Pass an explicit (possibly empty) list - there is
+    //!        no default that reuses \p permittedProcessPaths, because doing
+    //!        so would silently make every permitted process
+    //!        sandboxed-eligible.
+    CCommandProcessor(const TStrVec& permittedProcessPaths,
+                      const TStrVec& sandboxedProcessPaths,
+                      std::ostream& responseStream);
 
     //! Action commands read from the supplied \p commandStream until
     //! end-of-file is reached.
@@ -85,8 +93,12 @@ class CCommandProcessor {
     bool handleKill(std::uint32_t id, TStrVec tokens);
 
 private:
-    //! Used to spawn/kill the requested processes.
-    core::CDetachedProcessSpawner m_Spawner;
+    //! Used to spawn/kill the requested processes, and the single owner of
+    //! the "is this a configured sandboxed process path" predicate this
+    //! class queries via CProcessSpawnerRouter::isSandboxedProcessPath()
+    //! rather than keeping its own second copy of the list and the
+    //! std::find over it.
+    CProcessSpawnerRouter m_Spawner;
 
     //! Used to write responses in JSON format to the response stream.
     CResponseJsonWriter m_ResponseWriter;