add map reduce for top n for elk

Author: weyrick

reporting/top_n.elk

@@ -0,0 +1,93 @@
+    "top_n": {
+      "scripted_metric": {
+        "init_script": """
+          state.top_n = new HashMap();
+          state.top_n["dns_top_qname2"] = new LinkedHashMap();
+          state.top_n["dns_top_qname3"] = new LinkedHashMap();
+          state.top_n["dns_top_nxdomain"] = new LinkedHashMap();
+          state.top_n["dns_top_qtype"] = new LinkedHashMap();
+          state.top_n["dns_top_rcode"] = new LinkedHashMap();
+          state.top_n["dns_top_refused"] = new LinkedHashMap();
+          state.top_n["dns_top_srvfail"] = new LinkedHashMap();
+          state.top_n["dns_top_udp_ports"] = new LinkedHashMap();
+          state.top_n["packets_top_ASN"] = new LinkedHashMap();
+          state.top_n["packets_top_geoLoc"] = new LinkedHashMap();
+          state.top_n["packets_top_ipv4"] = new LinkedHashMap();
+          state.top_n["packets_top_ipv6"] = new LinkedHashMap();
+          state.top_n["xact_in_top_slow"] = new LinkedHashMap();
+          state.top_n["xact_out_top_slow"] = new LinkedHashMap();
+        """,
+        "map_script": """
+        long deep = doc["http.packets_deep_samples"][0].longValue();
+        long total = doc["http.packets_total"][0].longValue();
+        double adjust = 1.0;
+        if (total > 0L && deep > 0L) {
+          adjust = Math.round(1.0 / (deep.doubleValue() / total.doubleValue()));            
+        }
+        for (Map.Entry entry: state.top_n.entrySet()) {
+          for (int i = 0; i <= 9; i++) {
+            String name_key = "http." + entry.getKey() + "_" + String.valueOf(i) + "_name.raw";
+            String val_key = "http." + entry.getKey() + "_" + String.valueOf(i) + "_estimate";
+            if (doc.containsKey(name_key) && doc[name_key].size() > 0 && doc[val_key].size() > 0) {
+              String name = doc[name_key][0].toLowerCase();
+              long val = doc[val_key][0].longValue();
+              if (state.top_n[entry.getKey()].containsKey(name)) {
+                state.top_n[entry.getKey()][name] += (long)(val*adjust);              
+              }
+              else {
+                state.top_n[entry.getKey()][name] = (long)(val*adjust);
+              }
+            }
+          }
+        }
+        """,
+        "combine_script": """
+        for (Map.Entry entry: state.top_n.entrySet()) {
+          ArrayList list = state.top_n[entry.getKey()].entrySet().stream().sorted(Map.Entry.comparingByValue())
+          .collect(Collectors.toList());
+          Collections.reverse(list);
+          state.top_n[entry.getKey()].clear();
+          int i = 0;
+          for (Map.Entry subentry: list) {
+            i++;
+            if (i > 10)
+              break;
+            state.top_n[entry.getKey()].put(subentry.getKey(), subentry.getValue());
+          }
+        }
+        return state.top_n;
+        """,
+        "reduce_script": """
+        HashMap top_n = new HashMap();
+        for (shard_map in states) {
+          for (Map.Entry entry : shard_map.entrySet()) {
+            if (!top_n.containsKey(entry.getKey())) {
+              top_n[entry.getKey()] = new LinkedHashMap();              
+            }
+            for (Map.Entry subentry : entry.getValue().entrySet()) {
+              if (top_n[entry.getKey()].containsKey(subentry.getKey())) {
+                top_n[entry.getKey()][subentry.getKey()] += subentry.getValue();              
+              }
+              else {
+                top_n[entry.getKey()][subentry.getKey()] = subentry.getValue();  
+              }
+            }
+          }
+        }
+        for (Map.Entry entry: top_n.entrySet()) {
+          ArrayList list = top_n[entry.getKey()].entrySet().stream().sorted(Map.Entry.comparingByValue())
+          .collect(Collectors.toList());
+          Collections.reverse(list);
+          top_n[entry.getKey()].clear();
+          int i = 0;
+          for (Map.Entry subentry: list) {
+            i++;
+            if (i > 10)
+              break;
+            top_n[entry.getKey()].put(subentry.getKey(), subentry.getValue());
+          }
+        }        
+        return top_n;
+        """
+      }
+    }