Skip to content

Commit 0fd0e2e

Browse files
AWSHurneytAWSHurneyt
committed
[Backport 2.x] Implemented cross-cluster monitor support opensearch-project#584 (opensearch-project#586)
* Added clusters field to support cross cluster cluster metrics monitors. Signed-off-by: AWSHurneyt <hurneyt@amazon.com> * Fixed writeTo. Signed-off-by: AWSHurneyt <hurneyt@amazon.com> * Updated tests. Signed-off-by: AWSHurneyt <hurneyt@amazon.com> * Updated tests. Signed-off-by: AWSHurneyt <hurneyt@amazon.com> --------- Signed-off-by: AWSHurneyt <hurneyt@amazon.com>
1 parent 56c94d0 commit 0fd0e2e

File tree

6 files changed

+101
-57
lines changed

6 files changed

+101
-57
lines changed

‎src/main/kotlin/org/opensearch/commons/alerting/model/Alert.kt‎

Lines changed: 37 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ data class Alert(
4343
val aggregationResultBucket: AggregationResultBucket? = null,
4444
val executionId: String? = null,
4545
val associatedAlertIds: List<String>,
46+
val clusters: List<String>? = null,
4647
) : Writeable, ToXContent {
4748

4849
init {
@@ -61,6 +62,7 @@ data class Alert(
6162
chainedAlertTrigger: ChainedAlertTrigger,
6263
workflow: Workflow,
6364
associatedAlertIds: List<String>,
65+
clusters: List<String>? = null
6466
) : this(
6567
monitorId = NO_ID,
6668
monitorName = "",
@@ -82,7 +84,8 @@ data class Alert(
8284
executionId = executionId,
8385
workflowId = workflow.id,
8486
workflowName = workflow.name,
85-
associatedAlertIds = associatedAlertIds
87+
associatedAlertIds = associatedAlertIds,
88+
clusters = clusters
8689
)
8790

8891
constructor(
@@ -97,6 +100,7 @@ data class Alert(
97100
schemaVersion: Int = NO_SCHEMA_VERSION,
98101
executionId: String? = null,
99102
workflowId: String? = null,
103+
clusters: List<String>? = null
100104
) : this(
101105
monitorId = monitor.id,
102106
monitorName = monitor.name,
@@ -118,7 +122,8 @@ data class Alert(
118122
executionId = executionId,
119123
workflowId = workflowId ?: "",
120124
workflowName = "",
121-
associatedAlertIds = emptyList()
125+
associatedAlertIds = emptyList(),
126+
clusters = clusters
122127
)
123128

124129
constructor(
@@ -134,6 +139,7 @@ data class Alert(
134139
findingIds: List<String> = emptyList(),
135140
executionId: String? = null,
136141
workflowId: String? = null,
142+
clusters: List<String>? = null
137143
) : this(
138144
monitorId = monitor.id,
139145
monitorName = monitor.name,
@@ -155,7 +161,8 @@ data class Alert(
155161
executionId = executionId,
156162
workflowId = workflowId ?: "",
157163
workflowName = "",
158-
associatedAlertIds = emptyList()
164+
associatedAlertIds = emptyList(),
165+
clusters = clusters
159166
)
160167

161168
constructor(
@@ -172,6 +179,7 @@ data class Alert(
172179
findingIds: List<String> = emptyList(),
173180
executionId: String? = null,
174181
workflowId: String? = null,
182+
clusters: List<String>? = null
175183
) : this(
176184
monitorId = monitor.id,
177185
monitorName = monitor.name,
@@ -193,7 +201,8 @@ data class Alert(
193201
executionId = executionId,
194202
workflowId = workflowId ?: "",
195203
workflowName = "",
196-
associatedAlertIds = emptyList()
204+
associatedAlertIds = emptyList(),
205+
clusters = clusters
197206
)
198207

199208
constructor(
@@ -211,6 +220,7 @@ data class Alert(
211220
schemaVersion: Int = NO_SCHEMA_VERSION,
212221
executionId: String? = null,
213222
workflowId: String? = null,
223+
clusters: List<String>? = null
214224
) : this(
215225
id = id,
216226
monitorId = monitor.id,
@@ -233,7 +243,8 @@ data class Alert(
233243
executionId = executionId,
234244
workflowId = workflowId ?: "",
235245
workflowName = "",
236-
associatedAlertIds = emptyList()
246+
associatedAlertIds = emptyList(),
247+
clusters = clusters
237248
)
238249

239250
constructor(
@@ -248,6 +259,7 @@ data class Alert(
248259
schemaVersion: Int = NO_SCHEMA_VERSION,
249260
workflowId: String? = null,
250261
executionId: String?,
262+
clusters: List<String>? = null
251263
) : this(
252264
id = id,
253265
monitorId = monitor.id,
@@ -270,7 +282,8 @@ data class Alert(
270282
relatedDocIds = listOf(),
271283
workflowId = workflowId ?: "",
272284
executionId = executionId,
273-
associatedAlertIds = emptyList()
285+
associatedAlertIds = emptyList(),
286+
clusters = clusters
274287
)
275288

276289
enum class State {
@@ -311,7 +324,8 @@ data class Alert(
311324
actionExecutionResults = sin.readList(::ActionExecutionResult),
312325
aggregationResultBucket = if (sin.readBoolean()) AggregationResultBucket(sin) else null,
313326
executionId = sin.readOptionalString(),
314-
associatedAlertIds = sin.readStringList()
327+
associatedAlertIds = sin.readStringList(),
328+
clusters = sin.readOptionalStringList()
315329
)
316330

317331
fun isAcknowledged(): Boolean = (state == State.ACKNOWLEDGED)
@@ -349,6 +363,7 @@ data class Alert(
349363
}
350364
out.writeOptionalString(executionId)
351365
out.writeStringCollection(associatedAlertIds)
366+
out.writeOptionalStringArray(clusters?.toTypedArray())
352367
}
353368

354369
companion object {
@@ -379,6 +394,7 @@ data class Alert(
379394
const val ASSOCIATED_ALERT_IDS_FIELD = "associated_alert_ids"
380395
const val BUCKET_KEYS = AggregationResultBucket.BUCKET_KEYS
381396
const val PARENTS_BUCKET_PATH = AggregationResultBucket.PARENTS_BUCKET_PATH
397+
const val CLUSTERS_FIELD = "clusters"
382398
const val NO_ID = ""
383399
const val NO_VERSION = Versions.NOT_FOUND
384400

@@ -409,6 +425,7 @@ data class Alert(
409425
val actionExecutionResults: MutableList<ActionExecutionResult> = mutableListOf()
410426
var aggAlertBucket: AggregationResultBucket? = null
411427
val associatedAlertIds = mutableListOf<String>()
428+
val clusters = mutableListOf<String>()
412429
ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
413430
while (xcp.nextToken() != XContentParser.Token.END_OBJECT) {
414431
val fieldName = xcp.currentName()
@@ -475,6 +492,12 @@ data class Alert(
475492
AggregationResultBucket.parse(xcp)
476493
}
477494
}
495+
CLUSTERS_FIELD -> {
496+
ensureExpectedToken(XContentParser.Token.START_ARRAY, xcp.currentToken(), xcp)
497+
while (xcp.nextToken() != XContentParser.Token.END_ARRAY) {
498+
clusters.add(xcp.text())
499+
}
500+
}
478501
}
479502
}
480503

@@ -503,7 +526,8 @@ data class Alert(
503526
executionId = executionId,
504527
workflowId = workflowId,
505528
workflowName = workflowName,
506-
associatedAlertIds = associatedAlertIds
529+
associatedAlertIds = associatedAlertIds,
530+
clusters = if (clusters.size > 0) clusters else null
507531
)
508532
}
509533

@@ -553,6 +577,9 @@ data class Alert(
553577
.optionalTimeField(END_TIME_FIELD, endTime)
554578
.optionalTimeField(ACKNOWLEDGED_TIME_FIELD, acknowledgedTime)
555579
aggregationResultBucket?.innerXContent(builder)
580+
581+
if (!clusters.isNullOrEmpty()) builder.field(CLUSTERS_FIELD, clusters.toTypedArray())
582+
556583
builder.endObject()
557584
return builder
558585
}
@@ -576,7 +603,8 @@ data class Alert(
576603
BUCKET_KEYS to aggregationResultBucket?.bucketKeys?.joinToString(","),
577604
PARENTS_BUCKET_PATH to aggregationResultBucket?.parentBucketPath,
578605
FINDING_IDS to findingIds.joinToString(","),
579-
RELATED_DOC_IDS to relatedDocIds.joinToString(",")
606+
RELATED_DOC_IDS to relatedDocIds.joinToString(","),
607+
CLUSTERS_FIELD to clusters?.joinToString(",")
580608
)
581609
}
582610
}

‎src/main/kotlin/org/opensearch/commons/alerting/model/ClusterMetricsInput.kt‎

Lines changed: 44 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import org.opensearch.core.xcontent.XContentParser
1313
import org.opensearch.core.xcontent.XContentParserUtils
1414
import java.io.IOException
1515
import java.net.URI
16+
import java.net.URISyntaxException
1617

1718
val ILLEGAL_PATH_PARAMETER_CHARACTERS = arrayOf(':', '"', '+', '\\', '|', '?', '#', '>', '<', ' ')
1819

@@ -22,7 +23,8 @@ val ILLEGAL_PATH_PARAMETER_CHARACTERS = arrayOf(':', '"', '+', '\\', '|', '?', '
2223
data class ClusterMetricsInput(
2324
var path: String,
2425
var pathParams: String = "",
25-
var url: String
26+
var url: String,
27+
var clusters: List<String> = listOf()
2628
) : Input {
2729
val clusterMetricType: ClusterMetricType
2830
val constructedUri: URI
@@ -43,11 +45,10 @@ data class ClusterMetricsInput(
4345
"Invalid URI constructed from the path and path_params inputs, or the url input."
4446
}
4547

46-
if (url.isNotEmpty() && validateFieldsNotEmpty()) {
48+
if (url.isNotEmpty() && validateFieldsNotEmpty())
4749
require(constructedUri == constructUrlFromInputs()) {
4850
"The provided URL and URI fields form different URLs."
4951
}
50-
}
5152

5253
require(constructedUri.host.lowercase() == SUPPORTED_HOST) {
5354
"Only host '$SUPPORTED_HOST' is supported."
@@ -74,6 +75,7 @@ data class ClusterMetricsInput(
7475
.field(PATH_FIELD, path)
7576
.field(PATH_PARAMS_FIELD, pathParams)
7677
.field(URL_FIELD, url)
78+
.field(CLUSTERS_FIELD, clusters)
7779
.endObject()
7880
.endObject()
7981
}
@@ -87,6 +89,7 @@ data class ClusterMetricsInput(
8789
out.writeString(path)
8890
out.writeString(pathParams)
8991
out.writeString(url)
92+
out.writeStringArray(clusters.toTypedArray())
9093
}
9194

9295
companion object {
@@ -99,18 +102,19 @@ data class ClusterMetricsInput(
99102
const val PATH_PARAMS_FIELD = "path_params"
100103
const val URL_FIELD = "url"
101104
const val URI_FIELD = "uri"
105+
const val CLUSTERS_FIELD = "clusters"
102106

103107
val XCONTENT_REGISTRY = NamedXContentRegistry.Entry(Input::class.java, ParseField(URI_FIELD), CheckedFunction { parseInner(it) })
104108

105109
/**
106110
* This parse function uses [XContentParser] to parse JSON input and store corresponding fields to create a [ClusterMetricsInput] object
107111
*/
108-
@JvmStatic
109-
@Throws(IOException::class)
112+
@JvmStatic @Throws(IOException::class)
110113
fun parseInner(xcp: XContentParser): ClusterMetricsInput {
111114
var path = ""
112115
var pathParams = ""
113116
var url = ""
117+
val clusters = mutableListOf<String>()
114118

115119
XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, xcp.currentToken(), xcp)
116120

@@ -121,9 +125,17 @@ data class ClusterMetricsInput(
121125
PATH_FIELD -> path = xcp.text()
122126
PATH_PARAMS_FIELD -> pathParams = xcp.text()
123127
URL_FIELD -> url = xcp.text()
128+
CLUSTERS_FIELD -> {
129+
XContentParserUtils.ensureExpectedToken(
130+
XContentParser.Token.START_ARRAY,
131+
xcp.currentToken(),
132+
xcp
133+
)
134+
while (xcp.nextToken() != XContentParser.Token.END_ARRAY) clusters.add(xcp.text())
135+
}
124136
}
125137
}
126-
return ClusterMetricsInput(path, pathParams, url)
138+
return ClusterMetricsInput(path, pathParams, url, clusters)
127139
}
128140
}
129141

@@ -163,20 +175,17 @@ data class ClusterMetricsInput(
163175
if (pathParams.isNotEmpty()) {
164176
pathParams = pathParams.trim('/')
165177
ILLEGAL_PATH_PARAMETER_CHARACTERS.forEach { character ->
166-
if (pathParams.contains(character)) {
178+
if (pathParams.contains(character))
167179
throw IllegalArgumentException(
168-
"The provided path parameters contain invalid characters or spaces. Please omit: " + "${ILLEGAL_PATH_PARAMETER_CHARACTERS.joinToString(" ")}"
180+
"The provided path parameters contain invalid characters or spaces. Please omit: " + ILLEGAL_PATH_PARAMETER_CHARACTERS.joinToString(" ")
169181
)
170-
}
171182
}
172183
}
173184

174-
if (apiType.requiresPathParams && pathParams.isEmpty()) {
185+
if (apiType.requiresPathParams && pathParams.isEmpty())
175186
throw IllegalArgumentException("The API requires path parameters.")
176-
}
177-
if (!apiType.supportsPathParams && pathParams.isNotEmpty()) {
187+
if (!apiType.supportsPathParams && pathParams.isNotEmpty())
178188
throw IllegalArgumentException("The API does not use path parameters.")
179-
}
180189

181190
return pathParams
182191
}
@@ -192,13 +201,11 @@ data class ClusterMetricsInput(
192201
ClusterMetricType.values()
193202
.filter { option -> option != ClusterMetricType.BLANK }
194203
.forEach { option ->
195-
if (uriPath.startsWith(option.prependPath) || uriPath.startsWith(option.defaultPath)) {
204+
if (uriPath.startsWith(option.prependPath) || uriPath.startsWith(option.defaultPath))
196205
apiType = option
197-
}
198206
}
199-
if (apiType.isBlank()) {
207+
if (apiType.isBlank())
200208
throw IllegalArgumentException("The API could not be determined from the provided URI.")
201-
}
202209
return apiType
203210
}
204211

@@ -207,28 +214,36 @@ data class ClusterMetricsInput(
207214
* @return The constructed [URI].
208215
*/
209216
private fun constructUrlFromInputs(): URI {
210-
val uriBuilder = URIBuilder()
211-
.setScheme(SUPPORTED_SCHEME)
212-
.setHost(SUPPORTED_HOST)
213-
.setPort(SUPPORTED_PORT)
214-
.setPath(path + pathParams)
215-
return uriBuilder.build()
217+
/**
218+
* this try-catch block is required due to a httpcomponents 5.1.x library issue
219+
* it auto encodes path params in the url.
220+
*/
221+
return try {
222+
val formattedPath = if (path.startsWith("/") || path.isBlank()) path else "/$path"
223+
val formattedPathParams = if (pathParams.startsWith("/") || pathParams.isBlank()) pathParams else "/$pathParams"
224+
val uriBuilder = URIBuilder("$SUPPORTED_SCHEME://$SUPPORTED_HOST:$SUPPORTED_PORT$formattedPath$formattedPathParams")
225+
uriBuilder.build()
226+
} catch (ex: URISyntaxException) {
227+
val uriBuilder = URIBuilder()
228+
.setScheme(SUPPORTED_SCHEME)
229+
.setHost(SUPPORTED_HOST)
230+
.setPort(SUPPORTED_PORT)
231+
.setPath(path + pathParams)
232+
uriBuilder.build()
233+
}
216234
}
217235

218236
/**
219237
* If [url] field is empty, populates it with [constructedUri].
220238
* If [path] and [pathParams] are empty, populates them with values from [url].
221239
*/
222240
private fun parseEmptyFields() {
223-
if (pathParams.isEmpty()) {
241+
if (pathParams.isEmpty())
224242
pathParams = this.parsePathParams()
225-
}
226-
if (path.isEmpty()) {
243+
if (path.isEmpty())
227244
path = if (pathParams.isEmpty()) clusterMetricType.defaultPath else clusterMetricType.prependPath
228-
}
229-
if (url.isEmpty()) {
245+
if (url.isEmpty())
230246
url = constructedUri.toString()
231-
}
232247
}
233248

234249
/**

‎src/test/kotlin/org/opensearch/commons/alerting/AlertTests.kt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ class AlertTests {
2323
assertEquals(templateArgs[Alert.START_TIME_FIELD], alert.startTime.toEpochMilli(), "Template args start time does not")
2424
assertEquals(templateArgs[Alert.LAST_NOTIFICATION_TIME_FIELD], null, "Template args last notification time does not match")
2525
assertEquals(templateArgs[Alert.SEVERITY_FIELD], alert.severity, "Template args severity does not match")
26+
assertEquals(templateArgs[Alert.CLUSTERS_FIELD], alert.clusters?.joinToString(","), "Template args clusters does not match")
2627
}
2728

2829
@Test
@@ -40,6 +41,7 @@ class AlertTests {
4041
assertEquals(templateArgs[Alert.START_TIME_FIELD], alert.startTime.toEpochMilli(), "Template args start time does not")
4142
assertEquals(templateArgs[Alert.LAST_NOTIFICATION_TIME_FIELD], null, "Template args last notification time does not match")
4243
assertEquals(templateArgs[Alert.SEVERITY_FIELD], alert.severity, "Template args severity does not match")
44+
assertEquals(templateArgs[Alert.CLUSTERS_FIELD], alert.clusters?.joinToString(","), "Template args clusters does not match")
4345
assertEquals(
4446
templateArgs[Alert.BUCKET_KEYS],
4547
alert.aggregationResultBucket?.bucketKeys?.joinToString(","),

0 commit comments

Comments
 (0)