Merge pull request #16390 from Security-Onion-Solutions/foundry/mwaq014injcipm

Telegraf: fix EPS double counting and duplicate metric rows

salt/docker/defaults.yaml

@@ -103,6 +103,7 @@ docker:
         - 80:80
         - 443:443
         - 8443:8443
+        - 7788:7788
         - 7789:7789
       custom_bind_mounts: []
       extra_hosts: []

salt/firewall/defaults.yaml

@@ -118,6 +118,10 @@ firewall:
       tcp:
         - 5601
       udp: []
+    localrules:
+      tcp:
+        - 7788
+      udp: []
     nginx:
       tcp:
         - 80
@@ -196,6 +200,7 @@ firewall:
                 - postgres
                 - elasticsearch_rest
                 - elasticsearch_node
+                - localrules
             sensor:
               portgroups:
                 - beats_5044
@@ -386,6 +391,7 @@ firewall:
                 - elastic_agent_control
                 - elastic_agent_data
                 - elastic_agent_update
+                - localrules
                 - sensoroni
             fleet:
               portgroups:
@@ -605,6 +611,7 @@ firewall:
                 - elastic_agent_control
                 - elastic_agent_data
                 - elastic_agent_update
+                - localrules
                 - sensoroni
             fleet:
               portgroups:
@@ -822,6 +829,7 @@ firewall:
                 - elastic_agent_control
                 - elastic_agent_data
                 - elastic_agent_update
+                - localrules
                 - sensoroni
             fleet:
               portgroups:
@@ -1050,6 +1058,7 @@ firewall:
                 - elastic_agent_update
                 - endgame
                 - strelka_frontend
+                - localrules
             fleet:
               portgroups:
                 - elasticsearch_rest
@@ -1509,6 +1518,7 @@ firewall:
                 - elasticsearch_rest
                 - elasticsearch_node
                 - elastic_agent_control
+                - localrules
             sensor:
               portgroups:
                 - beats_5044

salt/firewall/soc_firewall.yaml

@@ -138,6 +138,9 @@ firewall:
     kibana:
       tcp: *tcpsettings
       udp: *udpsettings
+    localrules:
+      tcp: *tcpsettings
+      udp: *udpsettings
     nginx:
       tcp: *tcpsettings
       udp: *udpsettings

salt/manager/tools/sbin/so-yaml.py

@@ -20,7 +20,6 @@ def showUsage(args):
     print('    append           - Append a list item to a yaml key, if it exists and is a list. Requires KEY and LISTITEM args.', file=sys.stderr)
     print('    appendlistobject - Append an object to a yaml list key. Requires KEY and JSON_OBJECT args.', file=sys.stderr)
     print('    removelistitem   - Remove a list item from a yaml key, if it exists and is a list. Requires KEY and LISTITEM args.', file=sys.stderr)
-    print('    purgelistitem    - Remove every occurrence of a list item from all lists nested under a yaml key, if it exists. Requires KEY and LISTITEM args.', file=sys.stderr)
     print('    replacelistobject  - Replace a list object based on a condition. Requires KEY, CONDITION_FIELD, CONDITION_VALUE, and JSON_OBJECT args.', file=sys.stderr)
     print('    add              - Add a new key and set its value. Fails if key already exists. Requires KEY and VALUE args.', file=sys.stderr)
     print('    get [-r]         - Displays (to stdout) the value stored in the given key. Requires KEY arg. Use -r for raw output without YAML formatting.', file=sys.stderr)
@@ -200,33 +199,6 @@ def removelistitem(args):
     return 0
 
 
-def purgeListItem(content, listItem):
-    if isinstance(content, dict):
-        for value in content.values():
-            purgeListItem(value, listItem)
-    elif isinstance(content, list):
-        content[:] = [item for item in content if item != listItem]
-        for item in content:
-            purgeListItem(item, listItem)
-
-
-def purgelistitem(args):
-    if len(args) != 3:
-        print('Missing filename, key arg, or list item to purge', file=sys.stderr)
-        showUsage(None)
-        return 1
-
-    filename = args[0]
-    key = args[1]
-    listItem = args[2]
-
-    content = loadYaml(filename)
-    purgeListItem(getKeyValue(content, key), convertType(listItem))
-    writeYaml(filename, content)
-
-    return 0
-
-
 def replaceListObjectByCondition(content, key, conditionField, conditionValue, newObject):
     pieces = key.split(".", 1)
     if len(pieces) > 1:
@@ -413,7 +385,6 @@ def main():
         "append": append,
         "appendlistobject": appendlistobject,
         "removelistitem": removelistitem,
-        "purgelistitem": purgelistitem,
         "replacelistobject": replacelistobject,
         "get": get,
         "remove": remove,

salt/manager/tools/sbin/so-yaml_test.py

@@ -709,223 +709,6 @@ class TestRemoveListItem(unittest.TestCase):
                 self.assertEqual("The existing value for the given key is not a list. No action was taken on the file.\n", mock_stderr.getvalue())
 
 
-class TestPurgeListItem(unittest.TestCase):
-
-    def test_purgelistitem_missing_arg(self):
-        with patch('sys.exit', new=MagicMock()) as sysmock:
-            with patch('sys.stderr', new=StringIO()) as mock_stderr:
-                sys.argv = ["cmd", "help"]
-                soyaml.purgelistitem(["file", "key"])
-                sysmock.assert_called()
-                self.assertIn("Missing filename, key arg, or list item to purge", mock_stderr.getvalue())
-
-    def test_purgelistitem(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [a,b,c], key2: false}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "b"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- a\n- c\nkey2: false\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_nested_lists(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: { child1: [a,b,c], child2: { deep1: [b,d], deep2: 45 } }, key2: [b,e]}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "b"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n  child1:\n  - a\n  - c\n  child2:\n    deep1:\n    - d\n    deep2: 45\nkey2:\n- b\n- e\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_duplicates(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [b,a,b,c,b]}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "b"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- a\n- c\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_list_of_objects(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [{name: item1, values: [a,b]}, b, {name: item2, values: [b,c]}]}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "b"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- name: item1\n  values:\n  - a\n- name: item2\n  values:\n  - c\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_typed(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [1, '1', 2]}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "1"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- '1'\n- 2\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_item_not_in_list(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: { child1: [a,b,c] }}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "d"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n  child1:\n  - a\n  - b\n  - c\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_key_noexist(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [a,b,c]}")
-        file.close()
-
-        result = soyaml.purgelistitem([filename, "key2.child1", "b"])
-        self.assertEqual(result, 0)
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- a\n- b\n- c\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_key_noexist_deep(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: { child1: [a,b,c] }}")
-        file.close()
-
-        result = soyaml.purgelistitem([filename, "key1.child2.deep1", "b"])
-        self.assertEqual(result, 0)
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n  child1:\n  - a\n  - b\n  - c\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_scalar_and_null_values(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: { child1: b, child2: null, child3: [b] }}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "b"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n  child1: b\n  child2: null\n  child3: []\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_lists_of_lists(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [[a,b], [b,[b,c]], b]}")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "b"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- - a\n- - - c\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_file_object(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: [{name: x}, a, {name: y}, {name: x}]}")
-        file.close()
-
-        itemfile = "/tmp/so-yaml_test-purgelistitem-item.yaml"
-        file = open(itemfile, "w")
-        file.write("name: x\n")
-        file.close()
-
-        soyaml.purgelistitem([filename, "key1", "file:" + itemfile])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n- a\n- name: y\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_main(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: { child1: [a,b,c] }, key2: [b]}")
-        file.close()
-
-        with patch('sys.exit', new=MagicMock()) as sysmock:
-            sys.argv = ["cmd", "purgelistitem", filename, "key1", "b"]
-            soyaml.main()
-            sysmock.assert_called_once_with(0)
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n  child1:\n  - a\n  - c\nkey2:\n- b\n"
-        self.assertEqual(actual, expected)
-
-    def test_purgelistitem_with_remove(self):
-        filename = "/tmp/so-yaml_test-purgelistitem.yaml"
-        file = open(filename, "w")
-        file.write("{key1: { child1: { deep1: [a] }, child2: [b] }, key2: { group1: { items: [x,child1,y] }, group2: { items: [child1] } }}")
-        file.close()
-
-        soyaml.remove([filename, "key1.child1"])
-        soyaml.purgelistitem([filename, "key2", "child1"])
-
-        file = open(filename, "r")
-        actual = file.read()
-        file.close()
-
-        expected = "key1:\n  child2:\n  - b\nkey2:\n  group1:\n    items:\n    - x\n    - y\n  group2:\n    items: []\n"
-        self.assertEqual(actual, expected)
-
-
 class TestAppendListObject(unittest.TestCase):
 
     def test_appendlistobject_missing_arg(self):

salt/manager/tools/sbin/soup

@@ -527,7 +527,6 @@ preupgrade_changes() {
     [[ "$INSTALLEDVERSION" == "3.1.0" ]] && up_to_3.2.0
     [[ "$INSTALLEDVERSION" == "3.2.0" ]] && up_to_3.3.0
     [[ "$INSTALLEDVERSION" == "3.3.0" ]] && up_to_3.4.0
-    [[ "$INSTALLEDVERSION" == "3.4.0" ]] && up_to_3.5.0
     true
 }
 
@@ -1203,25 +1202,6 @@ add_detection_runtime_fields() {
 }
 ### 3.4.0 End ###
 
-### 3.5.0 Scripts ###
-up_to_3.5.0() {
-  remove_localrules_portgroup
-
-  INSTALLEDVERSION=3.5.0
-}
-
-remove_localrules_portgroup() {
-  local pillar_file
-
-  for pillar_file in /opt/so/saltstack/local/pillar/firewall/{soc,adv}_firewall.sls; do
-    grep -qs localrules "$pillar_file" || continue
-    echo "Removing the localrules portgroup from $pillar_file."
-    so-yaml.py remove "$pillar_file" firewall.portgroups.localrules
-    so-yaml.py purgelistitem "$pillar_file" firewall.role localrules
-  done
-}
-### 3.5.0 End ###
-
 
 repo_sync() {
   echo "Sync the local repo."

salt/nginx/enabled.sls

@@ -54,9 +54,9 @@ so-nginx:
       - /opt/so/log/nginx/:/var/log/nginx:rw
       - /opt/so/tmp/nginx/:/var/lib/nginx:rw
       - /opt/so/tmp/nginx/:/run:rw
-      - /nsm/elastic-fleet/artifacts/:/opt/socore/elastic-artifacts/artifacts:ro
-      {% if GLOBALS.is_manager %}
       - /nsm/elastic-fleet/so_agent-installers/:/opt/socore/html/packages
+      - /nsm/elastic-fleet/artifacts/:/opt/socore/html/artifacts 
+      {% if GLOBALS.is_manager %}
       - /etc/pki/managerssl.crt:/etc/pki/nginx/server.crt:ro
       - /etc/pki/managerssl.key:/etc/pki/nginx/server.key:ro
       # ATT&CK Navigator binds
@@ -64,6 +64,7 @@ so-nginx:
       - /opt/so/conf/navigator/config.json:/opt/socore/html/navigator/assets/config.json:ro
       - /nsm/repo:/opt/socore/html/repo:ro
       - /nsm/kernelrepo:/opt/socore/html/kernelrepo:ro
+      - /nsm/rules:/nsm/rules:ro
       {%   if NGINXMERGED.external_suricata %}
       - /opt/so/rules/nids/suri:/surirules:ro
       {%   endif %}

salt/nginx/etc/nginx.conf

@@ -4,8 +4,6 @@
 {%- set role = grains.id.split('_') | last %}
 {%- set influxpass = salt['pillar.get']('secrets:influx_pass') %}
 {%- set influxauth = ('so:' + influxpass) | base64_encode %}
-{%- set all_names = [GLOBALS.hostname, GLOBALS.url_base] + NGINXMERGED.ssl.alt_names %}
-{%- set full_server_name = all_names | unique | join(' ') %}
 
 worker_processes auto;
 error_log /var/log/nginx/error.log;
@@ -41,31 +39,52 @@ http {
 
 	include /etc/nginx/conf.d/*.conf;
 
+	{%- if role in ['fleet'] %}
+
+	server {
+		listen 8443;
+		server_name {{ GLOBALS.hostname }};
+		root /opt/socore/html;
+		location /artifacts/ {
+			try_files $uri =206;
+			proxy_read_timeout    90;
+			proxy_connect_timeout 90;
+			proxy_set_header      Host $host;
+			proxy_set_header      X-Real-IP $remote_addr;
+			proxy_set_header      X-Forwarded-For $proxy_add_x_forwarded_for;
+			proxy_set_header      Proxy "";
+			proxy_set_header      X-Forwarded-Proto $scheme;
+		}
+	}
+
+    {%- endif %}
+
 	{%- if GLOBALS.is_manager %}
+	{%- set all_names = [GLOBALS.hostname, GLOBALS.url_base] + NGINXMERGED.ssl.alt_names %}
+	{%- set full_server_name = all_names | unique | join(' ') %}
+
 	server {
 		listen 80 default_server;
 		server_name _;
 		return 307 https://{{ GLOBALS.url_base }}$request_uri;
 	}
-  {%- endif %}
-  {%- if GLOBALS.is_manager or role == 'fleet' %}
+
 	server {
 		listen 8443;
-    {%- if GLOBALS.is_manager %}
 		server_name {{ full_server_name }};
-    {%- else %}
-    server_name {{ GLOBALS.hostname }};
-    {%- endif %}
-		root /opt/socore/elastic-artifacts;
+		root /opt/socore/html;
 		location /artifacts/ {
-			try_files $uri =404;
-		}
-		location / {
-			return 404;
+			try_files $uri =206;
+			proxy_read_timeout    90;
+			proxy_connect_timeout 90;
+			proxy_set_header      Host $host;
+			proxy_set_header      X-Real-IP $remote_addr;
+			proxy_set_header      X-Forwarded-For $proxy_add_x_forwarded_for;
+			proxy_set_header      Proxy "";
+			proxy_set_header      X-Forwarded-Proto $scheme;
 		}
 	}
-  {%- endif %}
-  {%- if GLOBALS.is_manager %}
+
 	server {
 		listen 443 ssl default_server;
 		http2 on;
@@ -92,6 +111,21 @@ http {
 	{%- endif %}
 
 	{%- if GLOBALS.is_manager %}
+
+	server {
+		listen 7788;
+		server_name {{ full_server_name }};
+		root /nsm/rules;
+		location / {
+			allow all;
+			sendfile on;
+			sendfile_max_chunk 1m;
+			autoindex on;
+			autoindex_exact_size off;
+			autoindex_format html;
+			autoindex_localtime on;
+		}
+	}
 	{%-   if NGINXMERGED.external_suricata %}
     server {
         listen 7789 ssl;

salt/telegraf/etc/telegraf.conf

@@ -138,6 +138,10 @@
 #                            AGGREGATOR PLUGINS                               #
 ###############################################################################
 
+[[aggregators.merge]]
+  drop_original = true
+  namepass = ["system", "swap", "kafka_partition"]
+
 ###############################################################################
 #                            INPUT PLUGINS                                    #
 ###############################################################################
@@ -289,7 +293,7 @@
   password = "{{ salt['pillar.get']('elasticsearch:auth:users:so_logstash_user:pass') }}"
 {%- endif %}
 
-{% if grains.role in ['so-manager','so-managersearch','so-standalone','so-receiver'] and GLOBALS.pipeline == "KAFKA" -%}
+{% if grains.role in ['so-manager','so-managerhype','so-managersearch','so-standalone','so-receiver'] and GLOBALS.pipeline == "KAFKA" -%}
 [[inputs.jolokia2_agent]]
   name_prefix= "kafka_"
   urls = ["http://{{ NODEIP }}:8778/jolokia"]
@@ -346,6 +350,26 @@
    interval = "60s"
 {%- endif %}
 
+{#- These report state that changes slowly; collecting them every interval only stores duplicate rows #}
+{%- set SLOWSCRIPTS = [] %}
+{%- for script in ['agentstatus.sh', 'os.sh', 'features.sh', 'raid.sh', 'zeekloss.sh'] %}
+{%-   if script in TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]] %}
+{%-     do TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]].remove(script) %}
+{%-     do SLOWSCRIPTS.append(script) %}
+{%-   endif %}
+{%- endfor %}
+{%- if SLOWSCRIPTS | length > 0 %}
+[[inputs.exec]]
+   commands = [
+{%-   for script in SLOWSCRIPTS %}
+      ["/scripts/{{script}}"]{% if not loop.last %},{% endif %}
+{%-   endfor %}
+   ]
+   data_format = "influx"
+   timeout = "15s"
+   interval = "5m"
+{%- endif %}
+
 {%- if TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]] | length > 0 %}
 [[inputs.exec]]
    commands = [
@@ -358,12 +382,6 @@
   timeout = "15s"
 {%- endif %}
 
-{%- if salt['pillar.get']('healthcheck:enabled', False) %}
-[[inputs.file]]
-  files = ["/host/nsm/zeek/logs/zeek_restart.log"]
-  data_format = "influx"
-{%- endif %}
-
 [[inputs.file]]
   files = ["/etc/telegraf/node_config.json"]
   name_override = "node_config"

salt/telegraf/map.jinja

@@ -18,7 +18,7 @@
 
 {% if GLOBALS.pipeline != 'REDIS' %}
 {# When global pipeline is not REDIS remove redis.sh script. KAFKA metrics are collected via jolokia agent. Config in telegraf.conf #}
-{%   if GLOBALS.role in ['so-standalone', 'so-manager', 'so-managersearch', 'so-receiver', 'so-heavynode'] %}
+{%   if GLOBALS.role in ['so-standalone', 'so-manager', 'so-managerhype', 'so-managersearch', 'so-receiver', 'so-heavynode'] %}
 {%     do TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]].remove('redis.sh') %}
 {%   endif %}
 {% endif %}

salt/telegraf/scripts/checkfiles.sh

@@ -10,7 +10,7 @@
 # if this script isn't already running
 if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then
 
-    FILES=$(ls -1x /host/nsm/strelka/unprocessed | wc -l)
+    FILES=$(find /host/nsm/strelka/unprocessed -maxdepth 1 -type f | wc -l)
     echo "faffiles files=$FILES"
     
 fi

salt/telegraf/scripts/containers.sh

@@ -12,8 +12,9 @@ if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then
 
     CONTAINERSLOG=/var/log/somon/containers.log
     # the collector rewrites this every minute; report nothing rather than repeating a stale
-    # file as if it were current, in case a run was skipped or the collector is wedged
-    MAXAGE=150
+    # file as if it were current, in case a run was skipped or the collector is wedged.
+    # Kept under two collector cycles so a 60s telegraf interval can't emit the same snapshot twice.
+    MAXAGE=90
 
     if [ -r "$CONTAINERSLOG" ]; then
         AGE=$(( $(date +%s) - $(stat -c %Y "$CONTAINERSLOG") ))

salt/telegraf/scripts/eps.sh

@@ -11,27 +11,40 @@
 if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then
 
     PREVCOUNTFILE='/tmp/eps.txt'
-    EVENTCOUNTCURRENT="$(curl -s localhost:9600/_node/stats | jq '.events.in')"
+    STATS="$(curl -s localhost:9600/_node/stats)"
 
-    if [ ! -z "$EVENTCOUNTCURRENT" ]; then
-
-      if [ -f "$PREVCOUNTFILE" ]; then
-        EVENTCOUNTPREVIOUS=`cat $PREVCOUNTFILE`
-      else
-        echo "${EVENTCOUNTCURRENT}" > $PREVCOUNTFILE
-        exit 0
-      fi
-
-      echo "${EVENTCOUNTCURRENT}" > $PREVCOUNTFILE
-      # the division by 30 is because the agent interval is 30 seconds
-      EVENTS=$(((EVENTCOUNTCURRENT - EVENTCOUNTPREVIOUS)/30))
-      if [ "$EVENTS" -lt 0 ]; then
-        EVENTS=0
-      fi
-
-      echo "consumptioneps eps=${EVENTS%%.*}"
+    if echo "$STATS" | jq -e '.pipelines.search' > /dev/null 2>&1; then
+      STAGE=search
+      COUNT="$(echo "$STATS" | jq '.pipelines.search.events.out')"
+    elif echo "$STATS" | jq -e '.pipelines.receiver' > /dev/null 2>&1; then
+      STAGE=receiver
+      COUNT="$(echo "$STATS" | jq '.pipelines.receiver.events.in')"
+    else
+      exit 0
     fi
 
+    if ! [[ "$COUNT" =~ ^[0-9]+$ ]]; then
+      exit 0
+    fi
+
+    NOW=$(date +%s)
+    read -r PREVEPOCH PREVCOUNT 2>/dev/null < "$PREVCOUNTFILE"
+    echo "${NOW} ${COUNT}" > "$PREVCOUNTFILE"
+
+    if ! [[ "$PREVEPOCH" =~ ^[0-9]+$ && "$PREVCOUNT" =~ ^[0-9]+$ ]]; then
+      exit 0
+    fi
+
+    ELAPSED=$((NOW - PREVEPOCH))
+    DELTA=$((COUNT - PREVCOUNT))
+    if [ "$ELAPSED" -le 0 ] || [ "$DELTA" -lt 0 ]; then
+      EPS=0
+    else
+      EPS=$((DELTA / ELAPSED))
+    fi
+
+    echo "consumptioneps,stage=${STAGE} eps=${EPS}"
+
 fi
 
 exit 0

salt/telegraf/scripts/features.sh

@@ -10,8 +10,7 @@ if [[ ! "`pidof -x $(basename $0) -o %PPID`" ]]; then
     FPS_ENABLED=$(cat /var/log/sostatus/fps_enabled)
     LKS_ENABLED=$(cat /var/log/sostatus/lks_enabled)
 
-    echo "features fps=$FPS_ENABLED"
-    echo "features lks=$LKS_ENABLED"
+    echo "features fps=$FPS_ENABLED,lks=$LKS_ENABLED"
 fi
 
 exit 0