NMS-20027: Compute flow proportional sums with Painless scripts

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.
 ** A retention process to remove old flow indices so the disk does not fill up (for example, "keep _X_ days of flows").
 See <<deep-dive/elasticsearch/introduction.adoc#ga-elasticsearch-index-retention, Index retention>>.

docs/modules/releasenotes/pages/whatsnew.adoc

@@ -19,6 +19,15 @@ Most installations require no action, but there are behavior and compatibility c
 custom Spring Security configuration (SSO, LDAP), custom Camel routes, or plugins that use the OpenNMS persistence APIs.
 See xref:spring-hibernate-upgrade.adoc[Spring, Hibernate, and Camel Platform Upgrade] for details.
 
+=== 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`).

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,37 @@ 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<String, JsonElement> 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.
+            // 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));
+    }
+
     private void parseBuckets(JsonArray bucketsSource) {
         for (JsonElement bucket : bucketsSource) {
             JsonObject bucketObj = bucket.getAsJsonObject();