-
Notifications
You must be signed in to change notification settings - Fork 130
Add ActionExecutionPolicy #149
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.
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
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
114 changes: 114 additions & 0 deletions
114
alerting/src/main/kotlin/org/opensearch/alerting/model/action/ActionExecutionPolicy.kt
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| * | ||
| * Modifications Copyright OpenSearch Contributors. See | ||
| * GitHub history for details. | ||
| */ | ||
|
|
||
| package org.opensearch.alerting.model.action | ||
|
|
||
| import org.opensearch.common.io.stream.StreamInput | ||
| import org.opensearch.common.io.stream.StreamOutput | ||
| import org.opensearch.common.io.stream.Writeable | ||
| import org.opensearch.common.xcontent.ToXContent | ||
| import org.opensearch.common.xcontent.ToXContentObject | ||
| import org.opensearch.common.xcontent.XContentBuilder | ||
| import org.opensearch.common.xcontent.XContentParser | ||
| import org.opensearch.common.xcontent.XContentParser.Token | ||
| import org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken | ||
| import java.io.IOException | ||
|
|
||
| /** | ||
| * This class represents the container for various configurations which control Action behavior. | ||
| */ | ||
| // TODO: Should throttleEnabled be included in here as well? | ||
| data class ActionExecutionPolicy( | ||
| val throttle: Throttle? = null, | ||
| val actionExecutionScope: ActionExecutionScope | ||
| ) : Writeable, ToXContentObject { | ||
|
|
||
| @Throws(IOException::class) | ||
| constructor(sin: StreamInput) : this ( | ||
| sin.readOptionalWriteable(::Throttle), // throttle | ||
| ActionExecutionScope.readFrom(sin) // actionExecutionScope | ||
| ) | ||
|
|
||
| override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { | ||
| val xContentBuilder = builder.startObject() | ||
| if (throttle != null) { | ||
| xContentBuilder.field(THROTTLE_FIELD, throttle) | ||
| } | ||
| xContentBuilder.field(ACTION_EXECUTION_SCOPE, actionExecutionScope) | ||
| return xContentBuilder.endObject() | ||
| } | ||
|
|
||
| @Throws(IOException::class) | ||
| override fun writeTo(out: StreamOutput) { | ||
| if (throttle != null) { | ||
| out.writeBoolean(true) | ||
| throttle.writeTo(out) | ||
| } else { | ||
| out.writeBoolean(false) | ||
| } | ||
| if (actionExecutionScope is PerAlertActionScope) { | ||
| out.writeEnum(ActionExecutionScope.Type.PER_ALERT) | ||
| } else { | ||
| out.writeEnum(ActionExecutionScope.Type.PER_EXECUTION) | ||
| } | ||
| actionExecutionScope.writeTo(out) | ||
| } | ||
|
|
||
| companion object { | ||
| const val THROTTLE_FIELD = "throttle" | ||
| const val ACTION_EXECUTION_SCOPE = "action_execution_scope" | ||
|
|
||
| @JvmStatic | ||
| @Throws(IOException::class) | ||
| fun parse(xcp: XContentParser): ActionExecutionPolicy { | ||
| var throttle: Throttle? = null | ||
| lateinit var actionExecutionScope: ActionExecutionScope | ||
|
|
||
| ensureExpectedToken(Token.START_OBJECT, xcp.currentToken(), xcp) | ||
| while (xcp.nextToken() != Token.END_OBJECT) { | ||
| val fieldName = xcp.currentName() | ||
| xcp.nextToken() | ||
|
|
||
| when (fieldName) { | ||
| THROTTLE_FIELD -> { | ||
| throttle = if (xcp.currentToken() == Token.VALUE_NULL) null else Throttle.parse(xcp) | ||
| } | ||
| ACTION_EXECUTION_SCOPE -> actionExecutionScope = ActionExecutionScope.parse(xcp) | ||
| } | ||
| } | ||
|
|
||
| return ActionExecutionPolicy( | ||
| throttle, | ||
| requireNotNull(actionExecutionScope) { "Action execution scope is null" } | ||
| ) | ||
| } | ||
|
|
||
| @JvmStatic | ||
| @Throws(IOException::class) | ||
| fun readFrom(sin: StreamInput): ActionExecutionPolicy { | ||
| return ActionExecutionPolicy(sin) | ||
| } | ||
|
|
||
| /** | ||
| * The default [ActionExecutionPolicy] configuration. | ||
| * | ||
| * This is currently only used by Bucket-Level Monitors and was configured with that in mind. | ||
| * If Query-Level Monitors integrate the use of [ActionExecutionPolicy] then a separate default configuration | ||
| * might need to be made depending on the desired behavior. | ||
| */ | ||
| fun getDefaultConfiguration(): ActionExecutionPolicy { | ||
| val defaultActionExecutionScope = PerAlertActionScope( | ||
| actionableAlerts = setOf(AlertCategory.DEDUPED, AlertCategory.NEW) | ||
| ) | ||
| return ActionExecutionPolicy(throttle = null, actionExecutionScope = defaultActionExecutionScope) | ||
| } | ||
| } | ||
| } |
179 changes: 179 additions & 0 deletions
179
alerting/src/main/kotlin/org/opensearch/alerting/model/action/ActionExecutionScope.kt
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,179 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| * | ||
| * Modifications Copyright OpenSearch Contributors. See | ||
| * GitHub history for details. | ||
| */ | ||
|
|
||
| package org.opensearch.alerting.model.action | ||
|
|
||
| import org.opensearch.common.io.stream.StreamInput | ||
| import org.opensearch.common.io.stream.StreamOutput | ||
| import org.opensearch.common.io.stream.Writeable | ||
| import org.opensearch.common.xcontent.ToXContent | ||
| import org.opensearch.common.xcontent.ToXContentObject | ||
| import org.opensearch.common.xcontent.XContentBuilder | ||
| import org.opensearch.common.xcontent.XContentParser | ||
| import org.opensearch.common.xcontent.XContentParser.Token | ||
| import org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken | ||
| import java.io.IOException | ||
| import java.lang.IllegalArgumentException | ||
|
|
||
| /** | ||
| * This class represents configurations used to control the scope of Action executions when Alerts are created. | ||
| */ | ||
| sealed class ActionExecutionScope : Writeable, ToXContentObject { | ||
|
|
||
| enum class Type { PER_ALERT, PER_EXECUTION } | ||
|
|
||
| companion object { | ||
| const val PER_ALERT_FIELD = "per_alert" | ||
| const val PER_EXECUTION_FIELD = "per_execution" | ||
| const val ACTIONABLE_ALERTS_FIELD = "actionable_alerts" | ||
|
|
||
| @JvmStatic | ||
| @Throws(IOException::class) | ||
| fun parse(xcp: XContentParser): ActionExecutionScope { | ||
| var type: Type? = null | ||
| var actionExecutionScope: ActionExecutionScope? = null | ||
| val alertFilter = mutableSetOf<AlertCategory>() | ||
|
|
||
| ensureExpectedToken(Token.START_OBJECT, xcp.currentToken(), xcp) | ||
| while (xcp.nextToken() != Token.END_OBJECT) { | ||
| val fieldName = xcp.currentName() | ||
| xcp.nextToken() | ||
|
|
||
| // If the type field has already been set, the user has provided more than one type of schedule | ||
| if (type != null) { | ||
| throw IllegalArgumentException("You can only specify one type of action execution scope.") | ||
| } | ||
|
|
||
| when (fieldName) { | ||
| PER_ALERT_FIELD -> { | ||
| type = Type.PER_ALERT | ||
| while (xcp.nextToken() != Token.END_OBJECT) { | ||
| val perAlertFieldName = xcp.currentName() | ||
| xcp.nextToken() | ||
| when (perAlertFieldName) { | ||
| ACTIONABLE_ALERTS_FIELD -> { | ||
| ensureExpectedToken(Token.START_ARRAY, xcp.currentToken(), xcp) | ||
| val allowedCategories = AlertCategory.values().map { it.toString() } | ||
| while (xcp.nextToken() != Token.END_ARRAY) { | ||
| val alertCategory = xcp.text() | ||
| if (!allowedCategories.contains(alertCategory)) { | ||
| throw IllegalStateException("Actionable alerts should be one of $allowedCategories") | ||
| } | ||
| alertFilter.add(AlertCategory.valueOf(alertCategory)) | ||
| } | ||
| } | ||
|
qreshi marked this conversation as resolved.
|
||
| } | ||
| } | ||
| } | ||
| PER_EXECUTION_FIELD -> { | ||
| type = Type.PER_EXECUTION | ||
| while (xcp.nextToken() != Token.END_OBJECT) {} | ||
|
qreshi marked this conversation as resolved.
|
||
| } | ||
| else -> throw IllegalArgumentException("Invalid field [$fieldName] found in action execution scope.") | ||
| } | ||
| } | ||
|
|
||
| if (type == Type.PER_ALERT) { | ||
| actionExecutionScope = PerAlertActionScope(alertFilter) | ||
| } else if (type == Type.PER_EXECUTION) { | ||
| actionExecutionScope = PerExecutionActionScope() | ||
| } | ||
|
|
||
| return requireNotNull(actionExecutionScope) { "Action execution scope is null." } | ||
| } | ||
|
|
||
| @JvmStatic | ||
| @Throws(IOException::class) | ||
| fun readFrom(sin: StreamInput): ActionExecutionScope { | ||
| val type = sin.readEnum(ActionExecutionScope.Type::class.java) | ||
| return if (type == Type.PER_ALERT) { | ||
| PerAlertActionScope(sin) | ||
| } else { | ||
| PerExecutionActionScope(sin) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| abstract fun getExecutionScope(): Type | ||
| } | ||
|
|
||
| data class PerAlertActionScope( | ||
| val actionableAlerts: Set<AlertCategory> | ||
| ) : ActionExecutionScope() { | ||
|
|
||
| @Throws(IOException::class) | ||
| constructor(sin: StreamInput): this( | ||
| sin.readSet { si -> si.readEnum(AlertCategory::class.java) } // alertFilter | ||
| ) | ||
|
|
||
| override fun getExecutionScope(): Type = Type.PER_ALERT | ||
|
|
||
| override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { | ||
| builder.startObject() | ||
| .startObject(PER_ALERT_FIELD) | ||
| .field(ACTIONABLE_ALERTS_FIELD, actionableAlerts.toTypedArray()) | ||
| .endObject() | ||
| return builder.endObject() | ||
| } | ||
|
|
||
| @Throws(IOException::class) | ||
| override fun writeTo(out: StreamOutput) { | ||
| out.writeCollection(actionableAlerts) { o, v -> o.writeEnum(v) } | ||
| } | ||
|
|
||
| companion object { | ||
| @JvmStatic | ||
| @Throws(IOException::class) | ||
| fun readFrom(sin: StreamInput): PerAlertActionScope { | ||
| return PerAlertActionScope(sin) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| class PerExecutionActionScope() : ActionExecutionScope() { | ||
|
|
||
| @Throws(IOException::class) | ||
| constructor(sin: StreamInput) : this() | ||
|
|
||
| override fun hashCode(): Int { | ||
| return javaClass.hashCode() | ||
| } | ||
|
|
||
| // Creating an equals method that just checks class type rather than reference since this is currently stateless. | ||
| // Otherwise, it would have been a dataclass which would have handled this. | ||
| override fun equals(other: Any?): Boolean { | ||
| if (this === other) return true | ||
| if (other?.javaClass != javaClass) return false | ||
| return true | ||
| } | ||
|
|
||
| override fun getExecutionScope(): Type = Type.PER_EXECUTION | ||
|
|
||
| override fun toXContent(builder: XContentBuilder, params: ToXContent.Params): XContentBuilder { | ||
| builder.startObject() | ||
| .startObject(PER_EXECUTION_FIELD) | ||
| .endObject() | ||
| return builder.endObject() | ||
| } | ||
|
|
||
| @Throws(IOException::class) | ||
| override fun writeTo(out: StreamOutput) {} | ||
|
|
||
| companion object { | ||
| @JvmStatic | ||
| @Throws(IOException::class) | ||
| fun readFrom(sin: StreamInput): PerExecutionActionScope { | ||
| return PerExecutionActionScope(sin) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| enum class AlertCategory { DEDUPED, NEW, COMPLETED } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What happens to old Actions which doesn't have the actionExecutionPolicy? Is this change backward compatible?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So Query-Level Monitors don't actually use the
ActionExecutionPolicyas of now. If we make the change in the future to allow it, we'd update the default based on Monitor type to be compatible with both at that time.Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
To be on a safer side, we should try to keep the default behavior for existing query-level monitors as it is i.e. to not add any default field for
actionExecutionPolicyin the data class for query-level monitors. This will ensure both forward and backward compatibility issues.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah, that makes sense. I'll merge in this change for now since it doesn't reflect the full usage of
ActionExecutionPolicy. Once we get to the PR with the changes toMonitorRunner, I think we can updateactionExecutionPolicyto be nullable and set the default configuration at runtime if it is not set.