Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

separate doc-level monitor query indices for externally defined monitors #1699

Merged
merged 4 commits into from
Oct 16, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -374,6 +374,7 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
// Clean up any queries created by the dry run monitor
monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueriesOnDryRun(monitorMetadata)
}

// TODO: Update the Document as part of the Trigger and return back the trigger action result
return monitorResult.copy(triggerResults = triggerResults, inputResults = inputRunResults)
} catch (e: Exception) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -94,57 +94,69 @@ object DeleteMonitorService :

private suspend fun deleteDocLevelMonitorQueriesAndIndices(monitor: Monitor) {
try {
val metadata = MonitorMetadataService.getMetadata(monitor)
metadata?.sourceToQueryIndexMapping?.forEach { (_, queryIndex) ->
if (monitor.deleteQueryIndexInEveryRun == false) {
val metadata = MonitorMetadataService.getMetadata(monitor)
metadata?.sourceToQueryIndexMapping?.forEach { (_, queryIndex) ->

val indicesExistsResponse: IndicesExistsResponse =
client.suspendUntil {
client.admin().indices().exists(IndicesExistsRequest(queryIndex), it)
val indicesExistsResponse: IndicesExistsResponse =
client.suspendUntil {
client.admin().indices().exists(IndicesExistsRequest(queryIndex), it)
}
if (indicesExistsResponse.isExists == false) {
return
}
if (indicesExistsResponse.isExists == false) {
return
}
// Check if there's any queries from other monitors in this queryIndex,
// to avoid unnecessary doc deletion, if we could just delete index completely
val searchResponse: SearchResponse = client.suspendUntil {
search(
SearchRequest(queryIndex).source(
SearchSourceBuilder()
.size(0)
.query(
QueryBuilders.boolQuery().mustNot(
QueryBuilders.matchQuery("monitor_id", monitor.id)
// Check if there's any queries from other monitors in this queryIndex,
// to avoid unnecessary doc deletion, if we could just delete index completely
val searchResponse: SearchResponse = client.suspendUntil {
search(
SearchRequest(queryIndex).source(
SearchSourceBuilder()
.size(0)
.query(
QueryBuilders.boolQuery().mustNot(
QueryBuilders.matchQuery("monitor_id", monitor.id)
)
)
)
).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_HIDDEN),
it
)
}
if (searchResponse.hits.totalHits.value == 0L) {
val ack: AcknowledgedResponse = client.suspendUntil {
client.admin().indices().delete(
DeleteIndexRequest(queryIndex).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_HIDDEN),
).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_HIDDEN),
it
)
}
if (ack.isAcknowledged == false) {
log.error("Deletion of concrete queryIndex:$queryIndex is not ack'd!")
}
} else {
// Delete all queries added by this monitor
val response: BulkByScrollResponse = suspendCoroutine { cont ->
DeleteByQueryRequestBuilder(client, DeleteByQueryAction.INSTANCE)
.source(queryIndex)
.filter(QueryBuilders.matchQuery("monitor_id", monitor.id))
.refresh(true)
.execute(
object : ActionListener<BulkByScrollResponse> {
override fun onResponse(response: BulkByScrollResponse) = cont.resume(response)
override fun onFailure(t: Exception) = cont.resumeWithException(t)
}
if (searchResponse.hits.totalHits.value == 0L) {
val ack: AcknowledgedResponse = client.suspendUntil {
client.admin().indices().delete(
DeleteIndexRequest(queryIndex).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_HIDDEN),
it
)
}
if (ack.isAcknowledged == false) {
log.error("Deletion of concrete queryIndex:$queryIndex is not ack'd!")
}
} else {
// Delete all queries added by this monitor
val response: BulkByScrollResponse = suspendCoroutine { cont ->
DeleteByQueryRequestBuilder(client, DeleteByQueryAction.INSTANCE)
.source(queryIndex)
.filter(QueryBuilders.matchQuery("monitor_id", monitor.id))
.refresh(true)
.execute(
object : ActionListener<BulkByScrollResponse> {
override fun onResponse(response: BulkByScrollResponse) = cont.resume(response)
override fun onFailure(t: Exception) = cont.resumeWithException(t)
}
)
}
}
}
} else {
val ack: AcknowledgedResponse = client.suspendUntil {
client.admin().indices().delete(
DeleteIndexRequest(monitor.dataSources.queryIndex).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_HIDDEN),
it
)
}
if (ack.isAcknowledged == false) {
log.error("Deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd!")
}
}
} catch (e: Exception) {
// we only log the error and don't fail the request because if monitor document has been deleted successfully,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -719,13 +719,20 @@ class TransportIndexMonitorAction @Inject constructor(
Monitor.MonitorType.valueOf(currentMonitor.monitorType.uppercase(Locale.ROOT)) == Monitor.MonitorType.DOC_LEVEL_MONITOR
) {
updatedMetadata = MonitorMetadataService.recreateRunContext(metadata, currentMonitor)
client.suspendUntil<Client, BulkByScrollResponse> {
DeleteByQueryRequestBuilder(client, DeleteByQueryAction.INSTANCE)
.source(currentMonitor.dataSources.queryIndex)
.filter(QueryBuilders.matchQuery("monitor_id", currentMonitor.id))
.execute(it)
if (docLevelMonitorQueries.docLevelQueryIndexExists(currentMonitor.dataSources)) {
client.suspendUntil<Client, BulkByScrollResponse> {
DeleteByQueryRequestBuilder(client, DeleteByQueryAction.INSTANCE)
.source(currentMonitor.dataSources.queryIndex)
.filter(QueryBuilders.matchQuery("monitor_id", currentMonitor.id))
.execute(it)
}
}
indexDocLevelMonitorQueries(request.monitor, currentMonitor.id, updatedMetadata, request.refreshPolicy)
indexDocLevelMonitorQueries(
request.monitor,
currentMonitor.id,
updatedMetadata,
request.refreshPolicy
)
MonitorMetadataService.upsertMetadata(updatedMetadata, updating = true)
}
actionListener.onResponse(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import org.opensearch.action.admin.indices.settings.put.UpdateSettingsRequest
import org.opensearch.action.bulk.BulkRequest
import org.opensearch.action.bulk.BulkResponse
import org.opensearch.action.index.IndexRequest
import org.opensearch.action.support.IndicesOptions
import org.opensearch.action.support.WriteRequest.RefreshPolicy
import org.opensearch.action.support.master.AcknowledgedResponse
import org.opensearch.alerting.MonitorRunnerService.monitorCtx
Expand Down Expand Up @@ -181,6 +182,16 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
}
}

suspend fun deleteDocLevelQueryIndex(dataSources: DataSources): Boolean {
val ack: AcknowledgedResponse = client.suspendUntil {
client.admin().indices().delete(
DeleteIndexRequest(dataSources.queryIndex).indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_HIDDEN),
it
)
}
return ack.isAcknowledged
}

fun docLevelQueryIndexExists(dataSources: DataSources): Boolean {
val clusterState = clusterService.state()
return clusterState.metadata.hasAlias(dataSources.queryIndex)
Expand Down Expand Up @@ -347,7 +358,8 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
monitorMetadata,
updatedIndexName,
sourceIndexFieldLimit,
updatedProperties
updatedProperties,
indexTimeout
)

if (updateMappingResponse.isAcknowledged) {
Expand Down Expand Up @@ -434,6 +446,7 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
)
)
indexRequests.add(indexRequest)
log.debug("query $query added for execution of monitor $monitorId on index $sourceIndex")
}
log.debug("bulk inserting percolate [${queries.size}] queries")
if (indexRequests.isNotEmpty()) {
Expand All @@ -445,7 +458,7 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
}
bulkResponse.forEach { bulkItemResponse ->
if (bulkItemResponse.isFailed) {
log.debug(bulkItemResponse.failureMessage)
log.error(bulkItemResponse.failureMessage)
}
}
}
Expand Down Expand Up @@ -476,10 +489,16 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
monitorMetadata: MonitorMetadata,
sourceIndex: String,
sourceIndexFieldLimit: Long,
updatedProperties: MutableMap<String, Any>
updatedProperties: MutableMap<String, Any>,
indexTimeout: TimeValue
): Pair<AcknowledgedResponse, String> {
var targetQueryIndex = monitorMetadata.sourceToQueryIndexMapping[sourceIndex + monitor.id]
if (targetQueryIndex == null) {
if (
targetQueryIndex == null || (
targetQueryIndex != monitor.dataSources.queryIndex &&
monitor.deleteQueryIndexInEveryRun == true
)
) {
// queryIndex is alias which will always have only 1 backing index which is writeIndex
// This is due to a fact that that _rollover API would maintain only single index under alias
// if you don't add is_write_index setting when creating index initially
Expand Down Expand Up @@ -535,9 +554,48 @@ class DocLevelMonitorQueries(private val client: Client, private val clusterServ
}
}
} else {
log.debug("unknown exception during PUT mapping on queryIndex: $targetQueryIndex")
val unwrappedException = ExceptionsHelper.unwrapCause(e) as Exception
throw AlertingException.wrap(unwrappedException)
// retry with deleting query index
if (monitor.deleteQueryIndexInEveryRun == true) {
try {
log.error(
"unknown exception during PUT mapping on queryIndex: $targetQueryIndex, " +
"retrying with deletion of query index",
e
)
if (docLevelQueryIndexExists(monitor.dataSources)) {
val ack = monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueryIndex(monitor.dataSources)
if (!ack) {
log.error(
"Deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd! " +
"for monitor ${monitor.id}"
)
}
}
initDocLevelQueryIndex(monitor.dataSources)
indexDocLevelQueries(
monitor = monitor,
monitorId = monitor.id,
monitorMetadata,
indexTimeout = indexTimeout
)
} catch (e: Exception) {
log.error(
"Doc level monitor ${monitor.id}: unknown exception during " +
"PUT mapping on queryIndex: $targetQueryIndex",
e
)
val unwrappedException = ExceptionsHelper.unwrapCause(e) as Exception
throw AlertingException.wrap(unwrappedException)
}
} else {
log.error(
"Doc level monitor ${monitor.id}: unknown exception during " +
"PUT mapping on queryIndex: $targetQueryIndex",
e
)
val unwrappedException = ExceptionsHelper.unwrapCause(e) as Exception
throw AlertingException.wrap(unwrappedException)
}
}
}
// We did rollover, so try to apply mappings again on new targetQueryIndex
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1196,7 +1196,8 @@ class MonitorDataSourcesIT : AlertingSingleNodeTestCase() {
dataSources = DataSources(
queryIndex = customQueryIndex,
queryIndexMappingsByType = mapOf(Pair("text", mapOf(Pair("analyzer", analyzer)))),
)
),
owner = "alerting"
)
try {
createMonitor(monitor)
Expand Down Expand Up @@ -2381,7 +2382,9 @@ class MonitorDataSourcesIT : AlertingSingleNodeTestCase() {
val trigger = randomDocumentLevelTrigger(condition = ALWAYS_RUN)
var monitor = randomDocumentLevelMonitor(
inputs = listOf(docLevelInput),
triggers = listOf(trigger)
triggers = listOf(trigger),
dataSources = DataSources(),
owner = "alerting"
)
// This doc should create close to 10000 (limit) fields in index mapping. It's easier to add mappings like this then via api
val docPayload: StringBuilder = StringBuilder(100000)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,7 @@ class AlertingBackwardsCompatibilityIT : AlertingRestTestCase() {
val indexName = "test_bwc_index"
val bwcMonitorString = """
{
"owner": "alerting",
"type": "monitor",
"name": "test_bwc_monitor",
"enabled": true,
Expand Down
3 changes: 3 additions & 0 deletions core/src/main/resources/mappings/scheduled-jobs.json
Original file line number Diff line number Diff line change
Expand Up @@ -293,6 +293,9 @@
}
}
},
"delete_query_index_in_every_run": {
"type": "boolean"
},
"ui_metadata": {
"type": "object",
"enabled": false
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,7 @@ protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient
"id", null)), trigger1Serialized)),
Map.of(),
new DataSources(),
false,
"sample-remote-monitor-plugin"
);
IndexMonitorRequest indexMonitorRequest1 = new IndexMonitorRequest(
Expand Down Expand Up @@ -154,6 +155,7 @@ public void onFailure(Exception e) {
List.of(),
Map.of(),
new DataSources(),
false,
"sample-remote-monitor-plugin"
);
IndexMonitorRequest indexMonitorRequest2 = new IndexMonitorRequest(
Expand Down Expand Up @@ -237,6 +239,7 @@ public void onFailure(Exception e) {
"id", null)), trigger1Serialized)),
Map.of(),
new DataSources(),
false,
"sample-remote-monitor-plugin"
);
IndexMonitorRequest indexDocLevelMonitorRequest = new IndexMonitorRequest(
Expand Down
Loading