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

Telegraf: fix EPS double counting and duplicate metric rows

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