From b6f2914b168566930e76d8ef44639125be813b58 Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Mon, 5 Oct 2026 14:39:53 +0200 Subject: [PATCH 1/4] Generate streaming logs rpc code --- .../workflows/workflows/v1/worker_connect.py | 66 +++++++++++++++++++ .../workflows/workflows/v1/worker_pb2.py | 23 +++---- .../workflows/workflows/v1/worker_pb2.pyi | 1 + .../workflows/workflows/v1/worker_pb2_grpc.py | 58 +++++++++++++++- 4 files changed, 136 insertions(+), 12 deletions(-) diff --git a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_connect.py b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_connect.py index a350d39..82406c8 100644 --- a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_connect.py +++ b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_connect.py @@ -15,6 +15,7 @@ from connectrpc.request import Headers, RequestContext from connectrpc.server import ConnectASGIApplication, ConnectWSGIApplication, Endpoint, EndpointSync import google.protobuf.empty_pb2 as google_dot_protobuf_dot_empty__pb2 +import opentelemetry.proto.logs.v1.logs_pb2 as opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2 import tilebox.workflows.workflows.v1.core_pb2 as workflows_dot_v1_dot_core__pb2 import tilebox.workflows.workflows.v1.worker_pb2 as workflows_dot_v1_dot_worker__pb2 @@ -29,6 +30,9 @@ async def initialize_worker(self, request: workflows_dot_v1_dot_worker__pb2.Init async def execute_task(self, request: workflows_dot_v1_dot_core__pb2.Task, ctx: RequestContext) -> workflows_dot_v1_dot_worker__pb2.ExecuteTaskResponse: raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") + def watch_logs(self, request: google_dot_protobuf_dot_empty__pb2.Empty, ctx: RequestContext) -> AsyncIterator[opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord]: + raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") + async def shutdown_worker(self, request: google_dot_protobuf_dot_empty__pb2.Empty, ctx: RequestContext) -> google_dot_protobuf_dot_empty__pb2.Empty: raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") @@ -68,6 +72,16 @@ def __init__(self, service: WorkerService | AsyncGenerator[WorkerService], *, in ), function=svc.execute_task, ), + "/workflows.v1.WorkerService/WatchLogs": Endpoint.server_stream( + method=MethodInfo( + name="WatchLogs", + service_name="workflows.v1.WorkerService", + input=google_dot_protobuf_dot_empty__pb2.Empty, + output=opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord, + idempotency_level=IdempotencyLevel.UNKNOWN, + ), + function=svc.watch_logs, + ), "/workflows.v1.WorkerService/ShutdownWorker": Endpoint.unary( method=MethodInfo( name="ShutdownWorker", @@ -152,6 +166,26 @@ async def execute_task( timeout_ms=timeout_ms, ) + def watch_logs( + self, + request: google_dot_protobuf_dot_empty__pb2.Empty, + *, + headers: Headers | Mapping[str, str] | None = None, + timeout_ms: int | None = None, + ) -> AsyncIterator[opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord]: + return self.execute_server_stream( + request=request, + method=MethodInfo( + name="WatchLogs", + service_name="workflows.v1.WorkerService", + input=google_dot_protobuf_dot_empty__pb2.Empty, + output=opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord, + idempotency_level=IdempotencyLevel.UNKNOWN, + ), + headers=headers, + timeout_ms=timeout_ms, + ) + async def shutdown_worker( self, request: google_dot_protobuf_dot_empty__pb2.Empty, @@ -183,6 +217,8 @@ def initialize_worker(self, request: workflows_dot_v1_dot_worker__pb2.Initialize raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") def execute_task(self, request: workflows_dot_v1_dot_core__pb2.Task, ctx: RequestContext) -> workflows_dot_v1_dot_worker__pb2.ExecuteTaskResponse: raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") + def watch_logs(self, request: google_dot_protobuf_dot_empty__pb2.Empty, ctx: RequestContext) -> Iterator[opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord]: + raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") def shutdown_worker(self, request: google_dot_protobuf_dot_empty__pb2.Empty, ctx: RequestContext) -> google_dot_protobuf_dot_empty__pb2.Empty: raise ConnectError(Code.UNIMPLEMENTED, "Not implemented") @@ -221,6 +257,16 @@ def __init__(self, service: WorkerServiceSync, interceptors: Iterable[Intercepto ), function=service.execute_task, ), + "/workflows.v1.WorkerService/WatchLogs": EndpointSync.server_stream( + method=MethodInfo( + name="WatchLogs", + service_name="workflows.v1.WorkerService", + input=google_dot_protobuf_dot_empty__pb2.Empty, + output=opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord, + idempotency_level=IdempotencyLevel.UNKNOWN, + ), + function=service.watch_logs, + ), "/workflows.v1.WorkerService/ShutdownWorker": EndpointSync.unary( method=MethodInfo( name="ShutdownWorker", @@ -305,6 +351,26 @@ def execute_task( timeout_ms=timeout_ms, ) + def watch_logs( + self, + request: google_dot_protobuf_dot_empty__pb2.Empty, + *, + headers: Headers | Mapping[str, str] | None = None, + timeout_ms: int | None = None, + ) -> Iterator[opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord]: + return self.execute_server_stream( + request=request, + method=MethodInfo( + name="WatchLogs", + service_name="workflows.v1.WorkerService", + input=google_dot_protobuf_dot_empty__pb2.Empty, + output=opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord, + idempotency_level=IdempotencyLevel.UNKNOWN, + ), + headers=headers, + timeout_ms=timeout_ms, + ) + def shutdown_worker( self, request: google_dot_protobuf_dot_empty__pb2.Empty, diff --git a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.py b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.py index 7ee8c11..7cfb16c 100644 --- a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.py +++ b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.py @@ -23,13 +23,14 @@ from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 +from opentelemetry.proto.logs.v1 import logs_pb2 as opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2 from tilebox.datasets.tilebox.v1 import id_pb2 as tilebox_dot_v1_dot_id__pb2 from tilebox.workflows.workflows.v1 import core_pb2 as workflows_dot_v1_dot_core__pb2 from tilebox.workflows.workflows.v1 import task_pb2 as workflows_dot_v1_dot_task__pb2 from tilebox.workflows.workflows.v1 import workflows_pb2 as workflows_dot_v1_dot_workflows__pb2 -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x19workflows/v1/worker.proto\x12\x0cworkflows.v1\x1a\x1bgoogle/protobuf/empty.proto\x1a\x13tilebox/v1/id.proto\x1a\x17workflows/v1/core.proto\x1a\x17workflows/v1/task.proto\x1a\x1cworkflows/v1/workflows.proto\"\x99\x02\n\x17InitializeRunnerRequest\x12+\n\trunner_id\x18\x01 \x01(\x0b\x32\x0e.tilebox.v1.IDR\x08runnerId\x12!\n\x0ctrace_parent\x18\x02 \x01(\tR\x0btraceParent\x12/\n\x07\x63luster\x18\x03 \x01(\x0b\x32\x15.workflows.v1.ClusterR\x07\x63luster\x12\x32\n\x08workflow\x18\x04 \x01(\x0b\x32\x16.workflows.v1.WorkflowR\x08workflow\x12I\n\x0e\x61pi_connection\x18\x05 \x01(\x0b\x32\".workflows.v1.TileboxAPIConnectionR\rapiConnection\">\n\x14TileboxAPIConnection\x12\x10\n\x03url\x18\x01 \x01(\tR\x03url\x12\x14\n\x05token\x18\x02 \x01(\tR\x05token\"\x1a\n\x18InitializeRunnerResponse\"\x98\x01\n\x13\x45xecuteTaskResponse\x12?\n\rcomputed_task\x18\x01 \x01(\x0b\x32\x1a.workflows.v1.ComputedTaskR\x0c\x63omputedTask\x12@\n\x0b\x66\x61iled_task\x18\x02 \x01(\x0b\x32\x1f.workflows.v1.TaskFailedRequestR\nfailedTask2\xc8\x02\n\rWorkerService\x12L\n\x13ListRegisteredTasks\x12\x16.google.protobuf.Empty\x1a\x1d.workflows.v1.TaskIdentifiers\x12\x61\n\x10InitializeWorker\x12%.workflows.v1.InitializeRunnerRequest\x1a&.workflows.v1.InitializeRunnerResponse\x12\x44\n\x0b\x45xecuteTask\x12\x12.workflows.v1.Task\x1a!.workflows.v1.ExecuteTaskResponse\x12@\n\x0eShutdownWorker\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.EmptyBp\n\x10\x63om.workflows.v1B\x0bWorkerProtoP\x01\xa2\x02\x03WXX\xaa\x02\x0cWorkflows.V1\xca\x02\x0cWorkflows\\V1\xe2\x02\x18Workflows\\V1\\GPBMetadata\xea\x02\rWorkflows::V1b\x08\x65\x64itionsp\xe8\x07') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x19workflows/v1/worker.proto\x12\x0cworkflows.v1\x1a\x1bgoogle/protobuf/empty.proto\x1a&opentelemetry/proto/logs/v1/logs.proto\x1a\x13tilebox/v1/id.proto\x1a\x17workflows/v1/core.proto\x1a\x17workflows/v1/task.proto\x1a\x1cworkflows/v1/workflows.proto\"\x99\x02\n\x17InitializeRunnerRequest\x12+\n\trunner_id\x18\x01 \x01(\x0b\x32\x0e.tilebox.v1.IDR\x08runnerId\x12!\n\x0ctrace_parent\x18\x02 \x01(\tR\x0btraceParent\x12/\n\x07\x63luster\x18\x03 \x01(\x0b\x32\x15.workflows.v1.ClusterR\x07\x63luster\x12\x32\n\x08workflow\x18\x04 \x01(\x0b\x32\x16.workflows.v1.WorkflowR\x08workflow\x12I\n\x0e\x61pi_connection\x18\x05 \x01(\x0b\x32\".workflows.v1.TileboxAPIConnectionR\rapiConnection\">\n\x14TileboxAPIConnection\x12\x10\n\x03url\x18\x01 \x01(\tR\x03url\x12\x14\n\x05token\x18\x02 \x01(\tR\x05token\"\x1a\n\x18InitializeRunnerResponse\"\x98\x01\n\x13\x45xecuteTaskResponse\x12?\n\rcomputed_task\x18\x01 \x01(\x0b\x32\x1a.workflows.v1.ComputedTaskR\x0c\x63omputedTask\x12@\n\x0b\x66\x61iled_task\x18\x02 \x01(\x0b\x32\x1f.workflows.v1.TaskFailedRequestR\nfailedTask2\x97\x03\n\rWorkerService\x12L\n\x13ListRegisteredTasks\x12\x16.google.protobuf.Empty\x1a\x1d.workflows.v1.TaskIdentifiers\x12\x61\n\x10InitializeWorker\x12%.workflows.v1.InitializeRunnerRequest\x1a&.workflows.v1.InitializeRunnerResponse\x12\x44\n\x0b\x45xecuteTask\x12\x12.workflows.v1.Task\x1a!.workflows.v1.ExecuteTaskResponse\x12M\n\tWatchLogs\x12\x16.google.protobuf.Empty\x1a&.opentelemetry.proto.logs.v1.LogRecord0\x01\x12@\n\x0eShutdownWorker\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.EmptyBp\n\x10\x63om.workflows.v1B\x0bWorkerProtoP\x01\xa2\x02\x03WXX\xaa\x02\x0cWorkflows.V1\xca\x02\x0cWorkflows\\V1\xe2\x02\x18Workflows\\V1\\GPBMetadata\xea\x02\rWorkflows::V1b\x08\x65\x64itionsp\xe8\x07') _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) @@ -37,14 +38,14 @@ if not _descriptor._USE_C_DESCRIPTORS: _globals['DESCRIPTOR']._loaded_options = None _globals['DESCRIPTOR']._serialized_options = b'\n\020com.workflows.v1B\013WorkerProtoP\001\242\002\003WXX\252\002\014Workflows.V1\312\002\014Workflows\\V1\342\002\030Workflows\\V1\\GPBMetadata\352\002\rWorkflows::V1' - _globals['_INITIALIZERUNNERREQUEST']._serialized_start=174 - _globals['_INITIALIZERUNNERREQUEST']._serialized_end=455 - _globals['_TILEBOXAPICONNECTION']._serialized_start=457 - _globals['_TILEBOXAPICONNECTION']._serialized_end=519 - _globals['_INITIALIZERUNNERRESPONSE']._serialized_start=521 - _globals['_INITIALIZERUNNERRESPONSE']._serialized_end=547 - _globals['_EXECUTETASKRESPONSE']._serialized_start=550 - _globals['_EXECUTETASKRESPONSE']._serialized_end=702 - _globals['_WORKERSERVICE']._serialized_start=705 - _globals['_WORKERSERVICE']._serialized_end=1033 + _globals['_INITIALIZERUNNERREQUEST']._serialized_start=214 + _globals['_INITIALIZERUNNERREQUEST']._serialized_end=495 + _globals['_TILEBOXAPICONNECTION']._serialized_start=497 + _globals['_TILEBOXAPICONNECTION']._serialized_end=559 + _globals['_INITIALIZERUNNERRESPONSE']._serialized_start=561 + _globals['_INITIALIZERUNNERRESPONSE']._serialized_end=587 + _globals['_EXECUTETASKRESPONSE']._serialized_start=590 + _globals['_EXECUTETASKRESPONSE']._serialized_end=742 + _globals['_WORKERSERVICE']._serialized_start=745 + _globals['_WORKERSERVICE']._serialized_end=1152 # @@protoc_insertion_point(module_scope) diff --git a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.pyi b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.pyi index 2206c16..b952175 100644 --- a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.pyi +++ b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2.pyi @@ -1,4 +1,5 @@ from google.protobuf import empty_pb2 as _empty_pb2 +from opentelemetry.proto.logs.v1 import logs_pb2 as _logs_pb2 from tilebox.datasets.tilebox.v1 import id_pb2 as _id_pb2 from tilebox.workflows.workflows.v1 import core_pb2 as _core_pb2 from tilebox.workflows.workflows.v1 import task_pb2 as _task_pb2 diff --git a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2_grpc.py b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2_grpc.py index 87ce98f..c34721b 100644 --- a/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2_grpc.py +++ b/tilebox-workflows/tilebox/workflows/workflows/v1/worker_pb2_grpc.py @@ -3,6 +3,7 @@ import grpc from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 +from opentelemetry.proto.logs.v1 import logs_pb2 as opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2 from tilebox.workflows.workflows.v1 import core_pb2 as workflows_dot_v1_dot_core__pb2 from tilebox.workflows.workflows.v1 import worker_pb2 as workflows_dot_v1_dot_worker__pb2 @@ -32,6 +33,11 @@ def __init__(self, channel): request_serializer=workflows_dot_v1_dot_core__pb2.Task.SerializeToString, response_deserializer=workflows_dot_v1_dot_worker__pb2.ExecuteTaskResponse.FromString, _registered_method=True) + self.WatchLogs = channel.unary_stream( + '/workflows.v1.WorkerService/WatchLogs', + request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + response_deserializer=opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord.FromString, + _registered_method=True) self.ShutdownWorker = channel.unary_unary( '/workflows.v1.WorkerService/ShutdownWorker', request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, @@ -68,9 +74,27 @@ def ExecuteTask(self, request, context): context.set_details('Method not implemented!') raise NotImplementedError('Method not implemented!') + def WatchLogs(self, request, context): + """WatchLogs streams structured runtime logs to the local runner, independently of API log exports and raw + stdout/stderr. It is available before workflow import completes and before InitializeWorker is called. + The runner sends one request; the worker sends buffered startup records followed by live records in enqueue + order. The stream stays open until shutdown, cancellation, or a connection failure, even while no tasks run. + Only one subscriber is allowed per runtime; concurrent subscriptions fail with ALREADY_EXISTS. + Delivery is best-effort with bounded buffering: slow or disconnected subscribers must not block tasks. + Buffer overflow is reported as a warning record when delivery resumes. There is no acknowledgement or replay + of delivered records. Canceling this stream does not shut down the worker or disable its API log exports. + Messages use the log body, severity, trace/span IDs, and attributes from the OpenTelemetry log model; + exception details use exception.type, exception.message, and exception.stacktrace attributes. + buf:lint:ignore RPC_NO_SERVER_STREAMING + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') + def ShutdownWorker(self, request, context): """Gracefully shuts down the worker runtime. After receiving this request, the worker runtime will - cleanly shut down. + finish task cleanup, flush API logs, and drain buffered WatchLogs records within a bounded deadline. + The log stream ends before the worker stops its RPC server; an open subscription must not prevent shutdown. """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) context.set_details('Method not implemented!') @@ -94,6 +118,11 @@ def add_WorkerServiceServicer_to_server(servicer, server): request_deserializer=workflows_dot_v1_dot_core__pb2.Task.FromString, response_serializer=workflows_dot_v1_dot_worker__pb2.ExecuteTaskResponse.SerializeToString, ), + 'WatchLogs': grpc.unary_stream_rpc_method_handler( + servicer.WatchLogs, + request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, + response_serializer=opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord.SerializeToString, + ), 'ShutdownWorker': grpc.unary_unary_rpc_method_handler( servicer.ShutdownWorker, request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, @@ -192,6 +221,33 @@ def ExecuteTask(request, metadata, _registered_method=True) + @staticmethod + def WatchLogs(request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None): + return grpc.experimental.unary_stream( + request, + target, + '/workflows.v1.WorkerService/WatchLogs', + google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, + opentelemetry_dot_proto_dot_logs_dot_v1_dot_logs__pb2.LogRecord.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + @staticmethod def ShutdownWorker(request, target, From ce6a95e56ac1fa415914d1319abb84a007d09eaf Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Mon, 5 Oct 2026 16:48:17 +0200 Subject: [PATCH 2/4] Windows runner support --- .github/workflows/main.yml | 28 ++ CHANGELOG.md | 15 +- .../tests/observability/test_logging.py | 154 ++++----- .../tests/runner/test_log_stream.py | 164 +++++++++ .../tests/runner/test_runtime_logging.py | 325 +++++++----------- .../tests/runner/test_worker_server.py | 194 +++++++++++ .../tilebox/workflows/__init__.py | 9 + .../workflows/observability/_log_pipe.py | 124 ------- .../workflows/observability/_log_stream.py | 172 +++++++++ .../workflows/observability/logging.py | 57 +-- .../tilebox/workflows/runner/__main__.py | 15 +- .../tilebox/workflows/runner/worker_server.py | 115 +++++-- .../workflows/runner/worker_service.py | 80 ++++- 13 files changed, 986 insertions(+), 466 deletions(-) create mode 100644 tilebox-workflows/tests/runner/test_log_stream.py create mode 100644 tilebox-workflows/tests/runner/test_worker_server.py delete mode 100644 tilebox-workflows/tilebox/workflows/observability/_log_pipe.py create mode 100644 tilebox-workflows/tilebox/workflows/observability/_log_stream.py diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index eb7ea68..e52219a 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -60,6 +60,34 @@ jobs: paths: "test-report.xml" if: always() + windows-runtime: + name: Test Windows worker runtime + strategy: + matrix: + python-version: ["3.11", "3.14"] + runs-on: windows-latest + steps: + - uses: actions/checkout@v7 + with: + lfs: true + - name: Set up uv + uses: astral-sh/setup-uv@v10.1.0 + with: + python-version: ${{ matrix.python-version }} + enable-cache: true + cache-python: true + cache-dependency-glob: "uv.lock" + - name: Sync + run: uv sync --package tilebox-workflows --frozen + - name: Test worker lifecycle and logging + run: >- + uv run --frozen --package tilebox-workflows pytest + tilebox-workflows/tests/runner/test_worker_server.py + tilebox-workflows/tests/runner/test_worker_concurrency.py + tilebox-workflows/tests/runner/test_runtime_logging.py + tilebox-workflows/tests/runner/test_log_stream.py + tilebox-workflows/tests/observability/test_logging.py + minimum-dependencies: name: Test minimum dependencies runs-on: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 08d6b6e..4d692e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,9 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.64.0] - 2026-10-05 + +### Added + +- `tilebox-workflows`: Enable live workflow logs in the Windows CLI and improve local worker startup on Windows and Unix. + +### Changed + +- `tilebox-workflows`: Remove the legacy CLI log pipe. Older Unix CLIs can still execute workflows, but require an upgrade to receive structured live logs. + ### Fixed - `tilebox-workflows`: Fix logging initialization and workflow startup with OpenTelemetry 1.45 and newer. +- `tilebox-workflows`: Bound the final API log flush so stalled log exports do not delay CLI worker shutdown. +- `tilebox-workflows`: Preserve large integer attributes in live CLI logs as strings instead of dropping the log record. ## [0.63.2] - 2026-10-04 @@ -588,7 +600,8 @@ the first client that does not cache data (since it's already on the local file - Released under the [MIT](https://opensource.org/license/mit) license. - Released packages: `tilebox-datasets`, `tilebox-workflows`, `tilebox-storage`, `tilebox-grpc` -[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.63.2...HEAD +[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.64.0...HEAD +[0.64.0]: https://github.com/tilebox/tilebox-python/compare/v0.63.2...v0.64.0 [0.63.2]: https://github.com/tilebox/tilebox-python/compare/v0.63.1...v0.63.2 [0.63.1]: https://github.com/tilebox/tilebox-python/compare/v0.63.0...v0.63.1 [0.63.0]: https://github.com/tilebox/tilebox-python/compare/v0.62.0...v0.63.0 diff --git a/tilebox-workflows/tests/observability/test_logging.py b/tilebox-workflows/tests/observability/test_logging.py index 7fd254c..f46d4c3 100644 --- a/tilebox-workflows/tests/observability/test_logging.py +++ b/tilebox-workflows/tests/observability/test_logging.py @@ -21,6 +21,7 @@ from tilebox.workflows.observability import _logging as structured_logging from tilebox.workflows.observability import logging as observability from tilebox.workflows.observability import tracing +from tilebox.workflows.observability._log_stream import _LogQueue, _OTLPQueueHandler from tilebox.workflows.observability._logging import StructuredLogger, internal_logger, logger, root_logger, task_logger @@ -32,8 +33,7 @@ def isolated_logging(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureF monkeypatch.setattr(observability, "_api_handler", None) monkeypatch.setattr(observability, "_console_handlers", []) monkeypatch.setattr(observability, "_console_configured", False) - monkeypatch.setattr(observability, "_writer", None) - monkeypatch.delenv("TILEBOX_LOG_FD", raising=False) + monkeypatch.setattr(observability, "managed_log_queue", None) caplog.set_level(logging.INFO, logger=task_logger.name) caplog.set_level(logging.ERROR, logger=internal_logger.name) @@ -190,20 +190,20 @@ def test_export_attributes_capture_context_once_without_mutating_record(structur @pytest.mark.parametrize("stdlib", [False, True]) -@pytest.mark.parametrize("pipe_first", [False, True]) -def test_codecs_run_once_per_record_across_handlers(stdlib: bool, pipe_first: bool) -> None: +@pytest.mark.parametrize("stream_first", [False, True]) +def test_codecs_run_once_per_record_across_handlers(stdlib: bool, stream_first: bool) -> None: codec = registry.find(Affine) assert codec is not None encode = MagicMock(wraps=codec.encode) target = logging.Logger("normalized-records", logging.INFO) # noqa: LOG001 -- isolated from process-wide loggers - writer = MagicMock(spec=observability._PipeWriter) + queue = _LogQueue() exporters = [InMemoryLogRecordExporter(), InMemoryLogRecordExporter()] providers = [observability.LoggerProvider() for _ in exporters] - handlers: list[logging.Handler] = [observability._StructuredHandler(writer), tracing.SpanEventLoggingHandler()] + handlers: list[logging.Handler] = [_OTLPQueueHandler(queue), tracing.SpanEventLoggingHandler()] for provider, exporter in zip(providers, exporters, strict=True): provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) handlers.append(observability.OTELLoggingHandler(logger_provider=provider)) - for handler in handlers if pipe_first else reversed(handlers): + for handler in handlers if stream_first else reversed(handlers): target.addHandler(handler) tracer_provider = TracerProvider() value = {"transform": Affine(2, 3, 5, 7, 11, 13)} @@ -226,9 +226,14 @@ def test_codecs_run_once_per_record_across_handlers(stdlib: bool, pipe_first: bo assert event.attributes["input"] == '{"transform":[2.0,3.0,5.0,7.0,11.0,13.0]}' # Records hold snapshots, not references to task-owned mutable containers. value.clear() - for call in writer.submit.call_args_list: - assert call.args[0]["attributes"]["input"] == {"transform": [2, 3, 5, 7, 11, 13]} - assert writer.submit.call_count == 2 + queue.seal() + stream_records = list(queue.subscribe()) + assert [record.body.string_value for record in stream_records] == ["record-0", "record-1"] + for record in stream_records: + attributes = {item.key: item.value for item in record.attributes} + transform = attributes["input"].kvlist_value.values[0] + assert transform.key == "transform" + assert [item.double_value for item in transform.value.array_value.values] == [2, 3, 5, 7, 11, 13] for exporter in exporters: records = [item.log_record for item in exporter.get_finished_logs()] assert [record.body for record in records] == ["record-0", "record-1"] @@ -243,43 +248,34 @@ def test_codecs_run_once_per_record_across_handlers(stdlib: bool, pipe_first: bo @pytest.mark.usefixtures("isolated_logging") @pytest.mark.parametrize("external_first", [False, True]) -def test_external_exports_and_console_do_not_replace_api_or_pipe( - monkeypatch: pytest.MonkeyPatch, external_first: bool -) -> None: - read_fd, write_fd = os.pipe() - monkeypatch.setenv("TILEBOX_LOG_FD", str(write_fd)) +def test_external_exports_and_console_do_not_replace_api_or_stream(external_first: bool) -> None: + queue = _LogQueue() + root_logger.addHandler(_OTLPQueueHandler(queue)) api, external = InMemoryLogRecordExporter(), InMemoryLogRecordExporter() first, second = StringIO(), StringIO() - try: - with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(external)): - if external_first: - observability.configure_otel_logging(endpoint="https://external.example") - observability.configure_console_logging(stream=first) - with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(api)): - observability.initialize_logging("https://api.tilebox.com", "test-key") - with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(external)): - if not external_first: - observability.configure_otel_logging(endpoint="https://external.example") - observability.configure_console_logging(stream=first) - observability.configure_console_logging(stream=second, reconfigure=False) - observability.configure_log_level(logging.DEBUG) - assert internal_logger.level == logging.ERROR - task_logger.info("all outputs") - observability.configure_console_logging(enabled=False) - observability.initialize_logging("https://ignored.example", "ignored-key") - task_logger.debug("exports only") - assert first.getvalue().count("all outputs") == second.getvalue().count("all outputs") == 1 - assert "exports only" not in first.getvalue() + second.getvalue() - for exporter in (api, external): - assert [item.log_record.body for item in exporter.get_finished_logs()] == ["all outputs", "exports only"] - assert observability._writer is not None - observability._writer.close() - records = [msgspec.json.decode(line) for line in os.read(read_fd, 65536).splitlines()] - assert [record["message"] for record in records] == ["all outputs", "exports only"] - finally: - if observability._writer is not None: - observability._writer.close() - os.close(read_fd) + with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(external)): + if external_first: + observability.configure_otel_logging(endpoint="https://external.example") + observability.configure_console_logging(stream=first) + with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(api)): + observability.initialize_logging("https://api.tilebox.com", "test-key") + with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(external)): + if not external_first: + observability.configure_otel_logging(endpoint="https://external.example") + observability.configure_console_logging(stream=first) + observability.configure_console_logging(stream=second, reconfigure=False) + observability.configure_log_level(logging.DEBUG) + assert internal_logger.level == logging.ERROR + task_logger.info("all outputs") + observability.configure_console_logging(enabled=False) + observability.initialize_logging("https://ignored.example", "ignored-key") + task_logger.debug("exports only") + assert first.getvalue().count("all outputs") == second.getvalue().count("all outputs") == 1 + assert "exports only" not in first.getvalue() + second.getvalue() + for exporter in (api, external): + assert [item.log_record.body for item in exporter.get_finished_logs()] == ["all outputs", "exports only"] + queue.seal() + assert [record.body.string_value for record in queue.subscribe()] == ["all outputs", "exports only"] @pytest.mark.usefixtures("isolated_logging") @@ -320,40 +316,42 @@ def __str__(self) -> str: cyclic: list[Any] = [] cyclic.append(cyclic) - read_fd, write_fd = os.pipe() - monkeypatch.setenv("TILEBOX_LOG_FD", str(write_fd)) + queue = _LogQueue() + monkeypatch.setattr(observability, "managed_log_queue", queue) + root_logger.addHandler(_OTLPQueueHandler(queue)) exporter = InMemoryLogRecordExporter() - try: - with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(exporter)): - observability.initialize_logging("https://api.tilebox.com", "test-key") - StructuredLogger(task_logger).info( - "input", - input=Input(Affine(2, 3, 5, 7, 11, 13), Path("image.tif")), - counts={date(2026, 9, 16): 3}, - cyclic=cyclic, - unsupported=Broken(), - ) - assert observability._writer is not None - observability._writer.close() - record = msgspec.json.decode(os.read(read_fd, 65536)) - assert record["attributes"] == { - "input": {"transform": [2, 3, 5, 7, 11, 13], "path": "image.tif"}, - "counts": {"2026-09-16": 3}, - "cyclic": "[[...]]", - "unsupported": "", - } - attributes = exporter.get_finished_logs()[0].log_record.attributes - assert attributes is not None - assert isinstance(attributes["input"], str) - assert isinstance(attributes["counts"], str) - assert msgspec.json.decode(attributes["input"]) == record["attributes"]["input"] - assert msgspec.json.decode(attributes["counts"]) == record["attributes"]["counts"] - assert attributes["cyclic"] == "[[...]]" - assert attributes["unsupported"] == "" - finally: - if observability._writer is not None: - observability._writer.close() - os.close(read_fd) + with patch.object(observability, "_otel_log_exporter", return_value=SimpleLogRecordProcessor(exporter)): + observability.initialize_logging("https://api.tilebox.com", "test-key") + StructuredLogger(task_logger).info( + "input", + input=Input(Affine(2, 3, 5, 7, 11, 13), Path("image.tif")), + counts={date(2026, 9, 16): 3}, + cyclic=cyclic, + unsupported=Broken(), + ) + queue.seal() + [record] = queue.subscribe() + stream_attributes = {item.key: item.value for item in record.attributes} + input_attributes = {item.key: item.value for item in stream_attributes["input"].kvlist_value.values} + assert [item.double_value for item in input_attributes["transform"].array_value.values] == [2, 3, 5, 7, 11, 13] + assert input_attributes["path"].string_value == "image.tif" + counts = stream_attributes["counts"].kvlist_value.values + assert len(counts) == 1 + assert counts[0].key == "2026-09-16" + assert counts[0].value.int_value == 3 + assert stream_attributes["cyclic"].string_value == "[[...]]" + assert stream_attributes["unsupported"].string_value == "" + attributes = exporter.get_finished_logs()[0].log_record.attributes + assert attributes is not None + assert isinstance(attributes["input"], str) + assert isinstance(attributes["counts"], str) + assert msgspec.json.decode(attributes["input"]) == { + "transform": [2, 3, 5, 7, 11, 13], + "path": "image.tif", + } + assert msgspec.json.decode(attributes["counts"]) == {"2026-09-16": 3} + assert attributes["cyclic"] == "[[...]]" + assert attributes["unsupported"] == "" @pytest.mark.parametrize("without_otel", [False, True]) diff --git a/tilebox-workflows/tests/runner/test_log_stream.py b/tilebox-workflows/tests/runner/test_log_stream.py new file mode 100644 index 0000000..7ea531a --- /dev/null +++ b/tilebox-workflows/tests/runner/test_log_stream.py @@ -0,0 +1,164 @@ +import logging +import sys +import threading +import time + +import pytest +from opentelemetry.proto.logs.v1.logs_pb2 import LogRecord, SeverityNumber + +from tilebox.workflows.observability._log_stream import _LogQueue, _OTLPQueueHandler + + +def _record(message: str) -> LogRecord: + record = LogRecord() + record.body.string_value = message + return record + + +def test_full_sealed_queue_drains_without_sentinel() -> None: + records = _LogQueue(capacity=2) + records.submit(_record("one")) + records.submit(_record("two")) + records.seal() + + assert [record.body.string_value for record in records.subscribe()] == ["one", "two"] + assert list(records.subscribe()) == [] + + +def test_overflow_notice_is_emitted_without_another_submission() -> None: + records = _LogQueue(capacity=1) + records.submit(_record("kept")) + records.submit(_record("lost-one")) + records.submit(_record("lost-two")) + records.seal() + + assert [record.body.string_value for record in records.subscribe()] == [ + "kept", + "Dropped 2 local log records", + ] + + +def test_cancelled_subscription_reconnects_and_does_not_consume_full_queue() -> None: + records = _LogQueue(capacity=1) + records.submit(_record("buffered")) + assert list(records.subscribe(lambda: False)) == [] + records.seal() + assert [record.body.string_value for record in records.subscribe()] == ["buffered"] + + +def test_overflow_notice_precedes_new_records_after_delivery_resumes() -> None: + records = _LogQueue(capacity=2) + records.submit(_record("first")) + records.submit(_record("second")) + records.submit(_record("lost")) + subscriber = records.subscribe() + assert next(subscriber).body.string_value == "first" + assert next(subscriber).body.string_value == "second" + records.submit(_record("after loss")) + records.seal() + assert [record.body.string_value for record in subscriber] == [ + "Dropped 1 local log records", + "after loss", + ] + + +def test_concurrent_subscription_is_rejected() -> None: + records = _LogQueue() + subscribed = threading.Event() + release = threading.Event() + + def active() -> bool: + subscribed.set() + return not release.is_set() + + subscriber = threading.Thread(target=lambda: list(records.subscribe(active))) + subscriber.start() + assert subscribed.wait(timeout=1) + with pytest.raises(RuntimeError, match="already connected"): + next(records.subscribe()) + release.set() + subscriber.join(timeout=1) + assert not subscriber.is_alive() + + +def test_wait_empty_is_bounded_without_consumer() -> None: + records = _LogQueue() + records.submit(_record("buffered")) + started = time.monotonic() + assert not records.wait_empty(timeout=0.02) + assert time.monotonic() - started < 0.5 + + +def test_oversize_record_is_replaced_by_drop_notice() -> None: + records = _LogQueue(max_record_size=32) + records.submit(_record("x" * 100)) + records.seal() + assert [record.body.string_value for record in records.subscribe()] == ["Dropped 1 local log records"] + + +@pytest.mark.parametrize( + ("value", "field"), + [ + (True, "bool_value"), + (False, "bool_value"), + (0, "int_value"), + (2**63 - 1, "int_value"), + (-(2**63 - 1), "int_value"), + (2**63, "string_value"), + (-(2**63), "string_value"), + (-(2**63) - 1, "string_value"), + (2**80, "string_value"), + (-(2**80), "string_value"), + ], +) +def test_handler_preserves_integer_boundaries(value: int, field: str) -> None: + records = _LogQueue() + handler = _OTLPQueueHandler(records) + record = logging.LogRecord("test", logging.INFO, __file__, 1, "integer attribute", (), None) + record.tilebox_structured_log_attributes = {"value": value, "nested": {"items": [value]}} + handler.emit(record) + handler.emit(logging.LogRecord("test", logging.INFO, __file__, 1, "next record", (), None)) + records.seal() + + first, second = records.subscribe() + assert first.body.string_value == "integer attribute" + assert second.body.string_value == "next record" + attributes = {item.key: item.value for item in first.attributes} + nested = attributes["nested"].kvlist_value.values[0] + assert nested.key == "items" + for result in (attributes["value"], nested.value.array_value.values[0]): + assert result.WhichOneof("value") == field + assert getattr(result, field) == (str(value) if field == "string_value" else value) + + +def test_handler_preserves_structured_attributes_exception_trace_and_level() -> None: + records = _LogQueue() + handler = _OTLPQueueHandler(records) + + def fail() -> None: + raise ValueError("bad value") + + try: + fail() + except ValueError: + record = logging.LogRecord("test", logging.WARNING, __file__, 1, "failed", (), exc_info=sys.exc_info()) + record.tilebox_structured_log_attributes = { + "items": [1, "two"], + "mapping": {"nested": True}, + "missing": None, + "trace_id": "01" * 16, + "span_id": "02" * 8, + } + handler.emit(record) + records.seal() + [result] = records.subscribe() + attributes = {item.key: item.value for item in result.attributes} + assert result.severity_number == SeverityNumber.SEVERITY_NUMBER_WARN + assert result.severity_text == "WARN" + assert result.trace_id == b"\x01" * 16 + assert result.span_id == b"\x02" * 8 + assert [item.int_value or item.string_value for item in attributes["items"].array_value.values] == [1, "two"] + assert attributes["mapping"].kvlist_value.values[0].value.bool_value + assert attributes["missing"].WhichOneof("value") is None + assert attributes["exception.type"].string_value == "ValueError" + assert "ValueError: bad value" in attributes["exception.stacktrace"].string_value diff --git a/tilebox-workflows/tests/runner/test_runtime_logging.py b/tilebox-workflows/tests/runner/test_runtime_logging.py index e8f0635..69b4257 100644 --- a/tilebox-workflows/tests/runner/test_runtime_logging.py +++ b/tilebox-workflows/tests/runner/test_runtime_logging.py @@ -1,4 +1,3 @@ -import json import logging import os import subprocess @@ -12,8 +11,8 @@ from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter, SimpleLogRecordProcessor from tilebox.workflows import Client, Runner -from tilebox.workflows.observability import _log_pipe as runtime_logging from tilebox.workflows.observability import logging as observability +from tilebox.workflows.observability._log_stream import _LogQueue, _OTLPQueueHandler from tilebox.workflows.observability._logging import internal_logger, logger, root_logger, task_logger from tilebox.workflows.observability.logging import _configure_runtime_logging, configure_console_logging from tilebox.workflows.runner.worker_service import WorkerServiceServicer @@ -26,8 +25,7 @@ def isolated_handlers(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCapture monkeypatch.setattr(observability, "_api_handler", None) monkeypatch.setattr(observability, "_console_handlers", []) monkeypatch.setattr(observability, "_console_configured", False) - monkeypatch.setattr(observability, "_writer", None) - monkeypatch.delenv("TILEBOX_LOG_FD", raising=False) + monkeypatch.setattr(observability, "managed_log_queue", None) caplog.set_level(logging.INFO, logger=task_logger.name) caplog.set_level(logging.ERROR, logger=internal_logger.name) @@ -73,10 +71,65 @@ def test_direct_runner_initializes_on_connect_not_client_creation(capsys: pytest assert [line.split(": ", 2)[2] for line in capsys.readouterr().out.splitlines()] == expected -def test_absent_fd_adds_one_console_handler( - monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] -) -> None: - monkeypatch.delenv("TILEBOX_LOG_FD", raising=False) +@pytest.mark.parametrize("mode", ["runtime", "blocked-runtime", "direct"]) +def test_api_flush_and_process_exit(mode: str) -> None: + environment = {key: value for key, value in os.environ.items() if not key.startswith("TILEBOX_")} + if mode != "direct": + environment["TILEBOX_WORKER_ADDRESS"] = "127.0.0.1:0" + result = subprocess.run( # noqa: S603 -- interpreter, script, and mode are controlled by this test + [ + sys.executable, + "-c", + """ +import sys +import threading +from unittest.mock import patch +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, InMemoryLogRecordExporter +from tilebox.workflows.observability import logging as logs + +mode = sys.argv[1] +started = threading.Event() + +class Exporter(InMemoryLogRecordExporter): + def export(self, batch): + started.set() + if mode == "blocked-runtime": + threading.Event().wait(30) + print("exported", flush=True) + return super().export(batch) + +exporter = Exporter() +processor = BatchLogRecordProcessor(exporter, schedule_delay_millis=60000) +with patch.object(logs, '_otel_log_exporter', return_value=processor): + logs.initialize_logging('http://unused.example', None) +logs.get_logger().warning('queued') +if mode != "direct": + logs.flush_api_logging(timeout_millis=100) + assert started.wait(timeout=1) + if mode == "runtime": + assert [item.log_record.body for item in exporter.get_finished_logs()] == ['queued'] +print("main exited", flush=True) +""", + mode, + ], + env=environment, + capture_output=True, + text=True, + # Also catches a second, unbounded flush from OTEL/logging exit hooks. + timeout=10, + check=True, + ) + assert "main exited" in result.stdout + assert result.stderr == "" + if mode == "blocked-runtime": + assert "exported" not in result.stdout + elif mode == "direct": + assert result.stdout.index("main exited") < result.stdout.index("exported") + else: + assert result.stdout.index("exported") < result.stdout.index("main exited") + + +def test_unmanaged_runtime_adds_one_console_handler(capsys: pytest.CaptureFixture[str]) -> None: handler = logging.NullHandler() root_logger.addHandler(handler) assert _configure_runtime_logging() is None @@ -86,8 +139,7 @@ def test_absent_fd_adds_one_console_handler( assert capsys.readouterr().out.endswith(": INFO: console message\n") -def test_existing_console_is_reused(monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None: - monkeypatch.delenv("TILEBOX_LOG_FD", raising=False) +def test_existing_console_is_reused(capsys: pytest.CaptureFixture[str]) -> None: output = StringIO() configure_console_logging(stream=output) handlers = root_logger.handlers[:] @@ -134,43 +186,37 @@ def test_console_opt_out_persists(before_startup: bool, capsys: pytest.CaptureFi @pytest.mark.parametrize(("debug", "level"), [("false", "debug"), ("true", "error")]) -def test_independent_pipe_levels( - monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, debug: str, level: str -) -> None: - read_fd, write_fd = os.pipe() - monkeypatch.setenv("TILEBOX_LOG_FD", str(write_fd)) +def test_independent_stream_levels(caplog: pytest.LogCaptureFixture, debug: str, level: str) -> None: caplog.set_level(level.upper(), logger=task_logger.name) caplog.set_level(logging.DEBUG if debug == "true" else logging.ERROR, logger=internal_logger.name) - writer = _configure_runtime_logging() - assert writer is not None - assert _configure_runtime_logging() is writer + records = _LogQueue() + root_logger.addHandler(_OTLPQueueHandler(records)) + assert _configure_runtime_logging() is None workflow = task_logger for severity in (logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR): logger.log(severity, f"internal-{severity}", task_id="preserved") workflow.log(severity, f"workflow-{severity}") - writer.close() - records = [json.loads(line) for line in os.read(read_fd, 65536).decode().splitlines()] - os.close(read_fd) + records.seal() + exported = list(records.subscribe()) expected_internal = ( {"internal-10", "internal-20", "internal-30", "internal-40"} if debug == "true" else {"internal-40"} ) expected_workflow = ( {"workflow-10", "workflow-20", "workflow-30", "workflow-40"} if level == "debug" else {"workflow-40"} ) - assert {r["message"] for r in records} == expected_internal | expected_workflow - assert len(records) == len(expected_internal | expected_workflow) - assert all(r["attributes"]["task_id"] == "preserved" for r in records if r["message"].startswith("internal-")) + assert {record.body.string_value for record in exported} == expected_internal | expected_workflow + assert len(exported) == len(expected_internal | expected_workflow) + internal = [record for record in exported if record.body.string_value.startswith("internal-")] + assert all( + {item.key: item.value.string_value for item in record.attributes}["task_id"] == "preserved" + for record in internal + ) -def test_structured_pipe_is_ndjson_across_threads( - monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture -) -> None: - read_fd, write_fd = os.pipe() - monkeypatch.setenv("TILEBOX_LOG_FD", str(write_fd)) +def test_structured_stream_preserves_threaded_records(caplog: pytest.LogCaptureFixture) -> None: caplog.set_level(logging.WARNING, logger=task_logger.name) - writer = _configure_runtime_logging() - assert writer is not None - assert not os.get_inheritable(write_fd) + records = _LogQueue() + root_logger.addHandler(_OTLPQueueHandler(records)) local_logger = task_logger local_logger.info("filtered") @@ -188,74 +234,17 @@ def log_failure() -> None: except ValueError: local_logger.exception("failed", extra={"tilebox_structured_log_attributes": {"task": 1}}) - writer.close() - data = os.read(read_fd, 1_000_000).decode() - os.close(read_fd) - records = [json.loads(line) for line in data.splitlines()] - assert {record["message"] for record in records} == {*(f"message-{index}" for index in range(20)), "failed"} - failure = next(record for record in records if record["message"] == "failed") - assert failure["level"] == "error" - assert "ValueError: first line\nsecond line" in failure["exception"] - assert failure["attributes"] == {"task": 1} - - -def test_idle_writer_shutdown_wakes_blocking_get() -> None: - read_fd, write_fd = os.pipe() - waiting = threading.Event() - original_drain = runtime_logging._PipeWriter._drain - - def observe_get(writer: runtime_logging._PipeWriter) -> None: - original_get = writer._records.get - - def get() -> bytes | None: - waiting.set() - return original_get() - - with patch.object(writer._records, "get", side_effect=get): - original_drain(writer) - - with patch.object(runtime_logging._PipeWriter, "_drain", observe_get): - writer = runtime_logging._PipeWriter(write_fd) - try: - assert waiting.wait(timeout=2) - writer.close() - writer.close() - assert not writer._thread.is_alive() - finally: - writer.close() - os.close(read_fd) - - -@pytest.mark.parametrize("count", [3, 256]) -def test_shutdown_drains_queued_records_even_when_full(count: int) -> None: - read_fd, write_fd = os.pipe() - release = threading.Event() - original_drain = runtime_logging._PipeWriter._drain - - def delayed_drain(writer: runtime_logging._PipeWriter) -> None: - release.wait() - original_drain(writer) - - with ( - patch.object(runtime_logging._PipeWriter, "_drain", delayed_drain), - patch.object(runtime_logging.os, "write", wraps=os.write) as write, - ): - writer = runtime_logging._PipeWriter(write_fd) - try: - for index in range(count): - writer.submit({"i": index}) - # Keep draining paused until close has attempted to enqueue its sentinel. - writer.close() - release.set() - writer.close() - assert not writer._thread.is_alive() - records = [json.loads(line) for line in os.read(read_fd, 65536).splitlines()] - assert records == [{"i": index} for index in range(count)] - assert write.call_count == 1 - finally: - release.set() - writer.close() - os.close(read_fd) + records.seal() + exported = list(records.subscribe()) + assert {record.body.string_value for record in exported} == { + *(f"message-{index}" for index in range(20)), + "failed", + } + failure = next(record for record in exported if record.body.string_value == "failed") + assert failure.severity_text == "ERROR" + attributes = {item.key: item.value for item in failure.attributes} + assert "ValueError: first line\nsecond line" in attributes["exception.stacktrace"].string_value + assert attributes["task"].int_value == 1 @pytest.mark.parametrize("level", ["info", "error"]) @@ -269,119 +258,57 @@ def test_bootstrap_before_import(tmp_path: Path, level: str) -> None: "get_logger().error('import-error')\n" "runner = Runner(tasks=[])\n" ) - read_fd, write_fd = os.pipe() - try: - result = subprocess.run( - [ - sys.executable, - "-c", - """ -from unittest.mock import patch + result = subprocess.run( + [ + sys.executable, + "-c", + """ +from unittest.mock import MagicMock, patch from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter exporter = InMemoryLogRecordExporter() with patch('opentelemetry.exporter.otlp.proto.http._log_exporter.OTLPLogExporter', return_value=exporter) as factory: from tilebox.workflows.runner import __main__ as m from tilebox.workflows.observability import logging as logs + from tilebox.workflows.observability._log_stream import managed_log_queue from tilebox.workflows.observability._logging import task_logger, internal_logger import os assert task_logger.level == (20 if os.environ['TILEBOX_LOG_LEVEL'] == 'info' else 40) assert internal_logger.level == 40 - m.serve_runner = lambda _: None - m.main(['sample_runner:runner']) + server = MagicMock() + with patch.object(m, 'WorkerServer', return_value=server): + m.main(['sample_runner:runner']) + server.start.assert_called_once_with() + server.set_runner.assert_called_once() + assert server.wait.call_count == 2 + server.shutdown.assert_called_once_with() logs.initialize_logging('https://ignored.example', 'ignored-key') - logs._api_handler.flush() + logs._api_handler._logger_provider.force_flush() factory.assert_called_once() assert [item.log_record.body for item in exporter.get_finished_logs()] == ( ['import-info', 'import-error'] if task_logger.level == 20 else ['import-error'] ) + managed_log_queue.seal() + assert [record.body.string_value for record in managed_log_queue.subscribe()] == ( + ['import-info', 'import-error'] if task_logger.level == 20 else ['import-error'] + ) """, - ], - env=os.environ - | { - "TILEBOX_LOG_FD": str(write_fd), - "TILEBOX_RUNTIME_ID": runtime_id, - "TILEBOX_API_URL": "https://startup.example", - "TILEBOX_API_KEY": "startup-key", - "TILEBOX_LOG_LEVEL": level, - "TILEBOX_DEBUG": "false", - "PYTHONPATH": str(tmp_path), - }, - pass_fds=(write_fd,), - capture_output=True, - text=True, - timeout=10, - check=True, - ) - records = [json.loads(line) for line in os.read(read_fd, 65536).decode().splitlines()] - finally: - os.close(write_fd) - os.close(read_fd) + ], + env=os.environ + | { + "TILEBOX_RUNTIME_DIR": str(tmp_path), + "TILEBOX_RUNTIME_TOKEN": "runtime-token", + "TILEBOX_WORKER_ADDRESS": f"unix://{tmp_path / 'worker.sock'}", + "TILEBOX_RUNTIME_ID": runtime_id, + "TILEBOX_API_URL": "https://startup.example", + "TILEBOX_API_KEY": "startup-key", + "TILEBOX_LOG_LEVEL": level, + "TILEBOX_DEBUG": "false", + "PYTHONPATH": str(tmp_path), + }, + capture_output=True, + text=True, + timeout=10, + check=True, + ) assert result.stdout == runtime_id + "\n" assert result.stderr == "" - assert [record["message"] for record in records] == ( - ["import-info", "import-error"] if level == "info" else ["import-error"] - ) - - -def test_full_queue_skips_encoding_and_reports_drops() -> None: - read_fd, write_fd = os.pipe() - release = threading.Event() - original_drain = runtime_logging._PipeWriter._drain - - def delayed_drain(writer: runtime_logging._PipeWriter) -> None: - release.wait() - original_drain(writer) - - with patch.object(runtime_logging._PipeWriter, "_drain", delayed_drain): - writer = runtime_logging._PipeWriter(write_fd) - try: - for index in range(256): - writer.submit({"i": index}) - with patch.object(runtime_logging.msgspec.json, "encode") as encode: - writer.submit({"discarded": 1}) - writer.submit({"discarded": 2}) - encode.assert_not_called() - assert writer._records.qsize() == 256 - assert writer._records.get_nowait() == b'{"i":0}\n' - assert writer._records.get_nowait() == b'{"i":1}\n' - writer.submit({"resumed": True}) - release.set() - writer.close() - assert not writer._thread.is_alive() - records = [json.loads(line) for line in os.read(read_fd, 65536).splitlines()] - assert records == [ - *({"i": index} for index in range(2, 256)), - {"level": "warning", "message": "Dropped 2 local log records"}, - {"resumed": True}, - ] - finally: - release.set() - writer.close() - os.close(read_fd) - - -@pytest.mark.parametrize("disconnected", [False, True]) -def test_failed_pipe_releases_queue_and_skips_future_encoding(*, disconnected: bool) -> None: - read_fd, write_fd = os.pipe() - if disconnected: - os.close(read_fd) - writer = runtime_logging._PipeWriter(write_fd) - try: - # Larger than the pipe buffer: a stalled reader forces a partial write. - writer.submit({"message": "x" * (512 * 1024)}) - for index in range(8): - writer.submit({"i": index}) - writer._thread.join(timeout=2) - assert not writer._thread.is_alive() - assert writer._records.empty() - with patch.object(runtime_logging.msgspec.json, "encode") as encode: - writer.submit({"after_failure": True}) - encode.assert_not_called() - if not disconnected: - data = os.read(read_fd, 1024 * 1024) - assert data.startswith(b'{"message":"xxx') - assert b"\n" not in data # No later records appended to a truncated one. - finally: - writer.close() - if not disconnected: - os.close(read_fd) diff --git a/tilebox-workflows/tests/runner/test_worker_server.py b/tilebox-workflows/tests/runner/test_worker_server.py new file mode 100644 index 0000000..6c90278 --- /dev/null +++ b/tilebox-workflows/tests/runner/test_worker_server.py @@ -0,0 +1,194 @@ +import json +import os +import subprocess +import sys +import tempfile +import threading +import time +from pathlib import Path +from typing import cast +from unittest.mock import MagicMock + +import grpc +import pytest +from google.protobuf.empty_pb2 import Empty + +from tilebox.workflows import Runner +from tilebox.workflows.runner.worker_server import WorkerServer +from tilebox.workflows.runner.worker_service import WorkerServiceServicer +from tilebox.workflows.workflows.v1 import core_pb2, worker_pb2, worker_pb2_grpc + +_TOKEN = "test-runtime-token" # noqa: S105 -- local test credential +_AUTH = (("authorization", f"Bearer {_TOKEN}"),) + + +@pytest.mark.parametrize("address_from_environment", [False, True]) +def test_unmanaged_worker_serves_without_authentication_or_log_stream( + monkeypatch: pytest.MonkeyPatch, address_from_environment: bool +) -> None: + monkeypatch.delenv("TILEBOX_RUNTIME_DIR", raising=False) + monkeypatch.delenv("TILEBOX_RUNTIME_TOKEN", raising=False) + if address_from_environment: + monkeypatch.setenv("TILEBOX_WORKER_ADDRESS", "127.0.0.1:0") + server = WorkerServer(Runner(tasks=[]), None if address_from_environment else "127.0.0.1:0") + server.start() + try: + with grpc.insecure_channel(f"127.0.0.1:{server._port}") as channel: + worker = worker_pb2_grpc.WorkerServiceStub(channel) + assert not worker.ListRegisteredTasks(Empty(), timeout=2).identifiers + with pytest.raises(grpc.RpcError) as error: + next(worker.WatchLogs(Empty(), timeout=2)) + assert cast(grpc.Call, error.value).code() == grpc.StatusCode.UNAVAILABLE + worker.ShutdownWorker(Empty(), timeout=2) + finally: + server.shutdown() + server.wait() + + +def test_shutdown_waits_for_active_task_cleanup(monkeypatch: pytest.MonkeyPatch) -> None: + started, release, stopped = threading.Event(), threading.Event(), threading.Event() + service = WorkerServiceServicer(Runner(tasks=[]), stopped.set) + + def execute(_request: core_pb2.Task) -> worker_pb2.ExecuteTaskResponse: + started.set() + assert release.wait(timeout=2) + assert not stopped.is_set() + return worker_pb2.ExecuteTaskResponse() + + monkeypatch.setattr(service, "_execute_task", execute) + task = threading.Thread(target=service.ExecuteTask, args=(core_pb2.Task(), MagicMock())) + shutdown = threading.Thread(target=service.ShutdownWorker, args=(Empty(), MagicMock())) + task.start() + assert started.wait(timeout=1) + shutdown.start() + try: + assert not stopped.wait(timeout=0.05) + finally: + release.set() + task.join(timeout=2) + shutdown.join(timeout=2) + assert not task.is_alive() + assert not shutdown.is_alive() + assert stopped.is_set() + + +def _wait_for_file(path: Path, process: subprocess.Popen[str]) -> None: + deadline = time.monotonic() + 10 + while not path.exists(): + if process.poll() is not None: + stdout, stderr = process.communicate(timeout=2) + pytest.fail(f"Worker exited before {path.name}: {stdout}\n{stderr}") + assert time.monotonic() < deadline, f"Timed out waiting for {path}" + time.sleep(0.01) + + +def _assert_loading_rpc_contract(worker: worker_pb2_grpc.WorkerServiceStub) -> None: + for method, request in ( + (worker.ListRegisteredTasks, Empty()), + (worker.InitializeWorker, worker_pb2.InitializeRunnerRequest()), + (worker.ExecuteTask, core_pb2.Task()), + (worker.ShutdownWorker, Empty()), + ): + with pytest.raises(grpc.RpcError) as error: + method(request, timeout=2) + assert cast(grpc.Call, error.value).code() == grpc.StatusCode.UNAUTHENTICATED + with pytest.raises(grpc.RpcError) as error: + next(worker.WatchLogs(Empty(), timeout=2)) + assert cast(grpc.Call, error.value).code() == grpc.StatusCode.UNAUTHENTICATED + for method, request in ( + (worker.ListRegisteredTasks, Empty()), + (worker.InitializeWorker, worker_pb2.InitializeRunnerRequest()), + (worker.ExecuteTask, core_pb2.Task()), + ): + with pytest.raises(grpc.RpcError) as error: + method(request, metadata=_AUTH, timeout=2) + assert cast(grpc.Call, error.value).code() == grpc.StatusCode.UNAVAILABLE + + +def _wait_until_loaded(worker: worker_pb2_grpc.WorkerServiceStub) -> None: + deadline = time.monotonic() + 5 + while True: + try: + assert not worker.ListRegisteredTasks(Empty(), metadata=_AUTH, timeout=2).identifiers + except grpc.RpcError as error: + if cast(grpc.Call, error).code() != grpc.StatusCode.UNAVAILABLE: + raise + assert time.monotonic() < deadline + time.sleep(0.01) + else: + return + + +@pytest.mark.parametrize( + "transport", ["tcp", pytest.param("unix", marks=pytest.mark.skipif(os.name == "nt", reason="Unix socket"))] +) +@pytest.mark.parametrize("fail_import", [False, True]) +def test_managed_worker_bootstrap_logs_and_shutdown(tmp_path: Path, transport: str, fail_import: bool) -> None: + # Short Unix paths fit sockaddr_un even when pytest's temporary path is long. + with tempfile.TemporaryDirectory(prefix="tbw-", dir="/tmp" if transport == "unix" else None) as directory: # noqa: S108 -- private random directory, short enough for sockaddr_un + runtime_dir = Path(directory) + address = f"unix://{runtime_dir / 'worker.sock'}" if transport == "unix" else "127.0.0.1:0" + source = """ +import time +from pathlib import Path +from tilebox.workflows import Runner +from tilebox.workflows.observability.logging import get_logger + +get_logger().warning("before workflow loaded") +print("raw stdout", flush=True) +while not Path("load").exists(): + time.sleep(0.01) +""" + source += 'raise RuntimeError("broken workflow")\n' if fail_import else "runner = Runner(tasks=[])\n" + (tmp_path / "fixture.py").write_text(source) + environment = {key: value for key, value in os.environ.items() if not key.startswith("TILEBOX_")} + environment.update( + TILEBOX_WORKER_ADDRESS=address, + TILEBOX_RUNTIME_DIR=directory, + TILEBOX_RUNTIME_TOKEN=_TOKEN, + ) + with subprocess.Popen( + [sys.executable, "-m", "tilebox.workflows.runner", "fixture:runner"], + cwd=tmp_path, + env=environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) as process: + try: + announcement = runtime_dir / "endpoint.json" + _wait_for_file(announcement, process) + endpoint = json.loads(announcement.read_text()) + assert endpoint["version"] == 1 + assert _TOKEN not in announcement.read_text() + assert not (runtime_dir / "endpoint.json.tmp").exists() + if transport == "unix": + assert endpoint["address"] == address + else: + assert endpoint["address"].startswith("127.0.0.1:") + assert int(endpoint["address"].split(":")[1]) > 0 + with grpc.insecure_channel(endpoint["address"]) as channel: + worker = worker_pb2_grpc.WorkerServiceStub(channel) + _assert_loading_rpc_contract(worker) + logs = worker.WatchLogs(Empty(), metadata=_AUTH, timeout=10) + record = next(logs) + assert record.body.string_value == "before workflow loaded" + with pytest.raises(grpc.RpcError) as error: + next(worker.WatchLogs(Empty(), metadata=_AUTH, timeout=2)) + assert cast(grpc.Call, error.value).code() == grpc.StatusCode.ALREADY_EXISTS + (tmp_path / "load").touch() + if not fail_import: + _wait_until_loaded(worker) + worker.ShutdownWorker(Empty(), metadata=_AUTH, timeout=5) + assert list(logs) == [] + stdout, stderr = process.communicate(timeout=10) + assert stdout == "raw stdout\n" + if fail_import: + assert process.returncode != 0 + assert "broken workflow" in stderr + else: + assert process.returncode == 0, stderr + finally: + if process.poll() is None: + process.kill() + process.communicate(timeout=5) diff --git a/tilebox-workflows/tilebox/workflows/__init__.py b/tilebox-workflows/tilebox/workflows/__init__.py index c8d6426..511ed60 100644 --- a/tilebox-workflows/tilebox/workflows/__init__.py +++ b/tilebox-workflows/tilebox/workflows/__init__.py @@ -51,8 +51,17 @@ def __dir__() -> list[str]: def _initialize_logging_from_environment() -> None: # Stage 1: current CLIs provide credentials before importing workflow code, so # even import-time logs reach the API. Ordinary SDK imports stay lightweight. + runtime_dir = os.environ.get("TILEBOX_RUNTIME_DIR") + runtime_token = os.environ.get("TILEBOX_RUNTIME_TOKEN") + if bool(runtime_dir) != bool(runtime_token): + raise RuntimeError("TILEBOX_RUNTIME_DIR and TILEBOX_RUNTIME_TOKEN must be set together") + url = os.environ.get("TILEBOX_API_URL") token = os.environ.get("TILEBOX_API_KEY") + if runtime_dir and runtime_token: + # Importing installs the managed local stream before workflow modules can log. + from tilebox.workflows.observability import logging as _runtime_logging # noqa: F401, PLC0415 + if not url or not token: return diff --git a/tilebox-workflows/tilebox/workflows/observability/_log_pipe.py b/tilebox-workflows/tilebox/workflows/observability/_log_pipe.py deleted file mode 100644 index abbb20d..0000000 --- a/tilebox-workflows/tilebox/workflows/observability/_log_pipe.py +++ /dev/null @@ -1,124 +0,0 @@ -"""Local structured logging for CLI-managed workflow runtimes.""" - -import contextlib -import logging -import os -import queue -import select -import threading -import time -import traceback -from typing import Any - -import msgspec - -from tilebox.workflows.observability._logging import _record_attributes - - -class _PipeWriter: - def __init__(self, fd: int) -> None: - self._fd = fd - self._closing = threading.Event() - self._submit_lock = threading.Lock() - self._dropped = 0 - self._records: queue.Queue[bytes | None] = queue.Queue(maxsize=256) - self._thread = threading.Thread(target=self._run, name="tilebox-log-writer", daemon=True) - os.set_inheritable(fd, False) - os.set_blocking(fd, False) - self._thread.start() - - def submit(self, record: dict[str, Any]) -> None: - with self._submit_lock: - if self._closing.is_set(): - return - if self._records.full(): - self._dropped += 1 - return - data = msgspec.json.encode(record) + b"\n" - if len(data) > 1024 * 1024: - data = b'{"level":"warning","message":"Local log record exceeded 1 MiB; discarded"}\n' - try: - if self._dropped: - notice = {"level": "warning", "message": f"Dropped {self._dropped} local log records"} - self._records.put_nowait(msgspec.json.encode(notice) + b"\n") - self._dropped = 0 - self._records.put_nowait(data) - except queue.Full: - self._dropped += 1 - - def close(self) -> None: - with self._submit_lock: - if not self._closing.is_set(): - self._closing.set() - # Wake an idle writer. A full queue already keeps it awake until - # it drains, when the closing flag ends the loop instead. - with contextlib.suppress(queue.Full): - self._records.put_nowait(None) - self._thread.join(timeout=0.5) - - def _run(self) -> None: - try: - self._drain() - finally: - with self._submit_lock: - self._closing.set() - # Disconnected or stalled readers must not retain queued payloads. - while not self._records.empty(): - self._records.get_nowait() - with contextlib.suppress(OSError): - os.close(self._fd) - - def _drain(self) -> None: - while not self._closing.is_set() or not self._records.empty(): - item = self._records.get() - if item is None: - return - # Coalesce queued records without waiting for more. This avoids two - # syscalls per small message while keeping idle delivery immediate. - batch = [item] - size = len(item) - while size < 64 * 1024: - try: - item = self._records.get_nowait() - except queue.Empty: - break - if item is None: - break - batch.append(item) - size += len(item) - view = memoryview(b"".join(batch)) - deadline = time.monotonic() + 0.25 - while view and time.monotonic() < deadline: - try: - _, writable, _ = select.select([], [self._fd], [], 0.05) - if writable: - view = view[os.write(self._fd, view) :] - except (BrokenPipeError, OSError, ValueError): - return - if view: - # Never append another record after a partial timed-out write. - return - - -class _StructuredHandler(logging.Handler): - def __init__(self, writer: _PipeWriter) -> None: - super().__init__(logging.NOTSET) - self._writer = writer - - def emit(self, record: logging.LogRecord) -> None: - try: - self._emit(record) - except Exception: # noqa: BLE001 -- a local logging failure must not fail a task - self.handleError(record) - - def _emit(self, record: logging.LogRecord) -> None: - output: dict[str, Any] = { - "level": record.levelname.lower(), - "message": record.getMessage(), - } - if record.exc_info: - output["exception"] = "".join(traceback.format_exception(*record.exc_info)).rstrip("\n") - attributes = _record_attributes(record) - if attributes: - output["attributes"] = attributes - self._writer.submit(output) diff --git a/tilebox-workflows/tilebox/workflows/observability/_log_stream.py b/tilebox-workflows/tilebox/workflows/observability/_log_stream.py new file mode 100644 index 0000000..9103539 --- /dev/null +++ b/tilebox-workflows/tilebox/workflows/observability/_log_stream.py @@ -0,0 +1,172 @@ +"""Bounded OTLP log streaming for CLI-managed workflow runtimes.""" + +import logging +import os +import threading +import time +import traceback +from collections import deque +from collections.abc import Callable, Iterator +from typing import Any + +from opentelemetry.instrumentation.log_utils import std_to_otel +from opentelemetry.proto.common.v1.common_pb2 import AnyValue, ArrayValue, KeyValue, KeyValueList +from opentelemetry.proto.logs.v1.logs_pb2 import LogRecord, SeverityNumber +from opentelemetry.semconv.attributes import exception_attributes + +from tilebox.workflows.observability._logging import _record_attributes + +_MAX_RECORD_SIZE = 1024 * 1024 +_MANAGED_RUNTIME = bool(os.environ.get("TILEBOX_RUNTIME_DIR") and os.environ.get("TILEBOX_RUNTIME_TOKEN")) + + +def _any_value(value: Any) -> AnyValue: # noqa: PLR0911 -- each protobuf union arm is clearest explicitly + """Convert an already-normalized JSON value to OTLP.""" + if value is None: + return AnyValue() + if isinstance(value, bool): + return AnyValue(bool_value=value) + if isinstance(value, int): + if abs(value) > 2**63 - 1: + return AnyValue(string_value=str(value)) + return AnyValue(int_value=value) + if isinstance(value, float): + return AnyValue(double_value=value) + if isinstance(value, dict): + return AnyValue( + kvlist_value=KeyValueList(values=[KeyValue(key=key, value=_any_value(item)) for key, item in value.items()]) + ) + if isinstance(value, list): + return AnyValue(array_value=ArrayValue(values=[_any_value(item) for item in value])) + return AnyValue(string_value=value) + + +class _LogQueue: + """Non-blocking bounded queue with explicit sealed-and-empty termination.""" + + def __init__(self, capacity: int = 256, max_record_size: int = _MAX_RECORD_SIZE) -> None: + self._capacity = capacity + self._max_record_size = max_record_size + self._records: deque[LogRecord] = deque() + self._condition = threading.Condition() + self._subscriber = False + self._sealed = False + self._dropped = 0 + + def submit(self, record: LogRecord) -> None: + with self._condition: + if self._sealed: + return + if record.ByteSize() > self._max_record_size or len(self._records) >= self._capacity: + self._dropped += 1 + self._condition.notify_all() + return + if self._dropped: + self._records.append(_drop_record(self._dropped)) + self._dropped = 0 + if len(self._records) >= self._capacity: + self._dropped += 1 + self._condition.notify_all() + return + self._records.append(record) + self._condition.notify_all() + + def subscribe(self, active: Callable[[], bool] = lambda: True) -> Iterator[LogRecord]: + with self._condition: + if self._subscriber: + raise RuntimeError("A log subscriber is already connected") + self._subscriber = True + try: + while active(): + with self._condition: + if self._records: + record = self._records.popleft() + self._condition.notify_all() + elif self._dropped: + record = _drop_record(self._dropped) + self._dropped = 0 + self._condition.notify_all() + elif self._sealed: + return + else: + self._condition.wait(timeout=0.1) + continue + yield record + finally: + with self._condition: + self._subscriber = False + self._condition.notify_all() + + def seal(self) -> None: + with self._condition: + self._sealed = True + self._condition.notify_all() + + def wait_empty(self, timeout: float) -> bool: + deadline = time.monotonic() + timeout + with self._condition: + while self._records or self._dropped: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + self._condition.wait(timeout=remaining) + return True + + +def _drop_record(count: int) -> LogRecord: + return LogRecord( + time_unix_nano=time.time_ns(), + severity_number=_severity_number(logging.WARNING), + severity_text="WARN", + body=AnyValue(string_value=f"Dropped {count} local log records"), + ) + + +class _OTLPQueueHandler(logging.Handler): + def __init__(self, records: _LogQueue) -> None: + super().__init__(logging.NOTSET) + self._records = records + + def emit(self, record: logging.LogRecord) -> None: + try: + attributes = dict(_record_attributes(record)) + trace_id = _hex_id(attributes.pop("trace_id", ""), 16) + span_id = _hex_id(attributes.pop("span_id", ""), 8) + if record.exc_info: + exctype, value, tb = record.exc_info + if exctype is not None: + attributes[exception_attributes.EXCEPTION_TYPE] = exctype.__name__ + if value is not None: + attributes[exception_attributes.EXCEPTION_MESSAGE] = str(value) + if tb is not None: + attributes[exception_attributes.EXCEPTION_STACKTRACE] = "".join( + traceback.format_exception(*record.exc_info) + ) + self._records.submit( + LogRecord( + time_unix_nano=int(record.created * 1e9), + severity_number=_severity_number(record.levelno), + severity_text={"WARNING": "WARN", "CRITICAL": "FATAL"}.get(record.levelname, record.levelname), + body=_any_value(record.getMessage()), + attributes=[KeyValue(key=str(key), value=_any_value(value)) for key, value in attributes.items()], + trace_id=trace_id, + span_id=span_id, + ) + ) + except Exception: # noqa: BLE001 -- logging must never break workflow execution + self.handleError(record) + + +def _hex_id(value: Any, size: int) -> bytes: + try: + result = bytes.fromhex(value) if isinstance(value, str) else b"" + except ValueError: + return b"" + return result if len(result) == size else b"" + + +def _severity_number(level: int) -> SeverityNumber.ValueType: + return SeverityNumber.Value(f"SEVERITY_NUMBER_{std_to_otel(level).name}") + + +managed_log_queue = _LogQueue() if _MANAGED_RUNTIME else None diff --git a/tilebox-workflows/tilebox/workflows/observability/logging.py b/tilebox-workflows/tilebox/workflows/observability/logging.py index a378224..2aa7e2f 100644 --- a/tilebox-workflows/tilebox/workflows/observability/logging.py +++ b/tilebox-workflows/tilebox/workflows/observability/logging.py @@ -1,5 +1,4 @@ # allow the logging module name which shadows the builtin: -import atexit import contextlib import logging import os @@ -38,7 +37,7 @@ from opentelemetry.semconv.attributes import exception_attributes from tilebox.workflows._serialization import normalize_log_value -from tilebox.workflows.observability._log_pipe import _PipeWriter, _StructuredHandler +from tilebox.workflows.observability._log_stream import _OTLPQueueHandler, managed_log_queue from tilebox.workflows.observability._logging import ( StructuredLogger as StructuredLogger, # noqa: PLC0414 -- public compatibility alias ) @@ -138,7 +137,8 @@ def emit(self, record: logging.LogRecord) -> None: def flush(self) -> None: if callable(force_flush := getattr(self._logger_provider, "force_flush", None)): # Match OTEL's handler: flushing under the logging lock can deadlock. - threading.Thread(target=force_flush).start() + # Do not prolong process exit after a runtime's bounded final flush. + threading.Thread(target=force_flush, daemon=True).start() def _get_attributes(self, record: logging.LogRecord) -> dict[str, Any]: cached = getattr(record, "_tilebox_otel_attributes", None) @@ -166,10 +166,12 @@ def _get_attributes(self, record: logging.LogRecord) -> dict[str, Any]: _api_handler: OTELLoggingHandler | None = None _initialization_lock = threading.Lock() -_writer: _PipeWriter | None = None _console_handlers: list[logging.Handler] = [] _console_configured = False +if managed_log_queue is not None: + root_logger.addHandler(_OTLPQueueHandler(managed_log_queue)) + def _remove_console_handlers() -> None: for handler in _console_handlers: @@ -186,27 +188,13 @@ def _add_console_handler(level: int, stream: TextIO, formatter: logging.Formatte _console_handlers.append(handler) -def _configure_runtime_logging() -> _PipeWriter | None: - """Install the required CLI pipe, or an optional default console for direct runners. - - The CLI pipe is runtime-owned: public configuration can add outputs but cannot - disable or replace it. Console opt-out applies only to console handlers. - """ - global _writer # noqa: PLW0603 -- process-owned CLI pipe +def _configure_runtime_logging() -> None: + """Add an optional default console when logs aren't streamed to a managed CLI.""" with _initialization_lock: - fd_value = os.environ.get("TILEBOX_LOG_FD") - if fd_value is not None: - if _writer is None: - _writer = _PipeWriter(int(fd_value)) - root_logger.addHandler(_StructuredHandler(_writer)) - atexit.register(_writer.close) - if not _console_configured: - _remove_console_handlers() - elif not _console_configured and not _console_handlers: + if managed_log_queue is None and not _console_configured and not _console_handlers: _add_console_handler( logging.NOTSET, sys.stdout, logging.Formatter("%(process)d: %(levelname)s: %(message)s") ) - return _writer def configure_log_level(level: int = logging.INFO, *, tilebox_debug: bool = False) -> None: @@ -219,7 +207,7 @@ def configure_log_level(level: int = logging.INFO, *, tilebox_debug: bool = Fals filtered out by the logger. Every call overrides both startup settings: omitting tilebox_debug disables internal - DEBUG logging even if TILEBOX_DEBUG was enabled. Existing API exports, the CLI pipe, + DEBUG logging even if TILEBOX_DEBUG was enabled. Existing API exports, the CLI stream, and console handlers are unchanged and retain their own output thresholds. Args: @@ -243,7 +231,12 @@ def initialize_logging(url: str, token: str | None, service: str | None = None) if _api_handler is not None: return - provider = LoggerProvider(resource=_get_default_resource(service)) + # CLI runtimes own the final, bounded flush. An additional OTEL atexit + # shutdown could wait indefinitely on an exporter after that deadline. + provider = LoggerProvider( + resource=_get_default_resource(service), + shutdown_on_exit=not bool(os.environ.get("TILEBOX_WORKER_ADDRESS")), + ) processor = _otel_log_exporter(endpoint=url, headers={"Authorization": f"Bearer {token}"} if token else None) provider.add_log_record_processor(processor) handler = OTELLoggingHandler(level=logging.NOTSET, logger_provider=provider) @@ -251,6 +244,20 @@ def initialize_logging(url: str, token: str | None, service: str | None = None) _api_handler = handler +def flush_api_logging(timeout_millis: int = 1000) -> None: + """Flush the managed API exporter without waiting indefinitely.""" + handler = _api_handler + if handler is None: + return + force_flush = getattr(handler._logger_provider, "force_flush", None) # noqa: SLF001 + if callable(force_flush): + # BatchLogRecordProcessor currently ignores its timeout argument. + # Bound our wait without making the export thread block process exit. + flush = threading.Thread(target=force_flush, kwargs={"timeout_millis": timeout_millis}, daemon=True) + flush.start() + flush.join(timeout=timeout_millis / 1000) + + def _otel_log_exporter( endpoint: str | None = None, headers: dict[str, str] | None = None, @@ -294,7 +301,7 @@ def configure_otel_logging( This will configure a logging handler that will send log messages to an OTLP compatible endpoint using the open telemetry protocol for exporting logs. The logging handler will be attached to the root tilebox logger. All loggers created using `get_logger()` will therefore inherit this handler configuration. - Each call adds an export; Tilebox's API export, the CLI pipe, and console outputs remain installed. + Each call adds an export; Tilebox's API export, the CLI stream, and console outputs remain installed. Args: service: A string or a resource object to include in all traces. Used to identify the service being traced. @@ -425,7 +432,7 @@ def configure_console_logging( configured console logging handlers will be removed. If False, the existing handlers will be kept. Useful if you want to log to multiple consoles. enabled: If False, remove Tilebox-managed console handlers and disable automatic console output, even if - called before runner startup. API exports, the CLI log pipe, and user-installed handlers are unaffected. + called before runner startup. API exports, the CLI log stream, and user-installed handlers are unaffected. """ global _console_configured # noqa: PLW0603 -- explicit process-wide console policy with _initialization_lock: diff --git a/tilebox-workflows/tilebox/workflows/runner/__main__.py b/tilebox-workflows/tilebox/workflows/runner/__main__.py index 5ab0e82..64725ca 100644 --- a/tilebox-workflows/tilebox/workflows/runner/__main__.py +++ b/tilebox-workflows/tilebox/workflows/runner/__main__.py @@ -6,7 +6,7 @@ from tilebox.workflows.observability._logging import logger from tilebox.workflows.runner.runner import Runner -from tilebox.workflows.runner.worker_server import serve_runner +from tilebox.workflows.runner.worker_server import WorkerServer def main(argv: Sequence[str] | None = None) -> int: @@ -17,10 +17,15 @@ def main(argv: Sequence[str] | None = None) -> int: parser.add_argument("runner", help="Runner object import path, for example 'my_workflow.runner:runner'.") args = parser.parse_args(argv) - logger.debug(f"Starting Tilebox workflow runtime for runner {args.runner!r}") - runner = _import_runner(args.runner) - logger.debug(f"Imported runner {args.runner!r}; starting worker server") - serve_runner(runner) + server = WorkerServer() + server.start() + try: + runner = _import_runner(args.runner) + server.set_runner(runner) + server.wait() + finally: + server.shutdown() + server.wait() return 0 diff --git a/tilebox-workflows/tilebox/workflows/runner/worker_server.py b/tilebox-workflows/tilebox/workflows/runner/worker_server.py index 4cb86b8..5621f26 100644 --- a/tilebox-workflows/tilebox/workflows/runner/worker_server.py +++ b/tilebox-workflows/tilebox/workflows/runner/worker_server.py @@ -1,3 +1,4 @@ +import json import os import threading from concurrent import futures @@ -5,48 +6,108 @@ import grpc +from tilebox.workflows.observability._log_stream import managed_log_queue from tilebox.workflows.observability._logging import logger +from tilebox.workflows.observability.logging import flush_api_logging from tilebox.workflows.runner.runner import Runner from tilebox.workflows.runner.worker_service import WorkerServiceServicer from tilebox.workflows.workflows.v1 import worker_pb2_grpc WORKER_ADDRESS_ENV = "TILEBOX_WORKER_ADDRESS" +RUNTIME_DIR_ENV = "TILEBOX_RUNTIME_DIR" +RUNTIME_TOKEN_ENV = "TILEBOX_RUNTIME_TOKEN" # noqa: S105 -- environment variable name, not a credential -def serve_runner(runner: Runner, address: str | None = None) -> None: - address = address or os.environ.get(WORKER_ADDRESS_ENV) - if not address: - raise RuntimeError( - f"{WORKER_ADDRESS_ENV} is not set. Set it to a local gRPC address, for example " - f"'unix:///tmp/tilebox-worker.sock'." +class WorkerServer: + """Python worker server for a runner started from the Tilebox CLI. + + This server is the interface the CLI uses to communicate with the Python process. + """ + + def __init__(self, runner: Runner | None = None, address: str | None = None) -> None: + # Older CLIs and direct serve_runner() callers use the configured address + # without runtime-token authentication, endpoint discovery, or live log streaming. + self._managed = bool(os.environ.get(RUNTIME_DIR_ENV) and os.environ.get(RUNTIME_TOKEN_ENV)) + configured_address = address or os.environ.get(WORKER_ADDRESS_ENV) + if not configured_address: + raise RuntimeError( + f"{WORKER_ADDRESS_ENV} is not set. Set it to a local gRPC address, for example " + f"'unix:///tmp/tilebox-worker.sock'." + ) + self._address = configured_address + runtime_dir = os.environ.get(RUNTIME_DIR_ENV) + runtime_token = os.environ.get(RUNTIME_TOKEN_ENV) + if bool(runtime_dir) != bool(runtime_token): + raise RuntimeError(f"{RUNTIME_DIR_ENV} and {RUNTIME_TOKEN_ENV} must be set together") + self._bind_address = _normalize_grpc_address(configured_address) + if self._managed and not configured_address.startswith("unix://"): + host, _, port = configured_address.rpartition(":") + if host != "127.0.0.1" or not port.isdecimal() or int(port) > 65535: + raise RuntimeError("Managed worker TCP addresses must use the 127.0.0.1 loopback interface") + _unlink_stale_unix_socket(self._bind_address) + self._server = grpc.server(futures.ThreadPoolExecutor()) + self._service = WorkerServiceServicer( + runner, + self.shutdown, + managed_log_queue if self._managed else None, + os.environ.get(RUNTIME_TOKEN_ENV) if self._managed else None, ) + worker_pb2_grpc.add_WorkerServiceServicer_to_server(self._service, self._server) + self._port = self._server.add_insecure_port(self._bind_address) + if self._port == 0: + raise RuntimeError(f"Failed to bind worker server to {configured_address!r}") + self._shutdown_lock = threading.Lock() + self._shutdown_started = False + + def start(self) -> None: + self._server.start() + try: + if self._managed: + self._announce() + except BaseException: + self._server.stop(0).wait() + raise - bind_address = _normalize_grpc_address(address) - logger.debug(f"Starting worker server for address {bind_address!r}") - _unlink_stale_unix_socket(bind_address) + def set_runner(self, runner: Runner) -> None: + self._service.set_runner(runner) - logger.debug("Creating worker gRPC server") - server = grpc.server(futures.ThreadPoolExecutor()) + def wait(self) -> None: + self._server.wait_for_termination() - def shutdown() -> None: - logger.debug("Worker server shutdown requested") - # server.stop() is blocking, so we run it in a separate thread - # server.stop(5) means we stop accepting new requests immediately, but we give existing requests up to 5 - # seconds to finish before we forcefully terminate them - threading.Thread(target=server.stop, args=(5,), daemon=True).start() + def shutdown(self) -> None: + with self._shutdown_lock: + if self._shutdown_started: + return + self._shutdown_started = True + threading.Thread(target=self._shutdown, name="tilebox-worker-shutdown", daemon=True).start() - logger.debug("Registering worker service") - worker_pb2_grpc.add_WorkerServiceServicer_to_server(WorkerServiceServicer(runner, shutdown), server) - logger.debug(f"Binding worker server to {bind_address!r}") - port = server.add_insecure_port(bind_address) - if port == 0: - raise RuntimeError(f"Failed to bind worker server to {address!r}") + def _shutdown(self) -> None: + try: + flush_api_logging() + finally: + if self._managed and managed_log_queue is not None: + managed_log_queue.seal() + managed_log_queue.wait_empty(timeout=1.0) + self._server.stop(5).wait() - logger.debug("Starting worker gRPC server") + def _announce(self) -> None: + runtime_dir = Path(os.environ[RUNTIME_DIR_ENV]) + address = self._address + if not address.startswith("unix://"): + address = f"127.0.0.1:{self._port}" + temporary = runtime_dir / "endpoint.json.tmp" + temporary.write_text(json.dumps({"version": 1, "address": address}, separators=(",", ":"))) + temporary.replace(runtime_dir / "endpoint.json") + + +def serve_runner(runner: Runner, address: str | None = None) -> None: + server = WorkerServer(runner, address) server.start() - logger.debug("Worker gRPC server started; taking requests and waiting for termination") - server.wait_for_termination() - logger.debug("Worker gRPC server terminated") + try: + server.wait() + finally: + server.shutdown() + server.wait() def _normalize_grpc_address(address: str) -> str: diff --git a/tilebox-workflows/tilebox/workflows/runner/worker_service.py b/tilebox-workflows/tilebox/workflows/runner/worker_service.py index 502b230..9c2bd9d 100644 --- a/tilebox-workflows/tilebox/workflows/runner/worker_service.py +++ b/tilebox-workflows/tilebox/workflows/runner/worker_service.py @@ -1,12 +1,16 @@ -from collections.abc import Callable +import hmac +import threading +from collections.abc import Callable, Iterator import grpc from google.protobuf.empty_pb2 import Empty +from opentelemetry.proto.logs.v1.logs_pb2 import LogRecord from tilebox.datasets.uuid import uuid_message_to_uuid from tilebox.workflows.cache import NoCache from tilebox.workflows.client import Client from tilebox.workflows.data import Cluster, ComputedTask, FailedTask, Task +from tilebox.workflows.observability._log_stream import _LogQueue from tilebox.workflows.observability._logging import logger from tilebox.workflows.observability.logging import initialize_logging from tilebox.workflows.runner.executor import LazyStorageLocations, TaskExecutor @@ -18,16 +22,43 @@ class WorkerServiceServicer(worker_pb2_grpc.WorkerServiceServicer): def __init__( self, - runner: Runner, + runner: Runner | None, shutdown: Callable[[], None], + logs: _LogQueue | None = None, + token: str | None = None, ) -> None: self._runner = runner self._shutdown = shutdown self._executor: TaskExecutor | None = None + self._logs = logs + self._token = token + self._tasks = threading.Condition() + self._active_tasks = 0 + self._stopping = False + + def _authorize(self, context: grpc.ServicerContext) -> None: + if self._token is None: + return + metadata = dict(context.invocation_metadata()) + supplied = metadata.get("authorization", "") + if not hmac.compare_digest(supplied, f"Bearer {self._token}"): + context.abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid runtime authorization") + + def set_runner(self, runner: Runner) -> None: + self._runner = runner + + def _loaded_runner(self, context: grpc.ServicerContext) -> Runner: + runner = self._runner + if runner is None: + context.abort(grpc.StatusCode.UNAVAILABLE, "Worker runtime is still loading") + raise RuntimeError("Worker runtime is still loading") + return runner def ListRegisteredTasks(self, request: Empty, context: grpc.ServicerContext) -> core_pb2.TaskIdentifiers: # noqa: ARG002, N802 logger.debug("ListRegisteredTasks RPC called") - identifiers = [identifier.to_message() for identifier in self._runner.task_identifiers] + self._authorize(context) + runner = self._loaded_runner(context) + identifiers = [identifier.to_message() for identifier in runner.task_identifiers] logger.debug(f"ListRegisteredTasks RPC returning {len(identifiers)} task identifier(s)") return core_pb2.TaskIdentifiers(identifiers=identifiers) @@ -37,6 +68,8 @@ def InitializeWorker( # noqa: N802 context: grpc.ServicerContext, ) -> worker_pb2.InitializeRunnerResponse: logger.debug("InitializeWorker RPC called") + self._authorize(context) + runner = self._loaded_runner(context) runner_id = uuid_message_to_uuid(request.runner_id) if self._executor is not None: context.abort(grpc.StatusCode.FAILED_PRECONDITION, "Worker is already initialized") @@ -52,15 +85,17 @@ def InitializeWorker( # noqa: N802 # Stage 2: older CLIs supply API credentials in InitializeWorker rather than # startup environment variables. This is a no-op if stage 1 already ran. + # As of October 2026, keep this fallback for a few months; remove it once + # deployed CLIs all provide startup credentials. initialize_logging(**client._auth) # noqa: SLF001 - context_type = self._runner.context or RunnerContext + context_type = runner.context or RunnerContext runner_context = context_type(tracer) runner_context.storage_locations = LazyStorageLocations(client, runner_context) self._executor = TaskExecutor( - self._runner, - self._runner.cache or NoCache(), + runner, + runner.cache or NoCache(), tracer, task_logger, runner_context, @@ -74,9 +109,23 @@ def InitializeWorker( # noqa: N802 def ExecuteTask( # noqa: N802 self, request: core_pb2.Task, - context: grpc.ServicerContext, # noqa: ARG002 + context: grpc.ServicerContext, ) -> worker_pb2.ExecuteTaskResponse: logger.debug("ExecuteTask RPC called") + self._authorize(context) + self._loaded_runner(context) + with self._tasks: + if self._stopping: + context.abort(grpc.StatusCode.UNAVAILABLE, "Worker is shutting down") + self._active_tasks += 1 + try: + return self._execute_task(request) + finally: + with self._tasks: + self._active_tasks -= 1 + self._tasks.notify_all() + + def _execute_task(self, request: core_pb2.Task) -> worker_pb2.ExecuteTaskResponse: task = Task.from_message(request) executor = self._executor if executor is None: @@ -98,8 +147,25 @@ def ExecuteTask( # noqa: N802 return worker_pb2.ExecuteTaskResponse(failed_task=result.to_message()) raise TypeError(f"Unexpected task execution result: {type(result)}") + def WatchLogs(self, request: Empty, context: grpc.ServicerContext) -> Iterator[LogRecord]: # noqa: ARG002, N802 + self._authorize(context) + context.send_initial_metadata(()) + logs = self._logs + if logs is None: + context.abort(grpc.StatusCode.UNAVAILABLE, "Local log streaming is unavailable") + raise RuntimeError("Local log streaming is unavailable") + try: + yield from logs.subscribe(context.is_active) + except RuntimeError as error: + context.abort(grpc.StatusCode.ALREADY_EXISTS, str(error)) + def ShutdownWorker(self, request: Empty, context: grpc.ServicerContext) -> Empty: # noqa: ARG002, N802 + self._authorize(context) logger.debug("ShutdownWorker RPC called") + with self._tasks: + self._stopping = True + while self._active_tasks: + self._tasks.wait() self._shutdown() logger.debug("ShutdownWorker RPC returning") return Empty() From 82e7289437ea9e056a0c354441049cbc94873dab Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Mon, 5 Oct 2026 16:51:57 +0200 Subject: [PATCH 3/4] Rename managed flag to _legacy_cli --- .../tilebox/workflows/runner/worker_server.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tilebox-workflows/tilebox/workflows/runner/worker_server.py b/tilebox-workflows/tilebox/workflows/runner/worker_server.py index 5621f26..9d749b0 100644 --- a/tilebox-workflows/tilebox/workflows/runner/worker_server.py +++ b/tilebox-workflows/tilebox/workflows/runner/worker_server.py @@ -27,7 +27,7 @@ class WorkerServer: def __init__(self, runner: Runner | None = None, address: str | None = None) -> None: # Older CLIs and direct serve_runner() callers use the configured address # without runtime-token authentication, endpoint discovery, or live log streaming. - self._managed = bool(os.environ.get(RUNTIME_DIR_ENV) and os.environ.get(RUNTIME_TOKEN_ENV)) + self._legacy_cli = not (os.environ.get(RUNTIME_DIR_ENV) and os.environ.get(RUNTIME_TOKEN_ENV)) configured_address = address or os.environ.get(WORKER_ADDRESS_ENV) if not configured_address: raise RuntimeError( @@ -40,7 +40,7 @@ def __init__(self, runner: Runner | None = None, address: str | None = None) -> if bool(runtime_dir) != bool(runtime_token): raise RuntimeError(f"{RUNTIME_DIR_ENV} and {RUNTIME_TOKEN_ENV} must be set together") self._bind_address = _normalize_grpc_address(configured_address) - if self._managed and not configured_address.startswith("unix://"): + if not self._legacy_cli and not configured_address.startswith("unix://"): host, _, port = configured_address.rpartition(":") if host != "127.0.0.1" or not port.isdecimal() or int(port) > 65535: raise RuntimeError("Managed worker TCP addresses must use the 127.0.0.1 loopback interface") @@ -49,8 +49,8 @@ def __init__(self, runner: Runner | None = None, address: str | None = None) -> self._service = WorkerServiceServicer( runner, self.shutdown, - managed_log_queue if self._managed else None, - os.environ.get(RUNTIME_TOKEN_ENV) if self._managed else None, + None if self._legacy_cli else managed_log_queue, + None if self._legacy_cli else os.environ.get(RUNTIME_TOKEN_ENV), ) worker_pb2_grpc.add_WorkerServiceServicer_to_server(self._service, self._server) self._port = self._server.add_insecure_port(self._bind_address) @@ -62,7 +62,7 @@ def __init__(self, runner: Runner | None = None, address: str | None = None) -> def start(self) -> None: self._server.start() try: - if self._managed: + if not self._legacy_cli: self._announce() except BaseException: self._server.stop(0).wait() @@ -85,7 +85,7 @@ def _shutdown(self) -> None: try: flush_api_logging() finally: - if self._managed and managed_log_queue is not None: + if not self._legacy_cli and managed_log_queue is not None: managed_log_queue.seal() managed_log_queue.wait_empty(timeout=1.0) self._server.stop(5).wait() From 5e7bb9df9cbb426ad9df2af218189b599b972d9a Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Tue, 6 Oct 2026 11:43:28 +0200 Subject: [PATCH 4/4] Update default otel resource name --- .github/workflows/main.yml | 1 + CHANGELOG.md | 3 +- .../tests/observability/test_tracing.py | 88 ++++++++++++++++++- .../workflows/observability/logging.py | 27 ++---- .../workflows/observability/tracing.py | 18 ++-- 5 files changed, 104 insertions(+), 33 deletions(-) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index e52219a..05dd793 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -87,6 +87,7 @@ jobs: tilebox-workflows/tests/runner/test_runtime_logging.py tilebox-workflows/tests/runner/test_log_stream.py tilebox-workflows/tests/observability/test_logging.py + tilebox-workflows/tests/observability/test_tracing.py minimum-dependencies: name: Test minimum dependencies diff --git a/CHANGELOG.md b/CHANGELOG.md index 4d692e7..d05847a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -## [0.64.0] - 2026-10-05 +## [0.64.0] - 2026-10-06 ### Added @@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - `tilebox-workflows`: Fix logging initialization and workflow startup with OpenTelemetry 1.45 and newer. +- `tilebox-workflows`: Always apply Tilebox service metadata to logs and traces, with explicit resource attributes overriding those defaults. Initialize tracing with the same runtime identity as logging instead of relying on OpenTelemetry's defaults. - `tilebox-workflows`: Bound the final API log flush so stalled log exports do not delay CLI worker shutdown. - `tilebox-workflows`: Preserve large integer attributes in live CLI logs as strings instead of dropping the log record. diff --git a/tilebox-workflows/tests/observability/test_tracing.py b/tilebox-workflows/tests/observability/test_tracing.py index cc763f1..8f1014e 100644 --- a/tilebox-workflows/tests/observability/test_tracing.py +++ b/tilebox-workflows/tests/observability/test_tracing.py @@ -1,12 +1,19 @@ +import json +import os +import subprocess +import sys from collections.abc import Iterator +from importlib.metadata import version from unittest.mock import patch import pytest from opentelemetry.context import Context +from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import ReadableSpan, Span, TracerProvider -from opentelemetry.sdk.trace.export import SpanProcessor +from opentelemetry.sdk.trace.export import SimpleSpanProcessor, SpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from tilebox.workflows.observability import logging as observability_logging from tilebox.workflows.observability import tracing @@ -54,11 +61,12 @@ def force_flush(self, timeout_millis: int = 30000) -> bool: # noqa: ARG002 @pytest.fixture(autouse=True) def reset_tilebox_tracing() -> Iterator[None]: - tracing._set_tilebox_tracer_provider(TracerProvider()) + original_provider = tracing._get_tilebox_tracer_provider() tracing._workflow_tracers.clear() + tracing._set_tilebox_tracer_provider(TracerProvider(resource=observability_logging._get_default_resource())) yield - tracing._set_tilebox_tracer_provider(TracerProvider()) tracing._workflow_tracers.clear() + tracing._set_tilebox_tracer_provider(original_provider) @pytest.fixture @@ -74,6 +82,80 @@ def create_processor(*args: object, **kwargs: object) -> RecordingSpanProcessor: return processors +@pytest.mark.parametrize("otel_service_name", ["unknown_service:python", "unknown_service:python.exe", "other-default"]) +def test_exported_traces_use_tilebox_resource_defaults(otel_service_name: str) -> None: + # Exercise actual module initialization, independent of OTEL's platform-specific defaults. + instance_id = "66d615f3-7d53-4a31-bc94-94cbb9d9ffa2" + environment = {key: value for key, value in os.environ.items() if not key.startswith(("TILEBOX_", "OTEL_"))} + environment.update( + TILEBOX_RUNTIME_ID=instance_id, + OTEL_SERVICE_NAME=otel_service_name, + OTEL_RESOURCE_ATTRIBUTES="service.instance.id=otel-generated", + ) + result = subprocess.run( + [ + sys.executable, + "-c", + """ +import json +from unittest.mock import patch +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from tilebox.workflows.observability import tracing +from tilebox.workflows.observability.logging import _get_default_resource + +exporter = InMemorySpanExporter() +with patch.object(tracing, "_otel_span_exporter", return_value=SimpleSpanProcessor(exporter)): + tracer = tracing.WorkflowTracer(service=None, url="https://api.tilebox.com", token=None) + with tracer.span("task"): + pass +[span] = exporter.get_finished_spans() +print(json.dumps([dict(span.resource.attributes), dict(_get_default_resource().attributes)])) +""", + ], + env=environment, + check=True, + capture_output=True, + text=True, + timeout=10, + ) + expected = { + "service.name": "tilebox-python", + "service.namespace": "tilebox.workflows", + "service.version": version("tilebox-workflows"), + "service.instance.id": instance_id, + } + trace_resource, log_resource = json.loads(result.stdout) + assert {key: trace_resource.get(key) for key in expected} == expected + assert {key: log_resource.get(key) for key in expected} == expected + + +@pytest.mark.parametrize("service_name", [None, "custom-worker", "unknown_service:python.exe"]) +@pytest.mark.parametrize("explicit_resource", [False, True]) +def test_exported_traces_preserve_configured_resources(service_name: str | None, explicit_resource: bool) -> None: + attributes = {"service.instance.id": "custom-instance", "deployment": "test"} + if service_name is not None: + attributes["service.name"] = service_name + resource = Resource(attributes, schema_url="https://example.com/schema") + if not explicit_resource: + tracing._set_tilebox_tracer_provider(TracerProvider(resource=resource)) + exporter = InMemorySpanExporter() + with patch.object(tracing, "_otel_span_exporter", return_value=SimpleSpanProcessor(exporter)): + tracer = tracing.WorkflowTracer( + service=resource if explicit_resource else None, url="https://api.tilebox.com", token=None + ) + with tracer.span("task"): + pass + + [span] = exporter.get_finished_spans() + for result in (span.resource, observability_logging._get_default_resource(resource)): + assert {key: result.attributes.get(key) for key in attributes} == attributes + assert result.attributes["service.name"] == (service_name or "tilebox-python") + assert result.attributes["service.namespace"] == "tilebox.workflows" + assert result.attributes["service.version"] == version("tilebox-workflows") + assert result.schema_url == resource.schema_url + + def test_workflow_tracers_do_not_share_client_span_processors( span_processors: list[RecordingSpanProcessor], ) -> None: diff --git a/tilebox-workflows/tilebox/workflows/observability/logging.py b/tilebox-workflows/tilebox/workflows/observability/logging.py index 2aa7e2f..9f84986 100644 --- a/tilebox-workflows/tilebox/workflows/observability/logging.py +++ b/tilebox-workflows/tilebox/workflows/observability/logging.py @@ -64,13 +64,7 @@ def _get_default_resource(service: str | Resource | None = None) -> Resource: - if isinstance(service, Resource): # already a resource object - service_name = service.attributes.get(SERVICE_NAME) - if service_name is not None and service_name != "unknown_service": - # default value of SERVICE_NAME is "unknown_service", so if we have anything other than that we - # know it's already configured - return service - + """Build Tilebox defaults, then apply explicitly configured resource attributes.""" service_name = service if isinstance(service, str) else _DEFAULT_SERVICE_NAME workflows_version = "dev" @@ -78,7 +72,7 @@ def _get_default_resource(service: str | Resource | None = None) -> Resource: workflows_version = version("tilebox-workflows") uname = platform.uname() - return Resource.create( + resource = Resource.create( attributes={ SERVICE_NAMESPACE: "tilebox.workflows", SERVICE_NAME: service_name, @@ -90,6 +84,7 @@ def _get_default_resource(service: str | Resource | None = None) -> Resource: OS_TYPE: uname.system.lower(), } ) + return resource.merge(service) if isinstance(service, Resource) else resource def _sanitize_otel_attribute_value( @@ -304,11 +299,9 @@ def configure_otel_logging( Each call adds an export; Tilebox's API export, the CLI stream, and console outputs remain installed. Args: - service: A string or a resource object to include in all traces. Used to identify the service being traced. - If a string is provided, it will be used as the service name. If a resource object is provided, it will be - used as the resource. Defaults to a resource with the service name set to "tilebox.workflows-{process_id}", - the version set to the version of the package, and the service instance id set to a combination - of hostname and process id. + service: A service name or resource whose attributes override the Tilebox defaults. + Defaults to service name "tilebox-python", namespace "tilebox.workflows", the installed package version, + and the CLI runtime ID or a process-local UUID shared by logs and traces. level: The logging level to use for the OTEL handler. Only log messages with a level higher or equal to this will be sent to the endpoint. Defaults to logging.DEBUG. It is typically recommended to keep this at a lower level, since actual filtering of log messages to higher levels is typically done by the logger itself. @@ -348,11 +341,9 @@ def configure_otel_logging_axiom( configuration. Args: - service: A string or a resource object to include in all traces. Used to identify the service being traced. - If a string is provided, it will be used as the service name. If a resource object is provided, it will be - used as the resource. Defaults to a resource with the service name set to "tilebox.workflows-{process_id}", - the version set to the version of the package, and the service instance id set to a combination - of hostname and process id. + service: A service name or resource whose attributes override the Tilebox defaults. + Defaults to service name "tilebox-python", namespace "tilebox.workflows", the installed package version, + and the CLI runtime ID or a process-local UUID shared by logs and traces. level: The logging level to use for the Axiom log handler. Only log messages with a level higher or equal to this will be sent to the endpoint. Defaults to logging.DEBUG. It is typically recommended to keep this at a lower level, since actual filtering of log messages to higher levels is typically done by the logger itself. diff --git a/tilebox-workflows/tilebox/workflows/observability/tracing.py b/tilebox-workflows/tilebox/workflows/observability/tracing.py index 075f17f..32e0fda 100644 --- a/tilebox-workflows/tilebox/workflows/observability/tracing.py +++ b/tilebox-workflows/tilebox/workflows/observability/tracing.py @@ -45,7 +45,7 @@ # to extract trace_ids and spans, in case other runners / workflow clients have tracing configured. # So instead we use a tracer provider without any exporters, which will still create traces and spans, # but will not send them anywhere. -_tilebox_tracer_provider = TracerProvider() +_tilebox_tracer_provider = TracerProvider(resource=_get_default_resource()) _workflow_tracers = [] @@ -231,11 +231,9 @@ def configure_otel_tracing( Additionally, this will also configure a logging handler that will add log messages to active spans as span events. Args: - service: A string or a resource object to include in all traces. Used to identify the service being traced. - If a string is provided, it will be used as the service name. If a resource object is provided, it will be - used as the resource. Defaults to a resource with the service name set to "tilebox.workflows-{process_id}", - the version set to the version of the package, and the service instance id set to a combination - of hostname and process id. + service: A service name or resource whose attributes override the Tilebox defaults. + Defaults to service name "tilebox-python", namespace "tilebox.workflows", the installed package version, + and the CLI runtime ID or a process-local UUID shared by logs and traces. endpoint: The URL of the OTLP compatible endpoint to send traces and spans to. If not provided, the environment variable OTEL_TRACES_ENDPOINT will be used. If that is not set either, an error will be raised. OTLP compatible endpoints typically have the path name "/v1/traces". If the specified endpoint does not @@ -279,11 +277,9 @@ def configure_otel_tracing_axiom( send traces and spans to Axiom. Args: - service: A string or a resource object to include in all traces. Used to identify the service being traced. - If a string is provided, it will be used as the service name. If a resource object is provided, it will be - used as the resource. Defaults to a resource with the service name set to "tilebox.workflows-{process_id}", - the version set to the version of the package, and the service instance id set to a combination - of hostname and process id. + service: A service name or resource whose attributes override the Tilebox defaults. + Defaults to service name "tilebox-python", namespace "tilebox.workflows", the installed package version, + and the CLI runtime ID or a process-local UUID shared by logs and traces. dataset: The name of the Axiom dataset to ingest traces into. If not provided, the environment variable AXIOM_TRACES_DATASET will be used. If that is not set either, an error will be raised. api_key: The API key to use for authentication. If not provided, the environment variable AXIOM_API_KEY will be