From 03b6b6dd8cfb3d3b90094530d7b2d10439fe0a48 Mon Sep 17 00:00:00 2001 From: Marshall Massengill Date: Sun, 12 Jul 2026 10:40:57 -0400 Subject: [PATCH 1/5] NMS-20027: compute flow proportional sums with Painless instead of the drift plugin Replace the proportional_sum aggregation provided by the elasticsearch-drift-plugin with an equivalent scripted_metric aggregation built from inline Painless scripts. This removes the need to install a plugin build that exactly matches the Elasticsearch version, and makes flow queries work against stock Elasticsearch and OpenSearch clusters. The scripted_metric state is a flat double[] indexed by bucket when the window/step combination yields at most 4096 buckets (any UI-driven query), which benchmarked 2.3x faster than the plugin on whole-window series queries at 40M flows; a HashMap-based form with identical output covers arbitrarily fine-grained windows, since the dense state is allocated per terms cell up front. Field names are inlined into the script source: per-document params lookups benchmarked ~1.5x slower. The previous plugin-based queries remain available by setting proportionalSumStrategy=plugin in org.opennms.features.flows.persistence.elastic.cfg for clusters that disallow inline scripting and already run the plugin. The script-based implementation intentionally does not reproduce the defect tracked as NMS-20001: for query windows whose start is not a multiple of the step (the common case), the plugin mixes two bucket grids in series results and stops counting window totals at the first epoch-grid boundary after the window start, undercounting totals. Totals may therefore increase after this change. FlowQueryIT and AggregatedFlowQueryIT now run against a stock Elasticsearch container without the plugin installed; the drift plugin build-time dependency is removed. The testcontainers version in the flow itests is aligned with the version managed in the root pom. --- .../pages/deep-dive/flows/basic.adoc | 8 +- docs/modules/releasenotes/pages/whatsnew.adoc | 9 + .../elastic/ProportionalSumAggregation.java | 22 ++ .../flows/elastic/ProportionalSumQuery.java | 315 ++++++++++++++++++ .../flows/elastic/RawFlowQueryService.java | 7 +- .../flows/elastic/SearchQueryProvider.java | 28 +- .../agg/AggregatedFlowQueryService.java | 8 +- .../agg/AggregatedSearchQueryProvider.java | 21 ++ .../OSGI-INF/blueprint/blueprint.xml | 6 + .../netmgt/flows/elastic/agg/series_top_n.ftl | 26 +- .../flows/elastic/agg/series_totals.ftl | 26 +- .../flows/elastic/series_for_missing.ftl | 14 +- .../flows/elastic/series_for_others.ftl | 14 +- .../netmgt/flows/elastic/series_for_terms.ftl | 14 +- .../elastic/ProportionalSumQueryTest.java | 173 ++++++++++ features/flows/itests/pom.xml | 37 +- .../flows/elastic/AggregatedFlowQueryIT.java | 25 +- .../netmgt/flows/elastic/FlowQueryIT.java | 25 +- pom.xml | 1 - 19 files changed, 607 insertions(+), 172 deletions(-) create mode 100644 features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumQuery.java create mode 100644 features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java diff --git a/docs/modules/operation/pages/deep-dive/flows/basic.adoc b/docs/modules/operation/pages/deep-dive/flows/basic.adoc index bbfa2c761f60..874993ec15ac 100644 --- a/docs/modules/operation/pages/deep-dive/flows/basic.adoc +++ b/docs/modules/operation/pages/deep-dive/flows/basic.adoc @@ -11,9 +11,11 @@ Make sure that you have the following before you set up flows: * A configured {page-component-title} instance. * One or more devices that send flows visible to {page-component-title}, and that are monitored with SNMP. -* Elasticsearch cluster set up with the https://github.com/OpenNMS/elasticsearch-drift-plugin[Elasticsearch Drift plugin] installed on every Elasticsearch node. -** The Drift plugin persists and queries flows that {page-component-title} collects. -The Drift version must match the targeted Elasticsearch version. +* An Elasticsearch or OpenSearch cluster to persist and query the flows that {page-component-title} collects. +** Inline Painless scripting must be enabled on the cluster; it is enabled by default. +{page-component-title} uses inline scripts to distribute flow traffic proportionally over the time buckets of a graph. +** The https://github.com/OpenNMS/elasticsearch-drift-plugin[Elasticsearch Drift plugin] is no longer required. +Clusters that still have the plugin installed can keep using it by setting `proportionalSumStrategy=plugin` in `$\{OPENNMS_HOME}/etc/org.opennms.features.flows.persistence.elastic.cfg`; note that the plugin version must match the Elasticsearch version exactly, and that the plugin computes lower totals than the script-based default when a query window does not align with the graph resolution. ** (Optional) Configure Elasticsearch variables like `search.max_buckets` or maximum heap size `ES-JAVA_OPTS` if the default values are not sufficient for your volume of flows or number of nodes. ** (Optional) Create a job to clean the indices so that the disk does not fill up (for example, "keep _X_ days of flows"). Filled disks are a challenging problem to address for those who are not Elasticsearch experts. diff --git a/docs/modules/releasenotes/pages/whatsnew.adoc b/docs/modules/releasenotes/pages/whatsnew.adoc index 592bbf04bf76..1a0402be3e6b 100755 --- a/docs/modules/releasenotes/pages/whatsnew.adoc +++ b/docs/modules/releasenotes/pages/whatsnew.adoc @@ -9,6 +9,15 @@ == New features and important changes +=== Elasticsearch Drift plugin no longer required for flows +Flow series and total queries now compute proportional per-bucket sums with inline Painless scripts instead of the `proportional_sum` aggregation from the https://github.com/OpenNMS/elasticsearch-drift-plugin[Elasticsearch Drift plugin]. +This removes the need to install a plugin build that exactly matches the Elasticsearch version, and allows flow data to be stored on stock Elasticsearch or OpenSearch clusters. +Inline Painless scripting must be enabled on the cluster; it is enabled by default on both products. + +Clusters that already run the Drift plugin can keep using it by setting `proportionalSumStrategy=plugin` in `$\{OPENNMS_HOME}/etc/org.opennms.features.flows.persistence.elastic.cfg`. + +Note that flow totals and series values may increase after upgrading: the plugin undercounts traffic when a query window does not align with the graph resolution (which is the common case), and the script-based implementation corrects this. + === SNMP datacollection config moved to DB with new UI The contents of `etc/datacollection-config.xml` and `etc/datacollection/\*.xml` are migrated into the database during the upgrade process, and the original files are moved to `etc_archive/`. After upgrade, manage SNMP data collection through the *Administration -> Manage SNMP Data Collection Config* page (or the REST API under `/api/v2/datacollectionconf`). diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java index 83f93febfd9b..4f85049e8279 100644 --- a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java @@ -26,8 +26,10 @@ import static io.searchbox.core.search.aggregation.AggregationField.KEY; import static io.searchbox.core.search.aggregation.AggregationField.KEY_AS_STRING; +import java.util.Comparator; import java.util.LinkedList; import java.util.List; +import java.util.Map; import java.util.Objects; import com.google.gson.JsonArray; @@ -47,10 +49,30 @@ public class ProportionalSumAggregation extends Aggregation { public ProportionalSumAggregation(String name, JsonObject PartialSumAggregation) { super(name, PartialSumAggregation); if (PartialSumAggregation.has(String.valueOf(BUCKETS)) && PartialSumAggregation.get(String.valueOf(BUCKETS)).isJsonArray()) { + // Shape produced by the drift plugin's proportional_sum aggregation parseBuckets(PartialSumAggregation.get(String.valueOf(BUCKETS)).getAsJsonArray()); + } else if (PartialSumAggregation.has(String.valueOf(AggregationField.VALUE)) + && PartialSumAggregation.get(String.valueOf(AggregationField.VALUE)).isJsonObject()) { + // Shape produced by the scripted_metric replacement: a map of bucket start time -> sum + parseValueMap(PartialSumAggregation.get(String.valueOf(AggregationField.VALUE)).getAsJsonObject()); } } + private void parseValueMap(JsonObject valueMap) { + for (Map.Entry entry : valueMap.entrySet()) { + final long time; + try { + time = Long.parseLong(entry.getKey()); + } catch (NumberFormatException e) { + continue; // Skip entries without a parseable bucket key + } + final double value = entry.getValue().getAsDouble(); + // Per-bucket document counts are not tracked by the scripted variant + dateHistograms.add(new DateHistogram(valueMap, time, entry.getKey(), value, 0L)); + } + dateHistograms.sort(Comparator.comparing(DateHistogram::getTime)); + } + private void parseBuckets(JsonArray bucketsSource) { for (JsonElement bucket : bucketsSource) { JsonObject bucketObj = bucket.getAsJsonObject(); diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumQuery.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumQuery.java new file mode 100644 index 000000000000..08805c12c630 --- /dev/null +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumQuery.java @@ -0,0 +1,315 @@ +/* + * Licensed to The OpenNMS Group, Inc (TOG) under one or more + * contributor license agreements. See the LICENSE.md file + * distributed with this work for additional information + * regarding copyright ownership. + * + * TOG licenses this file to You under the GNU Affero General + * Public License Version 3 (the "License") or (at your option) + * any later version. You may not use this file except in + * compliance with the License. You may obtain a copy of the + * License at: + * + * https://www.gnu.org/licenses/agpl-3.0.txt + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific + * language governing permissions and limitations under the + * License. + */ +package org.opennms.netmgt.flows.elastic; + +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; + +/** + * Renders the aggregation used to distribute a flow's bytes proportionally over the + * time buckets its [start,end] range overlaps, clipped to the query window. + * + * Two strategies are supported: + * + * + * The Painless strategy has two internal forms with identical output: a dense one whose + * per-cell state is a flat double[] indexed by bucket (faster; used when the window/step + * combination yields at most {@link #DENSE_BUCKET_LIMIT} buckets) and a sparse HashMap-based + * fallback for arbitrarily fine-grained windows. + * + * Bucket semantics of the Painless variant: buckets are aligned to the window start, each + * document contributes {@code value * overlap(bucket, [start,end]) / (end - start)} to every + * bucket it overlaps within the window, documents with {@code start == end} contribute their + * whole value to the single bucket containing their start, and the value is multiplied by the + * sampling field when that field is present, finite and non-zero. Gaps between the first and + * last non-empty bucket are zero-filled, mirroring the plugin's min_doc_count=0 behavior. + */ +public final class ProportionalSumQuery { + + public enum Strategy { + PAINLESS, + PLUGIN; + + public static Strategy parse(String value) { + if (value != null && "plugin".equalsIgnoreCase(value.trim())) { + return PLUGIN; + } + return PAINLESS; + } + } + + /** + * With the dense scripts, every terms cell allocates a double[nBuckets] up front, and + * depth-first terms collection can hold many cells at once. Queries whose window/step + * combination produces more buckets than this fall back to the sparse (HashMap) scripts, + * which only allocate touched buckets. UI-driven queries stay far below this limit. + */ + private static final long DENSE_BUCKET_LIMIT = 4096; + + /** + * Shared prologue of both map scripts: doc-value extraction and window math. + * Field names are inlined into the script source at render time: a per-document + * {@code doc[params.field]} lookup benchmarked ~1.5x slower than {@code doc['literal']} + * on whole-window queries, and only three field combinations exist, so Elasticsearch's + * script cache still holds a handful of compiled scripts. The {@code @SAMPLING@} block + * is emitted only when a sampling field is configured. Documents with a missing or + * inverted range are skipped; the plugin fails the whole search in the inverted case, + * which is not worth preserving. + */ + private static final String MAP_PROLOGUE_TEMPLATE = "" + + "if (doc[@START@].size() > 0 && doc[@END@].size() > 0 && doc[@VALUE@].size() > 0) {\n" + + " def sv = doc[@START@].value;\n" + + " long rs = sv instanceof ZonedDateTime ? sv.toInstant().toEpochMilli() : ((Number) sv).longValue();\n" + + " def ev = doc[@END@].value;\n" + + " long re = ev instanceof ZonedDateTime ? ev.toInstant().toEpochMilli() : ((Number) ev).longValue();\n" + + " if (re >= rs && rs >= 0) {\n" + + " double v = ((Number) doc[@VALUE@].value).doubleValue();\n" + + "@SAMPLING@" + + " long interval = ((Number) params.interval).longValue();\n" + + " long qstart = ((Number) params.qstart).longValue();\n" + + " long qend = ((Number) params.qend).longValue();\n" + + " long off = qstart - (qstart / interval) * interval;\n" + + " long duration = re - rs;\n" + + " long first = ((((rs > qstart) ? rs : qstart) - off) / interval) * interval + off;\n" + + " long last = ((((re < qend) ? re : qend) - off) / interval) * interval + off;\n" + + " for (long b = first; b <= last; b += interval) {\n" + + " long nb = b + interval;\n" + + " double ratio;\n" + + " if (duration != 0) {\n" + + " long tin = (rs > nb || re < b) ? 0L : (((nb < re) ? nb : re) - ((b > rs) ? b : rs));\n" + + " ratio = tin / (double) duration;\n" + + " } else {\n" + + " ratio = 1.0;\n" + + " }\n"; + + private static final String SAMPLING_BLOCK_TEMPLATE = "" + + " if (doc.containsKey(@SAMPLING_FIELD@) && doc[@SAMPLING_FIELD@].size() > 0) {\n" + + " double smp = ((Number) doc[@SAMPLING_FIELD@].value).doubleValue();\n" + + " if (smp == smp && smp != Double.POSITIVE_INFINITY && smp != Double.NEGATIVE_INFINITY && smp != 0.0) {\n" + + " v *= smp;\n" + + " }\n" + + " }\n"; + + private static String mapPrologue(String startField, String endField, String valueField, + String samplingField) { + final String sampling = samplingField == null ? "" + : SAMPLING_BLOCK_TEMPLATE.replace("@SAMPLING_FIELD@", painlessStringLiteral(samplingField)); + return MAP_PROLOGUE_TEMPLATE + .replace("@START@", painlessStringLiteral(startField)) + .replace("@END@", painlessStringLiteral(endField)) + .replace("@VALUE@", painlessStringLiteral(valueField)) + .replace("@SAMPLING@", sampling); + } + + private static String painlessStringLiteral(String fieldName) { + // Field names come from our own query providers; refuse anything that could + // escape the string literal rather than attempting to quote it. + if (fieldName.indexOf('\'') >= 0 || fieldName.indexOf('\\') >= 0) { + throw new IllegalArgumentException("Unsupported characters in field name: " + fieldName); + } + return "'" + fieldName + "'"; + } + + // --- dense variant: per-cell double[] indexed by (bucket - qstart) / interval; the + // touched index range is tracked so the response keeps the plugin's min..max zero-fill. + + private static final String INIT_SCRIPT_DENSE = "" + + "state.sums = new double[(int) ((Number) params.nBuckets).longValue()];" + + " state.min = Integer.MAX_VALUE; state.max = -1;"; + + private static final String MAP_LOOP_DENSE = "" + + " int idx = (int) ((b - qstart) / interval);\n" + + " state.sums[idx] += v * ratio;\n" + + " if (idx < state.min) { state.min = idx; }\n" + + " if (idx > state.max) { state.max = idx; }\n" + + " }\n" + + " }\n" + + "}\n"; + + private static final String COMBINE_SCRIPT_DENSE = "" + + "List out = new ArrayList();\n" + + "out.add(state.min);\n" + + "out.add(state.max);\n" + + "if (state.max >= 0) {\n" + + " for (int i = (int) state.min; i <= (int) state.max; i++) {\n" + + " out.add(state.sums[i]);\n" + + " }\n" + + "}\n" + + "return out;\n"; + + private static final String REDUCE_SCRIPT_DENSE = "" + + "int gmin = Integer.MAX_VALUE;\n" + + "int gmax = -1;\n" + + "for (s in states) {\n" + + " if (s != null) {\n" + + " int mx = ((Number) s.get(1)).intValue();\n" + + " if (mx >= 0) {\n" + + " int mn = ((Number) s.get(0)).intValue();\n" + + " if (mn < gmin) { gmin = mn; }\n" + + " if (mx > gmax) { gmax = mx; }\n" + + " }\n" + + " }\n" + + "}\n" + + "Map m = new HashMap();\n" + + "if (gmax >= 0) {\n" + + " double[] tot = new double[gmax - gmin + 1];\n" + + " for (s in states) {\n" + + " if (s != null) {\n" + + " int mx = ((Number) s.get(1)).intValue();\n" + + " if (mx >= 0) {\n" + + " int mn = ((Number) s.get(0)).intValue();\n" + + " for (int i = mn; i <= mx; i++) {\n" + + " tot[i - gmin] += ((Number) s.get(2 + i - mn)).doubleValue();\n" + + " }\n" + + " }\n" + + " }\n" + + " }\n" + + " long interval = ((Number) params.interval).longValue();\n" + + " long qstart = ((Number) params.qstart).longValue();\n" + + " for (int i = 0; i < tot.length; i++) {\n" + + " m.put(String.valueOf(qstart + (long) (gmin + i) * interval), tot[i]);\n" + + " }\n" + + "}\n" + + "return m;\n"; + + // --- sparse variant: HashMap keyed by bucket start; allocation proportional to touched + // buckets only, used when nBuckets exceeds DENSE_BUCKET_LIMIT. + + private static final String INIT_SCRIPT_SPARSE = "state.sums = new HashMap();"; + + private static final String MAP_LOOP_SPARSE = "" + + " String k = String.valueOf(b);\n" + + " double prev = state.sums.containsKey(k) ? (double) state.sums.get(k) : 0.0;\n" + + " state.sums.put(k, prev + v * ratio);\n" + + " }\n" + + " }\n" + + "}\n"; + + private static final String COMBINE_SCRIPT_SPARSE = "return state.sums;"; + + private static final String REDUCE_SCRIPT_SPARSE = "" + + "Map merged = new HashMap();\n" + + "for (s in states) {\n" + + " if (s != null) {\n" + + " for (e in s.entrySet()) {\n" + + " String k = e.getKey();\n" + + " double val = ((Number) e.getValue()).doubleValue();\n" + + " double prev = merged.containsKey(k) ? (double) merged.get(k) : 0.0;\n" + + " merged.put(k, prev + val);\n" + + " }\n" + + " }\n" + + "}\n" + + "if (!merged.isEmpty()) {\n" + + " long interval = ((Number) params.interval).longValue();\n" + + " long minKey = Long.MAX_VALUE;\n" + + " long maxKey = Long.MIN_VALUE;\n" + + " for (k in merged.keySet()) {\n" + + " long key = Long.parseLong(k);\n" + + " if (key < minKey) { minKey = key; }\n" + + " if (key > maxKey) { maxKey = key; }\n" + + " }\n" + + " for (long b = minKey; b <= maxKey; b += interval) {\n" + + " String k = String.valueOf(b);\n" + + " if (!merged.containsKey(k)) { merged.put(k, 0.0); }\n" + + " }\n" + + "}\n" + + "return merged;\n"; + + private ProportionalSumQuery() { + } + + /** + * Renders the aggregation body (the JSON object placed after the aggregation name) + * for the given strategy. + * + * @param samplingField optional field the value is multiplied by; may be null + */ + public static String aggregationFor(Strategy strategy, long stepMs, long start, long end, + String startField, String endField, String valueField, + String samplingField) { + if (strategy == Strategy.PLUGIN) { + return pluginAggregation(stepMs, start, end, startField, endField, valueField, samplingField); + } + return painlessAggregation(stepMs, start, end, startField, endField, valueField, samplingField); + } + + private static String pluginAggregation(long stepMs, long start, long end, + String startField, String endField, String valueField, + String samplingField) { + final JsonArray fields = new JsonArray(); + fields.add(startField); + fields.add(endField); + fields.add(valueField); + if (samplingField != null) { + fields.add(samplingField); + } + final JsonObject proportionalSum = new JsonObject(); + proportionalSum.add("fields", fields); + proportionalSum.addProperty("interval", stepMs + "ms"); + proportionalSum.addProperty("start", start); + proportionalSum.addProperty("end", end); + final JsonObject agg = new JsonObject(); + agg.add("proportional_sum", proportionalSum); + return agg.toString(); + } + + private static String painlessAggregation(long stepMs, long start, long end, + String startField, String endField, String valueField, + String samplingField) { + final JsonObject params = new JsonObject(); + params.addProperty("interval", stepMs); + params.addProperty("qstart", start); + params.addProperty("qend", end); + + final long offset = start % stepMs; + final long lastBucket = ((end - offset) / stepMs) * stepMs + offset; + final long nBuckets = (lastBucket - start) / stepMs + 1; + + final String prologue = mapPrologue(startField, endField, valueField, samplingField); + final JsonObject scriptedMetric = new JsonObject(); + scriptedMetric.add("params", params); + if (nBuckets <= DENSE_BUCKET_LIMIT) { + params.addProperty("nBuckets", nBuckets); + scriptedMetric.addProperty("init_script", INIT_SCRIPT_DENSE); + scriptedMetric.addProperty("map_script", prologue + MAP_LOOP_DENSE); + scriptedMetric.addProperty("combine_script", COMBINE_SCRIPT_DENSE); + scriptedMetric.addProperty("reduce_script", REDUCE_SCRIPT_DENSE); + } else { + scriptedMetric.addProperty("init_script", INIT_SCRIPT_SPARSE); + scriptedMetric.addProperty("map_script", prologue + MAP_LOOP_SPARSE); + scriptedMetric.addProperty("combine_script", COMBINE_SCRIPT_SPARSE); + scriptedMetric.addProperty("reduce_script", REDUCE_SCRIPT_SPARSE); + } + final JsonObject agg = new JsonObject(); + agg.add("scripted_metric", scriptedMetric); + return agg.toString(); + } +} diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/RawFlowQueryService.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/RawFlowQueryService.java index 9489c81cff8b..e8088dbef18a 100644 --- a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/RawFlowQueryService.java +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/RawFlowQueryService.java @@ -66,10 +66,15 @@ public class RawFlowQueryService extends ElasticFlowQueryService { public static final String OTHER_NAME = "Other"; public static final String UNKNOWN_APPLICATION_NAME = "Unknown"; - private final SearchQueryProvider searchQueryProvider = new SearchQueryProvider(); + private final SearchQueryProvider searchQueryProvider; public RawFlowQueryService(ElasticRestClient client, IndexSelector indexSelector) { + this(client, indexSelector, null); + } + + public RawFlowQueryService(ElasticRestClient client, IndexSelector indexSelector, String proportionalSumStrategy) { super(client, indexSelector); + this.searchQueryProvider = new SearchQueryProvider(ProportionalSumQuery.Strategy.parse(proportionalSumStrategy)); } @Override diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/SearchQueryProvider.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/SearchQueryProvider.java index eae61a51b883..7ebdca98bc96 100644 --- a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/SearchQueryProvider.java +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/SearchQueryProvider.java @@ -59,7 +59,18 @@ public class SearchQueryProvider implements FilterVisitor { private final Configuration cfg = new Configuration(Configuration.VERSION_2_3_23); + private final ProportionalSumQuery.Strategy proportionalSumStrategy; + public SearchQueryProvider() { + this(ProportionalSumQuery.Strategy.PAINLESS); + } + + public SearchQueryProvider(String proportionalSumStrategy) { + this(ProportionalSumQuery.Strategy.parse(proportionalSumStrategy)); + } + + public SearchQueryProvider(ProportionalSumQuery.Strategy proportionalSumStrategy) { + this.proportionalSumStrategy = proportionalSumStrategy; // Setup Freemarker cfg.setClassForTemplateLoading(getClass(), ""); cfg.setDefaultEncoding(StandardCharsets.UTF_8.name()); @@ -69,6 +80,11 @@ public SearchQueryProvider() { .build()); } + private String proportionalSumAgg(long step, long start, long end) { + return ProportionalSumQuery.aggregationFor(proportionalSumStrategy, step, start, end, + "netflow.delta_switched", "netflow.last_switched", "netflow.bytes", "netflow.sampling_interval"); + } + public String getFlowCountQuery(List filters) { return render("flow_count.ftl", ImmutableMap.builder() .put("filters", getFilterQueries(filters)) @@ -106,7 +122,8 @@ public String getSeriesFromQuery(Collection from, long step, long start, .put("groupByTerm", groupByTerm) .put("step", step) .put("start", start) - .put("end", end); + .put("end", end) + .put("proportionalSum", proportionalSumAgg(step, start, end)); getSnmpInterfaceId(filters).ifPresent(iif -> builder.put("snmpInterfaceId", iif)); return render("series_for_terms.ftl", builder.build()); } @@ -119,7 +136,8 @@ public String getSeriesFromQuery(int size, long step, long start, long end, .put("groupByTerm", groupByTerm) .put("step", step) .put("start", start) - .put("end", end); + .put("end", end) + .put("proportionalSum", proportionalSumAgg(step, start, end)); getSnmpInterfaceId(filters).ifPresent(iif -> builder.put("snmpInterfaceId", iif)); return render("series_for_terms.ftl", builder.build()); } @@ -132,7 +150,8 @@ public String getSeriesFromMissingQuery(long step, long start, long end, String .put("keyForMissingTerm", keyForMissingTerm) .put("step", step) .put("start", start) - .put("end", end); + .put("end", end) + .put("proportionalSum", proportionalSumAgg(step, start, end)); getSnmpInterfaceId(filters).ifPresent(iif -> builder.put("snmpInterfaceId", iif)); return render("series_for_missing.ftl", builder.build()); } @@ -147,7 +166,8 @@ public String getSeriesFromOthersQuery(Collection from, long step, long .put("excludeMissing", excludeMissing) .put("step", step) .put("start", start) - .put("end", end); + .put("end", end) + .put("proportionalSum", proportionalSumAgg(step, start, end)); getSnmpInterfaceId(filters).ifPresent(iif -> builder.put("snmpInterfaceId", iif)); return render("series_for_others.ftl", builder.build()); } diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedFlowQueryService.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedFlowQueryService.java index 70153be22344..cc9c8a640c97 100644 --- a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedFlowQueryService.java +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedFlowQueryService.java @@ -55,6 +55,7 @@ import org.opennms.netmgt.flows.elastic.ElasticFlowQueryService; import org.opennms.netmgt.flows.elastic.GPath; import org.opennms.netmgt.flows.elastic.ProportionalSumAggregation; +import org.opennms.netmgt.flows.elastic.ProportionalSumQuery; import org.opennms.netmgt.flows.filter.api.DscpFilter; import org.opennms.netmgt.flows.filter.api.Filter; import org.opennms.netmgt.flows.filter.api.TimeRangeFilter; @@ -81,10 +82,15 @@ public class AggregatedFlowQueryService extends ElasticFlowQueryService { public static final String INDEX_NAME = "netflow_agg"; public static final String OTHER_NAME = "Other"; - private final AggregatedSearchQueryProvider searchQueryProvider = new AggregatedSearchQueryProvider(); + private final AggregatedSearchQueryProvider searchQueryProvider; public AggregatedFlowQueryService(ElasticRestClient client, IndexSelector indexSelector) { + this(client, indexSelector, null); + } + + public AggregatedFlowQueryService(ElasticRestClient client, IndexSelector indexSelector, String proportionalSumStrategy) { super(client, indexSelector); + this.searchQueryProvider = new AggregatedSearchQueryProvider(ProportionalSumQuery.Strategy.parse(proportionalSumStrategy)); } private boolean hasTosFilter(List filters) { diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedSearchQueryProvider.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedSearchQueryProvider.java index d7f61024bdd9..d48b6c3a3bf6 100644 --- a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedSearchQueryProvider.java +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/agg/AggregatedSearchQueryProvider.java @@ -28,6 +28,7 @@ import java.util.Map; import java.util.stream.Collectors; +import org.opennms.netmgt.flows.elastic.ProportionalSumQuery; import org.opennms.netmgt.flows.filter.api.DscpFilter; import org.opennms.netmgt.flows.filter.api.ExporterNodeFilter; import org.opennms.netmgt.flows.filter.api.Filter; @@ -53,13 +54,29 @@ public class AggregatedSearchQueryProvider implements FilterVisitor { private final Configuration cfg = new Configuration(Configuration.VERSION_2_3_23); + private final ProportionalSumQuery.Strategy proportionalSumStrategy; + public AggregatedSearchQueryProvider() { + this(ProportionalSumQuery.Strategy.PAINLESS); + } + + public AggregatedSearchQueryProvider(String proportionalSumStrategy) { + this(ProportionalSumQuery.Strategy.parse(proportionalSumStrategy)); + } + + public AggregatedSearchQueryProvider(ProportionalSumQuery.Strategy proportionalSumStrategy) { + this.proportionalSumStrategy = proportionalSumStrategy; // Setup Freemarker cfg.setClassForTemplateLoading(getClass(), ""); cfg.setDefaultEncoding(StandardCharsets.UTF_8.name()); cfg.setTemplateExceptionHandler(TemplateExceptionHandler.RETHROW_HANDLER); } + private String proportionalSumAgg(long step, long start, long end, String valueField) { + return ProportionalSumQuery.aggregationFor(proportionalSumStrategy, step, start, end, + "range_start", "range_end", valueField, null); + } + public String getFlowCountQuery(List filters) { return render("flow_count.ftl", ImmutableMap.builder() .put("filters", getFilterQueries(filters)) @@ -90,6 +107,8 @@ public String getSeriesFromTotalsQuery(GroupedBy groupedBy, long step, long star .put("step", step) .put("start", start) .put("end", end) + .put("proportionalSumIngress", proportionalSumAgg(step, start, end, "bytes_ingress")) + .put("proportionalSumEgress", proportionalSumAgg(step, start, end, "bytes_egress")) .build()); } @@ -103,6 +122,8 @@ public String getSeriesFromTopNQuery(int N, GroupedBy groupedBy, String aggregat .put("step", step) .put("start", start) .put("end", end) + .put("proportionalSumIngress", proportionalSumAgg(step, start, end, "bytes_ingress")) + .put("proportionalSumEgress", proportionalSumAgg(step, start, end, "bytes_egress")) .build()); } diff --git a/features/flows/elastic/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/features/flows/elastic/src/main/resources/OSGI-INF/blueprint/blueprint.xml index e47f88904112..3aa73080dec9 100644 --- a/features/flows/elastic/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/features/flows/elastic/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -55,6 +55,10 @@ + + @@ -157,6 +161,7 @@ + @@ -168,6 +173,7 @@ + diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_top_n.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_top_n.ftl index 574eaf4b1b3b..609ee9a01b4e 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_top_n.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_top_n.ftl @@ -38,30 +38,8 @@ ] }, "aggregations": { - "bytes_in": { - "proportional_sum": { - "fields": [ - "range_start", - "range_end", - "bytes_ingress" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - }, - "bytes_out": { - "proportional_sum": { - "fields": [ - "range_start", - "range_end", - "bytes_egress" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - }, + "bytes_in": ${proportionalSumIngress}, + "bytes_out": ${proportionalSumEgress}, "bytes_total": { "sum": { "field": "bytes_total" diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_totals.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_totals.ftl index 572f7424a3bc..96b80a9fcf5c 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_totals.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/agg/series_totals.ftl @@ -24,29 +24,7 @@ } }, "aggregations": { - "bytes_in": { - "proportional_sum": { - "fields": [ - "range_start", - "range_end", - "bytes_ingress" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - }, - "bytes_out": { - "proportional_sum": { - "fields": [ - "range_start", - "range_end", - "bytes_egress" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - } + "bytes_in": ${proportionalSumIngress}, + "bytes_out": ${proportionalSumEgress} } } diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl index f447e79c3c8d..7f80c87de284 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl @@ -37,19 +37,7 @@ "size": 2 }, "aggs": { - "bytes": { - "proportional_sum": { - "fields": [ - "netflow.delta_switched", - "netflow.last_switched", - "netflow.bytes", - "netflow.sampling_interval" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - }, + "bytes": ${proportionalSum}, <#-- netflow.ecn is a keyword -> max aggregation not possible; string comparison required--> "congestion_encountered": { "max": { diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl index 68cac3ca3562..67f934627b93 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl @@ -38,19 +38,7 @@ "size": 2 }, "aggs": { - "bytes": { - "proportional_sum": { - "fields": [ - "netflow.delta_switched", - "netflow.last_switched", - "netflow.bytes", - "netflow.sampling_interval" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - }, + "bytes": ${proportionalSum}, <#-- netflow.ecn is a keyword -> max aggregation not possible; string comparison required--> "congestion_encountered": { "max": { diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl index 6239e27a871c..8d35911b5b90 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl @@ -44,19 +44,7 @@ "size": 2 }, "aggs": { - "bytes": { - "proportional_sum": { - "fields": [ - "netflow.delta_switched", - "netflow.last_switched", - "netflow.bytes", - "netflow.sampling_interval" - ], - "interval": "${step?long?c}ms", - "start": ${start?long?c}, - "end": ${end?long?c} - } - }, + "bytes": ${proportionalSum}, <#-- netflow.ecn is a keyword -> max aggregation not possible; string comparison required--> "congestion_encountered": { "max": { diff --git a/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java b/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java new file mode 100644 index 000000000000..2171e8bf5c25 --- /dev/null +++ b/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java @@ -0,0 +1,173 @@ +/* + * Licensed to The OpenNMS Group, Inc (TOG) under one or more + * contributor license agreements. See the LICENSE.md file + * distributed with this work for additional information + * regarding copyright ownership. + * + * TOG licenses this file to You under the GNU Affero General + * Public License Version 3 (the "License") or (at your option) + * any later version. You may not use this file except in + * compliance with the License. You may obtain a copy of the + * License at: + * + * https://www.gnu.org/licenses/agpl-3.0.txt + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific + * language governing permissions and limitations under the + * License. + */ +package org.opennms.netmgt.flows.elastic; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.closeTo; +import static org.hamcrest.Matchers.contains; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasSize; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +import org.hamcrest.Matchers; +import org.junit.Test; +import org.opennms.netmgt.flows.filter.api.TimeRangeFilter; + +import com.google.gson.JsonArray; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; + +public class ProportionalSumQueryTest { + + @Test + public void pluginStrategyShouldRenderLegacyProportionalSum() { + final String agg = ProportionalSumQuery.aggregationFor(ProportionalSumQuery.Strategy.PLUGIN, + 300_000L, 1_500_000L, 3_000_000L, + "netflow.delta_switched", "netflow.last_switched", "netflow.bytes", "netflow.sampling_interval"); + final JsonObject json = JsonParser.parseString(agg).getAsJsonObject(); + final JsonObject proportionalSum = json.getAsJsonObject("proportional_sum"); + assertThat(proportionalSum.get("interval").getAsString(), equalTo("300000ms")); + assertThat(proportionalSum.get("start").getAsLong(), equalTo(1_500_000L)); + assertThat(proportionalSum.get("end").getAsLong(), equalTo(3_000_000L)); + final JsonArray fields = proportionalSum.getAsJsonArray("fields"); + assertThat(fields.size(), equalTo(4)); + assertThat(fields.get(0).getAsString(), equalTo("netflow.delta_switched")); + assertThat(fields.get(3).getAsString(), equalTo("netflow.sampling_interval")); + } + + @Test + public void painlessStrategyShouldRenderScriptedMetric() { + final String agg = ProportionalSumQuery.aggregationFor(ProportionalSumQuery.Strategy.PAINLESS, + 300_000L, 1_500_000L, 3_000_000L, + "range_start", "range_end", "bytes_ingress", null); + final JsonObject json = JsonParser.parseString(agg).getAsJsonObject(); + final JsonObject scriptedMetric = json.getAsJsonObject("scripted_metric"); + final JsonObject params = scriptedMetric.getAsJsonObject("params"); + assertThat(params.get("interval").getAsLong(), equalTo(300_000L)); + assertThat(params.get("qstart").getAsLong(), equalTo(1_500_000L)); + assertThat(params.get("qend").getAsLong(), equalTo(3_000_000L)); + for (String script : Arrays.asList("init_script", "map_script", "combine_script", "reduce_script")) { + assertThat(script + " must be present", scriptedMetric.has(script), equalTo(true)); + } + // Field names are inlined into the script source (per-doc params lookups are slow); + // without a sampling field the sampling block must not be emitted at all. + final String mapScript = scriptedMetric.get("map_script").getAsString(); + assertThat(mapScript, Matchers.containsString("doc['range_start']")); + assertThat(mapScript, Matchers.containsString("doc['bytes_ingress']")); + assertThat(mapScript, Matchers.not(Matchers.containsString("sampling"))); + assertThat(mapScript, Matchers.not(Matchers.containsString("params.startField"))); + // 6 buckets -> dense variant with an array state sized via the nBuckets param + assertThat(params.get("nBuckets").getAsLong(), equalTo(6L)); + assertThat(scriptedMetric.get("init_script").getAsString(), Matchers.containsString("new double[")); + } + + @Test + public void painlessStrategyShouldInlineSamplingFieldWhenConfigured() { + final String agg = ProportionalSumQuery.aggregationFor(ProportionalSumQuery.Strategy.PAINLESS, + 300_000L, 1_500_000L, 3_000_000L, + "netflow.delta_switched", "netflow.last_switched", "netflow.bytes", "netflow.sampling_interval"); + final String mapScript = JsonParser.parseString(agg).getAsJsonObject() + .getAsJsonObject("scripted_metric").get("map_script").getAsString(); + assertThat(mapScript, Matchers.containsString("doc['netflow.sampling_interval']")); + assertThat(mapScript, Matchers.containsString("doc['netflow.delta_switched']")); + } + + @Test + public void painlessStrategyShouldFallBackToSparseScriptsForHugeBucketCounts() { + // 30 days at a 1 second step -> ~2.6M buckets, far above the dense limit + final long start = 1_500_000L; + final long end = start + 30L * 24 * 3600 * 1000; + final String agg = ProportionalSumQuery.aggregationFor(ProportionalSumQuery.Strategy.PAINLESS, + 1000L, start, end, "range_start", "range_end", "bytes_ingress", null); + final JsonObject scriptedMetric = JsonParser.parseString(agg).getAsJsonObject() + .getAsJsonObject("scripted_metric"); + assertThat(scriptedMetric.getAsJsonObject("params").has("nBuckets"), equalTo(false)); + assertThat(scriptedMetric.get("init_script").getAsString(), Matchers.containsString("new HashMap")); + assertThat(scriptedMetric.get("map_script").getAsString(), Matchers.containsString("state.sums.put")); + } + + @Test + public void strategyParsingShouldDefaultToPainless() { + assertThat(ProportionalSumQuery.Strategy.parse(null), equalTo(ProportionalSumQuery.Strategy.PAINLESS)); + assertThat(ProportionalSumQuery.Strategy.parse(""), equalTo(ProportionalSumQuery.Strategy.PAINLESS)); + assertThat(ProportionalSumQuery.Strategy.parse("painless"), equalTo(ProportionalSumQuery.Strategy.PAINLESS)); + assertThat(ProportionalSumQuery.Strategy.parse("bogus"), equalTo(ProportionalSumQuery.Strategy.PAINLESS)); + assertThat(ProportionalSumQuery.Strategy.parse("plugin"), equalTo(ProportionalSumQuery.Strategy.PLUGIN)); + assertThat(ProportionalSumQuery.Strategy.parse(" Plugin "), equalTo(ProportionalSumQuery.Strategy.PLUGIN)); + } + + @Test + public void seriesTemplatesShouldRenderValidJsonForBothStrategies() { + for (ProportionalSumQuery.Strategy strategy : ProportionalSumQuery.Strategy.values()) { + final SearchQueryProvider provider = new SearchQueryProvider(strategy); + final List filters = + Collections.singletonList(new TimeRangeFilter(1_500_000L, 3_000_000L)); + for (String query : Arrays.asList( + provider.getSeriesFromQuery(Arrays.asList("http", "https"), 300_000L, 1_500_000L, 3_000_000L, + "netflow.application", filters), + provider.getSeriesFromQuery(10, 300_000L, 1_500_000L, 3_000_000L, + "netflow.application", filters), + provider.getSeriesFromMissingQuery(300_000L, 1_500_000L, 3_000_000L, + "netflow.application", "Unknown", filters), + provider.getSeriesFromOthersQuery(Arrays.asList("http", "https"), 300_000L, 1_500_000L, 3_000_000L, + "netflow.application", false, filters))) { + final JsonObject json = JsonParser.parseString(query).getAsJsonObject(); + final String expectedAggType = strategy == ProportionalSumQuery.Strategy.PLUGIN + ? "proportional_sum" : "scripted_metric"; + assertThat("query for strategy " + strategy + " must contain a " + expectedAggType + " aggregation: " + query, + query.contains("\"" + expectedAggType + "\""), equalTo(true)); + assertThat(json.has("aggs"), equalTo(true)); + } + } + } + + @Test + public void shouldParsePluginBucketsShape() { + final JsonObject agg = JsonParser.parseString("{\"buckets\": [" + + "{\"key\": 1500000, \"key_as_string\": \"1500000\", \"doc_count\": 2, \"value\": 42.5}," + + "{\"key\": 1800000, \"key_as_string\": \"1800000\", \"doc_count\": 0, \"value\": 0.0}]}") + .getAsJsonObject(); + final ProportionalSumAggregation parsed = new ProportionalSumAggregation("bytes", agg); + assertThat(parsed.getBuckets(), hasSize(2)); + assertThat(parsed.getBuckets().get(0).getTime(), equalTo(1_500_000L)); + assertThat(parsed.getBuckets().get(0).getValue(), closeTo(42.5, 1e-9)); + } + + @Test + public void shouldParseScriptedMetricValueShape() { + // Keys deliberately unordered: the parser must sort buckets by time + final JsonObject agg = JsonParser.parseString( + "{\"value\": {\"1800000\": 7.25, \"1500000\": 42.5, \"2100000\": 0.0}}").getAsJsonObject(); + final ProportionalSumAggregation parsed = new ProportionalSumAggregation("bytes", agg); + assertThat(parsed.getBuckets(), hasSize(3)); + assertThat(parsed.getBuckets().stream() + .map(ProportionalSumAggregation.DateHistogram::getTime) + .collect(Collectors.toList()), + contains(1_500_000L, 1_800_000L, 2_100_000L)); + assertThat(parsed.getBuckets().get(0).getValue(), closeTo(42.5, 1e-9)); + assertThat(parsed.getBuckets().get(1).getValue(), closeTo(7.25, 1e-9)); + } +} diff --git a/features/flows/itests/pom.xml b/features/flows/itests/pom.xml index 52894400ab3c..18308e9a05d8 100644 --- a/features/flows/itests/pom.xml +++ b/features/flows/itests/pom.xml @@ -16,7 +16,6 @@ 0.3.1 1.1.1 8.18.2 - 2.0.7 @@ -46,32 +45,6 @@ false - - - org.apache.maven.plugins - maven-dependency-plugin - - - copy-elasticsearch-plugins - generate-test-resources - - copy - - - - - org.opennms.elasticsearch - elasticsearch-drift-plugin-${elasticsearchTargetVersion} - ${elasticsearchDriftPluginVersion} - zip - ${project.build.directory}/elasticsearch-plugins - drift-plugin.zip - - - - - - org.codehaus.mojo @@ -320,12 +293,6 @@ ${elasticsearchNettyVersion} test - - org.opennms.elasticsearch - elasticsearch-drift-plugin-${elasticsearchTargetVersion} - ${elasticsearchDriftPluginVersion} - test - org.opennms.features.flows org.opennms.features.flows.kafka-persistence @@ -371,13 +338,13 @@ org.testcontainers testcontainers - 1.17.6 + ${testcontainers.version} test org.testcontainers elasticsearch - 1.17.6 + ${testcontainers.version} test diff --git a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/AggregatedFlowQueryIT.java b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/AggregatedFlowQueryIT.java index 1007c043524b..454a2e08d008 100644 --- a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/AggregatedFlowQueryIT.java +++ b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/AggregatedFlowQueryIT.java @@ -53,7 +53,6 @@ import org.hamcrest.collection.IsIterableContainingInOrder; import org.hamcrest.number.IsCloseTo; import org.junit.Before; -import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Ignore; import org.junit.Rule; @@ -116,7 +115,6 @@ public class AggregatedFlowQueryIT { // Elasticsearch version used for testing private static final String ES_VERSION = "8.18.2"; - private static final String DRIFT_PLUGIN_VERSION = "2.0.7"; protected ElasticFlowRepository flowRepository; protected SmartQueryService smartQueryService; @@ -124,30 +122,17 @@ public class AggregatedFlowQueryIT { protected AggregatedFlowQueryService aggFlowQueryService; protected DocumentEnricherImpl documentEnricher; + // Stock Elasticsearch, on purpose: the proportional sums used by the series/totals + // queries are computed with inline Painless scripts and must work without the + // elasticsearch-drift-plugin being installed. @ClassRule public static ElasticTestContainerWithPlugins elasticsearchContainer; static { try { - elasticsearchContainer = new ElasticTestContainerWithPlugins("docker.elastic.co/elasticsearch/elasticsearch:" + ES_VERSION) - // We only need to add the drift plugin - the Painless plugin is built into the Elasticsearch image - .withPlugin("org.opennms.elasticsearch", "elasticsearch-drift-plugin-" + ES_VERSION, DRIFT_PLUGIN_VERSION); - - LOG.info("Initialized ElasticsearchMavenPluginContainer using downloaded plugin from Maven"); + elasticsearchContainer = new ElasticTestContainerWithPlugins("docker.elastic.co/elasticsearch/elasticsearch:" + ES_VERSION); } catch (IOException e) { - throw new RuntimeException("Failed to initialize ElasticsearchMavenPluginContainer", e); - } - } - - @BeforeClass - public static void setUpClass() { - - // Verify plugins were correctly installed - boolean pluginsInstalled = elasticsearchContainer.verifyPluginsInstalled(); - LOG.info("Elasticsearch plugins successfully installed: {}", pluginsInstalled); - - if (!pluginsInstalled) { - throw new RuntimeException("Failed to install required Elasticsearch plugins. Test cannot continue."); + throw new RuntimeException("Failed to initialize Elasticsearch test container", e); } } diff --git a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/FlowQueryIT.java b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/FlowQueryIT.java index bed824692bbd..a6f20c070404 100644 --- a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/FlowQueryIT.java +++ b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/FlowQueryIT.java @@ -58,7 +58,6 @@ import org.hamcrest.collection.IsIterableContainingInOrder; import org.hamcrest.number.IsCloseTo; import org.junit.Before; -import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; import org.opennms.core.cache.CacheConfigBuilder; @@ -115,32 +114,18 @@ public class FlowQueryIT { // Elasticsearch version used for testing private static final String ES_VERSION = "8.18.2"; - private static final String DRIFT_PLUGIN_VERSION = "2.0.7"; + // Stock Elasticsearch, on purpose: the proportional sums used by the series/totals + // queries are computed with inline Painless scripts and must work without the + // elasticsearch-drift-plugin being installed. @ClassRule public static ElasticTestContainerWithPlugins elasticsearchContainer; static { try { - elasticsearchContainer = new ElasticTestContainerWithPlugins("docker.elastic.co/elasticsearch/elasticsearch:" + ES_VERSION) - // We only need to add the drift plugin - the Painless plugin is built into the Elasticsearch image - .withPlugin("org.opennms.elasticsearch", "elasticsearch-drift-plugin-" + ES_VERSION, DRIFT_PLUGIN_VERSION); - - LOG.info("Initialized ElasticsearchMavenPluginContainer using downloaded plugin from Maven"); + elasticsearchContainer = new ElasticTestContainerWithPlugins("docker.elastic.co/elasticsearch/elasticsearch:" + ES_VERSION); } catch (IOException e) { - throw new RuntimeException("Failed to initialize ElasticsearchMavenPluginContainer", e); - } - } - - @BeforeClass - public static void setUpClass() { - - // Verify plugins were correctly installed - boolean pluginsInstalled = elasticsearchContainer.verifyPluginsInstalled(); - LOG.info("Elasticsearch plugins successfully installed: {}", pluginsInstalled); - - if (!pluginsInstalled) { - throw new RuntimeException("Failed to install required Elasticsearch plugins. Test cannot continue."); + throw new RuntimeException("Failed to initialize Elasticsearch test container", e); } } diff --git a/pom.xml b/pom.xml index fe6860715968..f9f29c985abc 100644 --- a/pom.xml +++ b/pom.xml @@ -1795,7 +1795,6 @@ 2.5.1 0.3.0 ${netty4Version} - 2.0.5 7.17.9 8.18.1 2.25.0 From c1b4b88263d39f502941db23fa1d5c3320cff519 Mon Sep 17 00:00:00 2001 From: Marshall Massengill Date: Wed, 22 Jul 2026 15:23:55 -0400 Subject: [PATCH 2/5] NMS-20027: address review feedback Give each scripted_metric bucket its own JsonObject (key/key_as_string/ doc_count/value) in ProportionalSumAggregation.parseValueMap instead of passing the whole value map to every DateHistogram, so per-bucket equality and debugging behave sanely. Add ProportionalSumStrategyComparisonIT, which runs the plugin and Painless strategies side by side against the same data and asserts they produce the same series and totals on an aligned window. It is opt-in: it needs the drift plugin build, so it is disabled unless the flow-drift-comparison Maven profile is active (which downloads the plugin and sets org.opennms.flows.drift.comparison=true). --- .../elastic/ProportionalSumAggregation.java | 11 +- features/flows/itests/pom.xml | 62 ++++ .../ProportionalSumStrategyComparisonIT.java | 294 ++++++++++++++++++ 3 files changed, 365 insertions(+), 2 deletions(-) create mode 100644 features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java diff --git a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java index 4f85049e8279..27f9ab0a7e84 100644 --- a/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java +++ b/features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/ProportionalSumAggregation.java @@ -67,8 +67,15 @@ private void parseValueMap(JsonObject valueMap) { continue; // Skip entries without a parseable bucket key } final double value = entry.getValue().getAsDouble(); - // Per-bucket document counts are not tracked by the scripted variant - dateHistograms.add(new DateHistogram(valueMap, time, entry.getKey(), value, 0L)); + // Per-bucket document counts are not tracked by the scripted variant. + // Build a per-bucket JsonObject mirroring the drift plugin's bucket shape so each + // DateHistogram carries its own backing JSON rather than the whole value map. + final JsonObject bucketObj = new JsonObject(); + bucketObj.addProperty(String.valueOf(KEY), time); + bucketObj.addProperty(String.valueOf(KEY_AS_STRING), entry.getKey()); + bucketObj.addProperty(String.valueOf(DOC_COUNT), 0L); + bucketObj.addProperty(String.valueOf(AggregationField.VALUE), value); + dateHistograms.add(new DateHistogram(bucketObj, time, entry.getKey(), value, 0L)); } dateHistograms.sort(Comparator.comparing(DateHistogram::getTime)); } diff --git a/features/flows/itests/pom.xml b/features/flows/itests/pom.xml index 18308e9a05d8..d2cfaf8020a7 100644 --- a/features/flows/itests/pom.xml +++ b/features/flows/itests/pom.xml @@ -354,4 +354,66 @@ test + + + + flow-drift-comparison + + 2.0.7 + + + + + org.apache.maven.plugins + maven-failsafe-plugin + + + true + + + + + + org.apache.maven.plugins + maven-dependency-plugin + + + copy-elasticsearch-plugins + generate-test-resources + + copy + + + + + org.opennms.elasticsearch + elasticsearch-drift-plugin-${elasticsearchTargetVersion} + ${elasticsearchDriftPluginVersion} + zip + ${project.build.directory}/elasticsearch-plugins + drift-plugin.zip + + + + + + + + + + + org.opennms.elasticsearch + elasticsearch-drift-plugin-${elasticsearchTargetVersion} + ${elasticsearchDriftPluginVersion} + test + + + + diff --git a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java new file mode 100644 index 000000000000..0d6fda6c1f24 --- /dev/null +++ b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java @@ -0,0 +1,294 @@ +/* + * Licensed to The OpenNMS Group, Inc (TOG) under one or more + * contributor license agreements. See the LICENSE.md file + * distributed with this work for additional information + * regarding copyright ownership. + * + * TOG licenses this file to You under the GNU Affero General + * Public License Version 3 (the "License") or (at your option) + * any later version. You may not use this file except in + * compliance with the License. You may obtain a copy of the + * License at: + * + * https://www.gnu.org/licenses/agpl-3.0.txt + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, + * either express or implied. See the License for the specific + * language governing permissions and limitations under the + * License. + */ +package org.opennms.netmgt.flows.elastic; + +import static org.awaitility.Awaitility.await; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.closeTo; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.hasSize; +import static org.junit.Assume.assumeTrue; +import static org.mockito.Mockito.mock; +import static org.opennms.integration.api.v1.flows.Flow.Direction; + +import java.io.IOException; +import java.time.Instant; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import javax.script.ScriptEngineManager; + +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.opennms.core.cache.CacheConfigBuilder; +import org.opennms.features.elastic.client.ElasticRestClient; +import org.opennms.features.elastic.client.ElasticRestClientFactory; +import org.opennms.features.jest.client.index.IndexSelector; +import org.opennms.features.jest.client.index.IndexStrategy; +import org.opennms.features.jest.client.template.IndexSettings; +import org.opennms.netmgt.dao.mock.MockInterfaceToNodeCache; +import org.opennms.netmgt.dao.mock.MockIpInterfaceDao; +import org.opennms.netmgt.dao.mock.MockNodeDao; +import org.opennms.netmgt.dao.mock.MockSessionUtils; +import org.opennms.netmgt.flows.api.Directional; +import org.opennms.netmgt.flows.api.Flow; +import org.opennms.netmgt.flows.api.FlowSource; +import org.opennms.netmgt.flows.api.TrafficSummary; +import org.opennms.netmgt.flows.classification.FilterService; +import org.opennms.netmgt.flows.classification.internal.DefaultClassificationEngine; +import org.opennms.netmgt.flows.classification.persistence.api.RuleBuilder; +import org.opennms.netmgt.flows.elastic.agg.AggregatedFlowQueryService; +import org.opennms.netmgt.flows.filter.api.Filter; +import org.opennms.netmgt.flows.filter.api.SnmpInterfaceIdFilter; +import org.opennms.netmgt.flows.filter.api.TimeRangeFilter; +import org.opennms.netmgt.flows.processing.FlowBuilder; +import org.opennms.netmgt.flows.processing.impl.DocumentEnricherImpl; +import org.opennms.netmgt.flows.processing.impl.DocumentMangler; +import org.opennms.netmgt.telemetry.protocols.cache.NodeInfoCache; +import org.opennms.netmgt.telemetry.protocols.cache.NodeInfoCacheImpl; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.codahale.metrics.MetricRegistry; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Lists; +import com.google.common.collect.Table; + +/** + * Runs the two proportional-sum strategies side by side against the same Elasticsearch instance and + * the same flow data, and asserts they return the same series and totals. + * + *

The drift plugin's {@code proportional_sum} aggregation is the reference implementation; this + * test is the equivalence check requested in review of NMS-20027. It compares the {@code plugin} + * strategy against the {@code painless} ({@code scripted_metric}) strategy on the newest supported + * Elasticsearch version, for query windows whose start is aligned to the step. The two + * implementations intentionally diverge on unaligned windows (NMS-20001, where the plugin + * undercounts), so the comparison uses an aligned window (start {@code 0}, step {@code 10}).

+ * + *

Disabled by default: it requires the {@code elasticsearch-drift-plugin} build, which the + * default flow itests no longer download. Run it with the {@code flow-drift-comparison} Maven + * profile, which downloads the plugin and sets {@code org.opennms.flows.drift.comparison=true}.

+ */ +public class ProportionalSumStrategyComparisonIT { + + private static final Logger LOG = LoggerFactory.getLogger(ProportionalSumStrategyComparisonIT.class); + + // Match the version the default itests target, and the plugin build for it. + private static final String ES_VERSION = "8.18.2"; + private static final String DRIFT_PLUGIN_VERSION = "2.0.7"; + + private static final String ENABLE_PROPERTY = "org.opennms.flows.drift.comparison"; + + // Aligned window: start is a multiple of the step, the case where both strategies must agree. + private static final long STEP = 10L; + + // Cells are floating-point; the plugin and the script accumulate in different orders. + private static final double ERROR = 1e-8; + + private static ElasticTestContainerWithPlugins elasticsearchContainer; + + private SmartQueryService painless; + private SmartQueryService plugin; + + @BeforeClass + public static void setUpClass() throws IOException { + assumeTrue("disabled by default; run with -Pflow-drift-comparison (sets -D" + ENABLE_PROPERTY + "=true)", + Boolean.getBoolean(ENABLE_PROPERTY)); + elasticsearchContainer = new ElasticTestContainerWithPlugins("docker.elastic.co/elasticsearch/elasticsearch:" + ES_VERSION) + // Only the drift plugin needs installing; Painless is built into the image. + .withPlugin("org.opennms.elasticsearch", "elasticsearch-drift-plugin-" + ES_VERSION, DRIFT_PLUGIN_VERSION); + elasticsearchContainer.start(); + if (!elasticsearchContainer.verifyPluginsInstalled()) { + throw new IllegalStateException("The drift plugin was not installed in the test container; cannot compare strategies."); + } + LOG.info("Elasticsearch {} with drift plugin {} started for strategy comparison", ES_VERSION, DRIFT_PLUGIN_VERSION); + } + + @AfterClass + public static void tearDownClass() { + if (elasticsearchContainer != null) { + elasticsearchContainer.stop(); + } + } + + @Before + public void setUp() throws Exception { + final MetricRegistry metricRegistry = new MetricRegistry(); + final ElasticRestClientFactory factory = new ElasticRestClientFactory(elasticsearchContainer.getHttpHostAddress(), null, null); + final ElasticRestClient client = factory.createClient(); + + final IndexSettings settings = new IndexSettings(); + settings.setIndexPrefix("flows"); + final IndexSelector rawIndexSelector = new IndexSelector(settings, RawFlowQueryService.INDEX_NAME, IndexStrategy.MONTHLY, 120000); + + // Two query services over the same index, differing only in the proportional-sum strategy. + painless = smartQueryService(metricRegistry, client, rawIndexSelector, ProportionalSumQuery.Strategy.PAINLESS); + plugin = smartQueryService(metricRegistry, client, rawIndexSelector, ProportionalSumQuery.Strategy.PLUGIN); + + final ElasticFlowRepository flowRepository = new ElasticFlowRepository(metricRegistry, client, IndexStrategy.MONTHLY, + new MockIdentity(), new MockTracerRegistry(), settings, 0, 0); + + client.deleteIndex("flows*"); + new RawIndexInitializer(client, settings).initialize(); + + final List flows = getFlows(); + flowRepository.persist(documentEnricher().enrich(flows, new FlowSource("test", "127.0.0.1", null))); + await().atMost(60, TimeUnit.SECONDS).until(() -> painless.getFlowCount( + Collections.singletonList(new TimeRangeFilter(0, System.currentTimeMillis()))).get(), equalTo((long) flows.size())); + } + + @Test + public void applicationSeriesMatch() throws Exception { + final Set apps = ImmutableSet.of("http", "https"); + assertSeriesEqual( + plugin.getApplicationSeries(apps, STEP, true, getFilters()).get(), + painless.getApplicationSeries(apps, STEP, true, getFilters()).get()); + } + + @Test + public void topNApplicationSeriesMatch() throws Exception { + assertSeriesEqual( + plugin.getTopNApplicationSeries(10, STEP, true, getFilters()).get(), + painless.getTopNApplicationSeries(10, STEP, true, getFilters()).get()); + } + + @Test + public void hostSeriesMatch() throws Exception { + assertSeriesEqual( + plugin.getTopNHostSeries(10, STEP, true, getFilters()).get(), + painless.getTopNHostSeries(10, STEP, true, getFilters()).get()); + } + + @Test + public void topNApplicationSummariesMatch() throws Exception { + assertSummariesEqual( + plugin.getTopNApplicationSummaries(10, true, getFilters()).get(), + painless.getTopNApplicationSummaries(10, true, getFilters()).get()); + } + + @Test + public void applicationSummariesMatch() throws Exception { + final Set apps = ImmutableSet.of("http", "https"); + assertSummariesEqual( + plugin.getApplicationSummaries(apps, true, getFilters()).get(), + painless.getApplicationSummaries(apps, true, getFilters()).get()); + } + + private static void assertSeriesEqual(Table, Long, Double> expected, + Table, Long, Double> actual) { + assertThat("row keys", actual.rowKeySet(), containsInAnyOrder(expected.rowKeySet().toArray())); + assertThat("column (bucket) keys", actual.columnKeySet(), containsInAnyOrder(expected.columnKeySet().toArray())); + for (Table.Cell, Long, Double> cell : expected.cellSet()) { + final Double actualValue = actual.get(cell.getRowKey(), cell.getColumnKey()); + assertThat("cell " + cell.getRowKey() + "@" + cell.getColumnKey(), actualValue, closeTo(cell.getValue(), ERROR)); + } + } + + private static void assertSummariesEqual(List> expected, List> actual) { + assertThat(actual, hasSize(expected.size())); + for (int i = 0; i < expected.size(); i++) { + final TrafficSummary e = expected.get(i); + final TrafficSummary a = actual.get(i); + assertThat("entity at " + i, a.getEntity(), equalTo(e.getEntity())); + assertThat("bytesIn for " + e.getEntity(), a.getBytesIn(), equalTo(e.getBytesIn())); + assertThat("bytesOut for " + e.getEntity(), a.getBytesOut(), equalTo(e.getBytesOut())); + } + } + + private static SmartQueryService smartQueryService(MetricRegistry metricRegistry, ElasticRestClient client, + IndexSelector indexSelector, ProportionalSumQuery.Strategy strategy) { + final RawFlowQueryService raw = new RawFlowQueryService(client, indexSelector, strategy.name()); + final AggregatedFlowQueryService agg = mock(AggregatedFlowQueryService.class); + final SmartQueryService smart = new SmartQueryService(metricRegistry, raw, agg); + smart.setAlwaysUseRawForQueries(true); // Compare the raw-flow query path, which renders proportional sums. + return smart; + } + + private DocumentEnricherImpl documentEnricher() throws InterruptedException { + final var classificationEngine = new DefaultClassificationEngine(() -> Lists.newArrayList( + new RuleBuilder().withName("http").withDstPort("80").withProtocol("tcp,udp").build(), + new RuleBuilder().withName("https").withDstPort("443").withProtocol("tcp,udp").build(), + new RuleBuilder().withName("http").withSrcPort("80").withProtocol("tcp,udp").build(), + new RuleBuilder().withName("https").withSrcPort("443").withProtocol("tcp,udp").build()), + FilterService.NOOP); + final NodeInfoCache nodeInfoCache = new NodeInfoCacheImpl( + new CacheConfigBuilder() + .withName("nodeInfoCache") + .withMaximumSize(1000) + .withExpireAfterWrite(300) + .withExpireAfterRead(300) + .build(), + true, + new MetricRegistry(), + new MockNodeDao(), + new MockIpInterfaceDao(), + new MockInterfaceToNodeCache(), + new MockSessionUtils()); + return new DocumentEnricherImpl(new MockSessionUtils(), classificationEngine, 0, + new DocumentMangler(new ScriptEngineManager()), nodeInfoCache); + } + + private static List getFilters() { + return Lists.newArrayList(new TimeRangeFilter(0, System.currentTimeMillis()), new SnmpInterfaceIdFilter(98)); + } + + // The default flow set from FlowQueryIT: overlapping intervals across several buckets, so the + // proportional-sum aggregation actually has work to do. + private static List getFlows() { + return new FlowBuilder() + .withSnmpInterfaceId(98) + // 192.168.1.100:43444 <-> 10.1.1.11:80 (110 bytes in [3,15]) + .withDirection(Direction.INGRESS) + .withTos(4 + 64) + .withFlow(Instant.ofEpochMilli(3), Instant.ofEpochMilli(15), "192.168.1.100", 43444, "10.1.1.11", 80, 10) + .withDirection(Direction.EGRESS) + .withTos(8 + 128) + .withFlow(Instant.ofEpochMilli(3), Instant.ofEpochMilli(15), "10.1.1.11", 80, "192.168.1.100", 43444, 100) + // 192.168.1.100:43445 <-> 10.1.1.12:443 (1100 bytes in [13,26]) + .withDirection(Direction.INGRESS) + .withHostnames(null, "la.le.lu") + .withTos(16 + 64) + .withFlow(Instant.ofEpochMilli(13), Instant.ofEpochMilli(26), "192.168.1.100", 43445, "10.1.1.12", 443, 100) + .withDirection(Direction.EGRESS) + .withHostnames("la.le.lu", null) + .withTos(32 + 128) + .withFlow(Instant.ofEpochMilli(13), Instant.ofEpochMilli(26), "10.1.1.12", 443, "192.168.1.100", 43445, 1000) + // 192.168.1.101:43442 <-> 10.1.1.12:443 (1210 bytes in [14,45]) + .withDirection(Direction.INGRESS) + .withHostnames("ingress.only", "la.le.lu") + .withFlow(Instant.ofEpochMilli(14), Instant.ofEpochMilli(45), "192.168.1.101", 43442, "10.1.1.12", 443, 110) + .withDirection(Direction.EGRESS) + .withHostnames("la.le.lu", null) + .withFlow(Instant.ofEpochMilli(14), Instant.ofEpochMilli(45), "10.1.1.12", 443, "192.168.1.101", 43442, 1100) + // 192.168.1.102:50000 <-> 10.1.1.13:50001 (300 bytes in [50,52]) + .withDirection(Direction.INGRESS) + .withFlow(Instant.ofEpochMilli(50), Instant.ofEpochMilli(52), "192.168.1.102", 50000, "10.1.1.13", 50001, 200) + .withDirection(Direction.EGRESS) + .withFlow(Instant.ofEpochMilli(50), Instant.ofEpochMilli(52), "10.1.1.13", 50001, "192.168.1.102", 50000, 100) + .build(); + } +} From ebfbd292b21132abf6bcc9b003ba1c81e826738d Mon Sep 17 00:00:00 2001 From: Marshall Massengill Date: Wed, 22 Jul 2026 16:07:39 -0400 Subject: [PATCH 3/5] NMS-20027: pass SNMP interface id to the direction script as a parameter The unknown-direction handling in the raw flow series queries inlined the SNMP interface id into the Painless script source, producing a distinct script per interface. On busy clusters that generates a stream of one-off compilations and can trip the dynamic script compilation rate limit (observed as high script.compilation_limit_triggered counts in the field). Render the interface id via script params instead, so the source is identical across interfaces and Elasticsearch compiles and caches it once. Verified on Elasticsearch 7.13.x: 25 queries with distinct interface ids produce 0 additional compilations with the parameterized form versus 25 with the inlined form, and the aggregation still classifies unknown-direction flows as ingress/egress by matching input/output interface. --- .../opennms/netmgt/flows/elastic/common.ftl | 26 +++++++++++++------ .../flows/elastic/series_for_missing.ftl | 2 +- .../flows/elastic/series_for_others.ftl | 2 +- .../netmgt/flows/elastic/series_for_terms.ftl | 2 +- .../elastic/ProportionalSumQueryTest.java | 25 ++++++++++++++++++ 5 files changed, 46 insertions(+), 11 deletions(-) diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/common.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/common.ftl index 4dc10f4fb473..d33dc9fc031f 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/common.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/common.ftl @@ -1,22 +1,32 @@ <#-- - Painless script for dealing with unknown values the direction field - Used when aggregating flows into ingress vs egress buckets - When the direction is unknown, we treat flow records as ingress for the matching input interface, - and as egress to for the matching output interface. + Painless script clause for dealing with unknown values in the direction field. + Used when aggregating flows into ingress vs egress buckets. + When the direction is unknown, we treat flow records as ingress for the matching input + interface, and as egress for the matching output interface. + + The SNMP interface id is passed as a script parameter rather than inlined into the + source, so Elasticsearch compiles and caches a single script regardless of which + interface is queried. Inlining the id produced a distinct script per interface and + could trip the dynamic script compilation rate limit on busy clusters. + + Renders the complete "script": { ... } clause, so callers drop it directly in place of + a terms aggregation's "field" entry. --> <#function unknownDirectionScript snmpInterfaceId> - <#local script> + <#local source> if (doc['netflow.direction'].value != 'unknown') { return doc['netflow.direction'].value; } - if (doc['netflow.input_snmp'].value == ${snmpInterfaceId?long?c}) { + long snmpInterfaceId = ((Number) params.snmpInterfaceId).longValue(); + + if (doc['netflow.input_snmp'].value == snmpInterfaceId) { return 'ingress'; } - if (doc['netflow.output_snmp'].value == ${snmpInterfaceId?long?c}){ + if (doc['netflow.output_snmp'].value == snmpInterfaceId){ return 'egress'; } - <#return script> + <#return '"script": {"source": "' + source?json_string + '", "params": {"snmpInterfaceId": ' + snmpInterfaceId?long?c + '}}'> diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl index 7f80c87de284..aa2773c5c78d 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_missing.ftl @@ -30,7 +30,7 @@ "direction": { "terms": { <#if snmpInterfaceId??> - "script": "${onms.unknownDirectionScript(snmpInterfaceId)?json_string}", + ${onms.unknownDirectionScript(snmpInterfaceId)}, <#else> "field": "netflow.direction", diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl index 67f934627b93..2c8cab859606 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_others.ftl @@ -31,7 +31,7 @@ "direction": { "terms": { <#if snmpInterfaceId??> - "script": "${onms.unknownDirectionScript(snmpInterfaceId)?json_string}", + ${onms.unknownDirectionScript(snmpInterfaceId)}, <#else> "field": "netflow.direction", diff --git a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl index 8d35911b5b90..8bd227a0713f 100644 --- a/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl +++ b/features/flows/elastic/src/main/resources/org/opennms/netmgt/flows/elastic/series_for_terms.ftl @@ -37,7 +37,7 @@ "direction": { "terms": { <#if snmpInterfaceId??> - "script": "${onms.unknownDirectionScript(snmpInterfaceId)?json_string}", + ${onms.unknownDirectionScript(snmpInterfaceId)}, <#else> "field": "netflow.direction", diff --git a/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java b/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java index 2171e8bf5c25..ffec8c653d25 100644 --- a/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java +++ b/features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumQueryTest.java @@ -34,6 +34,8 @@ import org.hamcrest.Matchers; import org.junit.Test; +import org.opennms.netmgt.flows.filter.api.Filter; +import org.opennms.netmgt.flows.filter.api.SnmpInterfaceIdFilter; import org.opennms.netmgt.flows.filter.api.TimeRangeFilter; import com.google.gson.JsonArray; @@ -144,6 +146,29 @@ public void seriesTemplatesShouldRenderValidJsonForBothStrategies() { } } + @Test + public void interfaceScopedDirectionScriptShouldPassInterfaceIdAsParam() { + // The unknown-direction handling inlines the field name but passes the SNMP interface + // id via params, so the script source is identical for every interface and Elasticsearch + // compiles it once rather than per interface. + final int snmpInterfaceId = 4242; + final SearchQueryProvider provider = new SearchQueryProvider(ProportionalSumQuery.Strategy.PAINLESS); + final List filters = Arrays.asList(new TimeRangeFilter(1_500_000L, 3_000_000L), + new SnmpInterfaceIdFilter(snmpInterfaceId)); + for (String query : Arrays.asList( + provider.getSeriesFromQuery(10, 300_000L, 1_500_000L, 3_000_000L, "netflow.application", filters), + provider.getSeriesFromMissingQuery(300_000L, 1_500_000L, 3_000_000L, "netflow.application", "Unknown", filters), + provider.getSeriesFromOthersQuery(Arrays.asList("http", "https"), 300_000L, 1_500_000L, 3_000_000L, + "netflow.application", false, filters))) { + // Must be well-formed JSON with the id supplied as a parameter, not inlined into the source. + JsonParser.parseString(query).getAsJsonObject(); + assertThat(query, Matchers.containsString("\"snmpInterfaceId\": " + snmpInterfaceId)); + assertThat(query, Matchers.containsString("params.snmpInterfaceId")); + assertThat("interface id must not be inlined into the script source: " + query, + query.contains("== " + snmpInterfaceId), equalTo(false)); + } + } + @Test public void shouldParsePluginBucketsShape() { final JsonObject agg = JsonParser.parseString("{\"buckets\": [" + From 9994da3cf37b50138e6e1e49d233fd2489c38dd7 Mon Sep 17 00:00:00 2001 From: Marshall Massengill Date: Thu, 23 Jul 2026 08:19:01 -0400 Subject: [PATCH 4/5] NMS-20027: run the drift-vs-Painless comparison IT by default The drift plugin is staying supported for now, so run the equivalence check in CI instead of behind an opt-in profile. The flows itests pom downloads the drift plugin and enables the comparison; the other flow ITs still run plugin-free. The drift plugin version is now supplied via org.opennms.flows.drift.version so it can be pointed at the NMS-20001 bucketing fix without editing the test. Adds a gated unaligned-window agreement test: with the current plugin the strategies diverge on unaligned windows (NMS-20001), so it is skipped unless driftUnalignedFixed=true. When the drift fix ships, bump elasticsearchDriftPluginVersion and set driftUnalignedFixed=true to validate that the fixed plugin again matches the Painless reference. --- features/flows/itests/pom.xml | 104 +++++++----------- .../ProportionalSumStrategyComparisonIT.java | 44 +++++--- 2 files changed, 72 insertions(+), 76 deletions(-) diff --git a/features/flows/itests/pom.xml b/features/flows/itests/pom.xml index d2cfaf8020a7..cd19e0772468 100644 --- a/features/flows/itests/pom.xml +++ b/features/flows/itests/pom.xml @@ -16,6 +16,10 @@ 0.3.1 1.1.1 8.18.2 + + 2.0.7 + false @@ -40,11 +44,42 @@ ${test.storepass} ${test.keystore} ${test.storepass} + + true + ${elasticsearchDriftPluginVersion} + ${driftUnalignedFixed} 1 false + + + org.apache.maven.plugins + maven-dependency-plugin + + + copy-elasticsearch-plugins + generate-test-resources + + copy + + + + + org.opennms.elasticsearch + elasticsearch-drift-plugin-${elasticsearchTargetVersion} + ${elasticsearchDriftPluginVersion} + zip + ${project.build.directory}/elasticsearch-plugins + drift-plugin.zip + + + + + + org.codehaus.mojo @@ -353,67 +388,12 @@ snappy-java test + + + org.opennms.elasticsearch + elasticsearch-drift-plugin-${elasticsearchTargetVersion} + ${elasticsearchDriftPluginVersion} + test + - - - - flow-drift-comparison - - 2.0.7 - - - - - org.apache.maven.plugins - maven-failsafe-plugin - - - true - - - - - - org.apache.maven.plugins - maven-dependency-plugin - - - copy-elasticsearch-plugins - generate-test-resources - - copy - - - - - org.opennms.elasticsearch - elasticsearch-drift-plugin-${elasticsearchTargetVersion} - ${elasticsearchDriftPluginVersion} - zip - ${project.build.directory}/elasticsearch-plugins - drift-plugin.zip - - - - - - - - - - - org.opennms.elasticsearch - elasticsearch-drift-plugin-${elasticsearchTargetVersion} - ${elasticsearchDriftPluginVersion} - test - - - - diff --git a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java index 0d6fda6c1f24..0508b988b664 100644 --- a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java +++ b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java @@ -82,29 +82,23 @@ * Runs the two proportional-sum strategies side by side against the same Elasticsearch instance and * the same flow data, and asserts they return the same series and totals. * - *

The drift plugin's {@code proportional_sum} aggregation is the reference implementation; this - * test is the equivalence check requested in review of NMS-20027. It compares the {@code plugin} - * strategy against the {@code painless} ({@code scripted_metric}) strategy on the newest supported - * Elasticsearch version, for query windows whose start is aligned to the step. The two - * implementations intentionally diverge on unaligned windows (NMS-20001, where the plugin - * undercounts), so the comparison uses an aligned window (start {@code 0}, step {@code 10}).

- * - *

Disabled by default: it requires the {@code elasticsearch-drift-plugin} build, which the - * default flow itests no longer download. Run it with the {@code flow-drift-comparison} Maven - * profile, which downloads the plugin and sets {@code org.opennms.flows.drift.comparison=true}.

+ *

Aligned-window assertions run by default (the flows itests pom downloads the plugin and sets + * the {@code org.opennms.flows.drift.*} properties). Unaligned-window agreement only holds against a + * drift build that fixes NMS-20001, so it is gated behind {@code driftUnalignedFixed}.

*/ public class ProportionalSumStrategyComparisonIT { private static final Logger LOG = LoggerFactory.getLogger(ProportionalSumStrategyComparisonIT.class); - // Match the version the default itests target, and the plugin build for it. private static final String ES_VERSION = "8.18.2"; - private static final String DRIFT_PLUGIN_VERSION = "2.0.7"; + // Supplied by the pom so it can be bumped to the NMS-20001 fix without editing this test. + private static final String DRIFT_PLUGIN_VERSION = System.getProperty("org.opennms.flows.drift.version", "2.0.7"); private static final String ENABLE_PROPERTY = "org.opennms.flows.drift.comparison"; + private static final String UNALIGNED_FIXED_PROPERTY = "org.opennms.flows.drift.unalignedFixed"; - // Aligned window: start is a multiple of the step, the case where both strategies must agree. private static final long STEP = 10L; + private static final long UNALIGNED_START = 5L; // not a multiple of STEP // Cells are floating-point; the plugin and the script accumulate in different orders. private static final double ERROR = 1e-8; @@ -116,7 +110,7 @@ public class ProportionalSumStrategyComparisonIT { @BeforeClass public static void setUpClass() throws IOException { - assumeTrue("disabled by default; run with -Pflow-drift-comparison (sets -D" + ENABLE_PROPERTY + "=true)", + assumeTrue("requires the drift plugin; enabled by the itests pom (-D" + ENABLE_PROPERTY + "=true)", Boolean.getBoolean(ENABLE_PROPERTY)); elasticsearchContainer = new ElasticTestContainerWithPlugins("docker.elastic.co/elasticsearch/elasticsearch:" + ES_VERSION) // Only the drift plugin needs installing; Painless is built into the image. @@ -198,6 +192,24 @@ public void applicationSummariesMatch() throws Exception { painless.getApplicationSummaries(apps, true, getFilters()).get()); } + // Unaligned windows only agree on a drift build that fixes NMS-20001; skipped otherwise. + @Test + public void unalignedWindowSeriesMatchOnFixedPlugin() throws Exception { + assumeTrue("requires a drift build that fixes NMS-20001 (set -DdriftUnalignedFixed=true)", + Boolean.getBoolean(UNALIGNED_FIXED_PROPERTY)); + final List filters = unalignedFilters(); + final Set apps = ImmutableSet.of("http", "https"); + assertSeriesEqual( + plugin.getApplicationSeries(apps, STEP, true, filters).get(), + painless.getApplicationSeries(apps, STEP, true, filters).get()); + assertSeriesEqual( + plugin.getTopNApplicationSeries(10, STEP, true, filters).get(), + painless.getTopNApplicationSeries(10, STEP, true, filters).get()); + assertSummariesEqual( + plugin.getTopNApplicationSummaries(10, true, filters).get(), + painless.getTopNApplicationSummaries(10, true, filters).get()); + } + private static void assertSeriesEqual(Table, Long, Double> expected, Table, Long, Double> actual) { assertThat("row keys", actual.rowKeySet(), containsInAnyOrder(expected.rowKeySet().toArray())); @@ -256,6 +268,10 @@ private static List getFilters() { return Lists.newArrayList(new TimeRangeFilter(0, System.currentTimeMillis()), new SnmpInterfaceIdFilter(98)); } + private static List unalignedFilters() { + return Lists.newArrayList(new TimeRangeFilter(UNALIGNED_START, System.currentTimeMillis()), new SnmpInterfaceIdFilter(98)); + } + // The default flow set from FlowQueryIT: overlapping intervals across several buckets, so the // proportional-sum aggregation actually has work to do. private static List getFlows() { From d0abefcd00f8a85da239d10ea32a2af19a4d9eda Mon Sep 17 00:00:00 2001 From: Marshall Massengill Date: Thu, 23 Jul 2026 11:46:35 -0400 Subject: [PATCH 5/5] NMS-20027: treat NaN as equal in the strategy comparison Empty series buckets are filled with Double.NaN, and both strategies produce NaN in the same cells, but Hamcrest closeTo never matches NaN so the aligned comparison failed on empty buckets even though the drift and Painless results agree. Compare NaN cells as equal and keep the tolerance check for real values. --- .../elastic/ProportionalSumStrategyComparisonIT.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java index 0508b988b664..092744e6a955 100644 --- a/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java +++ b/features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ProportionalSumStrategyComparisonIT.java @@ -216,7 +216,14 @@ private static void assertSeriesEqual(Table, Long, Double> ex assertThat("column (bucket) keys", actual.columnKeySet(), containsInAnyOrder(expected.columnKeySet().toArray())); for (Table.Cell, Long, Double> cell : expected.cellSet()) { final Double actualValue = actual.get(cell.getRowKey(), cell.getColumnKey()); - assertThat("cell " + cell.getRowKey() + "@" + cell.getColumnKey(), actualValue, closeTo(cell.getValue(), ERROR)); + final Double expectedValue = cell.getValue(); + final String where = "cell " + cell.getRowKey() + "@" + cell.getColumnKey(); + // Empty buckets are filled with NaN; treat NaN as equal to NaN (closeTo never matches NaN). + if (expectedValue == null || expectedValue.isNaN()) { + assertThat(where, actualValue == null || actualValue.isNaN(), equalTo(true)); + } else { + assertThat(where, actualValue, closeTo(expectedValue, ERROR)); + } } }