[Obs AI Assistant] Distributed lock manager #216397

sorenlouv · 2025-03-29T06:07:54Z

Closes https://github.com/elastic/obs-ai-assistant-team/issues/238

LockManager is a lightweight, distributed locking solution. It ensures that critical operations, like re-indexing an index, are executed by only one node at a time, preventing race conditions and ensuring data consistency. By utilizing Elasticsearch's atomic updates, TTL-based expiration, and token-based safety checks, LockManager provides strong guarantees for safety and liveness without the complexity of full transactional systems, enabling reliable and efficient operations in distributed environments.

Usage:

const lockManagerService = new LockManagerService(coreSetup, logger);
lockManagerService.withLock('my_lock', () => {
 // perform operation
});

kibanamachine · 2025-03-29T10:05:07Z

Flaky Test Runner Stats

🎉 All tests passed! - kibana-flaky-test-suite-runner#8120

[✅] x-pack/test/api_integration/deployment_agnostic/configs/stateful/oblt.stateful.config.ts: 50/50 tests passed.
[✅] x-pack/test/api_integration/deployment_agnostic/configs/serverless/oblt.serverless.config.ts: 50/50 tests passed.

see run history

...m/plugins/shared/observability_ai_assistant/server/service/distributed_lock_manager/index.ts

...ic/apis/observability/ai_assistant/distributed_lock_manager/distributed_lock_manager.spec.ts

...m/plugins/shared/observability_ai_assistant/server/service/distributed_lock_manager/index.ts

...ic/apis/observability/ai_assistant/distributed_lock_manager/distributed_lock_manager.spec.ts

elasticmachine · 2025-04-01T10:47:39Z

Pinging @elastic/obs-ai-assistant (Team:Obs AI Assistant)

...ed/observability_ai_assistant/server/service/distributed_lock_manager/lock_manager_client.ts

kibanamachine · 2025-04-01T22:32:11Z

💚 All backports created successfully

Status	Branch	Result
✅	8.x

Note: Successful backport PRs will be merged automatically after passing CI.

Questions ?

Please refer to the Backport tool documentation

kibanamachine · 2025-04-01T23:06:42Z

Flaky Test Runner Stats

🟠 Some tests failed. - kibana-flaky-test-suite-runner#8128

[❌] x-pack/test/api_integration/deployment_agnostic/configs/stateful/oblt.stateful.config.ts: 0/50 tests passed.
[❌] x-pack/test/api_integration/deployment_agnostic/configs/serverless/oblt.serverless.config.ts: 0/50 tests passed.

see run history

kibanamachine · 2025-04-02T01:29:59Z

Flaky Test Runner Stats

🟠 Some tests failed. - kibana-flaky-test-suite-runner#8139

[❌] x-pack/test/api_integration/deployment_agnostic/configs/stateful/oblt.stateful.config.ts: 71/100 tests passed.

see run history

Tests added in #216397 are failing on MKI. Skipping temporarily in the affected environment ### Error ``` └- ✖ fail: Serverless Observability - Deployment-agnostic API integration tests observability AI Assistant LockManager Basic lock operations acquires the lock when not held │ ResponseError: security_exception │ Root causes: │ security_exception: action [indices:admin/create] is unauthorized for user [testing-internal] with effective roles [superuser] on restricted indices [.kibana_locks-000001], this action is granted by the index privileges [create_index,manage,all] ``` ### Root cause ```ts const es = getService('es'); es.deleteByQuery({ index: '.kibana_locks-000001', query: { match_all: {} }}); ```

# Backport This will backport the following commits from `main` to `8.x`: - [[Obs AI Assistant] Distributed lock manager (#216397)](#216397)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sorenlouv/backport)  Co-authored-by: Søren Louv-Jansen <[email protected]>

Related to #216397 Closes #216763 This change ensures that we do not send the `release` request and `extendTtl` request simultaneously in `withLock`. This caused a conflict causing tests to fail: ``` └-> "before all" hook for "should return the result of the callback" │ERROR Failed to release lock "my_lock_with_ttl_extension": version_conflict_engine_exception │ Root causes: │ version_conflict_engine_exception: [my_lock_with_ttl_extension]: version conflict, required seqNo [43], primary term [1]. current document has seqNo [44] and primary term [1] ``` Flaky tests: https://buildkite.com/elastic/kibana-flaky-test-suite-runner/builds/8142

Related to elastic#216397 Closes elastic#216763 This change ensures that we do not send the `release` request and `extendTtl` request simultaneously in `withLock`. This caused a conflict causing tests to fail: ``` └-> "before all" hook for "should return the result of the callback" │ERROR Failed to release lock "my_lock_with_ttl_extension": version_conflict_engine_exception │ Root causes: │ version_conflict_engine_exception: [my_lock_with_ttl_extension]: version conflict, required seqNo [43], primary term [1]. current document has seqNo [44] and primary term [1] ``` Flaky tests: https://buildkite.com/elastic/kibana-flaky-test-suite-runner/builds/8142 (cherry picked from commit 7275d2e)

) This PR replaces usage of the Kibana Task Manager with the [LockManager utility](#216397) that ensures strict mutual exclusion when running operations such as re-indexing or data migrations. --------- Co-authored-by: Viduni Wickramarachchi <[email protected]> Co-authored-by: kibanamachine <[email protected]>

### Background This PR is a proposal for a different way to structure the Streams code flow based on some challenges faced while working on elastic/streams-program#26 and discussed [here](elastic/streams-program#147) and [here](elastic/streams-program#55), mainly around finding it difficult to decide where to place certain validations that need access to the state as a whole. It is also in response to some expressed difficulty about how to add new stream types into the code base. It aims to achieve 3 goals: 1. It is easy to add new stream types and there is a clear place where changes (new validation, new logic) for existing stream types happen, making the code easier to evolve over time 2. It is easier to improve the robustness of the system because there are clear phases where problems can be caught, fixed and rolled back 3. It lays some ground work for features such as [bulk changes](elastic/streams-program#125), [dry runs](elastic/streams-program#138) and a [health endpoint](elastic/streams-program#139) In the future, this will most likely be handled by Elasticsearch to a large degree, as imagined in elastic/streams-program#30 The solution takes inspiration from the reconciliation / controller pattern that Kubernetes uses, where users specify a desired state and the system takes action towards reaching that step. But it is also somewhat more similar to how React's Virtual DOM works in that it happens in a single iteration. Another key pattern is the [Active Record pattern](https://www.martinfowler.com/eaaCatalog/activeRecord.html), we let each stream class contain all the logic for how to validate and modify that stream in Elasticsearch. The client and `State` class simply orchestrate the flow but defer all actual work and decision making to the stream classes. **Note:** This PoC ignores the management of assets ### Summary The process takes the following steps: 1. A route accepts a request (upsert / delete) and translates it into one or more (for bulk) `StreamChange` objects before passing these to `State.applyChanges` method (which also takes a toggle for dry runs) 2. The current state of Streams is loaded by using the `State` class 3. The changes are then applied to the current state to derive the desired state [1] 4. The desired state is then validated, this is done by asking each individual stream if given the desired state and starting state, from the perspective of that individual stream, is it in a valid state (upserted or deleted correctly) 5. If the state is invalid, we return those errors and stop 6. Else we continue, if it's a dry run, we ask the desired state object for what has changed and report that in the shape of the Elasticsearch actions that would be attempted 7. Else we proceed to commit the changes to Elasticsearch by asking each changed stream to determine which Elasticsearch actions need to be performed to reach the desired state 8. These actions are then combined and sent to the `ExecutionPlan` class which does planning (mainly for actions around Unwired streams) and then handles executing the actions in the most parallel way but in the safe order 9. If any error happens, we attempt to revert back to the starting state by taking the changed streams and marking each stream as created based on the starting state and then getting the Elasticsearch actions for that and applying those This PR also changes our `resync` endpoint to make use of the same rough strategy (load current state, mark all as created, get Elasticsearch actions and apply). [1] Applying changes: 1. The current state is first cloned 2. Then for each change we see if it is a deletion or an upsert 3. Based on this we either mark existing streams for deletion or create/update existing streams 10. When creating a new stream instance we use the helper `streamFromDefinition` which is the only mapping between the definition documents and the Active Record-style stream type classes 11. As part of this, each stream that changes is marked in the desired state 12. The stream is passed the desired and current state and should update itself based on the change 13. The stream can return a set of cascading changes (taking the same format as the requested changes) which are executed directly after but we have a limit for how many rounds of cascading changes can happen to avoid infinite loops ### Adding new stream types Key in all of this is that the client and `State` classes don't know anything about any of the specific stream types, they know only of the `StreamActiveRecord` interface. When adding a new stream type you need to implement this interface and update `streamFromDefinition` to create the right class for your new definition. Streams of different types should only interact with each other by creating cascading changes. ### Possible follow up tasks - Introduce a lazy Elasticsearch cluster state cache because multiple places in the code access the same stuff over and over again - Make API endpoints the consume `attemptChanges` pass back the `DesiredState` and planned `ElasticsearchActions` as debug information based on a flag (maybe also all cascading changes) - Don't run cascading changes by default but run them if _some_ flag is submitted based on elastic/streams-program#230 - Wrap `attemptChanges` and `resync` with the new LockManager #216397 - Unit test WiredStream, UnwiredStream and GroupStream - Clean up old sync helpers - Wrap ES calls to get better stack traces for errors ### Out of scope - Asset linking and content pack installation (it's probably okay for these to continue to use the asset client directly since there is less domain logic and no cascading changes involved) --------- Co-authored-by: Joe Reuter <[email protected]> Co-authored-by: kibanamachine <[email protected]>

Closes: #218944 The index template for the Lock Manager was not created, causing index mappings and settings to be incorrect. Root cause: the function responsible for creating the index template (`ensureTemplatesAndIndexCreated`) was never invoked - only during automated testing 🤦 ## Manual testing steps The mappings for the lock manager index (`.kibana_locks-000001`) should match this: ```ts { mappings: { dynamic: false, properties: { token: { type: 'keyword' }, metadata: { enabled: false }, createdAt: { type: 'date' }, expiresAt: { type: 'date' }, }, }, } ``` ### Test 1 In this test we make sure that the LockManager library can recover and fix the mappings if the existing index has invalid mappings #### Delete existing assets if they already exist ``` DELETE .kibana_locks-000001 DELETE _index_template/.kibana_locks-index-template DELETE _component_template/.kibana_locks-component ``` #### Create lock index. This is invalid because it does not have the correct mappings ``` PUT .kibana_locks-000001 ``` (Restart Kibana) #### Check mappings are correct ``` GET .kibana_locks-000001/_mapping ``` ### Test 2 In this test we make sure that out of the box, the LockManager library creates an index with the correct mappings ``` DELETE .kibana_locks-000001 DELETE _index_template/.kibana_locks-index-template DELETE _component_template/.kibana_locks-component ``` (Restart Kibana) #### Check mappings are correct ``` GET .kibana_locks-000001/_mapping ``` Related: #216916 #216397 --------- Co-authored-by: Viduni Wickramarachchi <[email protected]>

Tests added in elastic#216397 are failing on MKI. Skipping temporarily in the affected environment ### Error ``` └- ✖ fail: Serverless Observability - Deployment-agnostic API integration tests observability AI Assistant LockManager Basic lock operations acquires the lock when not held │ ResponseError: security_exception │ Root causes: │ security_exception: action [indices:admin/create] is unauthorized for user [testing-internal] with effective roles [superuser] on restricted indices [.kibana_locks-000001], this action is granted by the index privileges [create_index,manage,all] ``` ### Root cause ```ts const es = getService('es'); es.deleteByQuery({ index: '.kibana_locks-000001', query: { match_all: {} }}); ``` (cherry picked from commit 8bcce2e)

Closes: elastic#218944 The index template for the Lock Manager was not created, causing index mappings and settings to be incorrect. Root cause: the function responsible for creating the index template (`ensureTemplatesAndIndexCreated`) was never invoked - only during automated testing 🤦 The mappings for the lock manager index (`.kibana_locks-000001`) should match this: ```ts { mappings: { dynamic: false, properties: { token: { type: 'keyword' }, metadata: { enabled: false }, createdAt: { type: 'date' }, expiresAt: { type: 'date' }, }, }, } ``` In this test we make sure that the LockManager library can recover and fix the mappings if the existing index has invalid mappings ``` DELETE .kibana_locks-000001 DELETE _index_template/.kibana_locks-index-template DELETE _component_template/.kibana_locks-component ``` correct mappings ``` PUT .kibana_locks-000001 ``` (Restart Kibana) ``` GET .kibana_locks-000001/_mapping ``` In this test we make sure that out of the box, the LockManager library creates an index with the correct mappings ``` DELETE .kibana_locks-000001 DELETE _index_template/.kibana_locks-index-template DELETE _component_template/.kibana_locks-component ``` (Restart Kibana) ``` GET .kibana_locks-000001/_mapping ``` Related: elastic#216916 elastic#216397 --------- Co-authored-by: Viduni Wickramarachchi <[email protected]> (cherry picked from commit f684ea4)

@nchaulet

# Backport This will backport the following commits from `main` to `8.19`: - [[Obs AI Assistant] Skip lock tests in MKI temporarily (#216753)](#216753) - [[Obs AI Assistant] Replace Task Manager usage with Lock Manager (#216916)](#216916) - [[LockManager] Ensure index template are created (#218901)](#218901) - [[LockManager] Expose as package (#219220)](#219220)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sorenlouv/backport)  --------- Co-authored-by: kibanamachine <[email protected]>

…tic#216916) This PR replaces usage of the Kibana Task Manager with the [LockManager utility](elastic#216397) that ensures strict mutual exclusion when running operations such as re-indexing or data migrations. --------- Co-authored-by: Viduni Wickramarachchi <[email protected]> Co-authored-by: kibanamachine <[email protected]> (cherry picked from commit 6244105)

… Assistant] Replace Task Manager usage with Lock Manager (#216916) | update knowledge base installation flow (#214133) | [Obs AI Assistant] Specify embedding model during onboarding for the Knowledge Base (#218448) (#220041) # Backport This will backport the following commits from `main` to `8.19`: - [[Obs AI Assistant] Minor cleanup api tests (#215181)](#215181) - [[Obs AI Assistant] Replace Task Manager usage with Lock Manager (#216916)](#216916) - [update knowledge base installation flow (#214133)](#214133) - [[Obs AI Assistant] Specify embedding model during onboarding for the Knowledge Base (#218448)](#218448)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sorenlouv/backport)  --------- Co-authored-by: Arturo Lidueña <[email protected]> Co-authored-by: Viduni Wickramarachchi <[email protected]> Co-authored-by: kibanamachine <[email protected]>

…quired (#220476) Related: #216397 This fixes a bug in the Lock Manager where an expired lock can be acquired, but the token and metadata is not updated. This means that the lock cannot be released. Instead it is automatically released when the TTL expires.

…quired (elastic#220476) Related: elastic#216397 This fixes a bug in the Lock Manager where an expired lock can be acquired, but the token and metadata is not updated. This means that the lock cannot be released. Instead it is automatically released when the TTL expires. (cherry picked from commit 74e876d)

…quired (elastic#220476) Related: elastic#216397 This fixes a bug in the Lock Manager where an expired lock can be acquired, but the token and metadata is not updated. This means that the lock cannot be released. Instead it is automatically released when the TTL expires.

Closes: elastic#218944 The index template for the Lock Manager was not created, causing index mappings and settings to be incorrect. Root cause: the function responsible for creating the index template (`ensureTemplatesAndIndexCreated`) was never invoked - only during automated testing 🤦 ## Manual testing steps The mappings for the lock manager index (`.kibana_locks-000001`) should match this: ```ts { mappings: { dynamic: false, properties: { token: { type: 'keyword' }, metadata: { enabled: false }, createdAt: { type: 'date' }, expiresAt: { type: 'date' }, }, }, } ``` ### Test 1 In this test we make sure that the LockManager library can recover and fix the mappings if the existing index has invalid mappings #### Delete existing assets if they already exist ``` DELETE .kibana_locks-000001 DELETE _index_template/.kibana_locks-index-template DELETE _component_template/.kibana_locks-component ``` #### Create lock index. This is invalid because it does not have the correct mappings ``` PUT .kibana_locks-000001 ``` (Restart Kibana) #### Check mappings are correct ``` GET .kibana_locks-000001/_mapping ``` ### Test 2 In this test we make sure that out of the box, the LockManager library creates an index with the correct mappings ``` DELETE .kibana_locks-000001 DELETE _index_template/.kibana_locks-index-template DELETE _component_template/.kibana_locks-component ``` (Restart Kibana) #### Check mappings are correct ``` GET .kibana_locks-000001/_mapping ``` Related: elastic#216916 elastic#216397 --------- Co-authored-by: Viduni Wickramarachchi <[email protected]>

…quired (elastic#220476) Related: elastic#216397 This fixes a bug in the Lock Manager where an expired lock can be acquired, but the token and metadata is not updated. This means that the lock cannot be released. Instead it is automatically released when the TTL expires.

[Obs AI Assistant] Add distributed lock manager

a93e84b

sorenlouv force-pushed the add-distributed-lock-manager branch from e380b47 to a93e84b Compare March 29, 2025 06:25

sorenlouv changed the title ~~[Obs AI Assistant] Add distributed lock manager~~ [Obs AI Assistant] Add simple lock manager Mar 29, 2025

Add waitForLock and withLock

e48a65a

dgieselaar reviewed Mar 30, 2025

View reviewed changes