diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportAcknowledgeAlertAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportAcknowledgeAlertAction.kt index ae6581824..a21b924d8 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportAcknowledgeAlertAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportAcknowledgeAlertAction.kt @@ -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 @@ -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 @@ -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)) @@ -149,8 +151,8 @@ class TransportAcknowledgeAlertAction @Inject constructor( private suspend fun onSearchResponse(response: SearchResponse, monitor: Monitor) { val alertsHistoryIndex = monitor.dataSources.alertsHistoryIndex - val updateRequests = mutableListOf() - val copyRequests = mutableListOf() + val updateRequests = mutableListOf() + val copyRequests = mutableListOf() response.hits.forEach { hit -> val xcp = XContentHelper.createParser( xContentRegistry, LoggingDeprecationHandler.INSTANCE, @@ -165,41 +167,49 @@ 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}") @@ -207,8 +217,12 @@ class TransportAcknowledgeAlertAction @Inject constructor( } } - private suspend fun onBulkResponse(updateResponse: BulkResponse?, copyResponse: BulkResponse?, monitor: Monitor) { - val deleteRequests = mutableListOf() + private suspend fun onBulkResponse( + updateResponse: BulkDataObjectResponse?, + copyResponse: BulkDataObjectResponse?, + monitor: Monitor + ) { + val deleteRequests = mutableListOf() val missing = request.alertIds.toMutableSet() val acknowledged = mutableListOf() val failed = mutableListOf() @@ -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) {