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..029ccbc12 --- /dev/null +++ b/alerting/src/main/kotlin/org/opensearch/alerting/alerts/OrphanedAlertSweeper.kt @@ -0,0 +1,204 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +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 +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.lifecycle.LifecycleListener +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, LifecycleListener() { + + private val logger = LogManager.getLogger(OrphanedAlertSweeper::class.java) + 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) + private var scheduledSweep: Scheduler.Cancellable? = null + private var isClusterManager = false + + init { + clusterService.addListener(this) + clusterService.addLifecycleListener(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 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() + 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(