|
| 1 | +/* |
| 2 | + * SPDX-License-Identifier: Apache-2.0 |
| 3 | + * |
| 4 | + * The OpenSearch Contributors require contributions made to |
| 5 | + * this file be licensed under the Apache-2.0 license or a |
| 6 | + * compatible open source license. |
| 7 | + * |
| 8 | + * Modifications Copyright OpenSearch Contributors. See |
| 9 | + * GitHub history for details. |
| 10 | + */ |
| 11 | + |
| 12 | +package org.opensearch.ad.e2e; |
| 13 | + |
| 14 | +import static org.opensearch.ad.TestHelpers.toHttpEntity; |
| 15 | +import static org.opensearch.ad.settings.AnomalyDetectorSettings.BACKOFF_MINUTES; |
| 16 | +import static org.opensearch.ad.settings.AnomalyDetectorSettings.MAX_RETRY_FOR_UNRESPONSIVE_NODE; |
| 17 | + |
| 18 | +import java.io.File; |
| 19 | +import java.io.FileReader; |
| 20 | +import java.io.IOException; |
| 21 | +import java.io.InputStreamReader; |
| 22 | +import java.nio.charset.Charset; |
| 23 | +import java.time.Instant; |
| 24 | +import java.util.ArrayList; |
| 25 | +import java.util.List; |
| 26 | +import java.util.Locale; |
| 27 | +import java.util.Map; |
| 28 | + |
| 29 | +import org.apache.http.HttpHeaders; |
| 30 | +import org.apache.http.message.BasicHeader; |
| 31 | +import org.opensearch.ad.ODFERestTestCase; |
| 32 | +import org.opensearch.ad.TestHelpers; |
| 33 | +import org.opensearch.client.Request; |
| 34 | +import org.opensearch.client.RequestOptions; |
| 35 | +import org.opensearch.client.Response; |
| 36 | +import org.opensearch.client.RestClient; |
| 37 | +import org.opensearch.client.WarningsHandler; |
| 38 | +import org.opensearch.common.Strings; |
| 39 | +import org.opensearch.common.xcontent.XContentBuilder; |
| 40 | +import org.opensearch.common.xcontent.json.JsonXContent; |
| 41 | + |
| 42 | +import com.google.common.collect.ImmutableList; |
| 43 | +import com.google.gson.JsonArray; |
| 44 | +import com.google.gson.JsonObject; |
| 45 | +import com.google.gson.JsonParser; |
| 46 | + |
| 47 | +public class AbstractSyntheticDataTest extends ODFERestTestCase { |
| 48 | + |
| 49 | + /** |
| 50 | + * In real time AD, we mute a node for a detector if that node keeps returning |
| 51 | + * ResourceNotFoundException (5 times in a row). This is a problem for batch mode |
| 52 | + * testing as we issue a large amount of requests quickly. Due to the speed, we |
| 53 | + * won't be able to finish cold start before the ResourceNotFoundException mutes |
| 54 | + * a node. Since our test case has only one node, there is no other nodes to fall |
| 55 | + * back on. Here we disable such fault tolerance by setting max retries before |
| 56 | + * muting to a large number and the actual wait time during muting to 0. |
| 57 | + * |
| 58 | + * @throws IOException when failing to create http request body |
| 59 | + */ |
| 60 | + protected void disableResourceNotFoundFaultTolerence() throws IOException { |
| 61 | + XContentBuilder settingCommand = JsonXContent.contentBuilder(); |
| 62 | + |
| 63 | + settingCommand.startObject(); |
| 64 | + settingCommand.startObject("persistent"); |
| 65 | + settingCommand.field(MAX_RETRY_FOR_UNRESPONSIVE_NODE.getKey(), 100_000); |
| 66 | + settingCommand.field(BACKOFF_MINUTES.getKey(), 0); |
| 67 | + settingCommand.endObject(); |
| 68 | + settingCommand.endObject(); |
| 69 | + Request request = new Request("PUT", "/_cluster/settings"); |
| 70 | + request.setJsonEntity(Strings.toString(settingCommand)); |
| 71 | + |
| 72 | + adminClient().performRequest(request); |
| 73 | + } |
| 74 | + |
| 75 | + protected List<JsonObject> getData(String datasetFileName) throws Exception { |
| 76 | + JsonArray jsonArray = JsonParser |
| 77 | + .parseReader(new FileReader(new File(getClass().getResource(datasetFileName).toURI()), Charset.defaultCharset())) |
| 78 | + .getAsJsonArray(); |
| 79 | + List<JsonObject> list = new ArrayList<>(jsonArray.size()); |
| 80 | + jsonArray.iterator().forEachRemaining(i -> list.add(i.getAsJsonObject())); |
| 81 | + return list; |
| 82 | + } |
| 83 | + |
| 84 | + protected Map<String, Object> getDetectionResult(String detectorId, Instant begin, Instant end, RestClient client) { |
| 85 | + try { |
| 86 | + Request request = new Request( |
| 87 | + "POST", |
| 88 | + String.format(Locale.ROOT, "/_opendistro/_anomaly_detection/detectors/%s/_run", detectorId) |
| 89 | + ); |
| 90 | + request |
| 91 | + .setJsonEntity( |
| 92 | + String.format(Locale.ROOT, "{ \"period_start\": %d, \"period_end\": %d }", begin.toEpochMilli(), end.toEpochMilli()) |
| 93 | + ); |
| 94 | + return entityAsMap(client.performRequest(request)); |
| 95 | + } catch (Exception e) { |
| 96 | + throw new RuntimeException(e); |
| 97 | + } |
| 98 | + } |
| 99 | + |
| 100 | + protected void bulkIndexTrainData( |
| 101 | + String datasetName, |
| 102 | + List<JsonObject> data, |
| 103 | + int trainTestSplit, |
| 104 | + RestClient client, |
| 105 | + String categoryField |
| 106 | + ) throws Exception { |
| 107 | + Request request = new Request("PUT", datasetName); |
| 108 | + String requestBody = null; |
| 109 | + if (Strings.isEmpty(categoryField)) { |
| 110 | + requestBody = "{ \"mappings\": { \"properties\": { \"timestamp\": { \"type\": \"date\"}," |
| 111 | + + " \"Feature1\": { \"type\": \"double\" }, \"Feature2\": { \"type\": \"double\" } } } }"; |
| 112 | + } else { |
| 113 | + requestBody = String |
| 114 | + .format( |
| 115 | + Locale.ROOT, |
| 116 | + "{ \"mappings\": { \"properties\": { \"timestamp\": { \"type\": \"date\"}," |
| 117 | + + " \"Feature1\": { \"type\": \"double\" }, \"Feature2\": { \"type\": \"double\" }," |
| 118 | + + "\"%s\": { \"type\": \"keyword\"} } } }", |
| 119 | + categoryField |
| 120 | + ); |
| 121 | + } |
| 122 | + |
| 123 | + request.setJsonEntity(requestBody); |
| 124 | + setWarningHandler(request, false); |
| 125 | + client.performRequest(request); |
| 126 | + Thread.sleep(1_000); |
| 127 | + |
| 128 | + StringBuilder bulkRequestBuilder = new StringBuilder(); |
| 129 | + for (int i = 0; i < trainTestSplit; i++) { |
| 130 | + bulkRequestBuilder.append("{ \"index\" : { \"_index\" : \"" + datasetName + "\", \"_id\" : \"" + i + "\" } }\n"); |
| 131 | + bulkRequestBuilder.append(data.get(i).toString()).append("\n"); |
| 132 | + } |
| 133 | + TestHelpers |
| 134 | + .makeRequest( |
| 135 | + client, |
| 136 | + "POST", |
| 137 | + "_bulk?refresh=true", |
| 138 | + null, |
| 139 | + toHttpEntity(bulkRequestBuilder.toString()), |
| 140 | + ImmutableList.of(new BasicHeader(HttpHeaders.USER_AGENT, "Kibana")) |
| 141 | + ); |
| 142 | + Thread.sleep(1_000); |
| 143 | + waitAllSyncheticDataIngested(trainTestSplit, datasetName, client); |
| 144 | + } |
| 145 | + |
| 146 | + protected String createDetector( |
| 147 | + String datasetName, |
| 148 | + int intervalMinutes, |
| 149 | + RestClient client, |
| 150 | + String categoryField, |
| 151 | + long windowDelayInMins |
| 152 | + ) throws Exception { |
| 153 | + Request request = new Request("POST", "/_plugins/_anomaly_detection/detectors/"); |
| 154 | + String requestBody = null; |
| 155 | + if (Strings.isEmpty(categoryField)) { |
| 156 | + requestBody = String |
| 157 | + .format( |
| 158 | + Locale.ROOT, |
| 159 | + "{ \"name\": \"test\", \"description\": \"test\", \"time_field\": \"timestamp\"" |
| 160 | + + ", \"indices\": [\"%s\"], \"feature_attributes\": [{ \"feature_name\": \"feature 1\", \"feature_enabled\": " |
| 161 | + + "\"true\", \"aggregation_query\": { \"Feature1\": { \"sum\": { \"field\": \"Feature1\" } } } }, { \"feature_name\"" |
| 162 | + + ": \"feature 2\", \"feature_enabled\": \"true\", \"aggregation_query\": { \"Feature2\": { \"sum\": { \"field\": " |
| 163 | + + "\"Feature2\" } } } }], \"detection_interval\": { \"period\": { \"interval\": %d, \"unit\": \"Minutes\" } }, " |
| 164 | + + "\"window_delay\": { \"period\": {\"interval\": %d, \"unit\": \"MINUTES\"}}," |
| 165 | + + "\"schema_version\": 0 }", |
| 166 | + datasetName, |
| 167 | + intervalMinutes, |
| 168 | + windowDelayInMins |
| 169 | + ); |
| 170 | + } else { |
| 171 | + requestBody = String |
| 172 | + .format( |
| 173 | + Locale.ROOT, |
| 174 | + "{ \"name\": \"test\", \"description\": \"test\", \"time_field\": \"timestamp\"" |
| 175 | + + ", \"indices\": [\"%s\"], \"feature_attributes\": [{ \"feature_name\": \"feature 1\", \"feature_enabled\": " |
| 176 | + + "\"true\", \"aggregation_query\": { \"Feature1\": { \"sum\": { \"field\": \"Feature1\" } } } }, { \"feature_name\"" |
| 177 | + + ": \"feature 2\", \"feature_enabled\": \"true\", \"aggregation_query\": { \"Feature2\": { \"sum\": { \"field\": " |
| 178 | + + "\"Feature2\" } } } }], \"detection_interval\": { \"period\": { \"interval\": %d, \"unit\": \"Minutes\" } }, " |
| 179 | + + "\"category_field\": [\"%s\"], " |
| 180 | + + "\"window_delay\": { \"period\": {\"interval\": %d, \"unit\": \"MINUTES\"}}," |
| 181 | + + "\"schema_version\": 0 }", |
| 182 | + datasetName, |
| 183 | + intervalMinutes, |
| 184 | + categoryField, |
| 185 | + windowDelayInMins |
| 186 | + ); |
| 187 | + } |
| 188 | + |
| 189 | + request.setJsonEntity(requestBody); |
| 190 | + Map<String, Object> response = entityAsMap(client.performRequest(request)); |
| 191 | + String detectorId = (String) response.get("_id"); |
| 192 | + Thread.sleep(1_000); |
| 193 | + return detectorId; |
| 194 | + } |
| 195 | + |
| 196 | + protected void waitAllSyncheticDataIngested(int expectedSize, String datasetName, RestClient client) throws Exception { |
| 197 | + int maxWaitCycles = 3; |
| 198 | + do { |
| 199 | + Request request = new Request("POST", String.format(Locale.ROOT, "/%s/_search", datasetName)); |
| 200 | + request |
| 201 | + .setJsonEntity( |
| 202 | + String |
| 203 | + .format( |
| 204 | + Locale.ROOT, |
| 205 | + "{\"query\": {" |
| 206 | + + " \"match_all\": {}" |
| 207 | + + " }," |
| 208 | + + " \"size\": 1," |
| 209 | + + " \"sort\": [" |
| 210 | + + " {" |
| 211 | + + " \"timestamp\": {" |
| 212 | + + " \"order\": \"desc\"" |
| 213 | + + " }" |
| 214 | + + " }" |
| 215 | + + " ]}" |
| 216 | + ) |
| 217 | + ); |
| 218 | + // Make sure all of the test data has been ingested |
| 219 | + // Expected response: |
| 220 | + // "_index":"synthetic","_type":"_doc","_id":"10080","_score":null,"_source":{"timestamp":"2019-11-08T00:00:00Z","Feature1":156.30028000000001,"Feature2":100.211205,"host":"host1"},"sort":[1573171200000]} |
| 221 | + Response response = client.performRequest(request); |
| 222 | + JsonObject json = JsonParser |
| 223 | + .parseReader(new InputStreamReader(response.getEntity().getContent(), Charset.defaultCharset())) |
| 224 | + .getAsJsonObject(); |
| 225 | + JsonArray hits = json.getAsJsonObject("hits").getAsJsonArray("hits"); |
| 226 | + if (hits != null |
| 227 | + && hits.size() == 1 |
| 228 | + && expectedSize - 1 == hits.get(0).getAsJsonObject().getAsJsonPrimitive("_id").getAsLong()) { |
| 229 | + break; |
| 230 | + } else { |
| 231 | + request = new Request("POST", String.format(Locale.ROOT, "/%s/_refresh", datasetName)); |
| 232 | + client.performRequest(request); |
| 233 | + } |
| 234 | + Thread.sleep(1_000); |
| 235 | + } while (maxWaitCycles-- >= 0); |
| 236 | + } |
| 237 | + |
| 238 | + protected void setWarningHandler(Request request, boolean strictDeprecationMode) { |
| 239 | + RequestOptions.Builder options = RequestOptions.DEFAULT.toBuilder(); |
| 240 | + options.setWarningsHandler(strictDeprecationMode ? WarningsHandler.STRICT : WarningsHandler.PERMISSIVE); |
| 241 | + request.setOptions(options.build()); |
| 242 | + } |
| 243 | +} |
0 commit comments