add support for int key in bucket level trigger aggregations #323

sbcd90 · 2022-03-07T21:25:52Z

Signed-off-by: Subhobrata Dey [email protected]

Issue #, if available:
fixes #311

Description of changes:
Add support for int datatype as bucket keyField

CheckList:
[X] Commits are signed per the DCO using --signoff

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

codecov-commenter · 2022-03-07T21:35:06Z

Codecov Report

Merging #323 (de4ce3f) into main (d001abc) will increase coverage by 0.03%.
The diff coverage is 50.00%.

@@             Coverage Diff              @@
##               main     #323      +/-   ##
============================================
+ Coverage     76.83%   76.86%   +0.03%     
  Complexity      176      176              
============================================
  Files           166      166              
  Lines          8369     8370       +1     
  Branches       1231     1232       +1     
============================================
+ Hits           6430     6434       +4     
+ Misses         1334     1328       -6     
- Partials        605      608       +3

Impacted Files	Coverage Δ
...n/kotlin/org/opensearch/alerting/TriggerService.kt	`73.21% <50.00%> (+0.48%)`	⬆️
...ain/kotlin/org/opensearch/alerting/AlertService.kt	`77.99% <0.00%> (-0.48%)`	⬇️
...destinationmigration/DestinationConversionUtils.kt	`70.00% <0.00%> (+1.11%)`	⬆️
...ing/model/destination/DestinationContextFactory.kt	`78.57% <0.00%> (+3.57%)`	⬆️
...nationmigration/DestinationMigrationCoordinator.kt	`68.08% <0.00%> (+4.25%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update d001abc...de4ce3f. Read the comment docs.

alerting/src/test/kotlin/org/opensearch/alerting/TriggerServiceTests.kt

sbcd90 · 2022-03-09T02:57:44Z

Hi @qreshi @rishabhmaurya , fixed the code review comments. Added a test case for composite aggregation case.
Found a new issue while testing composite aggregation.

Stacktrace:

org.opensearch.action.search.SearchPhaseExecutionException: 
	at org.opensearch.action.search.AbstractSearchAsyncAction.onPhaseFailure(AbstractSearchAsyncAction.java:642) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase$1.onFailure(FetchSearchPhase.java:126) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.TimedRunnable.doRun(TimedRunnable.java:57) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:792) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:50) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630) ~[?:?]
	at java.lang.Thread.run(Thread.java:832) ~[?:?]
Caused by: java.lang.IllegalStateException: value must not be null
	at org.opensearch.alerting.aggregation.bucketselectorext.BucketSelectorExtAggregator.doReduce(BucketSelectorExtAggregator.kt:135) ~[?:?]
	at org.opensearch.search.aggregations.InternalAggregations.topLevelReduce(InternalAggregations.java:230) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.SearchPhaseController.reduceAggs(SearchPhaseController.java:561) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.SearchPhaseController.reducedQueryPhase(SearchPhaseController.java:532) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.QueryPhaseResultConsumer.reduce(QueryPhaseResultConsumer.java:151) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase.innerRun(FetchSearchPhase.java:135) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase.access$000(FetchSearchPhase.java:60) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase$1.doRun(FetchSearchPhase.java:121) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:50) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	... 6 more

Initial debugging suggests its an opensearch-core issue.

qreshi · 2022-03-09T20:57:48Z

Hi @qreshi @rishabhmaurya , fixed the code review comments. Added a test case for composite aggregation case. Found a new issue while testing composite aggregation.

Stacktrace:

org.opensearch.action.search.SearchPhaseExecutionException: 
	at org.opensearch.action.search.AbstractSearchAsyncAction.onPhaseFailure(AbstractSearchAsyncAction.java:642) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase$1.onFailure(FetchSearchPhase.java:126) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.TimedRunnable.doRun(TimedRunnable.java:57) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:792) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:50) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630) ~[?:?]
	at java.lang.Thread.run(Thread.java:832) ~[?:?]
Caused by: java.lang.IllegalStateException: value must not be null
	at org.opensearch.alerting.aggregation.bucketselectorext.BucketSelectorExtAggregator.doReduce(BucketSelectorExtAggregator.kt:135) ~[?:?]
	at org.opensearch.search.aggregations.InternalAggregations.topLevelReduce(InternalAggregations.java:230) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.SearchPhaseController.reduceAggs(SearchPhaseController.java:561) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.SearchPhaseController.reducedQueryPhase(SearchPhaseController.java:532) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.QueryPhaseResultConsumer.reduce(QueryPhaseResultConsumer.java:151) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase.innerRun(FetchSearchPhase.java:135) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase.access$000(FetchSearchPhase.java:60) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase$1.doRun(FetchSearchPhase.java:121) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:50) ~[opensearch-1.2.4-SNAPSHOT.jar:1.2.4-SNAPSHOT]
	... 6 more

Initial debugging suggests its an opensearch-core issue.

Looks like you're using opensearch-1.2.4-SNAPSHOT there. Any reason it's not on OS 2.0 or at least OS 1.3 snapshot?

sbcd90 · 2022-03-09T21:05:20Z

@qreshi sure I'll test with OS 1.3 snapshot & see if the same issue still exists.

sbcd90 · 2022-03-17T21:55:14Z

hi @qreshi , i produced the following issue with OS 1.3 & looks like the code that is causing this issue is still in the os-core main branch.

create index sample-http-responses. insert data from here: https://github.com/opensearch-project/anomaly-detection-dashboards-plugin/blob/main/server/sampleData/rawData/httpResponses.json.gz

In Opensearch-dashboards -> Create monitor -> per bucket monitor -> extraction query editor -> use query

{
          "size": 1,
          "query": {
            "match_all": {}
          },
          "aggs": {
            "status_code": {
                "composite": {
                    "sources": [
                        { "status_code": { "terms": { "field": "status_code" } } }
                    ]
                }
            }
          }
        }

& then add trigger -> trigger condition add this

{
          "buckets_path": {
            "_count":"_count",
            "_key": "_key"
          },
          "parent_bucket_path": "status_code",
          "script": {
            "source": "params._count > 0",
            "lang": "painless"
          }
        }

Now click on preview condition response gives error:

Empty list doesn't contain element at index 0.

Actual error in the logs:

org.opensearch.action.search.SearchPhaseExecutionException: 
	at org.opensearch.action.search.AbstractSearchAsyncAction.onPhaseFailure(AbstractSearchAsyncAction.java:642) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase$1.onFailure(FetchSearchPhase.java:126) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:52) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.common.util.concurrent.TimedRunnable.doRun(TimedRunnable.java:57) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:792) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:50) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]
	at java.lang.Thread.run(Thread.java:829) ~[?:?]
Caused by: org.opensearch.search.aggregations.AggregationExecutionException: buckets_path must reference either a number value or a single value numeric metric aggregation, got: [ArrayMap] at aggregation [_key]
	at org.opensearch.search.aggregations.pipeline.BucketHelpers.formatResolutionError(BucketHelpers.java:273) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.search.aggregations.pipeline.BucketHelpers.resolveBucketValue(BucketHelpers.java:202) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.search.aggregations.pipeline.BucketHelpers.resolveBucketValue(BucketHelpers.java:178) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.alerting.aggregation.bucketselectorext.BucketSelectorExtAggregator.doReduce(BucketSelectorExtAggregator.kt:132) ~[?:?]
	at org.opensearch.search.aggregations.InternalAggregations.topLevelReduce(InternalAggregations.java:230) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.SearchPhaseController.reduceAggs(SearchPhaseController.java:561) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.SearchPhaseController.reducedQueryPhase(SearchPhaseController.java:532) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.QueryPhaseResultConsumer.reduce(QueryPhaseResultConsumer.java:151) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase.innerRun(FetchSearchPhase.java:135) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase.access$000(FetchSearchPhase.java:60) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.action.search.FetchSearchPhase$1.doRun(FetchSearchPhase.java:121) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	at org.opensearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:50) ~[opensearch-1.3.0-SNAPSHOT.jar:1.3.0-SNAPSHOT]
	... 6 more
[2022-03-17T21:47:51,079][INFO ][o.o.a.TriggerService     ] [****] Error running trigger [KsLZmX8BYDwqmLCh73id] for monitor []
java.lang.IndexOutOfBoundsException: Empty list doesn't contain element at index 0.
	at kotlin.collections.EmptyList.get(Collections.kt:35) ~[kotlin-stdlib-1.3.72.jar:1.3.72-release-468 (1.3.72)]
	at kotlin.collections.EmptyList.get(Collections.kt:23) ~[kotlin-stdlib-1.3.72.jar:1.3.72-release-468 (1.3.72)]
	at org.opensearch.alerting.TriggerService.runBucketLevelTrigger(TriggerService.kt:64) [opensearch-alerting-1.3.0.0-SNAPSHOT.jar:1.3.0.0-SNAPSHOT]
	at org.opensearch.alerting.MonitorRunner.runBucketLevelMonitor(MonitorRunner.kt:384) [opensearch-alerting-1.3.0.0-SNAPSHOT.jar:1.3.0.0-SNAPSHOT]
	at org.opensearch.alerting.MonitorRunner$runBucketLevelMonitor$1.invokeSuspend(MonitorRunner.kt) [opensearch-alerting-1.3.0.0-SNAPSHOT.jar:1.3.0.0-SNAPSHOT]
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33) [kotlin-stdlib-1.3.72.jar:1.3.72-release-468 (1.3.72)]
	at kotlinx.coroutines.ResumeModeKt.resumeUninterceptedMode(ResumeMode.kt:46) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.internal.ScopeCoroutine.onCompletionInternal$kotlinx_coroutines_core(Scopes.kt:28) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.JobSupport.completeStateFinalization(JobSupport.kt:305) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.JobSupport.tryFinalizeSimpleState(JobSupport.kt:264) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.JobSupport.tryMakeCompleting(JobSupport.kt:762) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.JobSupport.makeCompletingOnce$kotlinx_coroutines_core(JobSupport.kt:742) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.AbstractCoroutine.resumeWith(AbstractCoroutine.kt:117) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:46) [kotlin-stdlib-1.3.72.jar:1.3.72-release-468 (1.3.72)]
	at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:285) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60) [kotlinx-coroutines-core-1.1.1.jar:?]
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:742) [kotlinx-coroutines-core-1.1.1.jar:?]

qreshi · 2022-03-18T17:13:47Z

@sbcd90 Thanks for the additional info. Also, what was the response of the extraction query itself?

sbcd90 · 2022-03-19T04:51:11Z

hi @qreshi , here is the response

{
    "_shards": {
        "total": 1,
        "failed": 0,
        "successful": 1,
        "skipped": 0
    },
    "hits": {
        "hits": [
            {
                "_index": "sample-http-responses",
                "_type": "http",
                "_source": {
                    "status_code": 100,
                    "http_4xx": 0,
                    "http_3xx": 0,
                    "http_5xx": 0,
                    "http_2xx": 0,
                    "timestamp": 100000,
                    "http_1xx": 1
                },
                "_id": "1",
                "_score": 1
            }
        ],
        "total": {
            "value": 4,
            "relation": "eq"
        },
        "max_score": 1
    },
    "took": 3,
    "timed_out": false,
    "aggregations": {
        "status_code": {
            "buckets": [
                {
                    "doc_count": 2,
                    "key": {
                        "status_code": 100
                    }
                },
                {
                    "doc_count": 1,
                    "key": {
                        "status_code": 102
                    }
                },
                {
                    "doc_count": 1,
                    "key": {
                        "status_code": 201
                    }
                }
            ],
            "after_key": {
                "status_code": 201
            }
        }
    }
}

sbcd90 · 2022-03-22T21:08:27Z

As per discussions with @qreshi & @rishabhmaurya & after debugging opensearch-core, looks like group by key is not supported as a feature in opensearch-core for composite aggregations.

However, there is a way to make group by key work in alerting using composite aggregations filters.

{
          "buckets_path": {
            "_count":"_count"
          },
          "parent_bucket_path": "status_code",
          "script": {
            "source": "params._count > 0",
            "lang": "painless"
          },
          "composite_agg_filter":{"status_code":{"include":[],"exclude":[]}}
}

Signed-off-by: Subhobrata Dey <[email protected]>

…rch-project#323) Signed-off-by: Angie Zhang <[email protected]>

sbcd90 requested a review from a team March 7, 2022 21:25

qreshi requested a review from rishabhmaurya March 7, 2022 21:56

qreshi reviewed Mar 7, 2022

View reviewed changes

alerting/src/test/kotlin/org/opensearch/alerting/TriggerServiceTests.kt Outdated Show resolved Hide resolved

alerting/src/test/kotlin/org/opensearch/alerting/TriggerServiceTests.kt Outdated Show resolved Hide resolved

sbcd90 force-pushed the bucket-trigger-int-key-issue branch from b21e70e to 750fef7 Compare March 9, 2022 02:44

qreshi approved these changes Mar 23, 2022

View reviewed changes

rishabhmaurya mentioned this pull request Apr 13, 2022

How to get access the key from aggregated bucket with bucket_level_trigger？ #311

Closed

rishabhmaurya approved these changes Apr 13, 2022

View reviewed changes

add support for int key in bucket level trigger aggregations

9e172e8

Signed-off-by: Subhobrata Dey <[email protected]>

sbcd90 force-pushed the bucket-trigger-int-key-issue branch from 750fef7 to 9e172e8 Compare June 6, 2022 23:50

sbcd90 added 2 commits June 7, 2022 00:15

add support for int key in bucket level trigger aggregations

7d82d0d

Signed-off-by: Subhobrata Dey <[email protected]>

add support for int key in bucket level trigger aggregations

d2629b7

Signed-off-by: Subhobrata Dey <[email protected]>

sbcd90 force-pushed the bucket-trigger-int-key-issue branch from b037151 to d2629b7 Compare June 7, 2022 00:22

add support for int key in bucket level trigger aggregations

de4ce3f

Signed-off-by: Subhobrata Dey <[email protected]>

sbcd90 merged commit 6d8a02c into opensearch-project:main Jun 7, 2022

Angie-Zhang pushed a commit to Angie-Zhang/alerting that referenced this pull request Jun 29, 2022

add support for int key in bucket level trigger aggregations (opensea…

97fe933

…rch-project#323) Signed-off-by: Angie Zhang <[email protected]>

Angie-Zhang pushed a commit to Angie-Zhang/alerting that referenced this pull request Jun 29, 2022

add support for int key in bucket level trigger aggregations (opensea…

e7f3b12

…rch-project#323) Signed-off-by: Angie Zhang <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add support for int key in bucket level trigger aggregations #323

add support for int key in bucket level trigger aggregations #323

sbcd90 commented Mar 7, 2022

codecov-commenter commented Mar 7, 2022 •

edited

Loading

sbcd90 commented Mar 9, 2022

qreshi commented Mar 9, 2022

sbcd90 commented Mar 9, 2022

sbcd90 commented Mar 17, 2022 •

edited

Loading

qreshi commented Mar 18, 2022

sbcd90 commented Mar 19, 2022

sbcd90 commented Mar 22, 2022

add support for int key in bucket level trigger aggregations #323

add support for int key in bucket level trigger aggregations #323

Conversation

sbcd90 commented Mar 7, 2022

codecov-commenter commented Mar 7, 2022 • edited Loading

Codecov Report

sbcd90 commented Mar 9, 2022

qreshi commented Mar 9, 2022

sbcd90 commented Mar 9, 2022

sbcd90 commented Mar 17, 2022 • edited Loading

qreshi commented Mar 18, 2022

sbcd90 commented Mar 19, 2022

sbcd90 commented Mar 22, 2022

codecov-commenter commented Mar 7, 2022 •

edited

Loading

sbcd90 commented Mar 17, 2022 •

edited

Loading