stormlog.telemetry_rollups

Derived compact rollups for append-only telemetry sinks.

Functions

`build_telemetry_rollups`(sessions, manifest, *)	Build compact rollups from loaded telemetry sessions.
`read_telemetry_rollups`(path)	Read a rollup sidecar if it exists and has a valid current shape.
`resolve_telemetry_rollup_path`(path)	Resolve a sink directory, manifest, segment, or rollup path to rollups.json.
`rollup_coverage_from_manifest`(manifest, *[, ...])	Build sidecar coverage fields from a sink manifest.
`telemetry_rollup_file_from_dict`(payload)	Deserialize a rollup payload from JSON.
`telemetry_rollup_file_to_dict`(rollups)	Serialize a rollup payload into a deterministic JSON-safe dictionary.
`write_telemetry_rollups`(root_dir, rollups)	Write rollups atomically next to a sink manifest.

Classes

`AlertSummary`([total_count, severity_counts, ...])	Alert counts by severity and event type.
`CollectorHealthSummary`([transition_count, ...])	Collector transition counts and degraded duration estimate.
`CounterSummary`([allocator_allocated_bytes, ...])	Peak memory counters for a session, rank, or window.
`OomSummary`([marker_count, event_count, ...])	OOM marker counts.
`PeakValue`([value, timestamp_ns, rank, device_id])	Peak value and source timestamp for a counter.
`RankRollup`(rank, local_rank, world_size, ...)	Per-rank compact summary.
`RollupCoverage`([retained_segment_filenames, ...])	Retained raw evidence covered by this rollup file.
`SessionRollup`(session, event_count, ...[, ...])	Compact derived summary for one telemetry session.
`TelemetryRollupFile`(schema_version, format, ...)	Top-level rollup sidecar payload.
`WindowRollup`(index, start_ns, end_ns, ...[, ...])	Fixed-duration session window summary.

class stormlog.telemetry_rollups.AlertSummary(total_count=0, severity_counts=<factory>, event_type_counts=<factory>)[source]

Bases: object

Alert counts by severity and event type.

Parameters:

total_count (int)
severity_counts (dict[str, int])
event_type_counts (dict[str, int])

total_count: int = 0

severity_counts: dict[str, int]

event_type_counts: dict[str, int]

class stormlog.telemetry_rollups.CollectorHealthSummary(transition_count=0, degraded_count=0, recovered_count=0, degraded_time_ns=0, last_status=None)[source]

Bases: object

Collector transition counts and degraded duration estimate.

Parameters:

transition_count (int)
degraded_count (int)
recovered_count (int)
degraded_time_ns (int)
last_status (str | None)

transition_count: int = 0

degraded_count: int = 0

recovered_count: int = 0

degraded_time_ns: int = 0

last_status: str | None = None

class stormlog.telemetry_rollups.CounterSummary(allocator_allocated_bytes=<factory>, allocator_reserved_bytes=<factory>, device_used_bytes=<factory>)[source]

Bases: object

Peak memory counters for a session, rank, or window.

Parameters:

allocator_allocated_bytes (PeakValue)
allocator_reserved_bytes (PeakValue)
device_used_bytes (PeakValue)

allocator_allocated_bytes: PeakValue

allocator_reserved_bytes: PeakValue

device_used_bytes: PeakValue

class stormlog.telemetry_rollups.OomSummary(marker_count=0, event_count=0, bundle_path_count=0)[source]

Bases: object

OOM marker counts.

Parameters:

marker_count (int)
event_count (int)
bundle_path_count (int)

marker_count: int = 0

event_count: int = 0

bundle_path_count: int = 0

class stormlog.telemetry_rollups.PeakValue(value=None, timestamp_ns=None, rank=None, device_id=None)[source]

Bases: object

Peak value and source timestamp for a counter.

Parameters:

value (int | None)
timestamp_ns (int | None)
rank (int | None)
device_id (int | None)

value: int | None = None

timestamp_ns: int | None = None

rank: int | None = None

device_id: int | None = None

class stormlog.telemetry_rollups.RankRollup(rank, local_rank, world_size, event_count, sample_count, first_timestamp_ns, last_timestamp_ns, counters, hidden_gap_first_bytes=None, hidden_gap_latest_bytes=None, hidden_gap_peak_bytes=None, hidden_gap_delta_bytes=None, hidden_gap_drift_bytes_per_second=None)[source]

Bases: object

Per-rank compact summary.

Parameters:

rank (int)
local_rank (int)
world_size (int)
event_count (int)
sample_count (int)
first_timestamp_ns (int | None)
last_timestamp_ns (int | None)
counters (CounterSummary)
hidden_gap_first_bytes (int | None)
hidden_gap_latest_bytes (int | None)
hidden_gap_peak_bytes (int | None)
hidden_gap_delta_bytes (int | None)
hidden_gap_drift_bytes_per_second (float | None)

rank: int

local_rank: int

world_size: int

event_count: int

sample_count: int

first_timestamp_ns: int | None

last_timestamp_ns: int | None

counters: CounterSummary

hidden_gap_first_bytes: int | None = None

hidden_gap_latest_bytes: int | None = None

hidden_gap_peak_bytes: int | None = None

hidden_gap_delta_bytes: int | None = None

hidden_gap_drift_bytes_per_second: float | None = None

class stormlog.telemetry_rollups.RollupCoverage(retained_segment_filenames=<factory>, retained_segment_count=0, retained_event_count=0, retained_bytes=0, pruned_segment_count=None, pruned_bytes=None)[source]

Bases: object

Retained raw evidence covered by this rollup file.

Parameters:

retained_segment_filenames (list[str])
retained_segment_count (int)
retained_event_count (int)
retained_bytes (int)
pruned_segment_count (int | None)
pruned_bytes (int | None)

retained_segment_filenames: list[str]

retained_segment_count: int = 0

retained_event_count: int = 0

retained_bytes: int = 0

pruned_segment_count: int | None = None

pruned_bytes: int | None = None

class stormlog.telemetry_rollups.SessionRollup(session, event_count, sample_count, first_timestamp_ns, last_timestamp_ns, source_count, sources_loaded, warnings, counters, alerts, collector_health, oom, ranks=<factory>, windows=<factory>)[source]

Bases: object

Compact derived summary for one telemetry session.

Parameters:

session (SessionSummary)
event_count (int)
sample_count (int)
first_timestamp_ns (int | None)
last_timestamp_ns (int | None)
source_count (int)
sources_loaded (list[str])
warnings (list[str])
counters (CounterSummary)
alerts (AlertSummary)
collector_health (CollectorHealthSummary)
oom (OomSummary)
ranks (list[RankRollup])
windows (list[WindowRollup])

session: SessionSummary

event_count: int

sample_count: int

first_timestamp_ns: int | None

last_timestamp_ns: int | None

source_count: int

sources_loaded: list[str]

warnings: list[str]

counters: CounterSummary

alerts: AlertSummary

collector_health: CollectorHealthSummary

oom: OomSummary

ranks: list[RankRollup]

windows: list[WindowRollup]

class stormlog.telemetry_rollups.TelemetryRollupFile(schema_version, format, generated_at_ns, source_manifest, source_manifest_schema_version, window_duration_ns, coverage, sessions=<factory>)[source]

Bases: object

Top-level rollup sidecar payload.

Parameters:

schema_version (int)
format (str)
generated_at_ns (int)
source_manifest (str | None)
source_manifest_schema_version (int | None)
window_duration_ns (int)
coverage (RollupCoverage)
sessions (list[SessionRollup])

schema_version: int

format: str

generated_at_ns: int

source_manifest: str | None

source_manifest_schema_version: int | None

window_duration_ns: int

coverage: RollupCoverage

sessions: list[SessionRollup]

class stormlog.telemetry_rollups.WindowRollup(index, start_ns, end_ns, event_count, sample_count, rank_count, counters, alert_count=0, collector_transition_count=0, oom_count=0)[source]

Bases: object

Fixed-duration session window summary.

Parameters:

index (int)
start_ns (int)
end_ns (int)
event_count (int)
sample_count (int)
rank_count (int)
counters (CounterSummary)
alert_count (int)
collector_transition_count (int)
oom_count (int)

index: int

start_ns: int

end_ns: int

event_count: int

sample_count: int

rank_count: int

counters: CounterSummary

alert_count: int = 0

collector_transition_count: int = 0

oom_count: int = 0

stormlog.telemetry_rollups.build_telemetry_rollups(sessions, manifest, *, window_duration_ns=60000000000, coverage=None)[source]

Build compact rollups from loaded telemetry sessions.

Parameters:

sessions (Sequence[LoadedTelemetrySession])
manifest (TelemetrySinkManifest | None)
window_duration_ns (int)
coverage (RollupCoverage | None)

Return type:

TelemetryRollupFile

stormlog.telemetry_rollups.read_telemetry_rollups(path)[source]

Read a rollup sidecar if it exists and has a valid current shape.

Parameters:: path (str | Path)
Return type:: TelemetryRollupFile | None

stormlog.telemetry_rollups.resolve_telemetry_rollup_path(path)[source]

Resolve a sink directory, manifest, segment, or rollup path to rollups.json.

Parameters:: path (str | Path)
Return type:: Path

stormlog.telemetry_rollups.rollup_coverage_from_manifest(manifest, *, pruned_segment_count=None, pruned_bytes=None)[source]

Build sidecar coverage fields from a sink manifest.

Parameters:

manifest (TelemetrySinkManifest | None)
pruned_segment_count (int | None)
pruned_bytes (int | None)

Return type:

RollupCoverage

stormlog.telemetry_rollups.telemetry_rollup_file_from_dict(payload)[source]

Deserialize a rollup payload from JSON.

Parameters:: payload (Mapping[str, Any])
Return type:: TelemetryRollupFile

stormlog.telemetry_rollups.telemetry_rollup_file_to_dict(rollups)[source]

Serialize a rollup payload into a deterministic JSON-safe dictionary.

Parameters:: rollups (TelemetryRollupFile)
Return type:: dict[str, Any]

stormlog.telemetry_rollups.write_telemetry_rollups(root_dir, rollups)[source]

Write rollups atomically next to a sink manifest.

Parameters:

root_dir (str | Path)
rollups (TelemetryRollupFile)

Return type:

Path