From ba7314357111be896bc0682d612a1c4fb41b9a37 Mon Sep 17 00:00:00 2001 From: Dennis Toepker Date: Wed, 3 Jun 2026 21:52:21 -0700 Subject: [PATCH 1/2] feat(alerting): Add OrphanedAlertSweeper and fix JobSweeper to deschedule deleted monitors 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. Configurable via: - plugins.alerting.orphaned_alert_sweep_period (default: 5m) - plugins.alerting.orphaned_alert_sweep_enabled (default: true) Includes integration tests verifying: - Active orphaned alerts are moved to history as DELETED - Acknowledged orphaned alerts are moved to history as DELETED - Non-orphaned alerts are not affected by the sweeper Signed-off-by: Dennis Toepker --- .../org/opensearch/alerting/AlertingPlugin.kt | 10 +- .../alerting/alerts/OrphanedAlertSweeper.kt | 180 ++++++++++++++++++ .../alerting/settings/AlertingSettings.kt | 12 ++ .../alerting/OrphanedAlertSweeperIT.kt | 179 +++++++++++++++++ .../opensearch/alerting/core/JobSweeper.kt | 9 + 5 files changed, 388 insertions(+), 2 deletions(-) create mode 100644 alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt create mode 100644 alerting/src/test/kotlin/org/opensearch/alerting/OrphanedAlertSweeperIT.kt diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt index 44c194b1e..ed4dc8a23 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingPlugin.kt @@ -16,6 +16,7 @@ import org.opensearch.alerting.action.SearchEmailAccountAction import org.opensearch.alerting.action.SearchEmailGroupAction import org.opensearch.alerting.alerts.AlertIndices import org.opensearch.alerting.alerts.AlertIndices.Companion.ALL_ALERT_INDEX_PATTERN +import org.opensearch.alerting.alerts.OrphanedAlertSweeper import org.opensearch.alerting.comments.CommentsIndices import org.opensearch.alerting.comments.CommentsIndices.Companion.ALL_COMMENTS_INDEX_PATTERN import org.opensearch.alerting.core.JobSweeper @@ -196,6 +197,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R lateinit var runner: MonitorRunnerService lateinit var scheduler: JobScheduler lateinit var sweeper: JobSweeper + lateinit var orphanedAlertSweeper: OrphanedAlertSweeper lateinit var scheduledJobIndices: ScheduledJobIndices lateinit var commentsIndices: CommentsIndices lateinit var docLevelMonitorQueries: DocLevelMonitorQueries @@ -359,6 +361,7 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R docLevelMonitorQueries = DocLevelMonitorQueries(client, clusterService) scheduler = JobScheduler(threadPool, runner) sweeper = JobSweeper(environment.settings(), client, clusterService, threadPool, xContentRegistry, scheduler, ALERTING_JOB_TYPES) + orphanedAlertSweeper = OrphanedAlertSweeper(environment.settings(), client, threadPool, clusterService) destinationMigrationCoordinator = DestinationMigrationCoordinator(client, clusterService, threadPool, scheduledJobIndices) this.threadPool = threadPool this.clusterService = clusterService @@ -421,7 +424,8 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R alertService, triggerService, sdkClient, - monitorJobPoller + monitorJobPoller, + orphanedAlertSweeper ) } @@ -527,7 +531,9 @@ internal class AlertingPlugin : PainlessExtension, ActionPlugin, ScriptPlugin, R AlertingSettings.JOB_QUEUE_ACCOUNT_PROVIDER_TYPE, AlertingSettings.TARGET_TYPE_TO_SERVICE_NAME, AlertingSettings.TENANT_ACCOUNT_ID_HEADER, - AlertingSettings.TENANT_RESOURCE_ID_HEADER + AlertingSettings.TENANT_RESOURCE_ID_HEADER, + AlertingSettings.ORPHANED_ALERT_SWEEP_PERIOD, + AlertingSettings.ORPHANED_ALERT_SWEEP_ENABLED ) } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt b/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt new file mode 100644 index 000000000..ca1151aca --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt @@ -0,0 +1,180 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.alerts + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.launch +import kotlinx.coroutines.withTimeout +import org.apache.logging.log4j.LogManager +import org.opensearch.action.search.SearchRequest +import org.opensearch.action.search.SearchResponse +import org.opensearch.alerting.alerts.AlertIndices.Companion.ALERT_INDEX +import org.opensearch.alerting.opensearchapi.suspendUntil +import org.opensearch.alerting.settings.AlertingSettings +import org.opensearch.cluster.ClusterChangedEvent +import org.opensearch.cluster.ClusterStateListener +import org.opensearch.cluster.service.ClusterService +import org.opensearch.common.settings.Settings +import org.opensearch.commons.alerting.model.Alert +import org.opensearch.commons.alerting.model.ScheduledJob +import org.opensearch.index.query.QueryBuilders +import org.opensearch.search.aggregations.bucket.terms.Terms +import org.opensearch.search.aggregations.bucket.terms.TermsAggregationBuilder +import org.opensearch.search.builder.SearchSourceBuilder +import org.opensearch.threadpool.Scheduler +import org.opensearch.threadpool.ThreadPool +import org.opensearch.transport.client.Client + +/** + * An anti-entropy background job that periodically scans for alerts whose monitors no longer + * exist in .opendistro-alerting-config ("orphaned alerts") and moves them to the alert history + * index with state = DELETED. + * + * Normally, when a monitor is deleted, the JobSweeper's IndexingOperationListener.postDelete + * callback fires, which deschedules the monitor and moves its alerts to history via AlertMover. + * However, postDelete does not always fire reliably — particularly when a monitor deletion + * occurs while one of its executions is in flight. In this case, + * alerts remain stranded in the active alerts index indefinitely. + * + * This sweeper is not a bug fix for postDelete — it is a safety net. It runs independently on the + * cluster manager node and cleans up any orphaned alerts that postDelete missed, ensuring the + * system eventually converges to a consistent state. + * + * Runs only on the cluster manager node to avoid duplicate work. + */ +class OrphanedAlertSweeper( + settings: Settings, + private val client: Client, + private val threadPool: ThreadPool, + private val clusterService: ClusterService +) : ClusterStateListener { + + private val logger = LogManager.getLogger(OrphanedAlertSweeper::class.java) + private val scope = CoroutineScope(Dispatchers.IO) + + @Volatile private var sweepPeriod = SWEEP_PERIOD.get(settings) + @Volatile private var enabled = ENABLED.get(settings) + private var scheduledSweep: Scheduler.Cancellable? = null + private var isClusterManager = false + + init { + clusterService.addListener(this) + clusterService.clusterSettings.addSettingsUpdateConsumer(SWEEP_PERIOD) { + sweepPeriod = it + if (clusterService.state().nodes.isLocalNodeElectedClusterManager && enabled) reschedule() + } + clusterService.clusterSettings.addSettingsUpdateConsumer(ENABLED) { + enabled = it + if (clusterService.state().nodes.isLocalNodeElectedClusterManager) { + if (enabled) reschedule() else cancel() + } + } + } + + override fun clusterChanged(event: ClusterChangedEvent) { + if (this.isClusterManager != event.localNodeClusterManager()) { + this.isClusterManager = event.localNodeClusterManager() + if (this.isClusterManager && enabled) { + reschedule() + } else { + cancel() + } + } + } + + private fun reschedule() { + cancel() + scheduledSweep = threadPool.scheduleWithFixedDelay( + { scope.launch { withTimeout(SWEEP_TIMEOUT_MS) { sweepOrphanedAlerts() } } }, + sweepPeriod, + ThreadPool.Names.MANAGEMENT + ) + logger.info("Orphaned alert sweeper scheduled with period: $sweepPeriod") + } + + private fun cancel() { + scheduledSweep?.cancel() + scheduledSweep = null + } + + internal suspend fun sweepOrphanedAlerts() { + try { + if (!clusterService.state().routingTable().hasIndex(ALERT_INDEX)) { + logger.debug("Alert index does not exist, skipping sweep") + return + } + + val alertMonitorIds = getAlertMonitorIds() + if (alertMonitorIds.isEmpty()) { + logger.debug("No alerts found in active index") + return + } + + // Check which monitors still exist in the config index + val existingIds = if (clusterService.state().routingTable().hasIndex(ScheduledJob.SCHEDULED_JOBS_INDEX)) { + getExistingMonitorIds(alertMonitorIds) + } else { + emptySet() + } + + val orphanedMonitorIds = alertMonitorIds - existingIds + if (orphanedMonitorIds.isEmpty()) { + logger.debug("No orphaned alerts found") + return + } + + logger.info("Found ${orphanedMonitorIds.size} orphaned monitor(s) with active alerts: $orphanedMonitorIds") + + // Move orphaned alerts to history + for (monitorId in orphanedMonitorIds) { + try { + AlertMover.moveAlerts(client, monitorId, null) + logger.info("Cleaned up orphaned alerts for monitor [$monitorId]") + } catch (e: Exception) { + logger.error("Failed to clean up orphaned alerts for monitor [$monitorId]", e) + } + } + } catch (e: Exception) { + logger.error("Error during orphaned alert sweep", e) + } + } + + // retrieves the monitor IDs referenced in the .opendistro-alerting-alerts index's alerts + // the possibility here is that some of these monitor IDs might no longer exist + private suspend fun getAlertMonitorIds(): Set { + val agg = TermsAggregationBuilder("monitor_ids") + .field(Alert.MONITOR_ID_FIELD) + .size(1000) + val searchSource = SearchSourceBuilder() + .size(0) + .aggregation(agg) + val searchRequest = SearchRequest(ALERT_INDEX).source(searchSource) + + val response: SearchResponse = client.suspendUntil { search(searchRequest, it) } + val termsAgg = response.aggregations?.get("monitor_ids") ?: return emptySet() + return termsAgg.buckets.map { it.keyAsString }.toSet() + } + + // retrieves the monitor IDs that currently exist in .opendistro-alerting-config + // this is the source of truth for what monitors do and do not exist + private suspend fun getExistingMonitorIds(monitorIds: Set): Set { + val searchSource = SearchSourceBuilder() + .size(monitorIds.size) + .query(QueryBuilders.idsQuery().addIds(*monitorIds.toTypedArray())) + .fetchSource(false) + val searchRequest = SearchRequest(ScheduledJob.SCHEDULED_JOBS_INDEX).source(searchSource) + + val response: SearchResponse = client.suspendUntil { search(searchRequest, it) } + return response.hits.hits.map { it.id }.toSet() + } + + companion object { + val SWEEP_PERIOD = AlertingSettings.ORPHANED_ALERT_SWEEP_PERIOD + val ENABLED = AlertingSettings.ORPHANED_ALERT_SWEEP_ENABLED + const val SWEEP_TIMEOUT_MS = 60_000L + } +} diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt b/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt index d5d0a84fa..8c526a297 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/settings/AlertingSettings.kt @@ -210,6 +210,18 @@ class AlertingSettings { Setting.Property.NodeScope, Setting.Property.Dynamic ) + val ORPHANED_ALERT_SWEEP_PERIOD = Setting.positiveTimeSetting( + "plugins.alerting.orphaned_alert_sweep_period", + TimeValue(5, TimeUnit.MINUTES), + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + + val ORPHANED_ALERT_SWEEP_ENABLED = Setting.boolSetting( + "plugins.alerting.orphaned_alert_sweep_enabled", + true, + Setting.Property.NodeScope, Setting.Property.Dynamic + ) + val REQUEST_TIMEOUT = Setting.positiveTimeSetting( "plugins.alerting.request_timeout", LegacyOpenDistroAlertingSettings.REQUEST_TIMEOUT, diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/OrphanedAlertSweeperIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/OrphanedAlertSweeperIT.kt new file mode 100644 index 000000000..f47f19279 --- /dev/null +++ b/alerting/src/test/kotlin/org/opensearch/alerting/OrphanedAlertSweeperIT.kt @@ -0,0 +1,179 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting + +import org.opensearch.alerting.alerts.AlertIndices +import org.opensearch.alerting.settings.AlertingSettings +import org.opensearch.client.Request +import org.opensearch.client.RequestOptions +import org.opensearch.client.WarningsHandler +import org.opensearch.common.settings.Settings +import org.opensearch.commons.alerting.model.Alert +import org.opensearch.commons.alerting.model.IntervalSchedule +import org.opensearch.commons.alerting.model.PPLTrigger +import org.opensearch.test.OpenSearchTestCase +import java.time.temporal.ChronoUnit.MINUTES +import java.util.concurrent.TimeUnit + +class OrphanedAlertSweeperIT : AlertingRestTestCase() { + + fun `test orphaned alert is moved to history after monitor deletion`() { + // Setup + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + indexDocFromSomeTimeAgo(2, MINUTES, "abc", 5) + + // Create and execute a monitor to generate an alert + val monitor = createMonitor( + randomPPLMonitor( + enabled = true, + schedule = IntervalSchedule(interval = 1, unit = MINUTES), + triggers = listOf( + randomPPLTrigger( + conditionType = PPLTrigger.ConditionType.NUMBER_OF_RESULTS, + numResultsCondition = PPLTrigger.NumResultsCondition.GREATER_THAN, + numResultsValue = 0L, + customCondition = null + ) + ), + query = "source = $TEST_INDEX_NAME | head 10" + ) + ) + executeMonitor(monitor.id) + val alerts = searchAlerts(monitor) + assertEquals("Alert should have been generated", 1, alerts.size) + assertEquals("Alert should be ACTIVE", Alert.State.ACTIVE, alerts.single().state) + + // Disable the JobSweeper so that postDelete doesn't fire and clean up alerts + client().updateSettings("plugins.scheduled_jobs.enabled", false) + + // Delete the monitor directly from the config index (simulating the case where + // JobSweeper.postDelete doesn't fire) + val deleteRequest = Request("DELETE", "/.opendistro-alerting-config/_doc/${monitor.id}?refresh=true") + val options = RequestOptions.DEFAULT.toBuilder() + options.setWarningsHandler(WarningsHandler.PERMISSIVE) + deleteRequest.options = options.build() + adminClient().performRequest(deleteRequest) + + // Verify the alert is still in the active index (orphaned) + val orphanedAlerts = searchAlerts(monitor, AlertIndices.ALERT_INDEX) + assertEquals("Alert should still be in active index (orphaned)", 1, orphanedAlerts.size) + + // Re-enable the JobSweeper and enable the orphaned alert sweeper with a short period + client().updateSettings("plugins.scheduled_jobs.enabled", true) + client().updateSettings(AlertingSettings.ORPHANED_ALERT_SWEEP_ENABLED.key, true) + client().updateSettings(AlertingSettings.ORPHANED_ALERT_SWEEP_PERIOD.key, "1s") + + // Wait for the sweeper to clean up the orphaned alert + OpenSearchTestCase.waitUntil({ + val activeAlerts = searchAlerts(monitor, AlertIndices.ALERT_INDEX) + activeAlerts.isEmpty() + }, 30, TimeUnit.SECONDS) + + // Verify the alert is gone from active index + val activeAlerts = searchAlerts(monitor, AlertIndices.ALERT_INDEX) + assertTrue("Orphaned alert should be removed from active index", activeAlerts.isEmpty()) + + // Verify the alert is in history with DELETED state + val historyAlerts = searchAlerts(monitor, AlertIndices.ALL_ALERT_INDEX_PATTERN) + assertEquals("Alert should be in history", 1, historyAlerts.size) + assertEquals("Alert should be in DELETED state", Alert.State.DELETED, historyAlerts.single().state) + } + + fun `test acknowledged orphaned alert is moved to history after monitor deletion`() { + // Setup + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + indexDocFromSomeTimeAgo(2, MINUTES, "abc", 5) + + // Create, execute, and acknowledge + val monitor = createMonitor( + randomPPLMonitor( + enabled = true, + schedule = IntervalSchedule(interval = 1, unit = MINUTES), + triggers = listOf( + randomPPLTrigger( + conditionType = PPLTrigger.ConditionType.NUMBER_OF_RESULTS, + numResultsCondition = PPLTrigger.NumResultsCondition.GREATER_THAN, + numResultsValue = 0L, + customCondition = null + ) + ), + query = "source = $TEST_INDEX_NAME | head 10" + ) + ) + executeMonitor(monitor.id) + val activeAlert = searchAlerts(monitor).single() + acknowledgeAlerts(monitor, activeAlert) + val ackedAlerts = searchAlerts(monitor) + assertEquals("Alert should be ACKNOWLEDGED", Alert.State.ACKNOWLEDGED, ackedAlerts.single().state) + + // Disable the JobSweeper so that postDelete doesn't fire and clean up alerts + client().updateSettings("plugins.scheduled_jobs.enabled", false) + + // Delete the monitor directly + val deleteRequest = Request("DELETE", "/.opendistro-alerting-config/_doc/${monitor.id}?refresh=true") + val options = RequestOptions.DEFAULT.toBuilder() + options.setWarningsHandler(WarningsHandler.PERMISSIVE) + deleteRequest.options = options.build() + adminClient().performRequest(deleteRequest) + + // Verify the alert is still in the active index (orphaned) + val orphanedAlerts = searchAlerts(monitor, AlertIndices.ALERT_INDEX) + assertEquals("Alert should still be in active index (orphaned)", 1, orphanedAlerts.size) + + // Re-enable the JobSweeper and enable orphaned alert sweeper + client().updateSettings("plugins.scheduled_jobs.enabled", true) + client().updateSettings(AlertingSettings.ORPHANED_ALERT_SWEEP_ENABLED.key, true) + client().updateSettings(AlertingSettings.ORPHANED_ALERT_SWEEP_PERIOD.key, "1s") + + // Wait for cleanup + OpenSearchTestCase.waitUntil({ + val remaining = searchAlerts(monitor, AlertIndices.ALERT_INDEX) + remaining.isEmpty() + }, 30, TimeUnit.SECONDS) + + // Verify moved to history as DELETED + val historyAlerts = searchAlerts(monitor, AlertIndices.ALL_ALERT_INDEX_PATTERN) + assertEquals("Alert should be in history", 1, historyAlerts.size) + assertEquals("Alert should be in DELETED state", Alert.State.DELETED, historyAlerts.single().state) + } + + fun `test non-orphaned alerts are not affected by sweeper`() { + // Setup + createIndex(TEST_INDEX_NAME, Settings.EMPTY, TEST_INDEX_MAPPINGS) + indexDocFromSomeTimeAgo(2, MINUTES, "abc", 5) + + // Create and execute a monitor (DON'T delete it) + val monitor = createMonitor( + randomPPLMonitor( + enabled = true, + schedule = IntervalSchedule(interval = 1, unit = MINUTES), + triggers = listOf( + randomPPLTrigger( + conditionType = PPLTrigger.ConditionType.NUMBER_OF_RESULTS, + numResultsCondition = PPLTrigger.NumResultsCondition.GREATER_THAN, + numResultsValue = 0L, + customCondition = null + ) + ), + query = "source = $TEST_INDEX_NAME | head 10" + ) + ) + executeMonitor(monitor.id) + assertEquals("Alert should exist", 1, searchAlerts(monitor).size) + + // Enable sweeper with short period + client().updateSettings(AlertingSettings.ORPHANED_ALERT_SWEEP_ENABLED.key, true) + client().updateSettings(AlertingSettings.ORPHANED_ALERT_SWEEP_PERIOD.key, "1s") + + // Wait a few sweep cycles + OpenSearchTestCase.waitUntil({ false }, 5, TimeUnit.SECONDS) + + // Alert should still be active (monitor still exists) + val alerts = searchAlerts(monitor) + assertEquals("Non-orphaned alert should remain", 1, alerts.size) + assertEquals("Alert should still be ACTIVE", Alert.State.ACTIVE, alerts.single().state) + } +} diff --git a/core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt b/core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt index 89913e2bf..d40d2ef30 100644 --- a/core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt +++ b/core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt @@ -308,6 +308,7 @@ class JobSweeper( // sweep the shard for new and updated jobs. Uses a search after query to paginate, assuming that any concurrent // updates and deletes are handled by the index operation listener. + val seenJobIds = mutableSetOf() var searchAfter: Long? = startAfter while (searchAfter != null) { val boolQueryBuilder = BoolQueryBuilder() @@ -342,6 +343,7 @@ class JobSweeper( } for (hit in response.hits) { if (shardNodes.isOwningNode(hit.id)) { + seenJobIds.add(hit.id) val xcp = XContentHelper.createParser( xContentRegistry, LoggingDeprecationHandler.INSTANCE, hit.sourceRef, XContentType.JSON @@ -351,6 +353,13 @@ class JobSweeper( } searchAfter = response.hits.lastOrNull()?.seqNo } + + // Deschedule any jobs that are in sweptJobs but no longer exist in the config index + currentJobs.keys.filter { shardNodes.isOwningNode(it) && !seenJobIds.contains(it) }.forEach { + logger.info("Descheduling orphaned job $it — no longer in config index") + scheduler.deschedule(it) + currentJobs.remove(it) + } } private fun sweep( From 162673ad8eb0f739aa1f2a6e517ba893696a745e Mon Sep 17 00:00:00 2001 From: Dennis Toepker Date: Fri, 5 Jun 2026 10:07:01 -0700 Subject: [PATCH 2/2] fix(alerting): Add lifecycle management and resilient coroutine scope to OrphanedAlertSweeper - Implement LifecycleListener with afterStart() to handle the case where the node is already cluster manager at listener registration time. - Add beforeStop() and beforeClose() for clean shutdown of the scheduled task and coroutine scope. - Use SupervisorJob to prevent a failed sweep coroutine from cancelling the scope and silently disabling future sweeps. - Add CoroutineExceptionHandler as a last-resort logger for uncaught exceptions that escape the sweep's try-catch. Signed-off-by: Dennis Toepker --- .../alerting/alerts/OrphanedAlertSweeper.kt | 28 +++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt b/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt index ca1151aca..029ccbc12 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt @@ -5,8 +5,11 @@ package org.opensearch.alerting.alerts +import kotlinx.coroutines.CoroutineExceptionHandler import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeout import org.apache.logging.log4j.LogManager @@ -18,6 +21,7 @@ import org.opensearch.alerting.settings.AlertingSettings import org.opensearch.cluster.ClusterChangedEvent import org.opensearch.cluster.ClusterStateListener import org.opensearch.cluster.service.ClusterService +import org.opensearch.common.lifecycle.LifecycleListener import org.opensearch.common.settings.Settings import org.opensearch.commons.alerting.model.Alert import org.opensearch.commons.alerting.model.ScheduledJob @@ -51,10 +55,13 @@ class OrphanedAlertSweeper( private val client: Client, private val threadPool: ThreadPool, private val clusterService: ClusterService -) : ClusterStateListener { +) : ClusterStateListener, LifecycleListener() { private val logger = LogManager.getLogger(OrphanedAlertSweeper::class.java) - private val scope = CoroutineScope(Dispatchers.IO) + private val exceptionHandler = CoroutineExceptionHandler { _, exception -> + logger.error("Uncaught exception in OrphanedAlertSweeper coroutine", exception) + } + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO + exceptionHandler) @Volatile private var sweepPeriod = SWEEP_PERIOD.get(settings) @Volatile private var enabled = ENABLED.get(settings) @@ -63,6 +70,7 @@ class OrphanedAlertSweeper( init { clusterService.addListener(this) + clusterService.addLifecycleListener(this) clusterService.clusterSettings.addSettingsUpdateConsumer(SWEEP_PERIOD) { sweepPeriod = it if (clusterService.state().nodes.isLocalNodeElectedClusterManager && enabled) reschedule() @@ -75,6 +83,22 @@ class OrphanedAlertSweeper( } } + override fun afterStart() { + if (clusterService.state().nodes.isLocalNodeElectedClusterManager && enabled) { + isClusterManager = true + reschedule() + } + } + + override fun beforeStop() { + cancel() + clusterService.removeListener(this) + } + + override fun beforeClose() { + scope.cancel() + } + override fun clusterChanged(event: ClusterChangedEvent) { if (this.isClusterManager != event.localNodeClusterManager()) { this.isClusterManager = event.localNodeClusterManager()