Add Executor ID to Executor logs and pass it all Function Executors

This allows us to understand easier what's going on in each Executor
machine.

indexify/pyproject.toml

@@ -24,7 +24,7 @@ grpcio = "1.70.0"
 pydantic = "2.10.4"
 httpx-sse = "^0.4.0"
 # Adds function-executor binary and utils lib.
-tensorlake = ">=0.1.14"
+tensorlake = ">=0.1.16"
 
 # CLI only
 rich = "^13.9.2"

indexify/src/indexify/cli/cli.py

@@ -223,14 +223,14 @@ def executor(
                 "At least one function must be specified when not running in development mode"
             )
 
-    logger = structlog.get_logger(module=__name__)
-    id = nanoid.generate()
     executor_version = version("indexify")
+    id = nanoid.generate()
+    logger = structlog.get_logger(module=__name__, executor_id=id)
+
     logger.info(
         "starting executor",
         server_addr=server_addr,
         config_path=config_path,
-        executor_id=id,
         executor_version=executor_version,
         executor_cache=executor_cache,
         ports=ports,

indexify/src/indexify/executor/executor.py

@@ -41,6 +41,7 @@ def __init__(
         self._base_url = f"{protocol}://{self._server_addr}"
         self._code_path = code_path
         self._task_runner = TaskRunner(
+            executor_id=id,
             function_executor_server_factory=function_executor_server_factory,
             base_url=self._base_url,
             config_path=config_path,

indexify/src/indexify/executor/function_executor/server/function_executor_server_factory.py

@@ -14,7 +14,8 @@ class FunctionExecutorServerConfiguration:
     configuration parameters or raise an exception if it can't implement
     them."""
 
-    def __init__(self, image_uri: Optional[str]):
+    def __init__(self, executor_id: str, image_uri: Optional[str]):
+        self.executor_id: str = executor_id
         # Container image URI of the Function Executor Server.
         self.image_uri: Optional[str] = image_uri
 

indexify/src/indexify/executor/function_executor/server/subprocess_function_executor_server_factory.py

@@ -28,6 +28,8 @@ async def create(
         try:
             port = self._allocate_port()
             args = [
+                "--executor-id",
+                config.executor_id,
                 "--address",
                 _server_address(port),
             ]

indexify/src/indexify/executor/function_executor/single_task_runner.py

@@ -25,13 +25,15 @@
 class SingleTaskRunner:
     def __init__(
         self,
+        executor_id: str,
         function_executor_state: FunctionExecutorState,
         task_input: TaskInput,
         function_executor_server_factory: FunctionExecutorServerFactory,
         base_url: str,
         config_path: Optional[str],
         logger: Any,
     ):
+        self._executor_id: str = executor_id
         self._state: FunctionExecutorState = function_executor_state
         self._task_input: TaskInput = task_input
         self._factory: FunctionExecutorServerFactory = function_executor_server_factory
@@ -76,6 +78,7 @@ async def _create_function_executor(self) -> FunctionExecutor:
         )
         config: FunctionExecutorServerConfiguration = (
             FunctionExecutorServerConfiguration(
+                executor_id=self._executor_id,
                 image_uri=self._task_input.task.image_uri,
             )
         )

indexify/src/indexify/executor/task_runner.py

@@ -18,11 +18,13 @@ class TaskRunner:
 
     def __init__(
         self,
+        executor_id: str,
         function_executor_server_factory: FunctionExecutorServerFactory,
         base_url: str,
         config_path: Optional[str],
         disable_automatic_function_executor_management: bool,
     ):
+        self._executor_id: str = executor_id
         self._factory: FunctionExecutorServerFactory = function_executor_server_factory
         self._base_url: str = base_url
         self._config_path: Optional[str] = config_path
@@ -88,6 +90,7 @@ async def _run_task(
         self, state: FunctionExecutorState, task_input: TaskInput, logger: Any
     ) -> TaskOutput:
         runner: SingleTaskRunner = SingleTaskRunner(
+            executor_id=self._executor_id,
             function_executor_state=state,
             task_input=task_input,
             function_executor_server_factory=self._factory,