[One Workflow] Data streams for executions dal - #279498
Draft
skynetigor wants to merge 134 commits into
Draft
Conversation
…rojectionField type and updating PlainIndexExecutionsDataAccess class definition
…access layer - Introduced new types: ExecutionsCountRequest, WorkflowExecutionsCountRequest, and StepExecutionsCountRequest. - Implemented the count method in PlainIndexExecutionsDataAccess to handle execution count requests. - Updated relevant type exports in the data access layer for better type safety and functionality.
…step and workflow executions - Updated the StepExecutionRepository and WorkflowExecutionRepository to use the new data access layer methods for fetching and manipulating step and workflow execution data. - Enhanced type safety by integrating the StepExecutionsDataAccess and WorkflowExecutionsDataAccess interfaces. - Removed direct Elasticsearch client calls in favor of the data access layer, improving code maintainability and readability. - Adjusted bulk upsert operations to align with the new data access layer structure.
- Introduced a new `create_executions_dal` implementation to streamline the creation of workflow and step executions data access. - Consolidated execution data access dependencies under `CreateExecutionsDataAccessDeps` for improved type safety. - Removed outdated `createWorkflowExecutionsDataAccess` and `createStepExecutionsDataAccess` implementations in favor of the new structure. - Updated validation functions to accommodate the new data access layer design. - Enhanced exports in the data access layer to reflect the changes and maintain consistency.
…gration - Implemented a new method for creating repositories in the WorkflowsExecutionEnginePlugin, consolidating the creation of WorkflowExecutionRepository and StepExecutionRepository. - Updated execution functions (runWorkflow, resumeWorkflow) to utilize the new repository creation method, improving code maintainability. - Enhanced type safety by ensuring the use of WorkflowExecutionsDataAccess and StepExecutionsDataAccess interfaces throughout the execution functions. - Removed redundant Elasticsearch client instantiation, streamlining the data access process.
- Added `executeIndexBulkCreate` and `executeIndexBulkUpdate` implementations for handling bulk create and update operations. - Introduced new types: `BulkCreateRequest` and `BulkUpdateRequest` to support the new operations. - Updated `PlainIndexExecutionsDataAccess` to include methods for bulk create and update. - Enhanced error handling with new functions for throwing bulk write errors and asserting success for bulk updates. - Improved response handling for bulk create operations with new utility functions.
…ration of concerns
… mock data generation
…ove handling of missing items
…ted interfaces to simplify execution retrieval
… execution retrieval and updating related interfaces for improved flexibility
- Guard against duplicate document IDs in a single bulk request; previously a duplicate would silently leave a result slot as undefined. - Hoist getMeta() / getDataStream out of the retry loop by resolving the write index once before the while loop and threading it into bulkGetFreshVersions via an optional param, eliminating one API call per retry iteration. - Stop forwarding retry_on_conflict to ES when if_seq_no/if_primary_term are present; ES silently ignores it in that case and the manual retry queue in bulk() is the sole retry mechanism for version-based CAS writes.
- Narrow getByIds signature from (string | { id, index })[] to string[],
matching the DataClient interface and PlainIndexDataClient; remove the
dead { id, index } object branch and simplify the mget doc-building loop.
- Fix count to query indexesToQuery (data stream + additional indexes) so
it covers the same document set as search during migration.
- Remove the dead else-if in resolveBulkItemVersions that checked item.index
— BulkItem has no index field so the condition was never true and caused
a TypeScript error.
- Compute indexesToQuery once in the constructor instead of inline per call.
Data streams are append-only so deleteByQuery is implemented as a soft-delete: search for matching documents, populate the version cache from the search hits explicitly, then bulk-update each with deleted:true. - Add deleted:boolean to both data stream mapping overrides so the field is indexed and can be filtered in queries. - Use _source:false in the search to avoid fetching source unnecessarily. - Explicitly call versionManager.setVersion for each hit before bulk so resolveBulkItemVersions reads versions from cache without an extra ES round-trip, rather than relying on the search method's side-effect which was gated on _source being truthy. - Remove seqNo/primaryTerm from bulk items — they were ignored by resolveBulkItemVersions and their presence was misleading.
…flict - Add BulkPlainItem/BulkUpdaterItem discriminated union to types; BulkUpdaterItem uses documentId + sourceFields + updater callback instead of a document patch - Add index? to BulkPlainItem so converted updater items carry their backing index - Rewrite retryOnConflicts: mgets updater items fresh each retry, converts them to plain items with seqNo/primaryTerm/index, maps raw ES response to BulkItemResponse; action signature now returns estypes.BulkResponse so mapping is centralised here - Simplify sharedBulk: delegates retry loop to retryOnConflicts, action is a raw esClient.bulk call; SharedBulkItem collapses to a type alias for BulkPlainItem
- sharedBulk: inline retry-on-conflict loop (handles both plain and updater items with mget + OCC via if_seq_no/if_primary_term); widen parameter type from SharedBulkRequestOptions to BulkRequestOptions; fix fetchFreshVersions to accept optional index and use it for targeted mget before falling back to fallbackIndexes; rename ids → docsToRefetch; add type imports - data_stream_data_client: route BulkUpdaterItem through sharedBulk with last-2 backing indexes as fallback; merge assignTimestampToItems into resolveBulkItemVersions (single pass: inject @timestamp for create/upsert, pre-fail creates missing the date field, resolve backing-index + seqNo); update version cache after successful writes via versionManager.setVersion - plain_index_data_client: split updater/plain items; handle updater items with getByIds + OCC conflict retry loop; plain items via sharedBulk - retry_on_conflicts: extract fetchFreshVersions helper
OCC conflict retry is handled inside sharedBulk via fetchFreshVersions; the fresh=true path (bulkGetFreshVersions) was never reached.
Add createRetryingEsClient (Proxy-based wrapper that intercepts all method calls and applies retryTransientEsErrors automatically, including nested namespaces like esClient.indices.*). Apply the wrapper once in DataStreamDataClientBundle and PlainIndexDataClientBundle so every downstream consumer — data clients, version manager, shared helpers — gets retry for free without explicit wrapping at each call site. Remove all retryTransientEsErrors wrappers from data_stream_data_client, document_version_manager, plain_index_data_client, shared_bulk, retry_on_conflicts, get_executions_by_ids, execute_script_update, and create_index.
bulk() calls bulkGetVersions internally, which resolves and caches versions for update items — the manual setVersion loop before the bulk call was a no-op duplicate.
Resolve plain items via resolveBulkItemVersions first, then merge sendable plain items and updater items into one array and pass them to a single sharedBulk call. sharedBulk already handles both item types internally, so splitting into two calls was unnecessary. Also fixes a bug where the plain-item bulk response was discarded (result[] was never populated for plain items).
Add workflowsExecutionEngine.storage.source config option
('plain_index' | 'data_stream', default: 'plain_index') so the
backing store can be switched without code changes.
Rename ExecutionStorageSource 'system_index' -> 'plain_index' to
align with the config value. Wire the plugin to use the config
instead of the hardcoded 'data_stream'. Update all test mocks.
Add template.lifecycle with a 90d data_retention default to both registerDataStream calls. Without a lifecycle entry the index template had no DLM policy, which caused the retention field to be uneditable in Index Management. Bump version to 2 so the updated template is applied on next startup.
Expose workflowsExecutionEngine.storage.dataRetention (default: 90d) in kibana.yml and wire it through CreateDataClientDeps into the data stream bundle's registerDataStream lifecycle, so operators can tune DLM retention without code changes.
Typed read-modify-write via bulk() works on both plain index and data stream storage, with testable updater logic and shared conflict handling.
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
2 participants
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.
Summary
Summarize your PR. If it involves visual changes include a screenshot or gif.
Checklist
Check the PR satisfies following conditions.
Reviewers should verify this PR satisfies this list as well.
release_note:breakinglabel should be applied in these situations.release_note:*label is applied per the guidelinesbackport:*labels.Identify risks
Does this PR introduce any risks? For example, consider risks like hard to test bugs, performance regression, potential of data loss.
Describe the risk, its severity, and mitigation for each identified risk. Invite stakeholders and evaluate how to proceed before merging.