Adds attribute logging to s3 tracker

We just had to wire it through -- this also adds app_id + partition_key
to the parameters.

burr/integrations/opentelemetry.py

@@ -427,6 +427,7 @@ def on_end(self, span: "Span") -> None:
                     span=cached_span.action_span,
                     tags={},  # TODO -- log
                     app_id=cached_span.app_id,
+                    partition_key=cached_span.partition_key,
                 )
 
 

burr/lifecycle/base.py

@@ -206,6 +206,8 @@ def do_log_attributes(
         action_sequence_id: int,
         span: Optional["ActionSpan"],
         tags: dict,
+        app_id: str,
+        partition_key: Optional[str],
         **future_kwargs: Any,
     ):
         pass
@@ -225,6 +227,8 @@ async def do_log_attributes(
         action_sequence_id: int,
         span: "ActionSpan",
         tags: dict,
+        app_id: str,
+        partition_key: Optional[str],
         **future_kwargs: Any,
     ):
         pass

burr/tracking/common/models.py

@@ -189,6 +189,8 @@ class EndSpanModel(IdentifyingModel):
 
     @property
     def sequence_id(self) -> int:
+        # so we have full backwards compatibility
+        # the server likes them all to be called sequence_id
         return self.action_sequence_id
 
 
@@ -203,3 +205,8 @@ class AttributeModel(IdentifyingModel):
     value: Union[dict, str, int, float, bool, list, None]
     tags: Dict[str, str]
     type: str = "attribute"
+
+    @property
+    def sequence_id(self) -> int:
+        # Ditto with the above
+        return self.action_sequence_id

burr/tracking/s3client.py

@@ -19,6 +19,7 @@
 from burr.tracking.common.models import (
     ApplicationMetadataModel,
     ApplicationModel,
+    AttributeModel,
     BeginEntryModel,
     BeginSpanModel,
     EndEntryModel,
@@ -84,7 +85,7 @@ def _allowed_project_name(project_name: str, on_windows: bool) -> bool:
     return bool(re.match(pattern, project_name))
 
 
-EventType = Union[BeginEntryModel, EndEntryModel, BeginSpanModel, EndSpanModel]
+EventType = Union[BeginEntryModel, EndEntryModel, BeginSpanModel, EndSpanModel, AttributeModel]
 
 
 def unique_ordered_prefix() -> str:
@@ -122,12 +123,22 @@ def do_log_attributes(
         action: str,
         action_sequence_id: int,
         span: Optional["ActionSpan"],
+        app_id: str,
+        partition_key: Optional[str],
         tags: dict,
         **future_kwargs: Any,
     ):
         # TODO -- log attributes to s3 as well
         # Coming up shortly
-        pass
+        for attribute_name, attribute in attributes.items():
+            attribute_model = AttributeModel(
+                key=attribute_name,
+                action_sequence_id=action_sequence_id,
+                span_id=span.uid if span is not None else None,
+                value=serde.serialize(attribute, **self.serde_kwargs),
+                tags=tags,
+            )
+            self.submit_log_event(attribute_model, app_id=app_id, partition_key=partition_key)
 
     def __init__(
         self,
@@ -305,7 +316,6 @@ def post_application_create(
                 else "None",
             },
         )
-        # TODO -- log parent relationship
 
     def pre_run_step(
         self,

burr/visibility/tracing.py

@@ -232,6 +232,8 @@ def log_attributes(self, **attributes):
             action=self.action,
             action_sequence_id=self.action_sequence_id,
             span=self.context_var.get(),
+            app_id=self.app_id,
+            partition_key=self.partition_key,
             tags={},  # TODO -- add tags
         )