Added test mode for influx push daemon.

* required for regression testing * added new confidence value (match by custom rule) * updated / tweaked grafana exported dashboard Signed-off-by: Toni Uhlig <matzeton@googlemail.com>
2026-01-27 10:19:45 +00:00 · 2024-02-08 00:58:40 +01:00
parent ea968180a2
commit 8949ba39e6
4 changed files with 3545 additions and 3401 deletions
--- a/README.md
+++ b/README.md
@@ -312,7 +312,7 @@ Format: `subopt` (unit, comment): description

 # test

-The recommended way to run integration / diff tests:
+The recommended way to run regression / diff tests:

 ```shell
 mkdir build
--- a/examples/c-influxd/c-influxd.c
+++ b/examples/c-influxd/c-influxd.c
@@ -21,6 +21,7 @@ static char * pidfile = NULL;
 static char * serv_optarg = NULL;
 static char * user = NULL;
 static char * group = NULL;
+static int test_mode = 0;
 static char * influxdb_interval = NULL;
 static nDPIsrvd_ull influxdb_interval_ull = 0uL;
 static char * influxdb_url = NULL;
@@ -179,6 +180,7 @@ static struct
        uint64_t flow_confidence_nbpf;
        uint64_t flow_confidence_by_ip;
        uint64_t flow_confidence_dpi_aggressive;
+        uint64_t flow_confidence_custom_rule;
        uint64_t flow_confidence_unknown;

        uint64_t flow_severity_low;
@@ -337,6 +339,7 @@ static struct global_map const confidence_map[] = {
    {"nBPF", INFLUXD_STATS_GAUGE_PTR(flow_confidence_nbpf)},
    {"Match by IP", INFLUXD_STATS_GAUGE_PTR(flow_confidence_by_ip)},
    {"DPI (aggressive)", INFLUXD_STATS_GAUGE_PTR(flow_confidence_dpi_aggressive)},
+    {"Match by custom rule", INFLUXD_STATS_GAUGE_PTR(flow_confidence_custom_rule)},
    {NULL, INFLUXD_STATS_GAUGE_PTR(flow_confidence_unknown)}};

 static struct global_map const severity_map[] = {{"Low", INFLUXD_STATS_GAUGE_PTR(flow_severity_low)},
@@ -509,7 +512,7 @@ static int serialize_influx_line(char * buf, size_t siz)
    bytes = snprintf(buf,
                     siz,
                     "%s " INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT()
-                         INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT_END(),
+                         INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT() INFLUXDB_FORMAT_END(),
                     "confidence",
                     INFLUXDB_VALUE_GAUGE(flow_confidence_by_port),
                     INFLUXDB_VALUE_GAUGE(flow_confidence_dpi_partial),
@@ -519,6 +522,7 @@ static int serialize_influx_line(char * buf, size_t siz)
                     INFLUXDB_VALUE_GAUGE(flow_confidence_nbpf),
                     INFLUXDB_VALUE_GAUGE(flow_confidence_by_ip),
                     INFLUXDB_VALUE_GAUGE(flow_confidence_dpi_aggressive),
+                     INFLUXDB_VALUE_GAUGE(flow_confidence_custom_rule),
                     INFLUXDB_VALUE_GAUGE(flow_confidence_unknown));
    CHECK_SNPRINTF_RET(bytes);

@@ -644,6 +648,7 @@ failure:
    INFLUXD_STATS_GAUGE_SUB(flow_confidence_nbpf);
    INFLUXD_STATS_GAUGE_SUB(flow_confidence_by_ip);
    INFLUXD_STATS_GAUGE_SUB(flow_confidence_dpi_aggressive);
+    INFLUXD_STATS_GAUGE_SUB(flow_confidence_custom_rule);
    INFLUXD_STATS_GAUGE_SUB(flow_confidence_unknown);

    INFLUXD_STATS_GAUGE_SUB(flow_severity_low);
@@ -1409,7 +1414,21 @@ static int mainloop(int epollfd, struct nDPIsrvd_socket * const sock)
                    return 1;
                }

-                start_influxdb_thread();
+                if (test_mode == 0)
+                {
+                    start_influxdb_thread();
+                }
+                else
+                {
+                    char stdout_buffer[BUFSIZ];
+
+                    if (serialize_influx_line(stdout_buffer, sizeof(stdout_buffer)) != 0)
+                    {
+                        logger(1, "%s", "Could not serialize influx buffer");
+                        return 1;
+                    }
+                    printf("%s", stdout_buffer);
+                }
            }
            else if (events[i].data.fd == sock->fd)
            {
@@ -1449,12 +1468,13 @@ static int parse_options(int argc, char ** argv, struct nDPIsrvd_socket * const
        "\t-u\tChange user.\n"
        "\t-g\tChange group.\n"
        "\t-i\tInterval between pushing statistics to an influxdb endpoint.\n"
+        "\t-t\tTest mode: Ignores `-U' / `-T' and prints stats to stdout.\n"
        "\t-U\tInfluxDB URL.\n"
        "\t  \tExample: http://127.0.0.1:8086/write?db=ndpi-daemon\n"
        "\t-T\tInfluxDB access token.\n"
        "\t  \tNot recommended, use environment variable INFLUXDB_AUTH_TOKEN instead.\n";

-    while ((opt = getopt(argc, argv, "hcdp:s:u:g:i:U:T:")) != -1)
+    while ((opt = getopt(argc, argv, "hcdp:s:u:g:i:tU:T:")) != -1)
    {
        switch (opt)
        {
@@ -1484,6 +1504,9 @@ static int parse_options(int argc, char ** argv, struct nDPIsrvd_socket * const
                free(influxdb_interval);
                influxdb_interval = strdup(optarg);
                break;
+            case 't':
+                test_mode = 1;
+                break;
            case 'U':
                free(influxdb_url);
                influxdb_url = strdup(optarg);
@@ -1498,6 +1521,15 @@ static int parse_options(int argc, char ** argv, struct nDPIsrvd_socket * const
        }
    }

+    if (test_mode != 0)
+    {
+        logger_early(1, "%s", "Test mode enabled: ignoring `-U' / `-T' command line parameters");
+        free(influxdb_url);
+        free(influxdb_token);
+        influxdb_url = NULL;
+        influxdb_token = NULL;
+    }
+
    if (serv_optarg == NULL)
    {
        serv_optarg = strdup(DISTRIBUTOR_UNIX_SOCKET);
@@ -1514,20 +1546,23 @@ static int parse_options(int argc, char ** argv, struct nDPIsrvd_socket * const
        return 1;
    }

-    if (influxdb_url == NULL)
+    if (test_mode == 0)
    {
-        logger_early(1, "%s", "Missing InfluxDB URL.");
-        return 1;
-    }
+        if (influxdb_url == NULL)
+        {
+            logger_early(1, "%s", "Missing InfluxDB URL.");
+            return 1;
+        }

-    if (influxdb_token == NULL && getenv("INFLUXDB_AUTH_TOKEN") != NULL)
-    {
-        influxdb_token = strdup(getenv("INFLUXDB_AUTH_TOKEN"));
-    }
-    if (influxdb_token == NULL)
-    {
-        logger_early(1, "%s", "Missing InfluxDB authentication token.");
-        return 1;
+        if (influxdb_token == NULL && getenv("INFLUXDB_AUTH_TOKEN") != NULL)
+        {
+            influxdb_token = strdup(getenv("INFLUXDB_AUTH_TOKEN"));
+        }
+        if (influxdb_token == NULL)
+        {
+            logger_early(1, "%s", "Missing InfluxDB authentication token.");
+            return 1;
+        }
    }

    if (nDPIsrvd_setup_address(&sock->address, serv_optarg) != 0)
@@ -1656,13 +1691,30 @@ int main(int argc, char ** argv)
        }
    }

-    curl_global_init(CURL_GLOBAL_ALL);
+    if (test_mode == 0)
+    {
+        curl_global_init(CURL_GLOBAL_ALL);
+    }

    logger_early(0, "%s", "Initialization succeeded.");
    retval = mainloop(epollfd, sock);
    logger_early(0, "%s", "Bye.");

-    curl_global_cleanup();
+    if (test_mode == 0)
+    {
+        curl_global_cleanup();
+    }
+    else
+    {
+        char stdout_buffer[BUFSIZ];
+
+        if (serialize_influx_line(stdout_buffer, sizeof(stdout_buffer)) != 0)
+        {
+            logger(1, "%s", "Could not serialize influx buffer");
+            return 1;
+        }
+        printf("%s", stdout_buffer);
+    }
 failure:
    nDPIsrvd_socket_free(&sock);
    close(influxd_timerfd);
--- a/examples/c-influxd/grafana-dashboard-simple.json
+++ b/examples/c-influxd/grafana-dashboard-simple.json
--- a/test/run_tests.sh
+++ b/test/run_tests.sh
@@ -14,6 +14,7 @@ FLOW_INFO="$(realpath "${5:-"${MYDIR}/../examples/py-flow-info/flow-info.py"}")"
 NDPISRVD_ANALYSED="$(realpath "${6:-"$(dirname ${nDPId_test_EXEC})/nDPIsrvd-analysed"}")"
 NDPISRVD_CAPTURED="$(realpath "${6:-"$(dirname ${nDPId_test_EXEC})/nDPIsrvd-captured"}")"
 NDPISRVD_COLLECTD="$(realpath "${6:-"$(dirname ${nDPId_test_EXEC})/nDPIsrvd-collectd"}")"
+NDPISRVD_INFLUXD="$(realpath "${6:-"$(dirname ${nDPId_test_EXEC})/nDPIsrvd-influxd"}")"
 IS_GIT=$(test -d "${MYDIR}/../.git" -o -f "${MYDIR}/../.git" && printf '1' || printf '0')

 function usage()
@@ -29,6 +30,7 @@ usage: ${0} [path-to-nDPI-source-root] \\
    path-to-nDPIsrvd-analysed defaults to       ${NDPISRVD_ANALYSED}
    path-to-nDPIsrvd-captured defaults to       ${NDPISRVD_CAPTURED}
    path-to-nDPIsrvd-collectd defaults to       ${NDPISRVD_COLLECTD}
+    path-to-nDPIsrvd-influxd defaults to        ${NDPISRVD_INFLUXD}
 EOF
 return 0
 }
@@ -483,6 +485,79 @@ fi

 cat <<EOF

+-----------------------------
+-- Influxd Statistics DIFF --
+-----------------------------
+
+EOF
+
+if [ -x "${NDPISRVD_INFLUXD}" ]; then
+    cd "${MYDIR}"
+    for out_file in results/*/*.out; do
+        if [ ! -r "${out_file}" ]; then
+            printf '%s: %s\n' "${0}" "${out_file} does not exist!"
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+            continue
+        fi
+        out_name="$(basename ${out_file})"
+        pcap_cfg="$(basename $(dirname ${out_file%.out}))"
+        stdout_file="/tmp/nDPId-test-stdout/${pcap_cfg}_${out_name}.influxd.new"
+        stderr_file="/tmp/nDPId-test-stderr/${out_name}"
+        result_file="${MYDIR}/results/influxd/${pcap_cfg}/${out_name}"
+        mkdir -p "$(dirname ${result_file})"
+        printf "%-${LINE_SPACES}s\t" "${out_name}"
+        cat "${out_file}" | grep -vE '^~~.*$' | ${NETCAT_EXEC} &
+        nc_pid=$!
+        while ! ss -x -t -n -l | grep -q "${NETCAT_SOCK}"; do sleep 0.1; printf '%s\n' "Waiting until socket ${NETCAT_SOCK} is available.." >>"${stderr_file}"; done
+        ${NDPISRVD_INFLUXD} -t -i 10 -c -s "${NETCAT_SOCK}" 2>>"${stderr_file}" 1>"${stdout_file}"
+        kill -SIGTERM ${nc_pid} 2>/dev/null
+        wait ${nc_pid} 2>/dev/null
+        while ss -x -t -n -l | grep -q "${NETCAT_SOCK}"; do sleep 0.1; printf '%s\n' "Waiting until socket ${NETCAT_SOCK} is not available anymore.." >>"${stderr_file}"; done
+
+        unknown_count="$(cat "${stdout_file}" | tr ' ' '\n' | tr ',' '\n' | grep -E '^flow.*_unknown' | wc -l || printf '%s' '0')"
+        if [ "${unknown_count}" -ne 5 ]; then
+            printf '%s: Unknown count: %s\n' '[INTERNAL]' "${unknown_count}"
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+        elif cat "${stdout_file}" | tr ' ' '\n' | tr ',' '\n' | grep -E '^flow.*_unknown' | grep -qvE '=0'; then
+            printf '%s\n' '[INTERNAL]'
+            cat "${stdout_file}" | tr ' ' '\n' | tr ',' '\n' | grep -E '^flow.*_unknown' | grep -vE '=0' || true
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+        elif [ ! -r "${result_file}" ]; then
+            printf '%s\n' '[NEW]'
+            test ${IS_GIT} -eq 1 && \
+                mv "${stdout_file}" "${result_file}"
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+        elif diff -u0 "${result_file}" "${stdout_file}" >/dev/null; then
+            printf '%s\n' '[OK]'
+            rm -f "${stdout_file}"
+        else
+            printf '%s\n' '[DIFF]'
+            diff -u0 "${result_file}" "${stdout_file}"
+            test ${IS_GIT} -eq 1 && \
+                mv "${stdout_file}" "${result_file}"
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+        fi
+    done
+
+    for out_file in ${MYDIR}/results/influxd/*/*.out; do
+        if [ ! -r "${out_file}" ]; then
+            printf '%s: %s\n' "${0}" "${out_file} does not exist!"
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+            continue
+        fi
+        result_file="$(basename ${out_file})"
+        pcap_cfg="$(basename $(dirname ${out_file%.out}))"
+        if [ ! -r "${MYDIR}/results/${pcap_cfg}/${result_file}" ]; then
+            printf "%-${LINE_SPACES}s\t%s\n" "${result_file}" "[MISSING][config: ${pcap_cfg}]"
+            TESTS_FAILED=$((TESTS_FAILED + 1))
+        fi
+    done
+else
+    printf '%s\n' "Not found or not executable: ${NDPISRVD_INFLUXD}"
+fi
+
+cat <<EOF
+
 --------------------------------
 -- SCHEMA/SEMANTIC Validation --
 --------------------------------