to_pandas conversion for job logs/traces

Author: lukasbindreiter

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.52.0]
+
+### Added
+
+- `tilebox-workflows`: Added automatic OpenTelemetry-based logging and tracing instrumentation to a Tilebox
+  observability backend.
+- `tilebox-workflows`: Added `ExecutionContext.logger` and `ExecutionContext.tracer` access for task implementations to
+  emit structured logs and custom spans during task execution.
+- `tilebox-workflows`: Added `JobClient.query_logs()` and `JobClient.query_spans()` to fetch telemetry for a job, with
+  list-like results that can be converted to pandas DataFrames via `to_pandas()`.
+
 ## [0.51.0] - 2026-04-07
 
 ### Changed
@@ -357,7 +368,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.51.0...HEAD
+[Unreleased]: https://github.com/tilebox/tilebox-python/compare/v0.52.0...HEAD
+[0.52.0]: https://github.com/tilebox/tilebox-python/compare/v0.51.0...v0.52.0
 [0.51.0]: https://github.com/tilebox/tilebox-python/compare/v0.50.1...v0.51.0
 [0.50.1]: https://github.com/tilebox/tilebox-python/compare/v0.50.0...v0.50.1
 [0.50.0]: https://github.com/tilebox/tilebox-python/compare/v0.49.0...v0.50.0

tilebox-workflows/tests/jobs/test_client.py

@@ -12,9 +12,11 @@
     Job,
     JobState,
     LogRecord,
+    LogRecords,
     QueryJobLogsResponse,
     QueryJobSpansResponse,
     Span,
+    Spans,
     uuid_message_to_uuid,
     uuid_to_uuid_message,
 )
@@ -70,13 +72,16 @@ def test_query_logs_paginates() -> None:
     ]
     telemetry_service = MagicMock()
     telemetry_service.query_job_logs.side_effect = [
-        QueryJobLogsResponse([log_records[0]], Pagination(starting_after=next_page_start)),
-        QueryJobLogsResponse([log_records[1]], Pagination()),
+        QueryJobLogsResponse(LogRecords([log_records[0]]), Pagination(starting_after=next_page_start)),
+        QueryJobLogsResponse(LogRecords([log_records[1]]), Pagination()),
     ]
     job_client = JobClient(MagicMock(), telemetry_service, NoopWorkflowTracer())
     job_id = uuid4()
 
-    assert job_client.query_logs(job_id) == log_records
+    logs = job_client.query_logs(job_id)
+
+    assert logs == log_records
+    assert list(logs.to_pandas()["body"]) == ["first", "second"]
     assert [call.args[1].starting_after for call in telemetry_service.query_job_logs.call_args_list] == [
         None,
         next_page_start,
@@ -115,13 +120,16 @@ def test_query_spans_paginates() -> None:
     ]
     telemetry_service = MagicMock()
     telemetry_service.query_job_spans.side_effect = [
-        QueryJobSpansResponse([spans[0]], Pagination(starting_after=next_page_start)),
-        QueryJobSpansResponse([spans[1]], Pagination()),
+        QueryJobSpansResponse(Spans([spans[0]]), Pagination(starting_after=next_page_start)),
+        QueryJobSpansResponse(Spans([spans[1]]), Pagination()),
     ]
     job_client = JobClient(MagicMock(), telemetry_service, NoopWorkflowTracer())
     job_id = uuid4()
 
-    assert job_client.query_spans(job_id) == spans
+    queried_spans = job_client.query_spans(job_id)
+
+    assert queried_spans == spans
+    assert list(queried_spans.to_pandas()["name"]) == ["first", "second"]
     assert [call.args[1].starting_after for call in telemetry_service.query_job_spans.call_args_list] == [
         None,
         next_page_start,

tilebox-workflows/tilebox/workflows/data.py

@@ -1,7 +1,7 @@
 import re
 import warnings
 from collections.abc import Callable
-from dataclasses import dataclass, field
+from dataclasses import asdict, dataclass, field
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 from functools import lru_cache
@@ -10,6 +10,7 @@
 from uuid import UUID
 
 import boto3
+import pandas as pd
 from google.cloud.storage import Client as GoogleStorageClient
 from google.cloud.storage.bucket import Bucket
 from google.protobuf.duration_pb2 import Duration
@@ -542,6 +543,12 @@ def to_message(self) -> logs_pb2.ResourceLogs:
         )
 
 
+class LogRecords(list[LogRecord]):
+    def to_pandas(self) -> Any:
+        """Convert log records to a pandas DataFrame."""
+        return pd.DataFrame([asdict(record) for record in self])
+
+
 @dataclass
 class Span:
     start_time: datetime
@@ -614,6 +621,17 @@ def to_message(self) -> trace_pb2.ResourceSpans:
         )
 
 
+class Spans(list[Span]):
+    def to_pandas(self) -> Any:
+        """Convert spans to a pandas DataFrame."""
+        rows = []
+        for span in self:
+            row = asdict(span)
+            row["duration"] = span.duration
+            rows.append(row)
+        return pd.DataFrame(rows)
+
+
 def _datetime_from_unix_nanos(unix_nanos: int) -> datetime:
     return datetime.fromtimestamp(unix_nanos / 1_000_000_000, tz=timezone.utc)
 
@@ -635,7 +653,7 @@ def _any_value_to_python(value: common_pb2.AnyValue) -> Any:  # noqa: PLR0911
         case "bytes_value":
             return value.bytes_value
         case "array_value":
-            return [_any_value_to_python(item) for item in value.array_value.values]
+            return [_any_value_to_python(item) for item in value.array_value.values]  # noqa: PD011
         case "kvlist_value":
             return _key_values_to_dict(value.kvlist_value.values)
         case _:
@@ -691,26 +709,26 @@ def _span_event_to_message(event: dict[str, Any]) -> trace_pb2.Span.Event:
 
 @dataclass(frozen=True)
 class QueryJobLogsResponse:
-    logs: list[LogRecord]
+    logs: LogRecords
     next_page: Pagination
 
     @classmethod
     def from_message(cls, page: Any) -> "QueryJobLogsResponse":
         return cls(
-            logs=[LogRecord.from_message(log_record) for log_record in page.resource_logs],
+            logs=LogRecords(LogRecord.from_message(log_record) for log_record in page.resource_logs),
             next_page=Pagination.from_message(page.next_page),
         )
 
 
 @dataclass(frozen=True)
 class QueryJobSpansResponse:
-    spans: list[Span]
+    spans: Spans
     next_page: Pagination
 
     @classmethod
     def from_message(cls, page: Any) -> "QueryJobSpansResponse":
         return cls(
-            spans=[Span.from_message(span) for span in page.resource_spans],
+            spans=Spans(Span.from_message(span) for span in page.resource_spans),
             next_page=Pagination.from_message(page.next_page),
         )
 

tilebox-workflows/tilebox/workflows/jobs/client.py

@@ -10,12 +10,12 @@
 from tilebox.workflows.data import (
     Job,
     JobState,
-    LogRecord,
+    LogRecords,
     QueryFilters,
     QueryJobLogsResponse,
     QueryJobSpansResponse,
     QueryJobsResponse,
-    Span,
+    Spans,
     TaskState,
 )
 from tilebox.workflows.jobs.service import JobService
@@ -138,7 +138,7 @@ def find(self, job_id: JobIDLike) -> Job:
         """
         return self._service.get_by_id(_to_uuid(job_id))
 
-    def query_logs(self, job_id: JobIDLike) -> list[LogRecord]:
+    def query_logs(self, job_id: JobIDLike) -> LogRecords:
         """Query logs emitted while running a job.
 
         Args:
@@ -154,12 +154,12 @@ def request(page: PaginationProtocol) -> QueryJobLogsResponse:
 
         pages = paginated_request(request, Pagination())
 
-        logs = []
+        logs = LogRecords()
         for page in pages:
             logs.extend(page.logs)
         return logs
 
-    def query_spans(self, job_id: JobIDLike) -> list[Span]:
+    def query_spans(self, job_id: JobIDLike) -> Spans:
         """Query spans emitted while running a job.
 
         Args:
@@ -175,7 +175,7 @@ def request(page: PaginationProtocol) -> QueryJobSpansResponse:
 
         pages = paginated_request(request, Pagination())
 
-        spans = []
+        spans = Spans()
         for page in pages:
             spans.extend(page.spans)
         return spans

tilebox-workflows/tilebox/workflows/runner/task_runner.py

@@ -1,5 +1,6 @@
 import contextlib
 import json
+import logging
 import random
 import signal
 import threading
@@ -540,7 +541,11 @@ def __init__(self, runner: TaskRunner, task: Task, job_cache: JobCache) -> None:
         self.job_cache = job_cache
         self._sub_tasks: list[FutureTask] = []
         self._progress_indicators: dict[str | None, ProgressUpdate] = {}
-        self._logger = runner.task_logger.bind(task_id=str(task.id))
+        if runner is None or task is None:
+            # Some tests instantiate an execution context only to exercise local subtask merging helpers.
+            self._logger = StructuredLogger(logging.getLogger("tilebox.workflows.noop"))
+        else:
+            self._logger = runner.task_logger.bind(task_id=str(task.id))
 
     def submit_subtask(
         self,