Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,10 @@ Key variables:
| `HARNESS_CONFIG_FILE` | Path to YAML config file (overrides env) |
| `HARNESS_GEN_AI_PAYLOAD_CAPTURE_ENABLED` | Capture LLM prompt/response payloads (default: off) |
| `HARNESS_GEN_AI_PAYLOAD_EVALUATION_ENABLED` | Run control plugins on GenAI spans |
| `HARNESS_SPAN_ATTRIBUTE_FLATTEN_ENABLED` | Flatten dict values from `set_span_attribute` into dot-notation keys (default: on; only `false` disables) |
| `HARNESS_SPAN_ATTRIBUTE_FLATTEN_MAX_DEPTH` | Max nesting depth when flattening dicts (default: 3) |
| `HARNESS_SPAN_ATTRIBUTE_FLATTEN_MAX_LEAVES` | Max leaf attributes emitted per dict (default: 32) |
| `HARNESS_SPAN_ATTRIBUTE_FLATTEN_RAW_JSON` | Also keep the original key as a JSON string when flattening (default: off) |

### Instrumentation opt-in (`HARNESS_` or `HA_` prefix; `HARNESS_` wins — no `AT_`/`TA_` aliases)

Expand Down
31 changes: 31 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,37 @@ For duplicate keys, the last write wins; instrumentation can overwrite a custome
if it writes the same key later. Because the span already exists, these attributes
cannot influence head sampling.

#### Dictionary values

Dictionaries are also accepted and are flattened into dot-notation attributes so each
leaf stays individually queryable in the backend:

```python
set_span_attribute("agent", {"action": "generate", "model": {"name": "gemini-2.0"}})
# exported as: agent.action="generate", agent.model.name="gemini-2.0"
```

Nothing is serialized when you call the helper — the dict is flattened at span end, just
before export. Flattening rules:

| Case | Result |
|---|---|
| `str` / `bool` / `int` / `float` leaf | kept as the native OTel type |
| `None` leaf | skipped |
| Any other object | `str(value)` |
| List of same-typed scalars | OTel array attribute at that key |
| List of dicts or mixed types | JSON string at that key |
| Nesting deeper than the configured max depth (default 3) | JSON string at the depth limit |
| Flattened key already set on the span | skipped — the explicit value wins |

At most `HARNESS_SPAN_ATTRIBUTE_FLATTEN_MAX_LEAVES` leaf attributes are emitted per
dictionary (default 32); the rest are dropped with a debug log. Override depth via
`HARNESS_SPAN_ATTRIBUTE_FLATTEN_MAX_DEPTH` (default 3). The original key (`agent`) is not set unless
`HARNESS_SPAN_ATTRIBUTE_FLATTEN_RAW_JSON=true`, which additionally stores the whole dict
as JSON there. Flattening is **enabled by default**; set
`HARNESS_SPAN_ATTRIBUTE_FLATTEN_ENABLED=false` to turn it off, in which case OTel rejects
dictionary values as it did before.

## Plugins

The SDK loads extensions via [setuptools entry points](https://setuptools.pypa.io/en/latest/userguide/entry_point.html). Each plugin has a **name** (the entry-point key). Names are listed in config or environment variables; only installed plugins are loaded, in the order you configure.
Expand Down
8 changes: 6 additions & 2 deletions src/harness_sdk/agent_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
from harness_sdk import constants
from harness_sdk.config import config_pb2
from harness_sdk.env import is_env_flag_enabled
from harness_sdk.flatten_dict_registry import is_flatten_enabled
from harness_sdk.flatten_dict_span_processor import FlattenDictSpanProcessor
from harness_sdk.otlp_reporting import (
compression_type_to_otlp_grpc,
compression_type_to_otlp_http,
Expand Down Expand Up @@ -97,8 +99,10 @@ def register_processor(self, processor) -> None:
def set_console_span_processor(self) -> None:
console_span_exporter = ConsoleSpanExporter(
service_name=self._config.config.service_name)
simple_export_span_processor = SimpleSpanProcessor(console_span_exporter)
trace.get_tracer_provider().add_span_processor(simple_export_span_processor)
processor = SimpleSpanProcessor(console_span_exporter)
if is_flatten_enabled():
processor = FlattenDictSpanProcessor(processor)
trace.get_tracer_provider().add_span_processor(processor)

def init_exporter(self, trace_reporter_type):
exporter_type = ''
Expand Down
122 changes: 122 additions & 0 deletions src/harness_sdk/flatten_dict_registry.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
"""Thread-safe hand-off of dict span attributes from the hot path to span end.

``set_span_attribute("agent", {...})`` cannot go through
``span.set_attribute``: OTel rejects mapping values outright. Serializing on
the caller's thread would put JSON encoding on the application hot path, so
dict values are parked here untouched and flattened later by
``FlattenDictSpanProcessor.on_end``.

Entries are keyed by span context rather than object identity because
``Span.end()`` hands ``on_end`` a fresh ``ReadableSpan`` snapshot, not the
recording ``Span`` the enrichment helper saw.
"""
import threading
from collections import OrderedDict

from harness_sdk.custom_logger import get_custom_logger
from harness_sdk.env import get_env_value, is_env_flag_enabled

logger = get_custom_logger(__name__)

FLATTEN_ENABLED_ENV = "SPAN_ATTRIBUTE_FLATTEN_ENABLED"
FLATTEN_RAW_JSON_ENV = "SPAN_ATTRIBUTE_FLATTEN_RAW_JSON"
FLATTEN_MAX_DEPTH_ENV = "SPAN_ATTRIBUTE_FLATTEN_MAX_DEPTH"
FLATTEN_MAX_LEAVES_ENV = "SPAN_ATTRIBUTE_FLATTEN_MAX_LEAVES"

DEFAULT_MAX_DEPTH = 3
DEFAULT_MAX_LEAVES = 32

# Bound on spans holding pending dicts. A span that is never ended would
# otherwise leak its entry forever; oldest entries are evicted instead.
_MAX_TRACKED_SPANS = 2048


def is_flatten_enabled():
"""Dict flattening is on by default; only explicit ``false`` disables it."""
value = get_env_value(FLATTEN_ENABLED_ENV)
if value is None:
return True
return value.strip().lower() != "false"


def is_raw_json_enabled():
"""Opt in to additionally keeping the original key as a JSON string."""
return is_env_flag_enabled(FLATTEN_RAW_JSON_ENV)


def _positive_int_env(env_key, default):
raw = (get_env_value(env_key) or "").strip()
if not raw:
return default
try:
value = int(raw)
return value if value > 0 else default
except ValueError:
return default


def get_flatten_max_depth():
return _positive_int_env(FLATTEN_MAX_DEPTH_ENV, DEFAULT_MAX_DEPTH)


def get_flatten_max_leaves():
return _positive_int_env(FLATTEN_MAX_LEAVES_ENV, DEFAULT_MAX_LEAVES)


def _span_key(span):
get_context = getattr(span, "get_span_context", None)
if get_context is None:
return None
context = get_context()
if context is None or not context.trace_id:
return None
return (context.trace_id, context.span_id)


class FlattenDictRegistry:
"""Maps span identity to the dict attributes awaiting flattening."""

def __init__(self, max_tracked_spans=_MAX_TRACKED_SPANS):
self._lock = threading.Lock()
self._pending = OrderedDict()
self._max_tracked_spans = max_tracked_spans

def register(self, span, key, value):
"""Park ``value`` under ``key`` for ``span``; last write wins."""
span_key = _span_key(span)
if span_key is None:
return
with self._lock:
attributes = self._pending.get(span_key)
if attributes is None:
attributes = OrderedDict()
self._pending[span_key] = attributes
attributes[key] = value
self._pending.move_to_end(span_key)
while len(self._pending) > self._max_tracked_spans:
evicted, _ = self._pending.popitem(last=False)
logger.debug(
"Flatten: evicted pending dict attributes for span %s "
"(registry limit %s reached)",
evicted,
self._max_tracked_spans,
)

def pop(self, span):
"""Remove and return the pending dict attributes for ``span``."""
span_key = _span_key(span)
if span_key is None:
return {}
with self._lock:
return self._pending.pop(span_key, {})

def clear(self):
with self._lock:
self._pending.clear()


_REGISTRY = FlattenDictRegistry()


def get_registry():
return _REGISTRY
148 changes: 148 additions & 0 deletions src/harness_sdk/flatten_dict_span_processor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
"""Span processor that expands dict span attributes into dot-notation keys.

``set_span_attribute("agent", {"action": "generate"})`` parks the dict in
``flatten_dict_registry`` instead of handing it to OTel, which would reject it.
This processor drains the registry at ``on_end`` and writes
``agent.action=generate`` so the backend gets individually queryable
attributes instead of an opaque JSON blob.

It must be the outermost processor: downstream scrubbing and exclusion logic
matches on attribute keys, so the flattened keys have to exist before those
run. Mutating the ended span works the same way ``GenAiPayloadScrubSpanProcessor``
relies on: ``ReadableSpan.attributes`` is a read-only view over the attribute
store the concrete SDK ``Span`` still owns at ``on_end`` time.
"""
import json
from typing import Mapping

from opentelemetry.sdk.trace import SpanProcessor

from harness_sdk.custom_logger import get_custom_logger
from harness_sdk.flatten_dict_registry import (
get_registry,
get_flatten_max_depth,
get_flatten_max_leaves,
is_raw_json_enabled,
)

logger = get_custom_logger(__name__)

_SCALAR_TYPES = (bool, int, float, str)


def _is_scalar(value):
return isinstance(value, _SCALAR_TYPES)


def _scalar_kind(value):
# bool is a subclass of int, but OTel treats them as distinct array types.
if isinstance(value, bool):
return bool
if isinstance(value, int):
return int
if isinstance(value, float):
return float
return str


def _to_json(value):
try:
return json.dumps(value, default=str)
except (TypeError, ValueError):
return str(value)


def _sequence_leaf(value):
"""Homogeneous scalar sequences stay arrays; anything else becomes JSON."""
items = tuple(value)
if not items:
return items
kinds = {_scalar_kind(item) for item in items if _is_scalar(item)}
if len(kinds) == 1 and all(_is_scalar(item) for item in items):
return items
return _to_json(value)


def _leaf_value(value):
"""Convert a non-mapping value to something OTel accepts, or None to skip."""
if value is None:
return None
if _is_scalar(value):
return value
if isinstance(value, (list, tuple, set, frozenset)):
return _sequence_leaf(value)
return str(value)


def _collect(prefix, mapping, depth, flattened, max_depth, max_leaves):
"""Walk ``mapping`` into ``flattened``; returns False once the cap is hit."""
for key, value in mapping.items():
if len(flattened) >= max_leaves:
return False
flat_key = f"{prefix}.{key}"
if isinstance(value, Mapping):
if depth < max_depth:
if not _collect(flat_key, value, depth + 1, flattened, max_depth, max_leaves):
return False
else:
flattened[flat_key] = _to_json(value)
continue
leaf = _leaf_value(value)
if leaf is not None:
flattened[flat_key] = leaf
return True


class FlattenDictSpanProcessor(SpanProcessor):
"""Flattens registered dict attributes onto the span before export."""

def __init__(self, processor):
self._processor = processor

def on_start(self, span, parent_context=None):
self._processor.on_start(span, parent_context)

def on_end(self, span):
pending = get_registry().pop(span)
if pending:
try:
self._flatten(span, pending)
except Exception as err: # pylint: disable=W0703
logger.debug(
"Flatten: failed to flatten dict attributes on span %s: %s",
getattr(span, "name", None),
err,
)
self._processor.on_end(span)

@staticmethod
def _flatten(span, pending):
attributes = getattr(span, "_attributes", None)
if attributes is None:
return
raw_json = is_raw_json_enabled()
max_depth = get_flatten_max_depth()
max_leaves = get_flatten_max_leaves()
for root_key, value in pending.items():
flattened = {}
if not _collect(root_key, value, 1, flattened, max_depth, max_leaves):
logger.debug(
"Flatten: dict attribute %r on span %s exceeded %s leaf "
"attributes; remaining entries dropped",
root_key,
getattr(span, "name", None),
max_leaves,
)
for flat_key, leaf in flattened.items():
# An explicitly set attribute always wins over a flattened one.
if flat_key in attributes:
continue
attributes[flat_key] = leaf
if raw_json and root_key not in attributes:
attributes[root_key] = _to_json(value)

def force_flush(self, timeout_millis=30000):
return self._processor.force_flush(timeout_millis)

def shutdown(self):
return self._processor.shutdown()
12 changes: 9 additions & 3 deletions src/harness_sdk/plugins/builtin/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
from harness_sdk.env import is_env_flag_enabled
from harness_sdk.excluded_by_attribute_span_processor import ExcludeByAttributeSpanProcessor
from harness_sdk.db_control_span_processor import DbControlSpanProcessor
from harness_sdk.flatten_dict_registry import is_flatten_enabled
from harness_sdk.flatten_dict_span_processor import FlattenDictSpanProcessor
from harness_sdk.gen_ai_payload_scrub_span_processor import GenAiPayloadScrubSpanProcessor

logger = get_custom_logger(__name__)
Expand Down Expand Up @@ -43,10 +45,14 @@ def create_span_processors(self, config: Any) -> List[SpanProcessor]:
excluded_value="nospan",
)
db_control_processor = DbControlSpanProcessor(filter_processor)
# Outermost: scrub GenAI payload attributes before any other on_end
# logic (control evaluation, filtering, batching) sees the span.
# Scrub GenAI payload attributes before control evaluation, filtering
# and batching see the span.
scrub_processor = GenAiPayloadScrubSpanProcessor(db_control_processor)
return [scrub_processor]
if not is_flatten_enabled():
return [scrub_processor]
# Outermost: dict attributes must be expanded into their flat keys
# before scrubbing and exclusion match on attribute keys.
return [FlattenDictSpanProcessor(scrub_processor)]

def shutdown(self) -> None:
pass
Expand Down
Loading
Loading