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.

Sign up for GitHub

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 #1664

Merged
merged 4 commits into from
Sep 25, 2024
Merged

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
452a51c
separate doc-level monitor query indices for externally defined monitors
sbcd90 Sep 25, 2024
b280761
separate doc-level monitor query indices for externally defined monitors
sbcd90 Sep 25, 2024
a15d5a1
update source to index mappings for detector upgrade scenarios
sbcd90 Sep 25, 2024
90620a0
fix ktlint
sbcd90 Sep 25, 2024
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
15 changes: 15 additions & 0 deletions alerting/src/main/kotlin/org/opensearch/alerting/DocumentLevelMonitorRunner.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,6 +374,13 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
// Clean up any queries created by the dry run monitor
monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueriesOnDryRun(monitorMetadata)
}

if (monitor.dataSources.queryIndex.contains("optimized")) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plz add a flag in monitor creation called deleteQueryIndexBeforeEveryRun (or something briefer)

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added as part of pr opensearch-project/common-utils#734

val ack = monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueryIndex(monitor.dataSources)
if (!ack) {
logger.error("Deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd!")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plz log monitor id

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed it.

}
}
// 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 All @@ -385,6 +392,14 @@ class DocumentLevelMonitorRunner : MonitorRunner() {
RestStatus.INTERNAL_SERVER_ERROR,
e
)
if (monitor.dataSources.queryIndex.contains("optimized") &&
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plz dont write this twice
move this to finally block

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed it.

monitorCtx.docLevelMonitorQueries!!.docLevelQueryIndexExists(monitor.dataSources)
) {
val ack = monitorCtx.docLevelMonitorQueries!!.deleteDocLevelQueryIndex(monitor.dataSources)
if (!ack) {
logger.error("Retry deletion of concrete queryIndex:${monitor.dataSources.queryIndex} is not ack'd!")
}
}
return monitorResult.copy(error = alertingException, inputResults = InputRunResults(emptyList(), alertingException))
} finally {
val endTime = System.currentTimeMillis()
Expand Down
96 changes: 54 additions & 42 deletions alerting/src/main/kotlin/org/opensearch/alerting/service/DeleteMonitorService.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,57 +95,69 @@ object DeleteMonitorService :

private suspend fun deleteDocLevelMonitorQueriesAndIndices(monitor: Monitor) {
try {
val metadata = MonitorMetadataService.getMetadata(monitor)
metadata?.sourceToQueryIndexMapping?.forEach { (_, queryIndex) ->
if (monitor.owner == "alerting") {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plz delete based on a boolean flag that defaults to false.
This is an anti pattern and makes code unmaintainable

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed it based on boolean flag passed from common-utils.

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
24 changes: 17 additions & 7 deletions alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexMonitorAction.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -537,7 +537,8 @@ class TransportIndexMonitorAction @Inject constructor(
if (
request.monitor.isMonitorOfStandardType() &&
Monitor.MonitorType.valueOf(request.monitor.monitorType.uppercase(Locale.ROOT)) ==
Monitor.MonitorType.DOC_LEVEL_MONITOR
Monitor.MonitorType.DOC_LEVEL_MONITOR &&
request.monitor.owner == "alerting"
) {
indexDocLevelMonitorQueries(request.monitor, indexResponse.id, metadata, request.refreshPolicy)
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

https://github.com/opensearch-project/alerting/pull/1664/files#diff-58c6e3aa339ebbcc7ee2452426f17164dc91d435c1f24ee55fdc4f06ec1a371eR546

here source to query mapping is updated but in case of new change query index doesnt even get created and wrong value would come up right?
this needs to be handled too

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the source to query mapping is dynamically updated during the first monitor run as shown in this logic. https://github.com/opensearch-project/alerting/blob/main/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt#L481

Copy link
Member

@eirsep eirsep Sep 25, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this method being called for the monitors with dedicated queryindex also?
https://github.com/opensearch-project/alerting/blob/main/alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt#L473C25-L473C49

this code snippet will always throw exception right?

  targetQueryIndex = getWriteIndexNameForAlias(monitor.dataSources.queryIndex)
            if (targetQueryIndex == null) {
                val message = "Failed to get write index for queryIndex alias:${monitor.dataSources.queryIndex}"
                log.error(message)
                throw AlertingException.wrap(
                    OpenSearchStatusException(message, RestStatus.INTERNAL_SERVER_ERROR)
                )
            }

because the queryIndex is not an alias anymore?

Expand Down Expand Up @@ -702,13 +703,22 @@ 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)
}
}
if (currentMonitor.owner == "alerting") {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

refactor to use flag

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed 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
11 changes: 11 additions & 0 deletions alerting/src/main/kotlin/org/opensearch/alerting/util/DocLevelMonitorQueries.kt
View file
Open in desktop
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 {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we do exists check before deleting

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed it.

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
7 changes: 5 additions & 2 deletions alerting/src/test/kotlin/org/opensearch/alerting/MonitorDataSourcesIT.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1195,7 +1195,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 @@ -2379,7 +2380,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"
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plz add 2 new tests-

  1. scenario where delete query index flag is false and monitor execution fails due to mapping conflict on alias rollover
  2. scenario where delete query index flag is true and monitor execution succeeds

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in first test plz also update monitor with flag false to true and do rollover again and this time assert monitor execution succeeeds

)
// 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
1 change: 1 addition & 0 deletions alerting/src/test/kotlin/org/opensearch/alerting/bwc/AlertingBackwardsCompatibilityIT.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,7 @@ class AlertingBackwardsCompatibilityIT : AlertingRestTestCase() {
val indexName = "test_bwc_index"
val bwcMonitorString = """
{
"owner": "alerting",
"type": "monitor",
"name": "test_bwc_monitor",
"enabled": true,
Expand Down
Loading