diff --git a/README.md b/README.md index 5a53261..5a2ca2b 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,19 @@ Multiple destinations can be enabled with a comma-separated list, for example Logging uses Application Default Credentials and requires permission to create log entries, typically through `roles/logging.logWriter`. +Request and operation logs include two timing views: + +- `timings_ms` and `timing_counts` are flat inclusive aggregates by segment + name, intended for quick scanning and compatibility with existing log + queries. +- `segment_tree` is an ordered nested view of segment occurrences. Repeated + sibling segments are preserved as separate entries, and safe scalar segment + attributes are included so callers can distinguish settings such as + `simulation_kind=baseline` versus `simulation_kind=reform`. + +Core structured log fields take precedence over caller-provided attributes with +the same keys. + On runtimes that do not provide Application Default Credentials, set `GCP_CREDENTIALS_JSON` to a service account JSON document. The Google Cloud Logging destination will materialize it into a temporary credentials file and diff --git a/changelog.d/15.added.md b/changelog.d/15.added.md new file mode 100644 index 0000000..967c096 --- /dev/null +++ b/changelog.d/15.added.md @@ -0,0 +1,2 @@ +Add ordered nested segment trees to request and operation structured logs. +Core structured log fields now take precedence over caller-provided attributes. diff --git a/policyengine_observability/context.py b/policyengine_observability/context.py index 8a8627f..bb2d464 100644 --- a/policyengine_observability/context.py +++ b/policyengine_observability/context.py @@ -24,6 +24,28 @@ def as_dict(self) -> dict[str, Any]: } +@dataclass +class SegmentTimingNode: + sequence: int + name: str + attrs: dict[str, Any] = field(default_factory=dict) + duration_ms: float | None = None + children: list[SegmentTimingNode] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + record: dict[str, Any] = { + "sequence": self.sequence, + "name": self.name, + } + if self.attrs: + record["attrs"] = dict(self.attrs) + if self.duration_ms is not None: + record["duration_ms"] = round(self.duration_ms, 3) + if self.children: + record["children"] = [child.as_dict() for child in self.children] + return record + + @dataclass class OperationObservabilityContext: config: ObservabilityConfig @@ -32,6 +54,8 @@ class OperationObservabilityContext: attributes: dict[str, Any] = field(default_factory=dict) timings_ms: dict[str, float] = field(default_factory=dict) timing_counts: dict[str, int] = field(default_factory=dict) + segment_tree: list[SegmentTimingNode] = field(default_factory=list) + segment_sequence: list[int] = field(default_factory=lambda: [0]) emit_log: bool = True record_metric: bool = True started_at: float = field(default_factory=time.perf_counter) @@ -94,6 +118,7 @@ def as_log_record( ) -> dict[str, Any]: event = "operation_failed" if self.error else "operation_completed" return { + **self.attributes, "schema_version": "policyengine.observability.operation.v1", "event": event, "service_name": self.config.service_name, @@ -107,7 +132,7 @@ def as_log_record( "duration_ms": round(self.duration_seconds() * 1000, 3), "timings_ms": dict(self.timings_ms), "timing_counts": dict(self.timing_counts), - **self.attributes, + "segment_tree": [node.as_dict() for node in self.segment_tree], "error": self.error.as_dict() if self.error else None, } @@ -129,6 +154,8 @@ class RequestObservabilityContext: attributes: dict[str, Any] = field(default_factory=dict) timings_ms: dict[str, float] = field(default_factory=dict) timing_counts: dict[str, int] = field(default_factory=dict) + segment_tree: list[SegmentTimingNode] = field(default_factory=list) + segment_sequence: list[int] = field(default_factory=lambda: [0]) status_code: int | None = None error: ErrorRecord | None = None emitted: bool = False @@ -206,6 +233,8 @@ def as_log_record( ) status_code = self.status_code or (500 if self.error else None) return { + **self.inbound, + **self.attributes, "schema_version": "policyengine.observability.request.v1", "event": event, "service_name": self.config.service_name, @@ -222,10 +251,9 @@ def as_log_record( "endpoint": self.endpoint, "status_code": status_code, "duration_ms": round(self.duration_seconds() * 1000, 3), - **self.inbound, "timings_ms": dict(self.timings_ms), "timing_counts": dict(self.timing_counts), - **self.attributes, + "segment_tree": [node.as_dict() for node in self.segment_tree], "error": self.error.as_dict() if self.error else None, } diff --git a/policyengine_observability/runtime.py b/policyengine_observability/runtime.py index 3be56fe..8fbda49 100644 --- a/policyengine_observability/runtime.py +++ b/policyengine_observability/runtime.py @@ -20,6 +20,7 @@ ErrorRecord, OperationObservabilityContext, RequestObservabilityContext, + SegmentTimingNode, _metric_attrs, ) from .destinations import LogDestinationManager @@ -58,6 +59,21 @@ "policyengine_observability_turn_start", default=None, ) +_SEGMENT_STACK: ContextVar[tuple[tuple[int, SegmentTimingNode], ...]] = ( + ContextVar( + "policyengine_observability_segment_stack", + default=(), + ) +) + +MAX_SEGMENT_ATTR_LENGTH = 200 +SENSITIVE_SEGMENT_ATTR_PARTS = ( + "authorization", + "credential", + "password", + "secret", + "token", +) class _NoOpInstrument: @@ -324,9 +340,18 @@ def _begin_request_operation( try: parent_operation = _OPERATION_CONTEXT.get() timings = context.timings_ms + timing_counts = context.timing_counts + segment_tree = context.segment_tree + segment_sequence = context.segment_sequence if context.internal_dispatch and parent_operation is not None: timings = parent_operation.timings_ms + timing_counts = parent_operation.timing_counts + segment_tree = parent_operation.segment_tree + segment_sequence = parent_operation.segment_sequence context.timings_ms = timings + context.timing_counts = timing_counts + context.segment_tree = segment_tree + context.segment_sequence = segment_sequence operation = OperationObservabilityContext( config=context.config, name=context.route, @@ -338,6 +363,9 @@ def _begin_request_operation( "path": context.path, }, timings_ms=timings, + timing_counts=timing_counts, + segment_tree=segment_tree, + segment_sequence=segment_sequence, emit_log=False, record_metric=False, ) @@ -513,6 +541,10 @@ def _segment_context(self, name: Any, **attrs: Any) -> Iterator[Any]: attrs, ) start = self._safe_perf_counter(f"segment.{segment_name}.start") + segment_tree_handle = self._start_segment_tree_node( + segment_name, + attrs, + ) span_attrs = self._segment_span_attributes(attrs) span_name = self._span_name(segment_name) error: BaseException | None = None @@ -521,11 +553,22 @@ def _segment_context(self, name: Any, **attrs: Any) -> Iterator[Any]: yield span except BaseException as exc: error = exc - self._record_segment_safely(segment_name, start, attrs) + self._record_segment_safely( + segment_name, + start, + attrs, + segment_tree_handle=segment_tree_handle, + ) raise else: - self._record_segment_safely(segment_name, start, attrs) + self._record_segment_safely( + segment_name, + start, + attrs, + segment_tree_handle=segment_tree_handle, + ) finally: + self._reset_segment_tree_stack(segment_tree_handle) self.end_operation(implicit_operation, error) @asynccontextmanager @@ -539,6 +582,10 @@ async def asegment(self, name: Any, **attrs: Any) -> AsyncIterator[Any]: attrs, ) start = self._safe_perf_counter(f"segment.{segment_name}.start") + segment_tree_handle = self._start_segment_tree_node( + segment_name, + attrs, + ) span_attrs = self._segment_span_attributes(attrs) span_name = self._span_name(segment_name) error: BaseException | None = None @@ -547,11 +594,22 @@ async def asegment(self, name: Any, **attrs: Any) -> AsyncIterator[Any]: yield span except BaseException as exc: error = exc - self._record_segment_safely(segment_name, start, attrs) + self._record_segment_safely( + segment_name, + start, + attrs, + segment_tree_handle=segment_tree_handle, + ) raise else: - self._record_segment_safely(segment_name, start, attrs) + self._record_segment_safely( + segment_name, + start, + attrs, + segment_tree_handle=segment_tree_handle, + ) finally: + self._reset_segment_tree_stack(segment_tree_handle) self.end_operation(implicit_operation, error) @contextmanager @@ -1460,11 +1518,130 @@ def _end_span( except BaseException as exc: self.log_observability_failure("otel.span_exit", exc) + def _start_segment_tree_node( + self, + name: str, + attrs: dict[str, Any], + ) -> dict[str, Any] | None: + try: + owner = self._segment_tree_owner() + if owner is None: + return None + owner.segment_sequence[0] += 1 + node = SegmentTimingNode( + sequence=owner.segment_sequence[0], + name=name, + attrs=self._safe_segment_tree_attrs(attrs), + ) + owner_id = id(owner.segment_tree) + stack = _SEGMENT_STACK.get() + if stack and stack[-1][0] == owner_id: + stack[-1][1].children.append(node) + else: + owner.segment_tree.append(node) + token = _SEGMENT_STACK.set((*stack, (owner_id, node))) + return {"node": node, "token": token} + except BaseException as exc: + self.log_observability_failure( + "segment.tree_start", + exc, + segment=name, + ) + return None + + def _finish_segment_tree_node( + self, + handle: dict[str, Any] | None, + duration_seconds: float, + ) -> None: + if not handle: + return + try: + node = handle.get("node") + if not isinstance(node, SegmentTimingNode): + return + node.duration_ms = duration_seconds * 1000 + except BaseException as exc: + self.log_observability_failure( + "segment.tree_finish", + exc, + ) + + def _reset_segment_tree_stack( + self, + handle: dict[str, Any] | None, + ) -> None: + if not handle: + return + token = handle.get("token") + if token is None: + return + try: + _SEGMENT_STACK.reset(token) + except BaseException as exc: + self.log_observability_failure("segment.tree_reset", exc) + + def _segment_tree_owner( + self, + ) -> RequestObservabilityContext | OperationObservabilityContext | None: + context = self.current_context() + if context is not None: + return context + return self.current_operation() + + def _safe_segment_tree_attrs( + self, + attrs: dict[str, Any], + ) -> dict[str, Any]: + safe_attrs: dict[str, Any] = {} + for key, value in attrs.items(): + key_text = self._safe_str(key) + key_lower = key_text.lower() + if any(part in key_lower for part in SENSITIVE_SEGMENT_ATTR_PARTS): + continue + if value is None: + continue + if hasattr(value, "value"): + value = value.value + if isinstance(value, bool | int | float): + safe_attrs[key_text] = value + elif isinstance(value, str): + safe_attrs[key_text] = value[:MAX_SEGMENT_ATTR_LENGTH] + return safe_attrs + + def _record_segment_flat_timing( + self, + context: RequestObservabilityContext | None, + operation: OperationObservabilityContext | None, + name: str, + duration_ms: float, + ) -> None: + seen_timing_ids: set[int] = set() + seen_count_ids: set[int] = set() + for target in (context, operation): + if target is None: + continue + timings_id = id(target.timings_ms) + if timings_id not in seen_timing_ids: + target.timings_ms[name] = round( + target.timings_ms.get(name, 0.0) + duration_ms, + 3, + ) + seen_timing_ids.add(timings_id) + counts_id = id(target.timing_counts) + if counts_id not in seen_count_ids: + target.timing_counts[name] = ( + target.timing_counts.get(name, 0) + 1 + ) + seen_count_ids.add(counts_id) + def _record_segment_safely( self, name: str, start: float | None, attrs: dict[str, Any], + *, + segment_tree_handle: dict[str, Any] | None = None, ) -> None: if start is None: return @@ -1473,6 +1650,7 @@ def _record_segment_safely( return try: duration = end - start + self._finish_segment_tree_node(segment_tree_handle, duration) self._record_timing(name, duration) context = self.current_context() operation = self.current_operation() @@ -1485,22 +1663,13 @@ def _record_segment_safely( ) } duration_ms = duration * 1000 - if context is not None: - context.timings_ms[name] = round( - context.timings_ms.get(name, 0.0) + duration_ms, - 3, - ) - context.timing_counts[name] = ( - context.timing_counts.get(name, 0) + 1 - ) + self._record_segment_flat_timing( + context, + operation, + name, + duration_ms, + ) if operation is not None: - operation.timings_ms[name] = round( - operation.timings_ms.get(name, 0.0) + duration_ms, - 3, - ) - operation.timing_counts[name] = ( - operation.timing_counts.get(name, 0) + 1 - ) metric_attributes = operation.metric_attributes( segment=name, **metric_extra, diff --git a/tests/test_runtime.py b/tests/test_runtime.py index db53acf..6dfaf7c 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -244,6 +244,74 @@ def test_operation_log_accumulates_repeated_segment_timings() -> None: assert payload["timing_counts"]["load"] == 2 +def test_operation_log_records_ordered_nested_segment_tree() -> None: + observed = runtime() + handle = observed.start_operation("job") + operation = handle["operation"] + + try: + with observed.segment(SegmentName.LOAD): + with observed.segment( + SegmentName.SAVE, + simulation_kind="baseline", + token="SECRET", + payload={"not": "safe"}, + ): + pass + with observed.segment( + SegmentName.SAVE, + simulation_kind="reform", + ): + pass + finally: + observed.end_operation(handle) + + payload = operation.as_log_record(trace_id=None, span_id=None) + tree = payload["segment_tree"] + assert len(tree) == 1 + assert tree[0]["sequence"] == 1 + assert tree[0]["name"] == "load" + assert "duration_ms" in tree[0] + assert "self_ms" not in tree[0] + + children = tree[0]["children"] + assert [child["sequence"] for child in children] == [2, 3] + assert [child["name"] for child in children] == ["save", "save"] + assert children[0]["attrs"] == {"simulation_kind": "baseline"} + assert children[1]["attrs"] == {"simulation_kind": "reform"} + assert "token" not in children[0].get("attrs", {}) + assert "payload" not in children[0].get("attrs", {}) + assert payload["timing_counts"]["save"] == 2 + + +def test_operation_log_reserved_fields_override_attributes() -> None: + observed = runtime() + handle = observed.start_operation( + "job", + operation="attribute-operation", + duration_ms="attribute-duration", + timings_ms="attribute-timings", + timing_counts="attribute-counts", + segment_tree="attribute-tree", + error="attribute-error", + ) + operation = handle["operation"] + + try: + with observed.segment(SegmentName.LOAD): + pass + finally: + observed.end_operation(handle) + + payload = operation.as_log_record(trace_id=None, span_id=None) + assert payload["operation"] == "job" + assert isinstance(payload["duration_ms"], float) + assert isinstance(payload["timings_ms"], dict) + assert isinstance(payload["timing_counts"], dict) + assert isinstance(payload["segment_tree"], list) + assert payload["error"] is None + + def test_async_segment_records_timing() -> None: async def run() -> dict[str, float]: observed = runtime() @@ -257,6 +325,42 @@ async def run() -> dict[str, float]: assert "save_ms" in timings +def test_async_segments_keep_independent_segment_tree_stacks() -> None: + async def run() -> list[dict[str, Any]]: + observed = runtime() + handle = observed.start_operation("job") + operation = handle["operation"] + + async def branch(branch_name: str) -> None: + async with observed.asegment(SegmentName.LOAD, branch=branch_name): + await asyncio.sleep(0) + async with observed.asegment( + SegmentName.SAVE, + branch=branch_name, + ): + await asyncio.sleep(0) + + try: + await asyncio.gather(branch("a"), branch("b")) + finally: + observed.end_operation(handle) + return operation.as_log_record(trace_id=None, span_id=None)[ + "segment_tree" + ] + + tree = asyncio.run(run()) + + assert [node["name"] for node in tree] == ["load", "load"] + assert [node["attrs"] for node in tree] == [ + {"branch": "a"}, + {"branch": "b"}, + ] + assert [node["children"][0]["attrs"] for node in tree] == [ + {"branch": "a"}, + {"branch": "b"}, + ] + + def test_segment_preserves_business_exception_and_records_timing() -> None: observed = runtime() @@ -268,6 +372,26 @@ def test_segment_preserves_business_exception_and_records_timing() -> None: assert "load_ms" in timings +def test_segment_tree_records_failed_segments_before_reraising() -> None: + observed = runtime() + handle = observed.start_operation("job") + operation = handle["operation"] + error = None + + try: + with observed.segment(SegmentName.LOAD): + raise ValueError("business failed") + except ValueError as exc: + error = exc + finally: + observed.end_operation(handle, error) + + payload = operation.as_log_record(trace_id=None, span_id=None) + assert payload["event"] == "operation_failed" + assert payload["segment_tree"][0]["name"] == "load" + assert "duration_ms" in payload["segment_tree"][0] + + def test_unregistered_segment_falls_back_without_throwing() -> None: class BrokenString: def __str__(self) -> str: @@ -475,6 +599,10 @@ def test_standalone_segment_creates_implicit_operation_metrics() -> None: observed.segment_duration = RecordingInstrument() observed.operation_duration = RecordingInstrument() observed.operations = RecordingInstrument() + emitted_payloads = [] + observed.emit_operation_log = lambda operation: emitted_payloads.append( + operation.as_log_record(trace_id=None, span_id=None) + ) with observed.segment(SegmentName.LOAD, flavor="cli", tool="loader"): pass @@ -486,6 +614,11 @@ def test_standalone_segment_creates_implicit_operation_metrics() -> None: assert segment_attributes["tool"] == "loader" assert operation_attributes["operation"] == "load" assert operation_attributes["flavor"] == "cli" + assert emitted_payloads[0]["segment_tree"][0]["name"] == "load" + assert emitted_payloads[0]["segment_tree"][0]["attrs"] == { + "flavor": "cli", + "tool": "loader", + } assert observed.current_operation() is None @@ -699,6 +832,10 @@ def test_request_log_accumulates_repeated_segment_timings() -> None: assert context.timing_counts["load"] == 2 payload = context.as_log_record(trace_id=None, span_id=None) assert payload["timing_counts"]["load"] == 2 + assert [node["name"] for node in payload["segment_tree"]] == [ + "load", + "load", + ] def test_internal_dispatch_segments_merge_into_parent_operation() -> None: @@ -729,7 +866,11 @@ def test_internal_dispatch_segments_merge_into_parent_operation() -> None: observed.teardown_request(None) assert context.timings_ms is parent_operation.timings_ms + assert context.timing_counts is parent_operation.timing_counts + assert context.segment_tree is parent_operation.segment_tree assert "load" in parent_operation.timings_ms + assert parent_operation.timing_counts["load"] == 1 + assert parent_operation.segment_tree[0].name == "load" assert observed.current_operation() is parent_operation finally: observed.end_operation(handle) @@ -764,7 +905,10 @@ def test_non_internal_request_timings_do_not_leak_to_parent_operation() -> ( observed.teardown_request(None) assert context.timings_ms is not parent_operation.timings_ms + assert context.segment_tree is not parent_operation.segment_tree assert "load" not in parent_operation.timings_ms + assert parent_operation.segment_tree == [] + assert context.segment_tree[0].name == "load" assert observed.current_operation() is parent_operation finally: observed.end_operation(handle) @@ -1621,6 +1765,45 @@ def test_request_log_emits_to_configured_destination_once() -> None: assert severity == "INFO" +def test_request_log_reserved_fields_override_inbound_and_attributes() -> None: + observed = runtime() + context = RequestObservabilityContext( + config=observed.config, + request_id="request-1", + method="GET", + route="/calculate", + path="/calculate", + endpoint="calculate", + query_keys=[], + content_length_bytes=None, + inbound={ + "request_id": "inbound-request", + "status_code": "inbound-status", + "duration_ms": "inbound-duration", + "segment_tree": "inbound-tree", + }, + attributes={ + "request_id": "attribute-request", + "status_code": "attribute-status", + "duration_ms": "attribute-duration", + "timings_ms": "attribute-timings", + "timing_counts": "attribute-counts", + "segment_tree": "attribute-tree", + "error": "attribute-error", + }, + status_code=204, + ) + + payload = context.as_log_record(trace_id=None, span_id=None) + assert payload["request_id"] == "request-1" + assert payload["status_code"] == 204 + assert isinstance(payload["duration_ms"], float) + assert isinstance(payload["timings_ms"], dict) + assert isinstance(payload["timing_counts"], dict) + assert payload["segment_tree"] == [] + assert payload["error"] is None + + def test_event_log_emits_to_configured_destination() -> None: observed = runtime() destination = RecordingLogDestination()