/ingestion/docs/design/ingestion-diagnostics.md

Ingestion Runtime Diagnostics

Status: Proposed Author: Sriharsha Chintalapani Date: 2026-05-15

---

TL;DR

OpenMetadata ingestion processes fail in production in two main ways: they hang (often silently) and they OOM. Today we have no built-in way to answer "what was it doing right before it died?", so we ship fixes based on hypotheses and hope. This doc proposes a small, always-available diagnostics subsystem inside the ingestion framework that — when an operator enables DEBUG logging on the workflow — captures enough runtime state (current operation, in-flight HTTP, memory growth, stage backpressure) that the root cause is in the logs every time. The cost when off: zero. The cost when on: ~500 KB and < 0.01% CPU.

The doc is organized so each component below can ship as an independent PR.

---

1. Why we need this — case studies from production

Case study 1: the "Snowflake hang" that was actually a logging recursion

Symptom (2026-05). Snowflake ingestion pods on a customer cluster were hanging indefinitely. From outside the pod:

• kubectl logs returned nothing recent.
• The OpenMetadata server saw no incoming requests from the connector.
• No queries appeared in Snowflake QUERY_HISTORY for the stuck window.
• The pod's CPU was pegged.

Hypothesis-driven fixes that did not help.

1. "HTTP pool is stale, no timeouts." Shipped PR #28131 to add request timeouts. Hang recurred.
2. "Snowflake driver hang." Considered OCSP, chunk-download, network. Could not falsify without data.
3. "Server-side backpressure." Plausible — but kubectl logs showed no inbound requests on the server, so the connector wasn't even reaching it.

What it actually was. A py-spy dump (which itself required kubectl debug --profile=sysadmin with SYS_PTRACE — a 30-minute setup) revealed MainThread sitting ~977 frames deep in StreamableLogHandler.emit(), alternating between logger.warning("queue is full") and logger.error("error in emit"). The handler was attached to the same logger it was calling, creating infinite recursion as soon as the bounded log queue filled. Every other thread — including the log-shipper itself — was blocked on the Python logging module's per-handler RLock held by MainThread. No HTTP request was in flight. No timeout could have helped.

The fix: route in-handler diagnostics through a separate non-propagating logger (PR #28160).

Time to root cause: ~6 hours of investigation, plus one shipped fix that addressed an entirely different theory. Time it should have taken: seconds, if the process had logged "MainThread stuck for 312 s in streamable_logger.emit" on its own.

Case study 2: connector OOMs without a clear cause

Symptom (ongoing). Several connectors — Snowflake, Postgres on large databases, S3-based connectors — die with OOMKill on Kubernetes. The pod restarts. The next run sometimes succeeds, sometimes OOMs again.

Why we cannot diagnose today. When the OOMKill fires:

• We get an exit code (137), no Python traceback, no last-state log.
• We have no idea which stage was running.
• We have no idea what type of object was occupying memory.
• We have no idea whether memory was growing slowly (leak) or spiked at a specific entity (one bad row).

Suspected contributors, none confirmed.

• Entity backlog: when one stage stalls, upstream stages keep producing and accumulate Pydantic entity objects.
• PyArrow ResultBatch objects held by reference longer than necessary.
• Unbounded internal caches in connector base classes (lineage cache, schema cache).
• Log queue with very large entries (multi-MB stack traces, formatted entity dumps).
• HTTP request bodies buffered for shipping batches.

Each of these is a plausible OOM source. Without per-stage memory growth tracking + a snapshot at OOM time, every fix is a guess. We have already shipped PRs to one or two of these on the suspicion that they were "the" cause.

Case study 3: silent slowness (no logs for hours)

Symptom. Long-running ingestion (DBT, lineage) shows no progress in logs for hours, then either completes or is killed by the workflow timeout. Customer asks: "is it stuck, or just slow?"

Today: we cannot answer. We tell them to wait, or kill and rerun.

What we need: every 30 seconds, a single structured log line saying exactly what entity is being processed and how long has been spent on it. The customer (and we) instantly know "it's making progress, table 4732 of 9000" vs "it hasn't moved in 4 hours, kill it."

---

2. The principle: stop guessing, start measuring

These cases share a single root cause: we ask the process to fail silently, then try to reconstruct what happened after the fact. The instrumentation we keep needing is the same every time:

• What thread was doing what, just before the failure?
• Was memory growing? At what rate? On which stage?
• What HTTP requests were in flight?
• What was the queue depth between stages?
• What was the last entity successfully yielded by the source / written by the sink?

This information is cheap to capture if we capture it as the process runs, instead of trying to recover it from a dead pod. The cost of always-on capture is tiny; the cost of ad-hoc post-mortem debugging is enormous. This doc designs the always-on capture layer.

---

3. Scope

Goals

• Every ingestion process running with loggerLevel == DEBUG produces a continuous, low-volume stream of structured diagnostic output (heartbeats, watchdog warnings) and supports on-demand dumps (signal-triggered).
• The dump from a hung or about-to-OOM process is enough to identify the stage, entity, operation, HTTP request, memory consumer, and stack involved.
• All of this works with no external tooling: no py-spy, no kubectl debug, no ptrace. The information is in kubectl logs.
• Zero overhead when off. Single existing knob (loggerLevel).

Non-goals

• A general-purpose APM/metrics system.
• Profiling for throughput optimization.
• Replacing the existing workflow status counters.
• Adding any new ingestion config field or env var.
• Catching every possible failure. We optimize for the common failure modes seen in production (hangs, OOMs, slowness).

What this is NOT

This is not a substitute for fixing bugs. The point is to identify which bug to fix. After this lands, when a customer reports a hang or OOM, we expect:

1. Look at kubectl logs.
2. Identify the operation/entity/memory pattern.
3. Open a focused bug fix.

…instead of "let's try adding timeouts and see if it helps."

---

4. Activation

Single knob: workflowConfig.loggerLevel.

loggerLevel	Diagnostics module behavior
DEBUG	Fully installed: signals, watchdog, heartbeat, memory tracker, HTTP introspection. Output to stderr + structured logs.
INFO (or anything else)	Module is dead code. operation() is a no-op contextmanager. No threads. No signal handlers. Memory cost ≈ 0.

We deliberately avoid new env vars or new config fields. Diagnostic toggle reuses an existing UI-controllable setting.

Trade-off: first occurrence of a non-reproducing hang on INFO is undiagnosed. Mitigation: flip log level → re-run → captured. One cycle on first occurrence, never afterward. Worth it to keep production runs strictly free of overhead.

---

5. Architecture

Five components, all under metadata/ingestion/diagnostics/. A single diagnostics.install(workflow) call in BaseWorkflow.execute() activates everything when loggerLevel == DEBUG.

5.1 Operation registry (registry.py)

Per-thread stack of "what am I doing right now":

from metadata.ingestion.diagnostics import operation

with operation("source.iter", entity_fqn=fqn):
    with operation("snowflake.query", sql=stmt[:500], query_id=qid):
        cursor.execute(stmt)
    with operation("sink.write", entity_type="table"):
        ometa.create_or_update(table)

State: {thread_id: [(name, kwargs, started_monotonic), ...]}.

The registry exposes snapshot() which, for each live thread, returns the stack with per-frame durations. This is the ground truth used by every other component (heartbeat, watchdog, signal dump).

Properties:

• Thread-safe via a single fine-grained lock.
• O(1) push/pop per operation() enter/exit.
• Kwargs are truncated to 2000 chars at registration (a 10 MB DDL string would stay referenced otherwise).
• Stack depth capped at 20 to guard against runaway nesting.
• The watchdog garbage-collects entries for dead thread.idents on each tick.

Cost: ~1 µs per enter/exit; ~15 KB at peak.

5.2 Signal handlers (signals.py)

On install, registers SIGUSR1 and SIGUSR2 plus faulthandler.register. On signal:

Signal	What gets dumped
SIGUSR1	Full: faulthandler.dump_traceback(all_threads=True) + operation registry snapshot + in-flight HTTP requests + memory state + workflow status
SIGUSR2	Incremental: registry + HTTP + memory only. Cheap. For periodic polling.

Operator workflow when something looks stuck:

kubectl exec <pod> -- bash -c 'pkill -SIGUSR1 -f "python.*main.py"'
kubectl logs <pod> --tail=500

No py-spy, no kubectl debug, no ptrace. Works on any pod, any cluster, any time.

faulthandler is stdlib; it captures both Python and native (C extension) frames, runs from a signal handler safely, and writes synchronously.

5.3 Watchdog (watchdog.py)

A daemon thread that wakes every 10 s. For each live thread:

• Look up the deepest active operation in the registry.
• If it has been on the same operation for > 60 s → log a structured warning line.
• If it has been on the same operation for > 300 s → trigger a full SIGUSR1-style dump.

Re-dump throttle: at most one full dump per (thread, operation_name) per 5 minutes.

This is the behavioral change that changes everything: the process logs its own hangs. No human has to be watching the pod for the data to be captured.

5.4 Heartbeat (heartbeat.py)

A daemon thread that emits one structured log line every 30 s while the workflow is running:

diag.heartbeat stage=source step=table progress=4732/9000 current_op=source.iter
  current_fqn=svc.db.schema.table4732 current_op_age=8s rss=412M rss_delta_30s=+2.1M
  threads=8 active_http=1 queue_source_to_sink=240/1000

This single line answers, for the operator looking at kubectl logs:

• Am I making progress? (compare two heartbeats)
• What stage am I on?
• What entity am I currently processing?
• Is memory growing? At what rate?
• Is one stage backed up behind another?

Output cadence is tunable as a constant; 30 s is a sensible default for ingestion that typically runs minutes to hours.

5.5 HTTP introspection (http_introspect.py)

Wraps the two HTTP surfaces we own:

Surface	Wrap point	What we record
OMetaClient	REST._request()	method, URL, started_at, request_id in an active-requests dict, plus an operation("ometa.http", ...) registry entry
Snowflake (and other DB connectors)	cursor.execute()	SQL (truncated), query_id, started_at, plus an operation("snowflake.query", ...) registry entry

Active-requests dict is keyed by (thread_id, request_id), entries removed in finally. On dump, rendered as a table:

diag.dump.http
  thread=log-shipper-xyz method=POST url=/api/v1/.../logs age=37s
  thread=MainThread       method=PUT  url=/api/v1/tables    age=2s

For surfaces we don't own (Snowflake's OCSP validator, chunk downloader), we rely on the faulthandler thread dump (native + Python frames) to identify the hang location. Those frames have characteristic library paths that are easy to recognize.

5.6 Memory tracker (memory.py) — NEW, motivated by case study 2

A daemon thread that samples memory every 30 s and maintains a small ring buffer (last ~10 samples). It records:

Metric	Source
rss	psutil.Process().memory_info().rss
rss_growth_per_sec	derivative over the ring buffer
cgroup_current / cgroup_max	/sys/fs/cgroup/memory.{current,max}
cgroup_oom_kill_count	/sys/fs/cgroup/memory.events oom_kill field
gc_collections	gc.get_stats() per generation
top_object_types	gc.get_objects() aggregated by type(obj).__name__, top 10 by count

On heartbeat, only rss and rss_delta_30s are emitted. On full dump (SIGUSR1 or watchdog auto-dump), all of the above are emitted including the top-10 object types.

The "top object types" is the OOM debugging breakthrough. Today we can't see which objects are growing. With gc.get_objects() aggregated into a Counter, a dump from a process that's growing toward OOM looks like:

diag.dump.memory
  rss=2841M rss_delta_30s=+45M cgroup=2841M/3072M oom_kills=0
  gc_gen0=23 gc_gen1=4 gc_gen2=1
  top_types:
    Table              182341
    Column             1843219
    Tag                94823
    LineageEdge        451233
    PyArrowResultBatch 47
    SnowflakeResultBatch 12

Now we know exactly what's accumulating. If Table is in the hundreds of thousands and growing per heartbeat, the sink is the bottleneck. If PyArrowResultBatch is growing, the source is holding references. We move from "OOM, unknown why" to "OOM, sink starved, here's the entity backlog."

Cost. gc.get_objects() is the one expensive call (it iterates all tracked objects). It's invoked only on dump (SIGUSR1 / watchdog auto-dump), not on heartbeat. On heartbeat we only sample rss. So the steady-state cost is psutil.memory_info() + arithmetic — sub-millisecond.

5.7 Stage backpressure visibility (stage_progress.py) — NEW

The topology runner moves entities through source → processor → sink stages via internal buffers. Today these buffers are invisible. A common OOM pattern: source produces faster than sink can drain → entities pile up in the in-memory buffer → OOM.

This component adds two small numbers to the heartbeat: queue depth and recent transition rate per inter-stage boundary. Implemented by hooking the topology runner's stage transitions to a tiny counter, not by changing the runner's logic.

Heartbeat then shows:

diag.heartbeat ... stage_queues=source→processor:14/100 processor→sink:240/1000

If processor→sink is at capacity for many heartbeats, the sink is starving — and you know to look at the OpenMetadata server / sink-side HTTP, not the source.

---

6. Output format (stable, grep-friendly)

All diagnostic output uses a diag. prefix:

diag.heartbeat    stage=... rss=... queue=...
diag.warn.stuck   thread=MainThread op=snowflake.query duration=72s kwargs={...}
diag.dump.begin   reason=watchdog trigger_op=snowflake.query duration=312s
diag.dump.threads ...
diag.dump.ops     ...
diag.dump.http    ...
diag.dump.memory  ...
diag.dump.queues  ...
diag.dump.end

Filter with grep '^diag\.'. Parse with any structured log tool. Same prefix appears in stderr-only output from internal loggers, so the operator sees everything together.

---

7. Wire-in points (existing framework code that needs to call operation())

File	Change	Why
metadata/workflow/base_workflow.py	Call diagnostics.install(self) if loggerLevel == DEBUG	Activation point
metadata/ingestion/api/steps.py	Wrap Source.iter() / Processor.run() / Sink.write_record() in operation(...)	The three stages
metadata/ingestion/topology_runner.py	Notify stage_progress on transitions	Backpressure visibility
metadata/ingestion/ometa/client.py	Wrap REST._request() in operation + active-HTTP register	OMetaClient introspection
metadata/ingestion/source/database/common_db_source.py (and DB connectors)	Wrap cursor.execute() in operation	DB query introspection

Connector authors writing new connectors do not need to know any of this. They get instrumentation for free at the framework seams.

---

8. Defaults (hardcoded; not user-configurable)

Setting	Value	Rationale
Activation	loggerLevel == DEBUG	Reuse existing config
Watchdog tick	10 s	Responsive, cheap
Stuck-warning threshold	60 s	Most legit ops < this
Auto-dump threshold	300 s	True hang
Re-dump throttle	5 min per (thread, op)	Prevent flooding
Heartbeat cadence	30 s	Useful, low-noise
Memory sample interval	30 s (heartbeat)	Cheap
Memory deep-snapshot	only on dump	gc.get_objects() is the only expensive call
Kwargs truncation	2000 chars	Captures normal SQL
Op stack depth cap	20	Defensive
Output target	stderr (also routed to /tmp/openmetadata-diag-<pid>.log)	Survives log rotation

Constants live in diagnostics/__init__.py. If reality shows them wrong, we change them in code — not config.

---

9. Memory and CPU budget

Off (INFO mode, every production run today): zero. Dead code.

On (DEBUG mode):

Component	Memory	CPU
Registry	~15 KB	~1 µs / op enter+exit
Active HTTP tracker	~5 KB	~1 µs / request
Watchdog thread	~100 KB RSS	~10 µs / 10 s
Heartbeat thread	~100 KB RSS	~100 µs / 30 s
Memory tracker	~10 KB ring buffer	~1 ms / 30 s (sample only)
Memory deep-snapshot	0 ongoing	~50–500 ms on each dump (rare)
Stage progress	~5 KB	~1 µs / transition
faulthandler buffer	~8 KB	0 until SIGUSR1
Total	~250 KB	< 0.01% CPU

The deep memory snapshot (gc.get_objects()) is the only expensive call and only fires on actual dump events.

---

10. Implementation plan — six small PRs

Each PR is independently reviewable, independently shippable, and each one improves visibility on its own.

PR 1 — Foundation (~350 LoC)

• diagnostics/__init__.py with public API (install, operation, is_active, dump)
• diagnostics/registry.py
• diagnostics/signals.py (SIGUSR1 / SIGUSR2 + faulthandler)
• Wire diagnostics.install() into BaseWorkflow.execute()
• Wrap Source.iter() and Sink.write_record() in operation()
• Wrap OMetaClient._request() + active-HTTP register
• Unit tests: registry, signal-triggered dump, no-op fallback when not installed

Ships value: kill -SIGUSR1 works for any DEBUG-enabled run. We can dump state on demand.

PR 2 — Watchdog + heartbeat (~250 LoC)

• diagnostics/watchdog.py with 60 s warn / 300 s auto-dump
• diagnostics/heartbeat.py with 30 s cadence
• Integration test: install diagnostics, sleep in an operation(), assert watchdog logs a stuck line and auto-dumps

Ships value: The process auto-detects its own hangs. Heartbeat confirms liveness.

PR 3 — Memory tracker (~200 LoC)

• diagnostics/memory.py: RSS + cgroup sampling on heartbeat, gc.get_objects() top-types on dump
• Heartbeat output gains rss= and rss_delta_30s= fields
• Dump output gains diag.dump.memory section
• Tests: simulate growth, assert top types are reported correctly

Ships value: OOMs become diagnosable. We see which object type is growing.

PR 4 — Stage backpressure visibility (~150 LoC)

• diagnostics/stage_progress.py: small counter hooked into topology runner
• Heartbeat output gains stage_queues= field
• Test: induce a slow sink, assert queue depth shows up in heartbeat

Ships value: We can tell whether the source, processor, or sink is the bottleneck.

PR 5 — DB connector instrumentation (~200 LoC)

• Wrap cursor.execute() in common_db_source.py so every SQL query is in the operation registry with query text + query_id
• Same pattern can be replicated per-connector when there's a need

Ships value: "Stuck on a SQL query" becomes a labeled, visible op.

PR 6 — Disk-persisted dumps (~50 LoC)

• Mirror SIGUSR1 dumps to /tmp/openmetadata-diag-<pid>.log so they survive container restarts and log rotation

Ships value: Post-mortem diagnosis works even if kubectl logs rolled over.

After PR 1 alone we can already diagnose the Snowflake-style hang on the next DEBUG run. After PR 2 we no longer need to be watching the pod. After PR 3 we can diagnose OOMs. PRs 4–6 are continuous improvement.

---

11. Worked examples — what would these PRs have told us?

The streamable_logger hang

Before: 6 hours of py-spy + manual analysis + one wrong-theory fix.

With this work installed at the time:

• Watchdog at 60 s: diag.warn.stuck thread=MainThread op=source.iter entity_fqn=svc.db.schema.table42 duration=72s — already pointing at a specific table.
• Heartbeat at 30 s: would stop firing because MainThread is wedged in the recursion (still useful: the last heartbeat tells us where it died).
• Watchdog auto-dump at 300 s: thread stack via faulthandler shows the same 977-deep emit -> warning -> emit chain, directly fingering streamable_logger.py as the culprit. Total time to root cause: ~5 minutes from log readout.

A future OOM (Case study 2 generalized)

Heartbeat history in kubectl logs:

T+0:00  diag.heartbeat ... rss=400M
T+5:00  diag.heartbeat ... rss=900M  rss_delta_30s=+50M  stage_queues=source→sink:1000/1000
T+8:00  diag.heartbeat ... rss=1.8G  rss_delta_30s=+60M  stage_queues=source→sink:1000/1000

The watchdog or the operator's SIGUSR1 gives:

diag.dump.memory
  rss=2.6G rss_delta_30s=+50M
  top_types:
    Table       423019
    Column      4109832
    LineageEdge 8431

Verdict: sink is at capacity, source keeps producing Table and Column objects, those accumulate. The problem is sink-side throughput, not the source. We open a focused PR on the sink — instead of three speculative ones across the source.

A "is it stuck or slow?" question

Customer reports an ingestion running for 4 hours with no logs. We say: kubectl logs | grep diag.heartbeat | tail -5. They paste:

T+3:50:00 diag.heartbeat ... stage=source step=table progress=4730/9000 rss=412M
T+3:50:30 diag.heartbeat ... stage=source step=table progress=4732/9000 rss=413M
T+3:51:00 diag.heartbeat ... stage=source step=table progress=4734/9000 rss=414M

It's making progress — slowly. We point them at the source's per-entity duration to identify which kind of table is slow. No "kill and rerun" needed.

---

12. Tradeoffs accepted

1. First occurrence of a non-reproducing hang on INFO will be undiagnosed. Mitigated by flipping the workflow's log level and re-running. One redeploy cycle on first occurrence, never afterward.
2. DEBUG runs emit more output (heartbeat every 30 s + HTTP per-request lines). Acceptable — DEBUG is opt-in for investigation.
3. No external knobs for thresholds. If 60 s / 300 s / 30 s prove wrong, we change them in code. We are explicitly trading user-tunability for fewer config knobs.
4. Native code in C extensions is captured only via faulthandler thread dumps, not the operation registry. Sufficient for identifying where a hang is; not for labeling what the C code is doing semantically.
5. gc.get_objects() is moderately expensive (10s of ms on processes with millions of objects). We only call it on dump, not heartbeat.

---

13. Open questions

1. Argo emissary at PID 1. In Argo workflow pods, the Python process is not PID 1. kill -SIGUSR1 1 would signal the emissary, not Python. Use pkill -SIGUSR1 -f python.*main.py instead. Document this in operator-facing notes.
2. Should heartbeat be at INFO level (visible by default) or DEBUG-only? Leaning toward "INFO when diagnostics is installed" — once an operator opted into DEBUG to install us, heartbeats are exactly what they want to see.
3. Should the operation registry also record args for the connector's _iter calls? Probably not — those args can be arbitrarily large (Table objects, schemas). Stick to entity_fqn only.
4. Auto-dump on MemoryError exception? Lean yes — extremely cheap to wire and exactly the moment we want a snapshot. To be decided in PR 3.
5. Disk-persisted dump location. /tmp/openmetadata-diag-<pid>.log is the default; users with custom log volume mounts may want a different path. Reasonable to make this one a hardcoded constant for now; revisit if customer asks.

---

14. Anti-patterns this design explicitly rejects

To stay honest, here is what we are NOT building and why:

Tempting feature	Rejected because
New env var to toggle diagnostics	Invisible to remote-deployed customers; needs pod redeploy to change
New ingestion config field for diagnostics	Permanent surface area for something almost no customer should touch; reuse loggerLevel
Periodic remote shipping of diagnostic state	Would re-create the streamable_logger failure mode (handler-on-itself recursion). All output stays local.
A web dashboard / sidecar	Far more complex than needed; kubectl logs already exists
Full APM instrumentation (every function call traced)	Overhead; not necessary for the hangs/OOMs we actually see
Optional drop-in flame graph generator	Different problem space (CPU profiling); see py-spy record for that

The principle throughout: minimum new surface, maximum information per line of log output.