depot/third_party/nixpkgs/nixos/tests/vector/nginx-clickhouse.nix

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

{
  name = "vector-nginx-clickhouse";
  meta.maintainers = [ pkgs.lib.maintainers.happysalada ];

  nodes = {
    clickhouse = { config, pkgs, ... }: {
      virtualisation.memorySize = 4096;

      # Clickhouse module can't listen on a non-loopback IP.
      networking.firewall.allowedTCPPorts = [ 6000 ];
      services.clickhouse.enable = true;

      # Exercise Vector sink->source for now.
      services.vector = {
        enable = true;

        settings = {
          sources = {
            vector_source = {
              type = "vector";
              address = "[::]:6000";
            };
          };

          sinks = {
            clickhouse = {
              type = "clickhouse";
              inputs = [ "vector_source" ];
              endpoint = "http://localhost:8123";
              database = "nginxdb";
              table = "access_logs";
              skip_unknown_fields = true;
            };
          };
        };
      };
    };

    nginx = { config, pkgs, ... }: {
      services.nginx = {
        enable = true;
        virtualHosts.localhost = {};
      };

      services.vector = {
        enable = true;

        settings = {
          sources = {
            nginx_logs = {
              type = "file";
              include = [ "/var/log/nginx/access.log" ];
              read_from = "end";
            };
          };

          sinks = {
            vector_sink = {
              type = "vector";
              inputs = [ "nginx_logs" ];
              address = "clickhouse:6000";
            };
          };
        };
      };

      systemd.services.vector.serviceConfig = {
        SupplementaryGroups = [ "nginx" ];
      };
    };
  };

  testScript =
  let
    # work around quote/substitution complexity by Nix, Perl, bash and SQL.
    databaseDDL = pkgs.writeText "database.sql" "CREATE DATABASE IF NOT EXISTS nginxdb";

    tableDDL = pkgs.writeText "table.sql" ''
      CREATE TABLE IF NOT EXISTS  nginxdb.access_logs (
        message String
      )
      ENGINE = MergeTree()
      ORDER BY tuple()
    '';

    # Graciously taken from https://clickhouse.com/docs/en/integrations/vector
    tableView = pkgs.writeText "table-view.sql" ''
      CREATE MATERIALIZED VIEW nginxdb.access_logs_view
      (
        RemoteAddr String,
        Client String,
        RemoteUser String,
        TimeLocal DateTime,
        RequestMethod String,
        Request String,
        HttpVersion String,
        Status Int32,
        BytesSent Int64,
        UserAgent String
      )
      ENGINE = MergeTree()
      ORDER BY RemoteAddr
      POPULATE AS
      WITH
       splitByWhitespace(message) as split,
       splitByRegexp('\S \d+ "([^"]*)"', message) as referer
      SELECT
        split[1] AS RemoteAddr,
        split[2] AS Client,
        split[3] AS RemoteUser,
        parseDateTimeBestEffort(replaceOne(trim(LEADING '[' FROM split[4]), ':', ' ')) AS TimeLocal,
        trim(LEADING '"' FROM split[6]) AS RequestMethod,
        split[7] AS Request,
        trim(TRAILING '"' FROM split[8]) AS HttpVersion,
        split[9] AS Status,
        split[10] AS BytesSent,
        trim(BOTH '"' from referer[2]) AS UserAgent
      FROM
        (SELECT message FROM nginxdb.access_logs)
    '';

    selectQuery = pkgs.writeText "select.sql" "SELECT * from nginxdb.access_logs_view";
  in
  ''
    clickhouse.wait_for_unit("clickhouse")
    clickhouse.wait_for_open_port(8123)

    clickhouse.wait_until_succeeds(
      "journalctl -o cat -u clickhouse.service | grep 'Started ClickHouse server'"
    )

    clickhouse.wait_for_unit("vector")
    clickhouse.wait_for_open_port(6000)

    clickhouse.succeed(
      "cat ${databaseDDL} | clickhouse-client"
    )

    clickhouse.succeed(
      "cat ${tableDDL} | clickhouse-client"
    )

    clickhouse.succeed(
      "cat ${tableView} | clickhouse-client"
    )

    nginx.wait_for_unit("nginx")
    nginx.wait_for_open_port(80)
    nginx.wait_for_unit("vector")
    nginx.wait_until_succeeds(
      "journalctl -o cat -u vector.service | grep 'Starting file server'"
    )

    nginx.succeed("curl http://localhost/")
    nginx.succeed("curl http://localhost/")

    nginx.wait_for_file("/var/log/nginx/access.log")
    nginx.wait_until_succeeds(
      "journalctl -o cat -u vector.service | grep 'Found new file to watch. file=/var/log/nginx/access.log'"
    )

    clickhouse.wait_until_succeeds(
      "cat ${selectQuery} | clickhouse-client | grep 'curl'"
    )
  '';
})