Add InfluxDB provisioning script

Signed-off-by: Magic_RB <magic_rb@redalder.org>

nixos/modules/influx-provisioning.nix

@@ -0,0 +1,119 @@
+{ config, pkgs, lib, ... }:
+let
+  cfg = config.services.influxdb2.provision;
+
+  inherit (lib)
+    mkEnableOption
+    mkOption
+    types
+    mdDoc
+    flip
+    mapAttrsToList
+    getExe
+    mkIf;
+
+  taskOptions =
+    { ... }:
+    {
+      options = {
+        cron = mkOption {
+          type = with types; nullOr str;
+          default = null;
+          description = mdDoc ''
+          '';
+        };
+
+        every = mkOption {
+          type = with types; nullOr str;
+          default = null;
+          description = mdDoc ''
+          '';
+        };
+
+        fluxFile = mkOption {
+          type = types.path;
+          description = mdDoc ''
+          '';
+        };
+
+        offset = mkOption {
+          type = types.str;
+          default = "0m";
+          description = mdDoc ''
+          '';
+        };
+      };
+    };
+
+  tasksFile =
+    (pkgs.formats.json {}).generate "tasks.json"
+      (flip mapAttrsToList cfg.tasks (name: value:
+        {
+          inherit name;
+          flux_file = value.fluxFile;
+          inherit (value)
+            every
+            cron
+            offset;
+        }
+      ));
+in
+{
+  options = {
+    services.influxdb2.provision = {
+      enable = mkEnableOption "Enable InfluxDB2 provisioning";
+
+      itpPackage = mkOption {
+        type = types.package;
+        default = pkgs.itp;
+        description = mdDoc ''
+        '';
+      };
+
+      stateFile = mkOption {
+        type = types.str;
+        description = mdDoc ''
+        '';
+      };
+
+      organization = mkOption {
+        type = types.str;
+        description = mdDoc ''
+        '';
+      };
+
+      tasks = mkOption {
+        type = with types; attrsOf (submodule taskOptions);
+        default = {};
+        description = mdDoc ''
+        '';
+      };
+    };
+  };
+
+  config = mkIf cfg.enable {
+    systemd.services.influxdb2-provision = {
+      after = [ "influxdb2.service" ];
+      wants = [ "influxdb2.service" ];
+      wantedBy = [ "multi-user.target" ];
+
+      restartIfChanged = true;
+
+      script = ''
+        ${getExe cfg.itpPackage} -s ${cfg.stateFile} -f ${tasksFile} -o ${cfg.organization}
+      '';
+
+      serviceConfig = {
+        Type = "oneshot";
+        Restart = "on-failure";
+        RestartSec = 3;
+      };
+    };
+
+    assertions = flip mapAttrsToList cfg.tasks
+      (n: v: {
+        assertion = (v.cron != null && v.every == null) || (v.cron == null && v.every != null);
+        message = "Exactly one of `services.influxdb2.provision.tasks.${n}.{cron, every}` must be non `null`";
+      });
+  };
+}

nixos/systems/blowhole/influx-tasks/system-memory.flux

@@ -0,0 +1,29 @@
+arcsize =
+    from(bucket: "metrics")
+        |> range(start: -duration(v: task.every))
+        |> filter(fn: (r) => r._measurement == "zfs" and r._field == "arcstats_size")
+        |> mean()
+
+used =
+    from(bucket: "metrics")
+        |> range(start: -duration(v: task.every))
+        |> filter(fn: (r) => r._measurement == "mem" and r._field == "used")
+        |> mean()
+
+free =
+    from(bucket: "metrics")
+        |> range(start: -duration(v: task.every))
+        |> filter(fn: (r) => r._measurement == "mem" and r._field == "available")
+        |> mean()
+
+union(tables: [arcsize, used, free])
+    |> group()
+    |> map(
+        fn: (r) =>
+            ({r with used: r.used - r.arcstats_size,
+                available: r.available + r.arcstats_size,
+                _time: r._start,
+            }),
+    )
+    |> group(columns: ["_time", "host"])
+    |> to(bucket: "metrics-preprocessed")

nixos/systems/blowhole/monitoring.nix

@@ -31,6 +31,15 @@ in
           };
         };
 
+        resource."influxdb-v2_bucket"."metrics_preprocessed_bucket" = {
+          name        = "metrics-preprocessed";
+          description = "Preprocessed bucket";
+          org_id      = "\${data.influxdb-v2_organization.redalder.id}";
+          retention_rules = {
+            every_seconds = 30 * 24 * 60 * 60; # days * h/d * m/h * s/m
+          };
+        };
+
         resource."influxdb-v2_bucket"."logs_bucket" = {
           org_id      = "\${data.influxdb-v2_organization.redalder.id}";
           name        = "logs";
@@ -66,7 +75,7 @@ in
 
         resource."influxdb-v2_authorization"."grafana_authorization" = {
           org_id      = "\${data.influxdb-v2_organization.redalder.id}";
-          description = "Token for Grefana";
+          description = "Token for Grafana";
           status      = "active";
           permissions = [
             {
@@ -77,6 +86,14 @@ in
                 type   = "buckets";
               };
             }
+            {
+              action = "read";
+              resource = {
+                id     = "\${influxdb-v2_bucket.metrics_preprocessed_bucket.id}";
+                org_id = "\${data.influxdb-v2_organization.redalder.id}";
+                type   = "buckets";
+              };
+            }
             {
               action = "read";
               resource = {
@@ -201,6 +218,15 @@ in
               'systemctl try-reload-or-restart grafana' || true
           '';
         }
+        {
+          source = pkgs.writeText "itp.env.vtmpl" ''
+            {{ with secret "kv/data/homelab-1/blowhole/monitor/itp" }}
+            INFLUX_HOST={{ .Data.data.host }}
+            INFLUX_TOKEN={{ .Data.data.token }}
+            {{ end }}
+          '';
+          destination = "/run/secrets/monitor/itp.env";
+        }
       ];
     };
 
@@ -311,7 +337,38 @@ in
 
           tagpass = {
             "grok_type" = [ "nginx" "apache" ];
-            "_field" = singleton "message";
+          };
+          namepass = [ "docker_log" ];
+        }
+        {
+          parse_fields = [ "message" ];
+          merge = "override";
+          data_format = "json_v2";
+
+          json_v2 = [
+            # the TOML generator won't create the structure required by telegraf without this
+            {}
+            {
+              object = [
+                {
+                  path = "@this";
+                  timestamp_key = "time";
+                  timestamp_format = "unix";
+                  tags = [
+                    "level"
+                    "server_name"
+                    "namespace"
+                    "level"
+                    "request"
+                  ];
+                  disable_prepend_keys = true;
+                }
+              ];
+            }
+          ];
+
+          tagpass = {
+            "grok_type" = [ "synapse" ];
           };
           namepass = [ "docker_log" ];
         }
@@ -397,6 +454,7 @@ in
         hashicorp-envoy
         telegraf
         grafana
+        influx-provisioning
       ];
 
       services.hashicorp-envoy.grafana = {
@@ -507,8 +565,21 @@ in
         extraConsulArgs = [ "-ignore-envoy-compatibility" ];
       };
 
+      systemd.services."influxdb2-provision".serviceConfig.EnvironmentFile = [
+        "/run/secrets/itp.env"
+      ];
+
       services.influxdb2 = {
         enable = true;
+        provision = {
+          stateFile = "/var/lib/influxdb2/itp.state";
+          organization = "redalder";
+
+          # tasks.test = {
+          #   every = "30s";
+          #   fluxFile = ./influx-tasks/system-memory.flux;
+          # };
+        };
         settings = {
           http-bind-address = "127.0.0.1:8086";
           hardening-enabled = true;

nixos/systems/blowhole/nixpkgs.nix

@@ -10,6 +10,7 @@
       emacsclient-remote
       zfs-relmount
       ical2org
+      itp
     ])
     ++
     (with inputs'.nixng.overlays; [

overlays/itp/default.nix

@@ -0,0 +1,14 @@
+{ inputs, ... }:
+{
+  flake.overlays.itp =
+    final: prev: {
+      itp = prev.writeShellApplication {
+        name = "itp";
+        runtimeInputs = with final; [
+          influxdb2-cli
+          jq
+        ];
+        text = builtins.readFile ./itp.sh;
+      };
+    };
+}

overlays/itp/itp.sh

@@ -0,0 +1,201 @@
+# -*- sh-basic-offset: 2 indent-tabs-mode: nil -*-
+
+declare -a _positional_args
+declare _dry_run=no
+declare _state_file
+declare _task_file
+declare _org
+
+while [[ $# -gt 0 ]]; do
+  case $1 in
+    -n|--dry-run)
+      _dry_run="yes"
+      shift # past argument
+      ;;
+    -s|--state)
+      if [[ -z "${2:-}" ]] ; then
+        echo "$1 takes one argument"
+        exit 2
+      fi
+
+      _state_file="$2"
+      shift # past argument
+      shift # past value
+      ;;
+    -f|--file)
+      if [[ -z "${2:-}" ]] ; then
+        echo "$1 takes one argument"
+        exit 2
+      fi
+
+      _task_file="$2"
+      shift # past argument
+      shift # past value
+      ;;
+    -o|--org)
+      if [[ -z "${2:-}" ]] ; then
+        echo "$1 takes one argument"
+        exit 2
+      fi
+
+      _org="$2"
+      shift # past argument
+      shift # past value
+      ;;
+    --*|-*)
+      echo "Unknown option $1"
+      exit 1
+      ;;
+    *)
+      _positional_args+=("$1") # save positional arg
+      shift # past argument
+      ;;
+  esac
+done
+
+set -- "${_positional_args[@]}"
+
+if [[ -z "${_state_file:-}" ]] ; then
+  echo "-s|--state must be specified"
+  exit 2
+fi
+
+if [[ -z "${_task_file:-}" ]] ; then
+  echo "-f|--file must be specified"
+  exit 2
+fi
+
+if ! [[ -f "$_task_file" ]] ; then
+  echo "$_task_file must exist and be writable"
+  exit 2
+fi
+
+if ! [[ -w "$_state_file" ]] ; then
+  touch "$_state_file" || ( echo "Couldn't create state file" ; exit 2 )
+  echo "[]" > "$_state_file"
+fi
+
+if [[ -z "${_org:-}" ]] ; then
+  echo "-o|--org must be specified"
+  exit 2
+fi
+
+if [[ -z "${INFLUX_HOST:-}" ]] ; then
+  echo "INFLUX_HOST must be set"
+  exit 2
+fi
+
+if [[ -z "${INFLUX_TOKEN:-}" ]] ; then
+  echo "INFLUX_TOKEN must be set"
+  exit 2
+fi
+
+
+if ! influx task list --org "$_org" >/dev/null 2>&1 ; then
+  echo "Failed to establish connection to InfuxDB"
+  exit 2
+fi
+
+declare -a _tasks_to_remove
+declare -a _tasks_to_add
+declare -a _tasks_to_update
+
+mapfile -t _tasks_to_remove < <(comm -23 <(jq -r '.[].name' < "$_state_file" | sort) <(jq -r '.[].name' < "$_task_file" | sort))
+mapfile -t _tasks_to_add < <(comm -13 <(jq -r '.[].name' < "$_state_file" | sort) <(jq -r '.[].name' < "$_task_file" | sort))
+mapfile -t _tasks_to_update < <(comm -12 <(jq -r '.[].name' < "$_state_file" | sort) <(jq -r '.[].name' < "$_task_file" | sort))
+
+function task_key_by_name() {
+  _file="$1"
+  _task="$2"
+  _key="$3"
+
+  # shellcheck disable=SC2086 # Intended splitting of JQ_OPTS
+  jq ${JQ_OPTS:-} '.[] | select(.name == "'"$_task"'") | .'"$_key" < "$_file"
+}
+
+function prepend_if_not_null() {
+  _prepend=$1
+  _input=$(cat)
+
+  if [[ "$_input" != "null" ]] ; then
+    echo "$_prepend $_input"
+  fi
+}
+
+function generate_flux_file() {
+  _task="$1"
+
+  cat <<EOF
+option task = {
+  name: $(task_key_by_name "$_task_file" "$task" "name")
+  $(task_key_by_name "$_task_file" "$task" "cron" | prepend_if_not_null ", cron: ")
+  $(JQ_OPTS="-r" task_key_by_name "$_task_file" "$task" "every" | prepend_if_not_null ", every: ")
+}
+
+$(cat "$(JQ_OPTS="-r" task_key_by_name "$_task_file" "$task" "flux_file")")
+
+EOF
+}
+
+function task_delete() {
+  _id="$1"
+
+  influx task delete --id "$_id"
+}
+
+function task_create() {
+  _flux_file="$1"
+
+  influx task create --org "$_org" --file <(cat "$_flux_file")
+}
+
+function task_update() {
+  _task="$1"
+  _flux_file="$2"
+
+  influx task update --id "$(JQ_OPTS="-r" task_key_by_name "$_state_file" "$_task" "id")" --file "$_flux_file"
+}
+
+for task in "${_tasks_to_remove[@]}" ; do
+  if [[ "$_dry_run" == "no" ]] ; then
+    echo "Removing $task"
+
+    task_delete "$(JQ_OPTS="-r" task_key_by_name "$_state_file" "$task" "id")"
+  else
+    echo "Would remove $task"
+  fi
+done
+
+for task in "${_tasks_to_add[@]}" ; do
+  if [[ "$_dry_run" == "no" ]] ; then
+    echo "Adding $task"
+    _flux_file=$(generate_flux_file "$task")
+
+    task_create <(cat <<<"$_flux_file")
+  else
+    echo "Would add $task"
+  fi
+done
+
+for task in "${_tasks_to_update[@]}" ; do
+  if [[ "$_dry_run" == "no" ]] ; then
+    echo "Updating $task"
+    _flux_file=$(generate_flux_file "$task")
+
+    _output="$(task_update "$task" <(cat <<<"$_flux_file") 2>&1)"
+
+    if [[ "$?" == "1" ]] && [[ "$_output" == *"404"* ]] ; then
+      echo "Failed to update, creating: $_output"
+      task_create <(cat <<<"$_flux_file")
+    fi
+  else
+    echo "Would update $task"
+  fi
+done
+
+_current_state=$(influx task list --org "$_org" --json)
+
+if [[ "$_dry_run" == "no" ]] ; then
+  echo "Saving new state"
+  cat "$_task_file" <(cat <<<"$_current_state") | jq -s '[.[0] as $new | .[1] as $old | $old[] | select(.name | IN($new[].name))]' > "$_state_file"
+fi

terranix/blowhole.nix

@@ -86,6 +86,10 @@ in
       path "${vaultKvMount}/data/homelab-1/blowhole/monitor/grafana" {
         capabilities = ["read"]
       }
+
+      path "${vaultKvMount}/data/homelab-1/blowhole/monitor/itp" {
+        capabilities = ["read"]
+      }
     '';
   };