diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt index 5de980bfa..643e79a69 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/MonitorRestApiIT.kt @@ -1559,4 +1559,31 @@ class MonitorRestApiIT : AlertingRestTestCase() { alertingStatsResponse[statsResponseOpenSearchSweeperEnabledField] ) } + + fun `test sweeper works with id field data disabled`() { + client().updateSettings(ScheduledJobSettings.SWEEPER_ENABLED.key, true) + val monitor = createRandomMonitor(refresh = true) + // Disable _id fielddata — this previously broke the sweeper + client().updateSettings("indices.id_field_data.enabled", false) + try { + val monitor2 = createRandomMonitor(refresh = true) + assertNotNull("Monitor was not created", monitor2.id) + val executeResponse = executeMonitor(monitor2.id) + assertEquals("Execute monitor failed", RestStatus.OK, executeResponse.restStatus()) + } finally { + client().updateSettings("indices.id_field_data.enabled", true) + } + } + + fun `test sweeper works after all monitors deleted`() { + client().updateSettings(ScheduledJobSettings.SWEEPER_ENABLED.key, true) + val monitor = createRandomMonitor(refresh = true) + client().makeRequest("DELETE", "$ALERTING_BASE_URI/${monitor.id}") + refreshIndex(ScheduledJob.SCHEDULED_JOBS_INDEX) + // Create a new monitor after index was emptied — sweeper should handle this + val monitor2 = createRandomMonitor(refresh = true) + assertNotNull("Monitor was not created", monitor2.id) + val executeResponse = executeMonitor(monitor2.id) + assertEquals("Execute monitor failed", RestStatus.OK, executeResponse.restStatus()) + } } 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 e08b1360f..89913e2bf 100644 --- a/core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt +++ b/core/src/main/kotlin/org/opensearch/alerting/core/JobSweeper.kt @@ -295,7 +295,7 @@ class JobSweeper( lastFullSweepTimeNano = System.nanoTime() } - private fun sweepShard(shardId: ShardId, shardNodes: ShardNodes, startAfter: String = "") { + private fun sweepShard(shardId: ShardId, shardNodes: ShardNodes, startAfter: Long = -1L) { val logger = Loggers.getLogger(javaClass, shardId) logger.debug("Sweeping shard $shardId") @@ -308,7 +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. - var searchAfter: String? = startAfter + var searchAfter: Long? = startAfter while (searchAfter != null) { val boolQueryBuilder = BoolQueryBuilder() sweepableJobTypes.forEach { boolQueryBuilder.should(QueryBuilders.existsQuery(it)) } @@ -318,18 +318,23 @@ class JobSweeper( .source( SearchSourceBuilder.searchSource() .version(true) + .seqNoAndPrimaryTerm(true) .sort( - FieldSortBuilder("_id") - .unmappedType("keyword") - .missing("_last") + FieldSortBuilder("_seq_no") + .unmappedType("long") ) .searchAfter(arrayOf(searchAfter)) .size(sweepPageSize) .query(boolQueryBuilder) ) - val response = sweepSearchBackoff.retry { - client.search(jobSearchRequest).actionGet(requestTimeout) + val response = try { + sweepSearchBackoff.retry { + client.search(jobSearchRequest).actionGet(requestTimeout) + } + } catch (e: Exception) { + logger.error("Aborting sweep of shard $shardId, will retry on next sweep cycle.", e) + return } if (response.status() != RestStatus.OK) { logger.error("Error sweeping shard $shardId.", response.firstFailureOrNull()) @@ -344,7 +349,7 @@ class JobSweeper( parseAndSweepJob(xcp, shardId, hit.id, hit.version, hit.sourceRef) } } - searchAfter = response.hits.lastOrNull()?.id + searchAfter = response.hits.lastOrNull()?.seqNo } }