feat(alerting): Add OrphanedAlertSweeper and fix JobSweeper to desche…

[toepkerd]

Description

Add OrphanedAlertSweeper: a scheduled background job that periodically scans for alerts whose monitors no longer exist in .opendistro-alerting-config. Orphaned alerts are moved to the alert history index with state DELETED.

Fix JobSweeper.sweepShard to deschedule monitors that are in the in-memory sweptJobs map but no longer exist in the config index. Previously, the background sweep only handled new/updated jobs and shard ownership changes, relying entirely on postDelete for deletions. If postDelete failed to fire (e.g., due to IndexingOperationListener not triggering), the monitor would remain scheduled indefinitely, producing repeated 'Can't find monitor' errors.

Note: this is not a bug fix for postDelete, but an anti entropy mechanism that maintains correct Alerting state while the bug gets root caused and fixed.

Related Issues

#2086

Check List

• [Y] New functionality includes testing.
• New functionality has been documented.
• API changes companion pull request created.
• [Y] Commits are signed per the DCO using --signoff.
• Public documentation issue/PR created.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[lezzago]

Coroutine scope can be permanently broken by an unhandled failure

This uses the default Job(). If any uncaught exception escapes withTimeout(...) in reschedule() (e.g. a TimeoutCancellationException thrown after the inner try/catch in sweepOrphanedAlerts returns, or any throwable raised before the try block is entered), the scope's job is cancelled and all future scheduled sweeps silently no-op — scope.launch { ... } on a cancelled scope is a no-op, and the user gets no signal that the sweeper has stopped working.

Suggestions:

• Use SupervisorJob(): CoroutineScope(SupervisorJob() + Dispatchers.IO) so a single failure doesn't tear down the scope.
• Or wrap the withTimeout call in a try/catch at the launch site so timeout/cancellation never propagates out of the coroutine.
• Add a CoroutineExceptionHandler for visibility either way.

[toepkerd]

Updated to use SupervisorJob + CoroutineExceptionHandler

[lezzago]

Boot-time race: init doesn't schedule a sweep

Scheduling is driven solely by clusterChanged. If the local node is already the elected cluster manager at the moment this listener is registered — uncommon but possible during a fast restart of a single-CM cluster, or when the plugin is installed on an already-running CM — clusterChanged may not fire a transition for that election and isClusterManager stays false forever, so the sweeper never starts.

JobSweeper avoids this via LifecycleListener.afterStart() (see JobSweeper.kt:124). I'd mirror that pattern here:

• Implement LifecycleListener, register with clusterService.addLifecycleListener(this), and bootstrap from afterStart() by checking clusterService.state().nodes.isLocalNodeElectedClusterManager.
• Also pair it with beforeStop/beforeClose to cancel the schedule, remove the cluster listener, and cancel scope — currently nothing tears these down on node shutdown.

[toepkerd]

Updated to make the sweeper also implement LifecycleListener

[engechas]

Is there a possibility that we deschedule valid jobs due to a partial shard failure in the search with this logic?

There should be an existing mechanism that handles descheduling jobs once they are deleted - I'm wondering why we need this extra layer of handling in the sweeper rather than relying on the DeleteMonitor flow

[toepkerd]

There is an existing mechanism to deschedule monitors on monitor deletion, here in JobSweeper.

The goal of this PR is to temporarily remediate a bug where this callback simply does not happen for a reason that is still being root caused. This is meant to be a short-medium term band-aid fix that maintains correct Alerting state for Alerting users while we root cause the true bug. Please see the description if you'd like more details.

[toepkerd]

To address the partial shard failure case, in the while loop above this, during the paginated shard search, the entire sweep returns early if any shards fail (reference). This means this descheduling block at the end is only reached if all shards succeed. If any shard at all failed, the retry is deferred to the next sweep cycle.

[engechas]

Do we know how alerts are getting orphaned? We shouldn't have orphans if the delete logic is working correctly so it seems like a bandaid to add a sweeper rather than closing the gaps in the delete logic (if any)

I would advocate that we should fix any bugs that lead to orphaned alerts rather than relying on an out-of-band cleanup process

[AWSHurneyt]

Is there a reason size is 1000 here?

This is an edge case, but the plugin does support the ALERTING_MAX_MONITORS cluster setting to potentially have more than 1000 monitors.
https://github.com/opensearch-project/alerting/blob/main/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt#L35

[AWSHurneyt]

Would it make sense to use one of the cluster settings here? E.g., BULK_TIMEOUT?
https://github.com/opensearch-project/alerting/blob/main/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt#L120

[lezzago]

🚫 BLOCKING — sweeper can incorrectly delete active workflow (chained) alerts

Per AlertService.kt:949, chained alerts can carry monitor_id="" and use workflow_id instead. The terms aggregation in getAlertMonitorIds() will return "" as a valid bucket key. That empty string is then:

1. Looked up via idsQuery().addIds("") against SCHEDULED_JOBS_INDEX — returns no hits.
2. Therefore added to orphanedMonitorIds.
3. Passed to AlertMover.moveAlerts(client, "", null), which runs termQuery(MONITOR_ID_FIELD, "") (AlertMover.kt:62) — matching every chained alert in the cluster whose monitor_id field is empty.

The result: chained alerts for active, non-deleted workflows get moved to history with state=DELETED on the very first sweep cycle. This is a regression risk, not just a coverage gap.

Required fix (one of):

• Filter the terms agg to non-empty monitor IDs (e.g. add a must(existsQuery) + mustNot(termQuery("monitor_id", "")) pre-filter), AND
• Add a parallel pass that aggregates Alert.WORKFLOW_ID_FIELD, reconciles against workflow IDs in SCHEDULED_JOBS_INDEX, and invokes the workflow variant of AlertMover.moveAlerts for genuine workflow orphans.

At minimum, please verify with a test that creates a workflow with chained alerts, leaves the workflow alive, runs the sweeper, and asserts the chained alerts are NOT moved to history.

[lezzago]

Nit: redundant isOwningNode check

Lines 304-307 earlier in sweepShard already remove non-owning jobs from currentJobs:

currentJobs.keys.filterNot { shardNodes.isOwningNode(it) }.forEach {
    scheduler.deschedule(it)
    currentJobs.remove(it)
}

By the time control reaches this new block, every key in currentJobs is owned by this node. The shardNodes.isOwningNode(it) && predicate is dead. Either drop it for clarity or add a comment noting it's defensive.

[lezzago]

Concurrent sweeps possible if cluster-manager-ship oscillates

cancel() only cancels the Scheduler.Cancellable — it doesn't track or cancel the in-flight coroutine Job returned by scope.launch { ... } at line 117. If CM-ship flips off and back on (rare but possible during partition flapping or network blips), the previous sweep's coroutine continues running while a new schedule is armed. Two sweeps can call AlertMover.moveAlerts for the same monitor IDs in parallel.

AlertMover uses versionType=EXTERNAL_GTE so the worst case is duplicated history-index writes / version-conflict log noise — no data corruption — but it's worth tightening:

private var inFlightSweep: Job? = null

private fun reschedule() {
    cancel()
    scheduledSweep = threadPool.scheduleWithFixedDelay(
        { inFlightSweep = scope.launch { withTimeout(SWEEP_TIMEOUT_MS) { sweepOrphanedAlerts() } } },
        sweepPeriod,
        ThreadPool.Names.MANAGEMENT
    )
}

private fun cancel() {
    scheduledSweep?.cancel()
    scheduledSweep = null
    inFlightSweep?.cancel()
    inFlightSweep = null
}

[lezzago]

60s timeout may be too tight for the first sweep after an outage

AlertMover.moveAlerts is called sequentially in a for loop, all under one withTimeout(SWEEP_TIMEOUT_MS). Each call is roughly 1 search + 1 bulk index + 1 bulk delete (3 round-trips). On the first sweep after an extended outage where many orphans accumulated, ~100 orphans could blow the 60s budget and lose progress mid-loop — and the next cycle starts from scratch with the same overload.

Options:

• Cap orphans-per-sweep (e.g. process up to N per cycle, drain across cycles).
• Make timeout proportional to orphan count.
• Process moves with bounded parallelism (e.g. coroutineScope { orphans.chunked(K).forEach { ... } }).
• Or expose SWEEP_TIMEOUT_MS as a cluster setting alongside the period/enabled flags so operators can tune it without a code change.

[toepkerd]

The proposed mitigation to cap orphans per sweep is already in place, just not very explicitly. moveAlerts() runs a search on the active alerts index without explicitly passing a size param (reference). This means it defaults to returning only 10 orphaned Alerts even within a single zombie Monitor. In this respect, this sweeper already cleans at most 10 alerts per orphaned monitor per sweep.

Even if within a 10 orphaned Alerts cleanup batch, the timeout can still be hit in the worst case but that should not be an issue:

• case: timeout hits after both the copy and the delete bulk requests are sent. In this case, the timeout simply prevents the thread from reading in the result, but the task is done. On the next sweep, the alerts will have already been cleaned up, and the sweeper moves on without issue
• case: timeout happens between the copy and the delete bulk requests. In this case, there are now 2 copies of the orphaned alerts in the active and history indices. On the next sweep, these are found again, but with EXTERNAL_GTE version checks, idempotency is enforced and no duplication happens, progress is still made. The only way this case scenario can stall progress is if on every single sweep, the timeout hits between the copy and delete bulk requests, which are back to back lines.
• case: timeout happens before copy or delete requests, stalls on the search for orphaned Active alerts. If the search to the active alerts concrete index takes more than 60s, the cluster has bigger problems than the alert sweeper not catching all Alerts. Once the cluster can breathe again, the above 2 cases kick in and make progress.