-
Notifications
You must be signed in to change notification settings - Fork 96
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
Implemented cross-cluster monitor support #584
Changes from 3 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -43,6 +43,7 @@ | |
val aggregationResultBucket: AggregationResultBucket? = null, | ||
val executionId: String? = null, | ||
val associatedAlertIds: List<String>, | ||
val clusters: List<String>? = null, | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. plz udpate the alert parsing unit tests to account for this field There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Revised. |
||
) : Writeable, ToXContent { | ||
|
||
init { | ||
|
@@ -61,6 +62,7 @@ | |
chainedAlertTrigger: ChainedAlertTrigger, | ||
workflow: Workflow, | ||
associatedAlertIds: List<String>, | ||
clusters: List<String>? = null | ||
) : this( | ||
monitorId = NO_ID, | ||
monitorName = "", | ||
|
@@ -82,7 +84,8 @@ | |
executionId = executionId, | ||
workflowId = workflow.id, | ||
workflowName = workflow.name, | ||
associatedAlertIds = associatedAlertIds | ||
associatedAlertIds = associatedAlertIds, | ||
clusters = clusters | ||
) | ||
|
||
constructor( | ||
|
@@ -97,6 +100,7 @@ | |
schemaVersion: Int = NO_SCHEMA_VERSION, | ||
executionId: String? = null, | ||
workflowId: String? = null, | ||
clusters: List<String>? = null | ||
) : this( | ||
monitorId = monitor.id, | ||
monitorName = monitor.name, | ||
|
@@ -118,7 +122,8 @@ | |
executionId = executionId, | ||
workflowId = workflowId ?: "", | ||
workflowName = "", | ||
associatedAlertIds = emptyList() | ||
associatedAlertIds = emptyList(), | ||
clusters = clusters | ||
) | ||
|
||
constructor( | ||
|
@@ -134,6 +139,7 @@ | |
findingIds: List<String> = emptyList(), | ||
executionId: String? = null, | ||
workflowId: String? = null, | ||
clusters: List<String>? = null | ||
) : this( | ||
monitorId = monitor.id, | ||
monitorName = monitor.name, | ||
|
@@ -155,7 +161,8 @@ | |
executionId = executionId, | ||
workflowId = workflowId ?: "", | ||
workflowName = "", | ||
associatedAlertIds = emptyList() | ||
associatedAlertIds = emptyList(), | ||
clusters = clusters | ||
) | ||
|
||
constructor( | ||
|
@@ -172,6 +179,7 @@ | |
findingIds: List<String> = emptyList(), | ||
executionId: String? = null, | ||
workflowId: String? = null, | ||
clusters: List<String>? = null | ||
) : this( | ||
monitorId = monitor.id, | ||
monitorName = monitor.name, | ||
|
@@ -193,7 +201,8 @@ | |
executionId = executionId, | ||
workflowId = workflowId ?: "", | ||
workflowName = "", | ||
associatedAlertIds = emptyList() | ||
associatedAlertIds = emptyList(), | ||
clusters = clusters | ||
) | ||
|
||
constructor( | ||
|
@@ -211,6 +220,7 @@ | |
schemaVersion: Int = NO_SCHEMA_VERSION, | ||
executionId: String? = null, | ||
workflowId: String? = null, | ||
clusters: List<String>? = null | ||
) : this( | ||
id = id, | ||
monitorId = monitor.id, | ||
|
@@ -233,7 +243,8 @@ | |
executionId = executionId, | ||
workflowId = workflowId ?: "", | ||
workflowName = "", | ||
associatedAlertIds = emptyList() | ||
associatedAlertIds = emptyList(), | ||
clusters = clusters | ||
) | ||
|
||
constructor( | ||
|
@@ -248,6 +259,7 @@ | |
schemaVersion: Int = NO_SCHEMA_VERSION, | ||
workflowId: String? = null, | ||
executionId: String?, | ||
clusters: List<String>? = null | ||
) : this( | ||
id = id, | ||
monitorId = monitor.id, | ||
|
@@ -270,7 +282,8 @@ | |
relatedDocIds = listOf(), | ||
workflowId = workflowId ?: "", | ||
executionId = executionId, | ||
associatedAlertIds = emptyList() | ||
associatedAlertIds = emptyList(), | ||
clusters = clusters | ||
) | ||
|
||
enum class State { | ||
|
@@ -311,7 +324,8 @@ | |
actionExecutionResults = sin.readList(::ActionExecutionResult), | ||
aggregationResultBucket = if (sin.readBoolean()) AggregationResultBucket(sin) else null, | ||
executionId = sin.readOptionalString(), | ||
associatedAlertIds = sin.readStringList() | ||
associatedAlertIds = sin.readStringList(), | ||
clusters = sin.readOptionalStringList() | ||
) | ||
|
||
fun isAcknowledged(): Boolean = (state == State.ACKNOWLEDGED) | ||
|
@@ -349,6 +363,7 @@ | |
} | ||
out.writeOptionalString(executionId) | ||
out.writeStringCollection(associatedAlertIds) | ||
out.writeOptionalStringArray(clusters?.toTypedArray()) | ||
} | ||
|
||
companion object { | ||
|
@@ -379,6 +394,7 @@ | |
const val ASSOCIATED_ALERT_IDS_FIELD = "associated_alert_ids" | ||
const val BUCKET_KEYS = AggregationResultBucket.BUCKET_KEYS | ||
const val PARENTS_BUCKET_PATH = AggregationResultBucket.PARENTS_BUCKET_PATH | ||
const val CLUSTERS_FIELD = "clusters" | ||
const val NO_ID = "" | ||
const val NO_VERSION = Versions.NOT_FOUND | ||
|
||
|
@@ -410,6 +426,7 @@ | |
val actionExecutionResults: MutableList<ActionExecutionResult> = mutableListOf() | ||
var aggAlertBucket: AggregationResultBucket? = null | ||
val associatedAlertIds = mutableListOf<String>() | ||
val clusters = mutableListOf<String>() | ||
ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp) | ||
while (xcp.nextToken() != XContentParser.Token.END_OBJECT) { | ||
val fieldName = xcp.currentName() | ||
|
@@ -476,6 +493,12 @@ | |
AggregationResultBucket.parse(xcp) | ||
} | ||
} | ||
CLUSTERS_FIELD -> { | ||
ensureExpectedToken(XContentParser.Token.START_ARRAY, xcp.currentToken(), xcp) | ||
while (xcp.nextToken() != XContentParser.Token.END_ARRAY) { | ||
clusters.add(xcp.text()) | ||
} | ||
} | ||
} | ||
} | ||
|
||
|
@@ -504,7 +527,8 @@ | |
executionId = executionId, | ||
workflowId = workflowId, | ||
workflowName = workflowName, | ||
associatedAlertIds = associatedAlertIds | ||
associatedAlertIds = associatedAlertIds, | ||
clusters = if (clusters.size > 0) clusters else null | ||
) | ||
} | ||
|
||
|
@@ -554,6 +578,9 @@ | |
.optionalTimeField(END_TIME_FIELD, endTime) | ||
.optionalTimeField(ACKNOWLEDGED_TIME_FIELD, acknowledgedTime) | ||
aggregationResultBucket?.innerXContent(builder) | ||
|
||
if (!clusters.isNullOrEmpty()) builder.field(CLUSTERS_FIELD, clusters.toTypedArray()) | ||
|
||
builder.endObject() | ||
return builder | ||
} | ||
|
@@ -577,7 +604,8 @@ | |
BUCKET_KEYS to aggregationResultBucket?.bucketKeys?.joinToString(","), | ||
PARENTS_BUCKET_PATH to aggregationResultBucket?.parentBucketPath, | ||
FINDING_IDS to findingIds.joinToString(","), | ||
RELATED_DOC_IDS to relatedDocIds.joinToString(",") | ||
RELATED_DOC_IDS to relatedDocIds.joinToString(","), | ||
CLUSTERS_FIELD to clusters?.joinToString(",") | ||
) | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -23,7 +23,8 @@ | |
data class ClusterMetricsInput( | ||
var path: String, | ||
var pathParams: String = "", | ||
var url: String | ||
var url: String, | ||
var clusters: List<String> = listOf() | ||
) : Input { | ||
val clusterMetricType: ClusterMetricType | ||
val constructedUri: URI | ||
|
@@ -74,6 +75,7 @@ | |
.field(PATH_FIELD, path) | ||
.field(PATH_PARAMS_FIELD, pathParams) | ||
.field(URL_FIELD, url) | ||
.field(CLUSTERS_FIELD, clusters) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. same.. update parsing tests for clusterMetricsInput model or plz add a test if non-existent There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Revised. |
||
.endObject() | ||
.endObject() | ||
} | ||
|
@@ -87,6 +89,7 @@ | |
out.writeString(path) | ||
out.writeString(pathParams) | ||
out.writeString(url) | ||
out.writeStringArray(clusters.toTypedArray()) | ||
} | ||
|
||
companion object { | ||
|
@@ -99,6 +102,7 @@ | |
const val PATH_PARAMS_FIELD = "path_params" | ||
const val URL_FIELD = "url" | ||
const val URI_FIELD = "uri" | ||
const val CLUSTERS_FIELD = "clusters" | ||
|
||
val XCONTENT_REGISTRY = NamedXContentRegistry.Entry(Input::class.java, ParseField(URI_FIELD), CheckedFunction { parseInner(it) }) | ||
|
||
|
@@ -110,6 +114,7 @@ | |
var path = "" | ||
var pathParams = "" | ||
var url = "" | ||
val clusters = mutableListOf<String>() | ||
|
||
XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp) | ||
|
||
|
@@ -120,9 +125,17 @@ | |
PATH_FIELD -> path = xcp.text() | ||
PATH_PARAMS_FIELD -> pathParams = xcp.text() | ||
URL_FIELD -> url = xcp.text() | ||
CLUSTERS_FIELD -> { | ||
XContentParserUtils.ensureExpectedToken( | ||
XContentParser.Token.START_ARRAY, | ||
xcp.currentToken(), | ||
xcp | ||
Check warning on line 132 in src/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsInput.kt Codecov / codecov/patchsrc/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsInput.kt#L129-L132
|
||
) | ||
while (xcp.nextToken() != XContentParser.Token.END_ARRAY) clusters.add(xcp.text()) | ||
} | ||
} | ||
} | ||
return ClusterMetricsInput(path, pathParams, url) | ||
return ClusterMetricsInput(path, pathParams, url, clusters) | ||
} | ||
} | ||
|
||
|
@@ -164,7 +177,7 @@ | |
ILLEGAL_PATH_PARAMETER_CHARACTERS.forEach { character -> | ||
if (pathParams.contains(character)) | ||
throw IllegalArgumentException( | ||
"The provided path parameters contain invalid characters or spaces. Please omit: " + "${ILLEGAL_PATH_PARAMETER_CHARACTERS.joinToString(" ")}" | ||
"The provided path parameters contain invalid characters or spaces. Please omit: " + ILLEGAL_PATH_PARAMETER_CHARACTERS.joinToString(" ") | ||
) | ||
} | ||
} | ||
|
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.
will this break security analytics constructor?? if yes, plz create PR for that too
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 is stored here? can you share an example list of opensource clusters?
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.
should this be called clusterNames or clusterIds
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.
Created this PR to update SAP opensearch-project/security-analytics#837