Skip to content

Use SdkClient for persistence in TransportAcknowledgeAlertAction #2089

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
AWSHurneyt merged 5 commits into from
Apr 10, 2026
Merged

Use SdkClient for persistence in TransportAcknowledgeAlertAction #2089

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
75f8376
Use SdkClient for persistence in TransportAcknowledgeAlertAction
Apr 7, 2026
16fadf8
chore: Retrigger CI after common-utils merge
Apr 9, 2026
26c8bb9
chore: Retrigger CI after SDK 3.6 backport merge
Apr 9, 2026
4679f55
chore: Retrigger CI
Apr 9, 2026
cf76f25
chore: Retrigger CI
Apr 9, 2026
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
152 changes: 85 additions & 67 deletions ...ting/src/main/kotlin/org/opensearch/alerting/transport/TransportAcknowledgeAlertAction.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,23 +11,17 @@ import kotlinx.coroutines.launch
import org.apache.logging.log4j.LogManager
import org.opensearch.ResourceNotFoundException
import org.opensearch.action.ActionRequest
import org.opensearch.action.bulk.BulkRequest
import org.opensearch.action.bulk.BulkResponse
import org.opensearch.action.delete.DeleteRequest
import org.opensearch.action.index.IndexRequest
import org.opensearch.action.search.SearchRequest
import org.opensearch.action.search.SearchResponse
import org.opensearch.action.support.ActionFilters
import org.opensearch.action.support.HandledTransportAction
import org.opensearch.action.update.UpdateRequest
import org.opensearch.alerting.opensearchapi.suspendUntil
import org.opensearch.alerting.settings.AlertingSettings
import org.opensearch.alerting.util.await
import org.opensearch.alerting.util.use
import org.opensearch.cluster.service.ClusterService
import org.opensearch.common.inject.Inject
import org.opensearch.common.settings.Settings
import org.opensearch.common.xcontent.LoggingDeprecationHandler
import org.opensearch.common.xcontent.XContentFactory
import org.opensearch.common.xcontent.XContentHelper
import org.opensearch.common.xcontent.XContentType
import org.opensearch.commons.alerting.action.AcknowledgeAlertRequest
Expand All @@ -42,10 +36,17 @@ import org.opensearch.commons.alerting.util.optionalTimeField
import org.opensearch.commons.utils.recreateObject
import org.opensearch.core.action.ActionListener
import org.opensearch.core.xcontent.NamedXContentRegistry
import org.opensearch.core.xcontent.ToXContentObject
import org.opensearch.core.xcontent.XContentParser
import org.opensearch.core.xcontent.XContentParserUtils
import org.opensearch.index.query.QueryBuilders
import org.opensearch.remote.metadata.client.BulkDataObjectRequest
import org.opensearch.remote.metadata.client.BulkDataObjectResponse
import org.opensearch.remote.metadata.client.DeleteDataObjectRequest
import org.opensearch.remote.metadata.client.PutDataObjectRequest
import org.opensearch.remote.metadata.client.SdkClient
import org.opensearch.remote.metadata.client.SearchDataObjectRequest
import org.opensearch.remote.metadata.client.UpdateDataObjectRequest
import org.opensearch.rest.RestRequest
import org.opensearch.search.builder.SearchSourceBuilder
import org.opensearch.search.fetch.subphase.FetchSourceContext
Expand Down Expand Up @@ -129,18 +130,19 @@ class TransportAcknowledgeAlertAction @Inject constructor(
val queryBuilder = QueryBuilders.boolQuery()
.filter(QueryBuilders.termQuery(Alert.MONITOR_ID_FIELD, request.monitorId))
.filter(QueryBuilders.termsQuery("_id", request.alertIds))
val searchRequest = SearchRequest()
val searchSourceBuilder = SearchSourceBuilder()
.query(queryBuilder)
.version(true)
.seqNoAndPrimaryTerm(true)
.size(request.alertIds.size)
val sdkSearchRequest = SearchDataObjectRequest.builder()
.indices(monitor.dataSources.alertsIndex)
.routing(request.monitorId)
.source(
SearchSourceBuilder()
.query(queryBuilder)
.version(true)
.seqNoAndPrimaryTerm(true)
.size(request.alertIds.size)
)
.searchSourceBuilder(searchSourceBuilder)
.build()
try {
val searchResponse: SearchResponse = client.suspendUntil { client.search(searchRequest, it) }
val searchResponse = sdkClient.searchDataObjectAsync(sdkSearchRequest).await()
.searchResponse() ?: throw RuntimeException("Unknown error loading alerts")
onSearchResponse(searchResponse, monitor)
} catch (t: Exception) {
actionListener.onFailure(AlertingException.wrap(t))
Expand All @@ -149,8 +151,8 @@ class TransportAcknowledgeAlertAction @Inject constructor(

private suspend fun onSearchResponse(response: SearchResponse, monitor: Monitor) {
val alertsHistoryIndex = monitor.dataSources.alertsHistoryIndex
val updateRequests = mutableListOf<UpdateRequest>()
val copyRequests = mutableListOf<IndexRequest>()
val updateRequests = mutableListOf<UpdateDataObjectRequest>()
val copyRequests = mutableListOf<PutDataObjectRequest>()
response.hits.forEach { hit ->
val xcp = XContentHelper.createParser(
xContentRegistry, LoggingDeprecationHandler.INSTANCE,
Expand All @@ -165,50 +167,62 @@ class TransportAcknowledgeAlertAction @Inject constructor(
alert.findingIds.isEmpty() ||
!isAlertHistoryEnabled
) {
val updateRequest = UpdateRequest(monitor.dataSources.alertsIndex, alert.id)
.routing(request.monitorId)
.setIfSeqNo(hit.seqNo)
.setIfPrimaryTerm(hit.primaryTerm)
.doc(
XContentFactory.jsonBuilder().startObject()
.field(Alert.STATE_FIELD, Alert.State.ACKNOWLEDGED.toString())
.optionalTimeField(Alert.ACKNOWLEDGED_TIME_FIELD, Instant.now())
.endObject()
)
updateRequests.add(updateRequest)
updateRequests.add(
UpdateDataObjectRequest.builder()
.index(monitor.dataSources.alertsIndex)
.id(alert.id)
.routing(request.monitorId)
.ifSeqNo(hit.seqNo)
.ifPrimaryTerm(hit.primaryTerm)
.dataObject(
ToXContentObject { builder, _ ->
builder.startObject()
.field(Alert.STATE_FIELD, Alert.State.ACKNOWLEDGED.toString())
.optionalTimeField(Alert.ACKNOWLEDGED_TIME_FIELD, Instant.now())
.endObject()
}
)
.build()
)
} else {
val copyRequest = IndexRequest(alertsHistoryIndex)
.routing(request.monitorId)
.id(alert.id)
.source(
alert.copy(state = Alert.State.ACKNOWLEDGED, acknowledgedTime = Instant.now())
.toXContentWithUser(XContentFactory.jsonBuilder())
)
copyRequests.add(copyRequest)
val ackedAlert = alert.copy(state = Alert.State.ACKNOWLEDGED, acknowledgedTime = Instant.now())
copyRequests.add(
PutDataObjectRequest.builder()
.index(alertsHistoryIndex)
.id(alert.id)
.routing(request.monitorId)
.overwriteIfExists(true)
.dataObject(ToXContentObject { builder, _ -> ackedAlert.toXContentWithUser(builder) })
.build()
)
}
}
}

try {
val updateResponse: BulkResponse? = if (updateRequests.isNotEmpty())
client.suspendUntil {
client.bulk(BulkRequest().add(updateRequests).setRefreshPolicy(request.refreshPolicy), it)
}
else null
val copyResponse: BulkResponse? = if (copyRequests.isNotEmpty())
client.suspendUntil {
client.bulk(BulkRequest().add(copyRequests).setRefreshPolicy(request.refreshPolicy), it)
}
else null
val updateResponse = if (updateRequests.isNotEmpty()) {
val bulkRequest = BulkDataObjectRequest(null)
updateRequests.forEach { bulkRequest.add(it) }
sdkClient.bulkDataObjectAsync(bulkRequest).await()
} else null
val copyResponse = if (copyRequests.isNotEmpty()) {
val bulkRequest = BulkDataObjectRequest(null)
copyRequests.forEach { bulkRequest.add(it) }
sdkClient.bulkDataObjectAsync(bulkRequest).await()
} else null
onBulkResponse(updateResponse, copyResponse, monitor)
} catch (t: Exception) {
log.error("ack error: ${t.message}")
actionListener.onFailure(AlertingException.wrap(t))
}
}

private suspend fun onBulkResponse(updateResponse: BulkResponse?, copyResponse: BulkResponse?, monitor: Monitor) {
val deleteRequests = mutableListOf<DeleteRequest>()
private suspend fun onBulkResponse(
updateResponse: BulkDataObjectResponse?,
copyResponse: BulkDataObjectResponse?,
monitor: Monitor
) {
val deleteRequests = mutableListOf<DeleteDataObjectRequest>()
val missing = request.alertIds.toMutableSet()
val acknowledged = mutableListOf<Alert>()
val failed = mutableListOf<Alert>()
Expand All @@ -220,39 +234,43 @@ class TransportAcknowledgeAlertAction @Inject constructor(
}
}

updateResponse?.items?.forEach { item ->
missing.remove(item.id)
updateResponse?.responses?.forEach { item ->
missing.remove(item.id())
if (item.isFailed) {
failed.add(alerts[item.id]!!)
failed.add(alerts[item.id()]!!)
} else {
acknowledged.add(alerts[item.id]!!)
acknowledged.add(alerts[item.id()]!!)
}
}

copyResponse?.items?.forEach { item ->
copyResponse?.responses?.forEach { item ->
log.info("got a copyResponse: $item")
missing.remove(item.id)
missing.remove(item.id())
if (item.isFailed) {
log.info("got a failureResponse: ${item.failureMessage}")
failed.add(alerts[item.id]!!)
log.info("got a failureResponse: ${item.cause()?.message}")
failed.add(alerts[item.id()]!!)
} else {
val deleteRequest = DeleteRequest(monitor.dataSources.alertsIndex, item.id)
.routing(request.monitorId)
deleteRequests.add(deleteRequest)
deleteRequests.add(
DeleteDataObjectRequest.builder()
.index(monitor.dataSources.alertsIndex)
.id(item.id())
.routing(request.monitorId)
.build()
)
}
}

if (deleteRequests.isNotEmpty()) {
try {
val deleteResponse: BulkResponse = client.suspendUntil {
client.bulk(BulkRequest().add(deleteRequests).setRefreshPolicy(request.refreshPolicy), it)
}
deleteResponse.items.forEach { item ->
missing.remove(item.id)
val bulkRequest = BulkDataObjectRequest(null)
deleteRequests.forEach { bulkRequest.add(it) }
val deleteResponse = sdkClient.bulkDataObjectAsync(bulkRequest).await()
deleteResponse.responses.forEach { item ->
missing.remove(item.id())
if (item.isFailed) {
failed.add(alerts[item.id]!!)
failed.add(alerts[item.id()]!!)
} else {
acknowledged.add(alerts[item.id]!!)
acknowledged.add(alerts[item.id()]!!)
}
}
} catch (t: Exception) {
Expand Down
Loading