feat(ingestion-progress): total assets, run-start emit, agent on discovery#29753
feat(ingestion-progress): total assets, run-start emit, agent on discovery#29753ulixius9 wants to merge 3 commits into
Conversation
…on discovery Follow-ups to the hierarchical progress SSE feature (PR #29365): - Add ProgressUpdate.totalAssetsIngested (leaf assets: tables + stored procedures) and send it on every update including the terminal one, so the final asset count is authoritative and survives scope pruning. - Emit an initial DISCOVERY progress update at run start so a run shorter than the reporting interval still surfaces live (a start + end event pair) instead of appearing to emit nothing. - Attach the IngestionPipeline entity to the DISCOVERY ServiceProgressEvent, loaded server-side once per run, so the UI can discover a newly created autopilot agent from the stream itself. Note: openmetadata-ui serviceProgressEvent.ts still needs `make update_typescript_types` to pick up the ingestionPipeline field (deferred to the UI-consumption work). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
❌ PR checklist incompleteThis PR cannot be merged until the following are addressed on its linked issue:
The fields live on the linked issue in the Shipping project (open the issue → right sidebar → Projects). After you set them, re-run this check (or push a commit) — issue/project changes do not re-trigger it automatically. Maintainers can bypass this check by adding the |
github-actions
Bot
added
Ingestion
safe to test
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Pull request overview
Refines the ingestion progress streaming contract (SSE + CLI/UI typing) to expose an authoritative run-level ingested-asset total, ensure an initial “run started” event is emitted for short runs, and optionally attach the owning ingestion pipeline entity to the run’s DISCOVERY event for agent discovery.
Changes:
- Add
totalAssetsIngestedtoProgressUpdate(schema + generated UI types) and populate it from the monotonic leaf counter in the progress registry/reporting path. - Emit an initial
DISCOVERYprogress update at the start ofBaseWorkflow.execute()to ensure sub-interval runs still show up live. - Add optional
ingestionPipelinetoServiceProgressEvent(schema) and attach it on DISCOVERY via backend repository lookup + tracker propagation.
Reviewed changes
Copilot reviewed 11 out of 13 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/serviceProgressEvent.ts | Updates generated ProgressUpdate typing to include totalAssetsIngested. |
| openmetadata-ui/src/main/resources/ui/src/generated/entity/services/ingestionPipelines/progressUpdate.ts | Updates generated ProgressUpdate typing to include totalAssetsIngested. |
| openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/serviceProgressEvent.json | Adds optional ingestionPipeline reference to service-scoped SSE event schema. |
| openmetadata-spec/src/main/resources/json/schema/entity/services/ingestionPipelines/progressUpdate.json | Adds totalAssetsIngested field to progress update schema. |
| openmetadata-service/src/main/java/org/openmetadata/service/monitoring/IngestionProgressTracker.java | Extends tracker API to optionally carry IngestionPipeline into ServiceProgressEvent. |
| openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/IngestionPipelineRepository.java | Loads pipeline entity on DISCOVERY (best-effort) and passes it to tracker for service stream. |
| openmetadata-service/src/test/java/org/openmetadata/service/monitoring/IngestionProgressTrackerTest.java | Adds unit tests asserting pipeline entity is present only when provided. |
| ingestion/src/metadata/workflow/workflow_status_mixin.py | Includes totalAssetsIngested in emitted progress payload. |
| ingestion/src/metadata/workflow/progress_render.py | Exposes monotonic assets_ingested() and uses it for rendering/payload. |
| ingestion/src/metadata/workflow/base.py | Emits initial DISCOVERY progress update at workflow start. |
| ingestion/tests/unit/workflow/test_status_mixin_progress.py | Adds tests for totalAssetsIngested presence on processing/terminal events and null when no registry. |
| ingestion/tests/unit/workflow/test_progress_rendering.py | Adds tests for reporter counting leaf types and surviving pruning. |
| ingestion/tests/unit/workflow/test_base_workflow.py | Updates ordering expectations and asserts initial DISCOVERY emit occurs before work. |
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
✅ TypeScript Types Auto-UpdatedThe generated TypeScript types have been automatically updated based on JSON schema changes in this PR. |
Jest test CoverageUI tests summary
|
🔴 Playwright Results — 1 failure(s), 15 flaky✅ 4504 passed · ❌ 1 failed · 🟡 15 flaky · ⏭️ 37 skipped
Genuine Failures (failed on all attempts)❌
|
Code Review ✅ Approved 1 resolved / 1 findingsRefines ingestion progress streaming by adding total asset counts, initial run-start events, and agent pipeline discovery. All findings were resolved, and no issues were found. ✅ 1 resolved✅ Quality: Generated UI type missing top-level ingestionPipeline field
OptionsDisplay: compact → Showing less information. Comment with these commands to change the behavior for this request:
Was this helpful? React with 👍 / 👎 | Gitar |
|
|
Describe your changes:
Follow-up to #29365 (hierarchical progress SSE).
I made the ingestion progress stream tell the UI three things it was missing: how many assets a run ingested, that a run has started (even very short runs), and which agent a run belongs to. These came out of testing the live SSE stream against an autopilot run, where short runs appeared to emit nothing, the final asset count was absent, and newly created agents weren't discoverable from the stream.
Type of change:
High-level design:
totalAssetsIngested— new additive field onprogressUpdate.json, populated from the registry's monotonic leaf-asset counter (tables + stored procedures) and sent on every update including the terminal one, so the final count is authoritative and survives scope pruning.BaseWorkflow.execute()now fires an initialDISCOVERYupdate immediately (the reporting timer's first tick is a full interval away, so sub-interval runs previously emitted only their terminal event).serviceProgressEvent.jsongains an optionalingestionPipeline($ref). The server loads the entity only on theDISCOVERYupdate (IngestionPipelineRepository.updateProgress→discoveredPipeline) andIngestionProgressTrackerattaches it to theServiceProgressEvent, so the UI can discover a newly created autopilot agent from the stream itself.getFields("")excludesopenMetadataServerConnection, so no secrets are exposed (same data the existing REST list already returns).Tests:
Unit tests
test_status_mixin_progress.py(totalAssetsIngested on processing + terminal),test_progress_rendering.py(reporter asset total),test_base_workflow.py(initial DISCOVERY before work).IngestionProgressTrackerTest(ServiceProgressEvent carries the pipeline on DISCOVERY, null otherwise).Backend integration tests
/progress/*SSE contract is unchanged except for additive fields.Ingestion integration tests
Playwright (UI) tests
Manual testing performed
totalAssetsIngested, and the agent entity onDISCOVERY.mvn spotless:check -pl openmetadata-serviceclean; Java tracker tests 14/14; ingestion workflow unit suite green.UI screen recording / screenshots:
Not applicable.
Checklist:
totalAssetsIngested, optionalingestionPipeline); no data migration needed.🤖 Generated with Claude Code
Greptile Summary
This PR improves ingestion progress events for live run tracking. The main changes are:
totalAssetsIngestedto progress updates.Confidence Score: 5/5
This looks safe to merge.
Important Files Changed
totalAssetsIngestedto the generated progress update type.ingestionPipelinepayload to the generated service event type.Reviews (3): Last reviewed commit: "Merge branch 'main' into port-louis" | Re-trigger Greptile
Context used (3)