hyperhive/nix/container-modules/swarm-journal-forwarder.nix

# A swarm service container's own journal forwarder: an otel collector that
# reads this container's journal and pushes it to the swarm collector by its
# gateway name, authenticated as the service's own OIDC client.
#
# Imported inside the container's own `config`. The host module that declares
# `containers.<name>` sets the options below: the container's evaluation is
# nested inside the host's, and reading the host side back from in here
# recurses.
#
# The client secret is fetched inside the container, under the container's own
# store identity, by `swarm-journal-forwarder-oidc`. That needs the container to
# reach the store, authelia and the swarm collector by name: the container needs
# a network and a resolver.
#
# ⚠️ The host must NOT start this container with `--link-journal=host`. That
# bind-mounts a host directory over the guest's /var/log/journal which journald
# does not write into, and the receiver below reads nothing while the collector
# stays healthy. nixpkgs' default `--link-journal=try-guest` keeps the journal
# in the guest, where this reads it.
{
  config,
  lib,
  pkgs,
  ...
}:
let
  cfg = config.services.hyperhive.swarmJournalForwarder;

  atomicWriteSecret = import ../host-modules/lib/atomic-write-secret.nix { };
  refreshConsumer = import ../host-modules/lib/refresh-consumer.nix { };
  storeRetry = import ../host-modules/lib/store-retry.nix { };
  storeRefreshTimer = import ../host-modules/lib/store-refresh-timer.nix { };

  fetchUnit = "swarm-journal-forwarder-oidc";
  # On the container's /run (tmpfs), so the secret never reaches persistent
  # disk. The file is not created ahead of the fetch: `LoadCredential` refuses a
  # missing source, so the collector stays down until the store answers and
  # then resumes from its saved journal cursor.
  secretDir = "/run/${fetchUnit}";
  secretFile = "${secretDir}/client-secret";

  # The collector runs under `DynamicUser` and opens `client_secret_file`
  # itself. `LoadCredential` reads the file as root and re-exposes it under
  # `${CREDENTIALS_DIRECTORY}/<id>`, whichever uid the unit got.
  credentialId = "oidc-client-secret";
  authName = "oauth2client/swarm";
in
{
  options.services.hyperhive.swarmJournalForwarder = {
    enable = lib.mkEnableOption "this container's own journal forwarder to the swarm collector";

    serviceName = lib.mkOption {
      type = lib.types.str;
      example = "nats";
      description = ''
        `service.name` stamped on every record. The display identity a
        dashboard selects on, matching the service's scrape job name. No
        `hive` label: the swarm collector stamps `swarm` from the receiver
        that accepted the push.
      '';
    };

    endpoint = lib.mkOption {
      type = lib.types.str;
      example = "https://otel.swarm.example/swarm-nats";
      description = ''
        Base URL of this service's own receiver on the swarm collector:
        `https://''${swarm.otel.domain}/<producer>`, a route
        {option}`services.hyperhive.swarm.otel.serviceProducers` declares.
        The exporter appends `/v1/logs`.
      '';
    };

    clientId = lib.mkOption {
      type = lib.types.str;
      description = ''
        OAuth2 client the forwarder mints its token as, and the audience it
        asks for. Must equal the `clientId` of this service's
        {option}`services.hyperhive.swarm.otel.serviceProducers` entry, which
        is the one audience that receiver accepts.
      '';
    };

    tokenUrl = lib.mkOption {
      type = lib.types.str;
      example = "https://auth.swarm.example/api/oidc/token";
      description = "Authelia's token endpoint.";
    };

    telemetryPort = lib.mkOption {
      type = lib.types.port;
      description = ''
        Loopback port for the collector's own metrics (refused and dropped
        records, exporter failures). Required because the container may share
        the host netns, where the binary's default 8888 is the hive tier's.
      '';
    };

    store = {
      package = lib.mkOption {
        type = lib.types.package;
        description = "The `bao` CLI the fetch unit logs in and reads with.";
      };
      addr = lib.mkOption {
        type = lib.types.str;
        description = "The swarm secret store's address, by name.";
      };
      secretPath = lib.mkOption {
        type = lib.types.str;
        example = "secret/swarm/services/swarm-nats/oidc/client";
        description = ''
          KV path of the client secret. The container's store identity must
          hold a read grant on it.
        '';
      };
      clientCertFile = lib.mkOption {
        type = lib.types.str;
        description = ''
          The container's store identity, as a path inside the container. A
          path, never a value.
        '';
      };
      clientKeyFile = lib.mkOption {
        type = lib.types.str;
        description = "Private key for `store.clientCertFile`. A path, never a value.";
      };
      caFile = lib.mkOption {
        type = lib.types.nullOr lib.types.str;
        default = null;
        description = "CA the store's server certificate chains to; null means the system trust store.";
      };
    };
  };

  config = lib.mkIf cfg.enable {
    assertions = [
      {
        # With a volatile journal the receiver finds an empty directory and the
        # collector starts clean and ships nothing.
        assertion =
          !(lib.elem config.services.journald.storage [
            "volatile"
            "none"
          ]);
        message = ''
          This container forwards its own journal to the swarm collector, but
          services.journald.storage is "${config.services.journald.storage}".
          The forwarder reads /var/log/journal, which journald only writes when
          it stores persistently.
        '';
      }
    ];

    services.opentelemetry-collector = {
      enable = true;
      # `journald` is a contrib receiver.
      package = pkgs.opentelemetry-collector-contrib;
      # A parser, not a wiring check: it accepts an extension that is never
      # started.
      validateConfigFile = true;
      settings = {
        # The receiver's read cursor, so a restart resumes rather than skipping
        # what was written while it was down. `create_directory`, because
        # `otelcol validate` runs in the build sandbox before systemd's
        # `StateDirectory=` exists.
        extensions.file_storage = {
          directory = "/var/lib/opentelemetry-collector";
          create_directory = true;
        };

        receivers.journald = {
          # ⚠️ Stated: the receiver's default is the runtime journal under
          # /run/log/journal, which is empty when journald stores persistently.
          directory = "/var/log/journal";
          # Without `--merge`, `journalctl --follow` over a journal with no
          # entry for the current boot crash-loops the receiver.
          merge = true;
          storage = "file_storage";
        };

        processors.resource.attributes = [
          {
            key = "service.name";
            value = cfg.serviceName;
            action = "upsert";
          }
        ];

        exporters.otlphttp = {
          endpoint = cfg.endpoint;
          # The receiver accepts an OIDC token and nothing else.
          auth.authenticator = authName;
        };

        extensions.${authName} = {
          client_id = cfg.clientId;
          # A path systemd resolves at runtime, so neither the secret nor the
          # credentials directory is rendered into the nix store.
          client_secret_file = "\${env:CREDENTIALS_DIRECTORY}/${credentialId}";
          token_url = cfg.tokenUrl;
          # ⚠️ Requested, not merely registered: a token minted without asking
          # carries `aud: []` and the receiver refuses it.
          endpoint_params.audience = cfg.clientId;
        };

        service.pipelines.logs = {
          receivers = [ "journald" ];
          processors = [ "resource" ];
          exporters = [ "otlphttp" ];
        };

        # ⚠️ `readers`, not `metrics.address`: this collector version rejects
        # `address` at startup.
        service.telemetry.metrics.readers = [
          {
            pull.exporter.prometheus = {
              host = "127.0.0.1";
              port = cfg.telemetryPort;
            };
          }
        ];

        # An extension not listed here is inert: the receiver's storage would
        # name a component that never starts, and the exporter would push
        # unauthenticated.
        service.extensions = [
          "file_storage"
          authName
        ];
      };
    };

    # `Restart` is nixpkgs' `"always"`; a second definition conflicts. Only
    # `RestartSec` is ours: the 100ms default burns the start-rate limit while
    # the secret is still on its way, leaving the unit failed instead of
    # retrying.
    systemd.services.opentelemetry-collector.serviceConfig = {
      LoadCredential = [ "${credentialId}:${secretFile}" ];
      RestartSec = 15;
    };

    # Reads the client secret with the container's own store identity. A store
    # or authelia that is not up yet is a failed run that retries
    # (../host-modules/lib/store-retry.nix), and a collector that restarts until the file lands.
    systemd.services.${fetchUnit} = {
      description = "fetch this container's journal forwarder OIDC client secret from the swarm secret store";
      wantedBy = [ "multi-user.target" ];
      before = [ "opentelemetry-collector.service" ];
      path = [
        cfg.store.package
        pkgs.coreutils
        pkgs.systemd
      ];
      inherit (storeRetry) startLimitBurst startLimitIntervalSec;
      serviceConfig = storeRetry.serviceConfig // {
        # No `RemainAfterExit`: the timer below starts this again, and starting
        # an active unit is a no-op.
        Type = "oneshot";
        SyslogIdentifier = fetchUnit;
        TimeoutStartSec = 30;
      };
      environment = {
        BAO_ADDR = cfg.store.addr;
        BAO_CLIENT_CERT = cfg.store.clientCertFile;
        BAO_CLIENT_KEY = cfg.store.clientKeyFile;
      }
      // lib.optionalAttrs (cfg.store.caFile != null) {
        BAO_CACERT = cfg.store.caFile;
      };
      script = ''
        set -euo pipefail

        ${atomicWriteSecret}
        ${refreshConsumer}

        err="$(mktemp)"
        trap 'rm -f "$err"' EXIT

        # `-token-only` answers on stdout and skips the token helper, which
        # needs a `sh` this unit's `path` does not carry.
        if ! BAO_TOKEN="$(bao login -method=cert -token-only 2>"$err")"; then
          echo "could not log in to the swarm secret store with this container's certificate." >&2
          cat "$err" >&2
          exit 1
        fi
        export BAO_TOKEN

        if ! secret="$(bao kv get -field=value ${lib.escapeShellArg cfg.store.secretPath} 2>"$err")"; then
          echo "the store did not return ${cfg.store.secretPath}: the forwarder has no client secret yet." >&2
          cat "$err" >&2
          exit 1
        fi

        if [ -z "$secret" ]; then
          echo "the store returned an empty ${cfg.store.secretPath}." >&2
          exit 1
        fi

        # `printf` inside `atomic_write_secret` is bash's own, so the value
        # never becomes an argument in /proc.
        install -d -m 0700 ${lib.escapeShellArg secretDir}
        if secret_differs ${lib.escapeShellArg secretFile} "$secret"; then
          atomic_write_secret 0400 root:root ${lib.escapeShellArg secretFile} "$secret"
        fi

        # `LoadCredential` copies the file at start only: a rotated secret
        # reaches the collector by restarting it.
        refresh_unit opentelemetry-collector.service ${lib.escapeShellArg secretFile}
      '';
    };

    systemd.timers.${fetchUnit} =
      storeRefreshTimer "re-read this container's journal forwarder OIDC client secret from the swarm secret store";
  };
}