add a feature that flattens custom result index when enabled #1401
Conversation
This file contains 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
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
requested review from
jmazanec15,
jngz-es,
kaituo,
saratvemulapalli,
ohltyler,
vamshin,
VijayanB,
ylwu-amzn,
amitgalitz,
sean-zheng-amazon,
dbwiddis,
owaiskazi19 and
joshpalis
as code owners
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
3 times, most recently
from
dc1c604 to
9f0dcf6
Compare
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
from
d0d60f0 to
352075b
Compare
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
from
c947793 to
bf552b5
Compare
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
from
bf552b5 to
9fced72
Compare
|
resolved |
kaituo
reviewed
Jan 24, 2025
src/main/java/org/opensearch/ad/settings/AnomalyDetectorSettings.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/indices/IndexManagement.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/ad/transport/ADResultBulkTransportAction.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/indices/IndexManagement.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/test/java/org/opensearch/ad/rest/AnomalyDetectorRestApiIT.java
Outdated
Show resolved
Hide resolved
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
2 times, most recently
from
2fd6bc2 to
8b12fd7
Compare
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
from
8b12fd7 to
3ce0734
Compare
Signed-off-by: Jackie Han <[email protected]>
Signed-off-by: Jackie Han <[email protected]>
|
checking the failed bwc test... |
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
4 times, most recently
from
ac8f122 to
9ffc24f
Compare
kaituo
reviewed
Jan 27, 2025
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
| .format(Locale.ROOT, "/opensearch-ad-plugin-result-test_flattened_%s", id.toLowerCase(Locale.ROOT)); | ||
| // wait for the detector starts writing result | ||
| try { | ||
| Thread.sleep(60 * 1000); |
There was a problem hiding this comment.
This will make our build slow down 1 minute. Can you periodically check for results and wait for 1 second if not?
| Map<String, Object> flattenedResultIndex = entityAsMap(getIndexResponse); | ||
|
|
||
| String indexKey = flattenedResultIndex.keySet().stream().findFirst().orElse(null); | ||
| Map<String, Object> indexDetails = (Map<String, Object>) flattenedResultIndex.get(indexKey); |
There was a problem hiding this comment.
Can you add checks for the concrete value? For example, in original result index, entity.name is foo and value is bar, you will see entity.name.foo: bar. Also, in original result index, feature data name is bytes and value is blah. In flattened result index, we see feature_data_feature_bytes: blah.
There was a problem hiding this comment.
I added a concrete check for feature_data -
the painless script works the same for every nested fields, so if feature_data can be flattened, then entity fields can be flattened successfully as well. But I can certainly add more assertion checks here
There was a problem hiding this comment.
please add more assertion. containsKey may not be enough.
There was a problem hiding this comment.
Did you address the latest comment?
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
3 times, most recently
from
4400f44 to
b876a22
Compare
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
2 times, most recently
from
f1e21cf to
ad3d99c
Compare
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
from
ad3d99c to
c844922
Compare
|
bwc test failure is similar to opensearch-project/OpenSearch#15234 |
kaituo
reviewed
Jan 28, 2025
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Outdated
Show resolved
Hide resolved
Signed-off-by: Jackie Han <[email protected]>
kaituo
reviewed
Jan 28, 2025
| Map<String, Object> responseMap = entityAsMap(response); | ||
| String id = (String) responseMap.get("_id"); | ||
| List<Feature> features = detector.getFeatureAttributes(); | ||
| long expectedFeatures = features.stream().filter(Feature::getEnabled).count(); |
There was a problem hiding this comment.
Where do we check equality of expectedFeatures?
There was a problem hiding this comment.
Did you address this comment?
src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
Show resolved
Hide resolved
Signed-off-by: Jackie Han <[email protected]>
kaituo
reviewed
Jan 28, 2025
| client.admin().indices().getAliases(getAliasesRequest, ActionListener.wrap(getAliasesResponse -> { | ||
| Set<String> indices = getAliasesResponse.getAliases().keySet(); | ||
| if (indices.isEmpty()) { | ||
| return; |
There was a problem hiding this comment.
you need to return listener
|
|
||
| private void getFlattenResultAliasIndex(Config existingConfig, ActionListener<T> listener, String id, boolean indexingDryRun) { | ||
| String flattenResultIndexAlias = timeSeriesIndices | ||
| .getFlattenedResultIndexAlias(existingConfig.getCustomResultIndexOrAlias(), existingConfig.getId()); |
There was a problem hiding this comment.
existingConfig.getCustomResultIndexOrAlias can be an alias. Running get alias call on an alias can cause exception.
PUT my-data-stream
POST _aliases
{
"actions": [
{
"add": {
"index": "my-data-stream",
"alias": "my-alias"
}
}
]
}
GET my-alias/_alis
Result:
{
"error" : "no handler found for uri [/my-alias/_alis] and method [GET]"
}
| return; | ||
| } | ||
| String indexName = indices.iterator().next(); | ||
| deleteAlias(indexName, flattenResultIndexAlias, existingConfig, listener, id, indexingDryRun); |
There was a problem hiding this comment.
Why do you want to delete alias? Unselecting flatten option causes data deletion does not seem to be a good user experience.
| .format(Locale.ROOT, "/opensearch-ad-plugin-result-test_flattened_%s", id.toLowerCase(Locale.ROOT)); | ||
| // wait for the detector starts writing result | ||
| try { | ||
| Thread.sleep(60 * 1000); |
There was a problem hiding this comment.
This would still wait for 1 minute, right?
| Map<String, Object> responseMap = entityAsMap(response); | ||
| String id = (String) responseMap.get("_id"); | ||
| List<Feature> features = detector.getFeatureAttributes(); | ||
| long expectedFeatures = features.stream().filter(Feature::getEnabled).count(); |
There was a problem hiding this comment.
Did you address this comment?
| Map<String, Object> flattenedResultIndex = entityAsMap(getIndexResponse); | ||
|
|
||
| String indexKey = flattenedResultIndex.keySet().stream().findFirst().orElse(null); | ||
| Map<String, Object> indexDetails = (Map<String, Object>) flattenedResultIndex.get(indexKey); |
There was a problem hiding this comment.
Did you address the latest comment?
…ult index Signed-off-by: Jackie Han <[email protected]>
kaituo
reviewed
Jan 29, 2025
Comment on lines
+1058
to
+1064
| public String getFlattenedResultIndexAlias(String indexOrAliasName, String configId) { | ||
| return indexOrAliasName + "_flattened_" + configId.toLowerCase(Locale.ROOT); | ||
| } | ||
|
|
||
| public String getFlattenResultIndexIngestPipelineId(String configId) { | ||
| return "flatten_result_index_ingest_pipeline" + configId.toLowerCase(Locale.ROOT); | ||
| } |
There was a problem hiding this comment.
I would suggest move these to Config and it is a fixed value when creating the config so we don't need to repeatedly reconstruct it.
Comment on lines
138
to
142
| private void addToFlattenedIndexIfExists(BulkRequest bulkRequest, AnomalyResult result, String resultIndex) { | ||
| String flattenedResultIndexAlias = resultIndex + "_flattened_" + result.getDetectorId().toLowerCase(Locale.ROOT); | ||
| if (clusterService.state().metadata().hasAlias(flattenedResultIndexAlias)) { | ||
| addResult(bulkRequest, result, flattenedResultIndexAlias); | ||
| } |
There was a problem hiding this comment.
because methods in ADResultBulkTransportAction are async, you cannot use async method like nodeStateManager.getConfig. The caller of ADResultBulkTransportAction.prepareBulkRequest does not expect async call and is synchronously waiting for the result. How about adding
ResultWriteRequest:
String flattenIndex;
public ResultWriteRequest(long expirationEpochMs, String configId, RequestPriority priority, ResultType result, String resultIndex, String flattenIndex) {
super(expirationEpochMs, configId, priority);
this.result = result;
this.resultIndex = resultIndex;
this.flattenIndex = flattenIndex;
}
public String getFlattenIndex() {
return flattenIndex;
}
and then in ADResultBulkTransportAction, you just check if getFlattenIndex() is a null or not.
Signed-off-by: Jackie Han <[email protected]>
jackiehanyang
force-pushed
the
ingestPipelineFlatten
branch
from
590fb5a to
d557c50
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Description
wip - sending implementation out first.
Related Issues
Resolves #[Issue number to be closed when this PR is merged]
Check List
--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.