Skip to content

[BUG] ExecuteMonitor inserting metadata doc during dry run #758

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

Merged
merged 3 commits into from
Feb 3, 2023
Merged

[BUG] ExecuteMonitor inserting metadata doc during dry run #758

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9e92fb5
execute monitor bugfix
petardz Jan 16, 2023
e61b19c
added IT
petardz Jan 17, 2023
95d3be7
fixed created retval when skipIndex=true
petardz Jan 17, 2023
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
8 changes: 6 additions & 2 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 @@ -65,6 +65,7 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
dryrun: Boolean
): MonitorRunResult<DocumentLevelTriggerRunResult> {
logger.debug("Document-level-monitor is running ...")
val isTempMonitor = dryrun || monitor.id == Monitor.NO_ID
var monitorResult = MonitorRunResult<DocumentLevelTriggerRunResult>(monitor.name, periodStart, periodEnd)

try {
Expand All @@ -84,7 +85,11 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
return monitorResult.copy(error = AlertingException.wrap(e))
}

var (monitorMetadata, _) = MonitorMetadataService.getOrCreateMetadata(monitor, createWithRunContext = false)
var (monitorMetadata, _) = MonitorMetadataService.getOrCreateMetadata(
monitor = monitor,
createWithRunContext = false,
skipIndex = isTempMonitor
)

monitorCtx.docLevelMonitorQueries!!.initDocLevelQueryIndex(monitor.dataSources)
monitorCtx.docLevelMonitorQueries!!.indexDocLevelQueries(
Expand All @@ -98,7 +103,6 @@ object DocumentLevelMonitorRunner : MonitorRunner() {
val index = docLevelMonitorInput.indices[0]
val queries: List<DocLevelQuery> = docLevelMonitorInput.queries

val isTempMonitor = dryrun || monitor.id == Monitor.NO_ID
val lastRunContext = if (monitorMetadata.lastRunContext.isNullOrEmpty()) mutableMapOf()
else monitorMetadata.lastRunContext.toMutableMap() as MutableMap<String, MutableMap<String, Any>>

Expand Down
12 changes: 10 additions & 2 deletions alerting/src/main/kotlin/org/opensearch/alerting/MonitorMetadataService.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,15 +110,23 @@ object MonitorMetadataService :
}
}

suspend fun getOrCreateMetadata(monitor: Monitor, createWithRunContext: Boolean = true): Pair<MonitorMetadata, Boolean> {
suspend fun getOrCreateMetadata(
monitor: Monitor,
createWithRunContext: Boolean = true,
skipIndex: Boolean = false
): Pair<MonitorMetadata, Boolean> {
try {
val created = true
val metadata = getMetadata(monitor)
return if (metadata != null) {
metadata to !created
} else {
val newMetadata = createNewMetadata(monitor, createWithRunContext = createWithRunContext)
upsertMetadata(newMetadata, updating = false) to created
if (skipIndex) {
newMetadata to created
} else {
upsertMetadata(newMetadata, updating = false) to created
}
}
} catch (e: Exception) {
throw AlertingException.wrap(e)
Expand Down
2 changes: 1 addition & 1 deletion alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportExecuteMonitorAction.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ class TransportExecuteMonitorAction @Inject constructor(
docLevelMonitorQueries.initDocLevelQueryIndex(monitor.dataSources)
log.info("Central Percolation index ${ScheduledJob.DOC_LEVEL_QUERIES_INDEX} created")
}
val (metadata, _) = MonitorMetadataService.getOrCreateMetadata(monitor)
val (metadata, _) = MonitorMetadataService.getOrCreateMetadata(monitor, skipIndex = true)
Copy link
Member

Choose a reason for hiding this comment

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

This should only be true if the dryrun param is true

Copy link
Contributor Author

Choose a reason for hiding this comment

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

even if monitorId is null ?

Copy link
Contributor Author

@petardz petardz Jan 17, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

I think if monitorId is null we definitely can't save metadata (there's no monitor to tie it to).

Dryrun can be true for existing monitor and we take that into account before indexing metadata inside DocumentLevelMonitorRunner here

Copy link
Member

Choose a reason for hiding this comment

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

Oh I missed that part of the logic, good call out.

docLevelMonitorQueries.indexDocLevelQueries(
monitor,
monitor.id,
Expand Down
2 changes: 2 additions & 0 deletions alerting/src/test/kotlin/org/opensearch/alerting/DocumentMonitorRunnerIT.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,8 @@ class DocumentMonitorRunnerIT : AlertingRestTestCase() {
}
}

refreshAllIndices()

val alerts = searchAlertsWithFilter(monitor)
assertEquals("Alert saved for test monitor", 2, alerts.size)

Expand Down
42 changes: 42 additions & 0 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 @@ -19,6 +19,7 @@ import org.opensearch.alerting.action.SearchMonitorAction
import org.opensearch.alerting.action.SearchMonitorRequest
import org.opensearch.alerting.alerts.AlertIndices
import org.opensearch.alerting.core.ScheduledJobIndices
import org.opensearch.alerting.model.DocumentLevelTriggerRunResult
import org.opensearch.alerting.transport.AlertingSingleNodeTestCase
import org.opensearch.common.settings.Settings
import org.opensearch.common.xcontent.XContentType
Expand Down Expand Up @@ -341,6 +342,47 @@ class MonitorDataSourcesIT : AlertingSingleNodeTestCase() {
assertEquals("Didn't match all 4 queries", 1, findings[0].docLevelQueries.size)
}

fun `test execute monitor without create when no monitors exists`() {
val docQuery = DocLevelQuery(query = "test_field:\"us-west-2\"", name = "3")
val docLevelInput = DocLevelMonitorInput("description", listOf(index), listOf(docQuery))
val trigger = randomDocumentLevelTrigger(condition = ALWAYS_RUN)
val customQueryIndex = "custom_alerts_index"
val analyzer = "whitespace"
var monitor = randomDocumentLevelMonitor(
inputs = listOf(docLevelInput),
triggers = listOf(trigger),
dataSources = DataSources(
queryIndex = customQueryIndex,
queryIndexMappingsByType = mapOf(Pair("text", mapOf(Pair("analyzer", analyzer)))),
)
)
var executeMonitorResponse = executeMonitor(monitor, null)
val testTime = DateTimeFormatter.ISO_OFFSET_DATE_TIME.format(ZonedDateTime.now().truncatedTo(MILLIS))
val testDoc = """{
"message" : "This is an error from IAD region",
"test_strict_date_time" : "$testTime",
"test_field" : "us-west-2"
}"""

assertIndexNotExists(SCHEDULED_JOBS_INDEX)

val createMonitorResponse = createMonitor(monitor)

assertIndexExists(SCHEDULED_JOBS_INDEX)

indexDoc(index, "1", testDoc)

executeMonitorResponse = executeMonitor(monitor, createMonitorResponse?.id, dryRun = false)

Assert.assertEquals(executeMonitorResponse!!.monitorRunResult.monitorName, monitor.name)
Assert.assertEquals(executeMonitorResponse.monitorRunResult.triggerResults.size, 1)
Assert.assertEquals(
(executeMonitorResponse.monitorRunResult.triggerResults.iterator().next().value as DocumentLevelTriggerRunResult)
.triggeredDocs.size,
1
)
}

fun `test execute monitor with custom query index and custom field mappings`() {
val docQuery = DocLevelQuery(query = "test_field:\"us-west-2\"", name = "3")
val docLevelInput = DocLevelMonitorInput("description", listOf(index), listOf(docQuery))
Expand Down
2 changes: 1 addition & 1 deletion alerting/src/test/kotlin/org/opensearch/alerting/transport/AlertingSingleNodeTestCase.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ abstract class AlertingSingleNodeTestCase : OpenSearchSingleNodeTestCase() {
return getIndexResponse.indices().toList()
}

protected fun executeMonitor(monitor: Monitor, id: String, dryRun: Boolean = true): ExecuteMonitorResponse? {
protected fun executeMonitor(monitor: Monitor, id: String?, dryRun: Boolean = true): ExecuteMonitorResponse? {
val request = ExecuteMonitorRequest(dryRun, TimeValue(Instant.now().toEpochMilli()), id, monitor)
return client().execute(ExecuteMonitorAction.INSTANCE, request).get()
}
Expand Down