nixpkgs/nixos/tests/vector/syslog-quickwit.nix

import ../make-test-python.nix ({ lib, pkgs, ... }:

# Based on https://quickwit.io/docs/log-management/send-logs/using-vector

{
  name = "vector-syslog-quickwit";
  meta.maintainers = [ pkgs.lib.maintainers.happysalada ];

  nodes = {
    quickwit = { config, pkgs, ... }: {
      environment.systemPackages = [ pkgs.jq ];

      networking.firewall.allowedTCPPorts = [ 7280 ];

      services.quickwit = {
        enable = true;
        settings = {
          listen_address = "::";
        };
      };
    };

    syslog = { config, pkgs, ... }: {
      services.vector = {
        enable = true;

        settings = {
          sources = {
            generate_syslog = {
              type = "demo_logs";
              format = "syslog";
              interval = 0.5;
            };
          };

          transforms = {
            remap_syslog = {
              inputs = ["generate_syslog"];
              type = "remap";
              source = ''
                structured = parse_syslog!(.message)
                .timestamp_nanos = to_unix_timestamp!(structured.timestamp, unit: "nanoseconds")
                .body = structured
                .service_name = structured.appname
                .resource_attributes.source_type = .source_type
                .resource_attributes.host.hostname = structured.hostname
                .resource_attributes.service.name = structured.appname
                .attributes.syslog.procid = structured.procid
                .attributes.syslog.facility = structured.facility
                .attributes.syslog.version = structured.version
                .severity_text = if includes(["emerg", "err", "crit", "alert"], structured.severity) {
                  "ERROR"
                } else if structured.severity == "warning" {
                  "WARN"
                } else if structured.severity == "debug" {
                  "DEBUG"
                } else if includes(["info", "notice"], structured.severity) {
                  "INFO"
                } else {
                 structured.severity
                }
                .scope_name = structured.msgid
                del(.message)
                del(.host)
                del(.timestamp)
                del(.service)
                del(.source_type)
              '';
            };
          };

          sinks = {
            #emit_syslog = {
            #  inputs = ["remap_syslog"];
            #  type = "console";
            #  encoding.codec = "json";
            #};
            quickwit_logs = {
              type = "http";
              method = "post";
              inputs = [ "remap_syslog" ];
              encoding.codec = "json";
              framing.method = "newline_delimited";
              uri = "http://quickwit:7280/api/v1/otel-logs-v0_7/ingest";
            };
          };
        };
      };
    };
  };

  testScript =
  let
    aggregationQuery = pkgs.writeText "aggregation-query.json" ''
      {
        "query": "*",
        "max_hits": 0,
        "aggs": {
          "count_per_minute": {
            "histogram": {
                "field": "timestamp_nanos",
                "interval": 60000000
            },
            "aggs": {
              "severity_text_count": {
                "terms": {
                  "field": "severity_text"
                }
              }
            }
          }
        }
      }
    '';
  in
  ''
    quickwit.wait_for_unit("quickwit")
    quickwit.wait_for_open_port(7280)
    quickwit.wait_for_open_port(7281)

    quickwit.wait_until_succeeds(
      "journalctl -o cat -u quickwit.service | grep 'transitioned to ready state'"
    )

    syslog.wait_for_unit("vector")
    syslog.wait_until_succeeds(
      "journalctl -o cat -u vector.service | grep 'Vector has started'"
    )

    quickwit.wait_until_succeeds(
      "journalctl -o cat -u quickwit.service | grep 'publish-new-splits'"
    )

    # Wait for logs to be generated
    # Test below aggregates by the minute
    syslog.sleep(60 * 2)

    quickwit.wait_until_succeeds(
      "curl -sSf -XGET http://127.0.0.1:7280/api/v1/otel-logs-v0_7/search?query=severity_text:ERROR |"
      + " jq '.num_hits' | grep -v '0'"
    )

    quickwit.wait_until_succeeds(
      "journalctl -o cat -u quickwit.service | grep 'SearchRequest'"
    )

    quickwit.wait_until_succeeds(
      "curl -sSf -XPOST -H 'Content-Type: application/json' http://127.0.0.1:7280/api/v1/otel-logs-v0_7/search --data @${aggregationQuery} |"
      + " jq '.num_hits' | grep -v '0'"
    )

    quickwit.wait_until_succeeds(
      "journalctl -o cat -u quickwit.service | grep 'count_per_minute'"
    )
  '';
})