Skip to content

Refactor TransportAcknowledgeAlertAction to extend SecureTransportAction. #14

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

Draft
AWSHurneyt wants to merge 1 commit into
base: add-filter-by-backend-roles-setting-copy
Choose a base branch
from
Draft

Refactor TransportAcknowledgeAlertAction to extend SecureTransportAction. #14

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
76 changes: 44 additions & 32 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 @@ -8,16 +8,13 @@ package org.opensearch.alerting.transport
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.newSingleThreadContext
import kotlinx.coroutines.withContext
import org.apache.logging.log4j.LogManager
import org.opensearch.ResourceNotFoundException
import org.opensearch.action.ActionRequest
import org.opensearch.action.search.SearchResponse
import org.opensearch.action.support.ActionFilters
import org.opensearch.action.support.HandledTransportAction
import org.opensearch.alerting.AlertingPlugin
import org.opensearch.alerting.opensearchapi.suspendUntil
import org.opensearch.alerting.settings.AlertingSettings
import org.opensearch.alerting.util.await
import org.opensearch.alerting.util.use
Expand All @@ -30,13 +27,13 @@ import org.opensearch.common.xcontent.XContentType
import org.opensearch.commons.alerting.action.AcknowledgeAlertRequest
import org.opensearch.commons.alerting.action.AcknowledgeAlertResponse
import org.opensearch.commons.alerting.action.AlertingActions
import org.opensearch.commons.alerting.action.GetMonitorRequest
import org.opensearch.commons.alerting.action.GetMonitorResponse
import org.opensearch.commons.alerting.model.Alert
import org.opensearch.commons.alerting.model.Monitor
import org.opensearch.commons.alerting.model.ScheduledJob
import org.opensearch.commons.alerting.util.AlertingException
import org.opensearch.commons.alerting.util.optionalTimeField
import org.opensearch.commons.utils.TenantContext
import org.opensearch.commons.utils.currentTenantId
import org.opensearch.commons.utils.recreateObject
import org.opensearch.core.action.ActionListener
import org.opensearch.core.xcontent.NamedXContentRegistry
Expand All @@ -47,13 +44,12 @@ 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.GetDataObjectRequest
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
import org.opensearch.tasks.Task
import org.opensearch.transport.TransportService
import org.opensearch.transport.client.Client
Expand All @@ -74,13 +70,18 @@ class TransportAcknowledgeAlertAction @Inject constructor(
val sdkClient: SdkClient
) : HandledTransportAction<ActionRequest, AcknowledgeAlertResponse>(
AlertingActions.ACKNOWLEDGE_ALERTS_ACTION_NAME, transportService, actionFilters, ::AcknowledgeAlertRequest
) {
),
SecureTransportAction {

@Volatile override var filterByEnabled = AlertingSettings.FILTER_BY_BACKEND_ROLES.get(settings)
@Volatile override var filterByAccessStrategy = AlertingSettings.FILTER_BY_BACKEND_ROLES_ACCESS_STRATEGY.get(settings)

@Volatile
private var isAlertHistoryEnabled = AlertingSettings.ALERT_HISTORY_ENABLED.get(settings)

init {
clusterService.clusterSettings.addSettingsUpdateConsumer(AlertingSettings.ALERT_HISTORY_ENABLED) { isAlertHistoryEnabled = it }
listenFilterBySettingChange(clusterService)
}

override fun doExecute(
Expand All @@ -90,30 +91,18 @@ class TransportAcknowledgeAlertAction @Inject constructor(
) {
val request = acknowledgeAlertRequest as? AcknowledgeAlertRequest
?: recreateObject(acknowledgeAlertRequest) { AcknowledgeAlertRequest(it) }
val tenantId = client.threadPool().threadContext.getHeader(AlertingPlugin.TENANT_ID_HEADER)
val user = readUserFromThreadContext(client)

if (!validateUserBackendRoles(user, actionListener)) {
return
}

val tenantId = client.threadPool().threadContext.getHeader(AlertingPlugin.TENANT_ID_HEADER)
client.threadPool().threadContext.stashContext().use {
scope.launch(TenantContext(tenantId)) {
try {
val singleThreadContext = newSingleThreadContext("TransportAcknowledgeAlertAction")
var getMonitorResponse: GetMonitorResponse? = null
withContext(singleThreadContext) {
// in the case where filter by backend role is enabled, we need the user context
// which is stashed via `stashContext()` above to be restored here so that it will be
// used in the get monitor request.
it.restore()
getMonitorResponse =
transportGetMonitorAction.client.suspendUntil {
val getMonitorRequest = GetMonitorRequest(
monitorId = request.monitorId,
-3L,
RestRequest.Method.GET,
FetchSourceContext.FETCH_SOURCE
)
execute(AlertingActions.GET_MONITOR_ACTION_TYPE, getMonitorRequest, it)
}
}
if (getMonitorResponse == null || getMonitorResponse?.monitor == null) {
val monitor = getMonitor(request.monitorId)
if (monitor == null) {
actionListener.onFailure(
AlertingException.wrap(
ResourceNotFoundException(
Expand All @@ -125,10 +114,14 @@ class TransportAcknowledgeAlertAction @Inject constructor(
)
)
)
} else {
getMonitorResponse?.let { getMonitorResponseSafe ->
AcknowledgeHandler(client, actionListener, request).start(getMonitorResponseSafe.monitor!!)
}
return@launch
}

val canAccess = user == null || !doFilterForUser(user) ||
checkUserPermissionsWithResource(user, monitor.user, actionListener, "monitor", request.monitorId)

if (canAccess) {
AcknowledgeHandler(client, actionListener, request).start(monitor)
}
} catch (e: Exception) {
log.error("Failed to launch acknowledge handler", e)
Expand All @@ -138,6 +131,25 @@ class TransportAcknowledgeAlertAction @Inject constructor(
}
}

private suspend fun getMonitor(monitorId: String): Monitor? {
val tenantId = currentTenantId()
val getRequest = GetDataObjectRequest.builder()
.index(ScheduledJob.SCHEDULED_JOBS_INDEX)
.id(monitorId)
.tenantId(tenantId)
.build()
val response = sdkClient.getDataObject(getRequest)
val getResponse = response.getResponse()
if (getResponse == null || !getResponse.isExists) {
return null
}
val xcp = XContentHelper.createParser(
xContentRegistry, LoggingDeprecationHandler.INSTANCE,
getResponse.sourceAsBytesRef, XContentType.JSON
)
return ScheduledJob.parse(xcp, getResponse.id, getResponse.version) as Monitor
}

inner class AcknowledgeHandler(
private val client: Client,
private val actionListener: ActionListener<AcknowledgeAlertResponse>,
Expand Down
Loading