Conversation
Adds application_sdk.observability.cgroup alongside resource_sampler. The existing sampler reads the process (/proc/self/stat RSS, getrusage CPU), which is right for App Vitals and wrong for sizing decisions: - RSS is not what the OOM killer acts on; memory.current is. - Start/end point samples miss the peak a tier has to cover. - CPU seconds cannot tell a cheap activity from a throttled one. Peak memory takes the cheapest instrument that works: the kernel's memory.peak watermark is reset on entry and read on exit (two file reads per activity), and a background poller is used only when that reset cannot be *proven* to have taken effect. Proof rather than kernel-version sniffing, because memory.peak is only writable from Linux 6.8 and a write can succeed as a no-op. Every reader returns None rather than raising or guessing, and None stays distinguishable from 0 — sizing on a silent 0 would pick the smallest tier for every activity. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…eptor Adds SizingTelemetryInterceptor, wired by create_worker and gated on APPLICATION_SDK_ENABLE_SIZING_TELEMETRY (default off), so a version bump alone changes nothing on any tenant. An interceptor rather than a decorator: a decorator has to be applied by every app author to every task method, and the teams that forget are exactly the ones with no sizing data — the dataset would be biased towards teams who already care about resource usage. create_worker already attaches SDK interceptors to every activity in every v3 app, so this is the only hook that is uniform by construction, and it needs nothing from the activity signature. A sibling of MetricsInterceptor, not an extension: this one is gated and that one is unconditional, and the App Vitals metrics path should not gain a background task and a set of cgroup reads as a side effect of a sizing rollout. Emits four OTel histograms (peak memory MiB and fraction, CPU throttled fraction, mean cores) plus one structured activity_sizing_observation log line per execution for offline tier fitting. Labels are bounded — no workflow_id. Nothing here reads or decides a tier: measurement that depended on routing would make the calibration circular. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds APPLICATION_SDK_SIZING_TELEMETRY_ACTIVITIES, a comma-separated list of activity names to measure. Empty collects nothing. Sizing data is only worth collecting for activities whose resource use varies with the data they process. Most activities are fixed-cost bookkeeping, and measuring them adds rows to the dataset the tier table is fitted from without adding information — so this is opt-in by name, not a default-on sweep. Empty is the fail-closed direction: a tenant that sets the enable flag and forgets the list gets no telemetry rather than telemetry on everything. Since that is silent, the worker warns at startup when it happens. '*' selects every activity, for a discovery pass on a test tenant. The filter runs before the tracker is constructed, so an unselected activity costs a set lookup — no cgroup reads, no poller. A test asserts on the tracker rather than on the record, because filtering later would also record nothing while still paying setup on every activity in the app. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A v3 activity registers with Temporal as "{app_name}:{task_name}", so
activity.info().activity_type is "automation-engine:merge" — but an app author
reading their own source sees '@task async def merge' and will write "merge".
Matching only the qualified form meant the obvious spelling silently collected
nothing: the config looks right, the worker logs the activities it is measuring,
and the dataset comes back empty. Both forms now match, and a qualified entry
still narrows to one app rather than matching the same task name elsewhere.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
📜 Docstring Coverage ReportRESULT: PASSED (minimum: 30.0%, actual: 79.0%) Detailed Coverage ReportThis message was truncated. Download full message |
Contributor
📦 Trivy Vulnerability Scan Results
Report Summary
Scan Result Detailspackages/conformance/uv.lockrequirements.txtuv.lock |
Contributor
📦 Trivy Secret Scan Results
Report Summary
Scan Result Detailspackages/conformance/uv.lockrequirements.txtuv.lock |
Contributor
☂️ Code Coverage
Overall Coverage
New Files
Modified Files
|
Same code, less prose. Keeps the load-bearing warnings as one-liners — proven reset vs version-sniffing, nulls not zeros, filter before the tracker, read the trace after the tracker exits — and drops the surrounding exposition. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…able Peak memory alone says a tier is wrong; it cannot say what to key the tier on. Adds input_bytes / input_file_count / input_basis to the observation, plus an activity.sizing.input_mib histogram and a peak_per_input_byte ratio. Two sources, in order: 1. FileReference fields on the Input — zero config, and measured rather than reported. Sized in the interceptor's finally, not at entry: the SDK materialises durable refs at the top of the activity, so by then the bytes are on local disk and this is a stat instead of an object-store call per activity. 2. A sizing_input_bytes() hook on the Input — the escape hatch for apps that pass raw object-store paths. 53 org repos use FileReference, but AE is not one of them: merge takes input_prefixes: list[str] and would otherwise report nothing, which is the flagship case. basis travels with the number because measured and self-reported bytes are not the same quantity, and fitting one rule to a silent mix of them fits it to neither. None means unknown, never 0 — a zero would fit a rule to inputs nobody sized. The directory walk is capped and reports truncation rather than passing a partial count off as complete. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Replaces the sizing_input_bytes() model hook with report_input_bytes(), and calls it from _download_files — the one function every SDK Parquet/JSON read goes through, covering both the local-hit and downloaded paths. The hook had no usable consumer. AE's merge takes input_prefixes: list[str], has no FileReference, and learns its byte counts inside the read loop rather than on the Input model — so the flagship case could not have used it. Instrumenting the reader instead means merge (and any app on those readers) contributes the driver variable with no app code at all. The collector is created by the interceptor and mutated in place, following OutputInterceptor's pattern: a ContextVar *set* inside the activity may not be visible to the interceptor across a thread or context boundary, whereas a shared object is. Reported bytes win over a FileReference walk — reported is what the activity read, a ref is only what it was handed. Note ParquetFileReader is deprecated in favour of FileReference, so the two sources cover today and where the SDK is heading. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…y pipeline Adds a durable per-execution record. The OTel histograms answer 'is this tier wrong'; fitting a rule that predicts a tier needs the rows, because a histogram bucket cannot give you peak_per_input_byte per execution. Rides AtlanObservability rather than adding a sink. That base already does batching, hive partitioning by year/month/day/hour, gzipped NDJSON, retention cleanup, and upload to the deployment store *and* the upstream Atlan store when ENABLE_ATLAN_UPLOAD is set. NDJSON rather than a bespoke Parquet writer for the same reason: ~400 lines of proven upload/partition/cleanup machinery already exists, DuckDB reads gzipped NDJSON natively, and diverging would put this signal somewhere no existing tooling looks. That upstream leg also largely answers cross-tenant curation — records from every tenant already land under one partitioned prefix — so app/deployment are stamped on each row: a row that cannot name its tenant cannot fit that tenant's tiers, and pooling tenants blindly would be wrong anyway since data volume is the thing being measured. Every row carries schema_version, because these are read months later mixed across SDK versions and 'which keys are present' is not a contract. Also fixes a latent trap the new signal exposed: LOCAL_OBS_SUBDIR_MAP and OBSERVABILITY_S3_PREFIX_MAP are separate maps, and a signal in only one writes to other/ on disk while uploading to sizing/. A test now asserts they agree. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Adds started_at, pod, concurrency_max and is_attributable, and forbids the memory.peak watermark reset whenever more than one activity is in flight. Schema version 2 — a v1 row cannot say whether its peak was pod-wide, so v1 and v2 must not be pooled. A cgroup reading is pod-wide, and worker concurrency defaults to 100. Rather than force concurrency to 1 — a throughput change on apps that have nothing to do with tiering — each row now records the maximum concurrency it saw, so the analysis can pick a model instead of pooling two: concurrency 1 fits per-activity, above 1 fits per-pod by joining rows that overlap on (pod, started_at, duration). That join is why the three fields are enough; no in-process bookkeeping of who-ran-with-whom is needed. Two correctness points found while building it: - The census counts EVERY activity, not just allow-listed ones. What invalidates attribution is another activity using the pod's memory, not whether we were measuring it; counting only the allow-list would let a measured merge sharing a pod with unmeasured work report concurrency_max=1. - leave() is idempotent and peak() reads without deregistering. Two callers release each execution, and an unconditional decrement undercounted concurrency for everything else still running. Per-thread CPU is deliberately NOT included: loop.run_in_executor sits in the innermost interceptor, so the whole interceptor chain runs on the event loop and time.thread_time() there measures the loop, not the activity. Attributing it would have been wrong in the direction that looks right. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ng dataset Two silent failures on the write path, both of which would have shown up only as an empty prefix after a week of collecting nothing. 1. ATLAN_ENABLE_OBSERVABILITY_STORE_SINK gates logs, metrics and sizing together, and falls back to ATLAN_ENABLE_OBSERVABILITY_DAPR_SINK. AE sets that to false to stop shipping logs and metrics, so it resolved to false and _flush_records returned early — no local file, no upload. _store_sink_enabled() is now an overridable hook and the sizing sink returns True: collection is already gated twice, by APPLICATION_SDK_ENABLE_SIZING_TELEMETRY and the per-activity allow-list, so nothing is written unless an operator asked for it by name. Other signals still respect the flag; a test pins that. 2. _flush_records partitions on record["timestamp"], which process_record never emitted — KeyError per batch, swallowed as best-effort telemetry. Now set from the execution's started_at, so a row lands in the hour the activity ran rather than the hour it was flushed. Both were found by one end-to-end flush test rather than by mocking the flush. It asserts on the upload, not on a leftover local file: the base uploads and then deletes the staged file, so a local-file check would only pass if the flush had crashed before cleanup. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A full day of collection on three tenants wrote nothing to the object store. The
log lines were there and the measurements were sound, but every buffered row died
with its pod.
add_record only evaluates its flush condition when a record ARRIVES:
len(buffer) >= batch_size or (now - last_flush) >= flush_interval
On this workload -- maxConcurrentActivities 1, a merge every 15-30 minutes, KEDA
scaling pods to zero in between -- a pod typically sees ONE record in its lifetime,
so neither branch is ever re-checked. AtlanTracesAdapter, AtlanLoggerAdapter and
AtlanMetricsAdapter all start _periodic_flush in __init__ for exactly this reason;
this sink inherited the base and skipped the one part of the pattern that makes the
interval real.
Three changes:
- start _periodic_flush, using the traces adaptor's loop-or-daemon-thread shape
- add drain(), for the gap between the last record and process exit
- promote the flush line to INFO. The base logs success at DEBUG, which every
deployment filters, so whether the sink was writing could only be settled by
exec-ing into a pod and forcing a flush by hand.
The new tests fail against the pre-fix __init__ and pass after it, verified by
reverting the fix and watching them go red -- a test for a silent bug that cannot
fail is worth nothing.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Changelog
Collects per-activity resource usage so worker tier envelopes can be re-derived from measured data instead of estimated. Collection only — nothing here reads or decides a tier. Ships off.
application_sdk/observability/cgroup.py(new) — container-level readers:memory.current/memory.max/memory.peakandcpu.stat(usage_usec,nr_periods,nr_throttled,throttled_usec), cgroup v2 with v1 fallbacks, plustrack_container_usage()as an async context manager.application_sdk/observability/sizing.py(new) —SizingObservationrecord, four OTel histograms (activity.sizing.peak_memory_mib/peak_memory_fraction/cpu_throttled_fraction/mean_cpu_cores), and one structuredactivity_sizing_observationlog line per execution.interceptors/sizing.py(new) —SizingTelemetryInterceptor, wired bycreate_worker, gated on an allow-list of activity names.execution/settings.py—APPLICATION_SDK_ENABLE_SIZING_TELEMETRY(defaultfalse),APPLICATION_SDK_SIZING_TELEMETRY_ACTIVITIES(default empty),APPLICATION_SDK_SIZING_TELEMETRY_POLL_SECONDS(default1.0).Why not extend
resource_samplerIt reads the process, which is the right instrument for App Vitals and the wrong one for sizing:
memory.current, which includes page cache and any child process. An activity reading a large Parquet file through the page cache is under-measured, in the direction that makes a too-small tier look safe.cpu.stat's throttling counters are the only signal that separates the two.Design notes for review
Peak memory takes the cheapest instrument that works.
memory.peakis reset on entry and read on exit (two file reads per activity, catches spikes of any duration); the background poller is used only when that reset cannot be proven. Proof rather than kernel-version sniffing —memory.peakis only writable from Linux 6.8, a write can succeed as a no-op, and the v1/v2 split makes any version table unreliable — so it reads the watermark back and requires it to have dropped. An unproven reset would report the pod's lifetime watermark as this activity's peak.peak_sourceis recorded on every observation because a watermark peak and a polled peak have different blind spots.A 1-second poll default is affordable because there is no RPC. Deliberately unlike AE's
report_memory_pressure, whose per-tickactivity.heartbeat()is a network call — that one is a safety device whose readings must reach the workflow; this one only has to reach the local process.An interceptor, not a decorator. A decorator has to be applied by every app author to every task method, and the teams that forget are exactly the ones with no sizing data — so the dataset would skew towards teams who already care about resource usage.
create_workeralready attaches SDK interceptors to every activity in every v3 app.A sibling of
MetricsInterceptor, not an extension. This one is gated and that one is unconditional; the App Vitals path should not gain a background task and a set of cgroup reads as a side effect of a sizing rollout. Not exported from theinterceptorspackage and not accepted viacreate_worker(interceptors=...), which makes double-registration — and therefore two pollers per activity — impossible by construction rather than by a guard list.Empty allow-list collects nothing. Sizing data is only worth collecting for activities whose resource use varies with the data they process; most activities are fixed-cost bookkeeping, and measuring them adds rows without adding information. Empty is also the fail-closed direction, so a tenant that sets the enable flag and forgets the list gets no telemetry rather than telemetry on everything. Since that is silent, the worker warns at startup.
"*"selects everything, for a discovery pass on a test tenant.The allow-list matches the bare task name as well as the qualified one. A v3 activity registers as
"{app_name}:{task_name}", soactivity_typeis"automation-engine:merge"— but an author reading@task async def mergewill writemerge. Matching only the qualified form silently collected nothing: config looks right, worker logs the activities it is measuring, dataset comes back empty. A qualified entry still narrows to one app.Nulls stay distinguishable from zeros. Every reader returns
Nonerather than raising or guessing, and an observation with nothing measured is dropped rather than emitted — a null read downstream as a zero would fit the smallest tier to an activity nobody measured.Telemetry never fails the activity it measures. The setup block, the poller-cancel path and the finalisation are each guarded; tests pin all three, including that the block's own exception still propagates.
Additional context
6cf5f0bc). AE builds its own worker and so owns its own interceptor list, meaning SDK interceptors reach the whole fleet except AE — it needs the interceptor attached by hand. That change is inert until this ships, since AE pins>=3.15.1and 3.15.1 has neither the interceptor nor the settings fields.input_bytesdriver variable, a durable columnar sink (the structured log line is the only sink today), cross-tenant curation, and the analysis that emits the tier table.Checklist
Warning
Not verified on a real cgroup. Everything is tested against fake hierarchies on disk (macOS has no cgroup). Which peak mode actually engages on a tenant —
watermarkorpoll— depends on the node kernel, andmemory.peakis only resettable from 6.8, sopollis the likely path on current GKE/EKS nodes.peak_sourceis on every observation so this is checkable the moment collection is switched on. This is the main reason the PR is a draft.Note
Conformance. 9 findings remain in the new files (broad-except guards,
contextlib.suppress, loop-swallowedOSError) — the same intentional-and-documented class as the existing observability interceptors, which report the same codes despite carrying# conformance: ignore[...]comments (those are advisory, not machine-enforced). Repo total went 523 → 519, because conformance caught four real issues in this code: an assign-onlyexcept,json→orjson, and a missingexc_info=True.Copyleft License Compliance
🤖 Generated with Claude Code