Update task span attributes

Author: lukasbindreiter

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
     hooks:
       - id: sync-with-uv
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.15.7
+    rev: v0.15.12
     hooks:
       - id: ruff-check
         args: [--fix, --exit-non-zero-on-fix]

tilebox-datasets/tilebox/datasets/protobuf_conversion/field_types.py

@@ -254,6 +254,8 @@ def to_proto(self, value: tuple[float, float, float]) -> LatLonAlt | None:
 
 def infer_field_type(field: FieldDescriptor) -> ProtobufFieldType:
     if field.type == FieldDescriptor.TYPE_MESSAGE:
+        if field.message_type is None:
+            raise ValueError("Expected message type for field but got None")
         message_name = field.message_type.full_name
         if message_name not in _MESSAGE_NAMES_TO_FIELDS:
             raise ValueError(f"Unsupported message type {message_name}")
@@ -281,6 +283,9 @@ def enum_mapping_from_field_descriptor(field: FieldDescriptor) -> dict[int, str]
     if field.type != FieldDescriptor.TYPE_ENUM:
         raise ValueError("Expected field to be of type FieldDescriptor.TYPE_ENUM")
 
+    if field.enum_type is None:
+        raise ValueError("Expected enum type for field but got None")
+
     # remove the enum type prefix from the enum values
     # e.g. FLIGHT_DIRECTION_ASCENDING of the FlightDirection enum will result in a value of ASCENDING
     enum_type_prefix = _camel_to_uppercase(field.enum_type.name) + "_"

tilebox-datasets/tilebox/datasets/protobuf_conversion/protobuf_xarray.py

@@ -4,7 +4,7 @@
 
 import contextlib
 from collections.abc import Sequence
-from typing import Any, TypeVar
+from typing import Any, TypeVar, cast
 
 import numpy as np
 import xarray as xr
@@ -352,7 +352,7 @@ def _create_field_converters(message: Message, buffer_size: int) -> dict[str, _F
     for field in message.DESCRIPTOR.fields:
         # if we have an unsupported field type we will get a ValueError, so we just skip those fields
         with contextlib.suppress(ValueError):
-            converter = _create_field_converter(field)
+            converter = _create_field_converter(cast(FieldDescriptor, field))
             converters[field.name] = converter
 
     if buffer_size > 0:

tilebox-datasets/tilebox/datasets/protobuf_conversion/to_protobuf.py

@@ -1,11 +1,12 @@
 from collections import defaultdict
 from collections.abc import Collection, Iterable, Iterator, Mapping
-from typing import Any
+from typing import Any, cast
 from uuid import UUID
 
 import numpy as np
 import pandas as pd
 import xarray as xr
+from google.protobuf.descriptor import Descriptor
 from google.protobuf.message import Message
 
 from tilebox.datasets.protobuf_conversion.field_types import (
@@ -40,7 +41,7 @@ def to_messages(  # noqa: C901, PLR0912
 
     ignore = set(ignore_fields or [])
 
-    field_descriptors_by_name = message_type.DESCRIPTOR.fields_by_name
+    field_descriptors_by_name = cast(Descriptor, message_type.DESCRIPTOR).fields_by_name
 
     # let's validate our fields, to make sure that they are all known fields for the given protobuf message
     # and that they are all lists of the same length

tilebox-workflows/pyproject.toml

@@ -33,6 +33,10 @@ dependencies = [
     "ipywidgets>=8.1.7",
     "python-dateutil>=2.9.0.post0",
     "obstore>=0.8.2",
+    "opentelemetry-proto>=1.30.0",
+    "structlog>=25.5.0",
+    # grpcio 1.80.0 contains unwanted log message spam: https://github.com/grpc/grpc/issues/42293
+    "grpcio<1.80.0",
 ]
 
 [dependency-groups]

tilebox-workflows/tests/automations/test_storage_event.py

@@ -7,6 +7,7 @@
 
 from tilebox.workflows.automations import StorageEventTask
 from tilebox.workflows.data import StorageEventType, StorageLocation
+from tilebox.workflows.observability.tracing import NoopWorkflowTracer
 from tilebox.workflows.task import RunnerContext
 
 
@@ -43,7 +44,7 @@ def test_storage_event_task_de_serialization_roundtrip(storage_location: Storage
     triggered_task = task.once(storage_location, "FM171/apid.json", StorageEventType.CREATED)
 
     # serialized task only contains a bucket id, so for deserialization we need to provide a bucket lookup table
-    context = RunnerContext(storage_locations=[storage_location])
+    context = RunnerContext(NoopWorkflowTracer(), storage_locations=[storage_location])
 
     serialized = triggered_task._serialize()
     assert ExampleStorageEventTask._deserialize(serialized, context) == triggered_task
@@ -55,7 +56,7 @@ def test_storage_event_task_de_serialization_roundtrip_protobuf(storage_location
     triggered_task = task.once(storage_location, "FM171/apid.json")
 
     # serialized task only contains a bucket id, so for deserialization we need to provide a bucket lookup table
-    context = RunnerContext(storage_locations=[storage_location])
+    context = RunnerContext(NoopWorkflowTracer(), storage_locations=[storage_location])
 
     serialized = triggered_task._serialize()
     assert ExampleProtoStorageEventTask._deserialize(serialized, context) == triggered_task

tilebox-workflows/tests/jobs/test_client.py

@@ -10,6 +10,7 @@
 from tilebox.workflows.data import Job, JobState, uuid_message_to_uuid, uuid_to_uuid_message
 from tilebox.workflows.jobs.client import JobClient
 from tilebox.workflows.jobs.service import JobService
+from tilebox.workflows.observability.tracing import NoopWorkflowTracer
 from tilebox.workflows.task import ExecutionContext, Task
 from tilebox.workflows.workflows.v1.core_pb2 import Job as JobMessage
 from tilebox.workflows.workflows.v1.core_pb2 import JobState as JobStateEnum
@@ -111,7 +112,7 @@ def __init__(self) -> None:
         super().__init__()
         service = JobService(MagicMock())
         service.service = MockJobService()  # mock the gRPC service
-        self.job_client = JobClient(service)
+        self.job_client = JobClient(service, NoopWorkflowTracer())
         self.count_total_submitted = 0
 
     queued_jobs: Bundle[Job] = Bundle("queued_jobs")

tilebox-workflows/tests/runner/test_runner.py

@@ -10,6 +10,7 @@
 from tilebox.workflows.cache import InMemoryCache, JobCache
 from tilebox.workflows.client import Client
 from tilebox.workflows.data import JobState, ProgressIndicator, RunnerContext, TaskState
+from tilebox.workflows.observability.tracing import NoopWorkflowTracer
 from tilebox.workflows.runner.task_runner import TaskRunner
 
 
@@ -206,10 +207,10 @@ def test_runner_disallow_duplicate_task_identifiers() -> None:
         MagicMock(),
         "dummy-cluster",
         InMemoryCache(),
-        None,
+        NoopWorkflowTracer(),
         None,
         MagicMock(),
-        RunnerContext(),
+        RunnerContext(NoopWorkflowTracer()),
     )
 
     runner.register(FlakyTask)

tilebox-workflows/tilebox/workflows/client.py

@@ -12,27 +12,30 @@
 )
 from tilebox.workflows.jobs.client import JobClient
 from tilebox.workflows.jobs.service import JobService
-from tilebox.workflows.observability.tracing import (
-    WorkflowTracer,
-)
+from tilebox.workflows.observability.tracing import WorkflowTracer
 from tilebox.workflows.runner.task_runner import TaskRunner, _LeaseRenewer
 from tilebox.workflows.runner.task_service import TaskService
 
 
 class Client:
-    def __init__(self, *, url: str = "https://api.tilebox.com", token: str | None = None) -> None:
+    def __init__(
+        self, *, url: str = "https://api.tilebox.com", token: str | None = None, name: str | None = None
+    ) -> None:
         """
         Create a Tilebox workflows client.
 
         Args:
             url: Tilebox API Url. Defaults to "https://api.tilebox.com".
             token: The API Key to authenticate with. If not set the `TILEBOX_API_KEY` environment variable will be used.
+            name: An optional name of the client, used as service.name for telemetry. If not set, defaults to
+                the service name provided by `tilebox.workflows.observability.tracing.configure_otel_tracing`,
+                or "tilebox-python" if no external tracer is configured.
         """
         token = _token_from_env(url, token)
         self._auth: dict[str, str] = {"token": token, "url": url}
         self._channel = open_channel(url, token)
+        self._tracer = WorkflowTracer(service=name, url=url, token=token)
 
-        self._tracer: WorkflowTracer | None = None
         self._logger: logging.Logger | None = None
 
     def configure_tracing(self, tracer: WorkflowTracer) -> None:
@@ -90,8 +93,6 @@ def runner(
         if cache is None:
             cache = NoCache()  # a no-op cache that will raise an error if it's used
 
-        tracer = self._tracer or WorkflowTracer()
-
         found_cluster = self.clusters().find(to_cluster_slug(cluster or ""))
 
         try:
@@ -103,7 +104,7 @@ def runner(
 
         runner_context_type = context or RunnerContext
         runner_context = runner_context_type(
-            tracer._tracer,  # noqa: SLF001
+            self._tracer,
             datasets_client=DatasetsClient(**self._auth),  # ty: ignore[invalid-argument-type]
             storage_locations=storage_locations,
         )
@@ -112,7 +113,7 @@ def runner(
             TaskService(self._channel),
             found_cluster.slug,
             cache,
-            tracer,
+            self._tracer,
             self._logger,
             _LeaseRenewer(**self._auth),
             runner_context,

tilebox-workflows/tilebox/workflows/data.py

@@ -30,9 +30,9 @@
     from mypy_boto3_s3.client import S3Client
 except ModuleNotFoundError:
     from typing import Any as S3Client
-from opentelemetry.trace import ProxyTracerProvider, Tracer
 
 from tilebox.datasets.sync.client import Client as DatasetsClient
+from tilebox.workflows.observability.tracing import NoopWorkflowTracer, WorkflowTracer
 from tilebox.workflows.workflows.v1 import automation_pb2 as automation_pb
 from tilebox.workflows.workflows.v1 import core_pb2, job_pb2, task_pb2
 
@@ -675,14 +675,13 @@ def to_message(self) -> automation_pb.AutomationPrototype:
 class RunnerContext:
     def __init__(
         self,
-        tracer: Tracer | None = None,
+        tracer: WorkflowTracer | None = None,
         datasets_client: DatasetsClient | None = None,
         storage_locations: list[StorageLocation] | None = None,
     ) -> None:
         if tracer is None:
-            self.tracer = ProxyTracerProvider().get_tracer("tilebox.workflows.RunnerContext")
-        else:
-            self.tracer = tracer
+            tracer = NoopWorkflowTracer()
+        self.tracer = tracer
         self.datasets_client = datasets_client
         self.storage_locations = {
             sl.id: sl._with_runner_context(self)  # noqa: SLF001