sensu: fix rabbitmq user updates

Due to performance reasons we did not always change the passwords,
only for new users.

This caused inconsistencies more often than we can tolerate.

This now fixes it properly by parallelizing the updates, limiting to
CPU_COUNT-1 and also deleting superfluous users.

Fixes PL-132945

nixos/services/sensu/configure-sensu-clients.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3.8
+
+import asyncio
+import multiprocessing
+import sys
+import textwrap
+from pathlib import Path
+
+
+class UserConfig:
+    known_users: set
+    user_db: list
+    errors: list
+
+    protected_users = set(["fc-sensu", "fc-telegraf", "sensu-server"])
+
+    def __init__(self, known_users_file):
+        self.errors = []
+        self.known_users = set()
+        for line in known_users_file.open():
+            line = line.strip()
+            if not line:
+                continue
+            if "[" not in line:
+                # ignore headers:
+                # Listing users ...
+                # user      tags
+                continue
+            user = line.split("\t")[0]
+            self.known_users.add(user)
+
+        self.user_db = []
+        self.users_to_delete = set(self.known_users)
+        for line in Path("/var/lib/rabbitmq/sensu-clients").open():
+            line = line.strip()
+            if not line:
+                continue
+            nodename, user, password = line.split(":")
+            self.user_db.append((nodename, user, password))
+            if user in self.users_to_delete:
+                self.users_to_delete.remove(user)
+
+        self.users_to_delete = self.users_to_delete - self.protected_users
+        print(f"{len(self.users_to_delete)} users to delete")
+        print(f"{len(self.user_db)} users to configure")
+
+    async def configure(self):
+        self.max_jobs = asyncio.Semaphore(
+            max([1, multiprocessing.cpu_count() - 1])
+        )
+        jobs = []
+        for user in self.users_to_delete:
+            jobs.append(self.delete_user(user))
+        for nodename, user, password in self.user_db:
+            jobs.append(self.configure_user(nodename, user, password))
+        await asyncio.gather(*jobs, return_exceptions=True)
+        return bool(self.errors)
+
+    async def delete_user(self, user):
+        async with self.max_jobs:
+            print(f"Deleting {user}")
+            await self.run(f"rabbitmqctl delete_user {user}")
+
+    async def configure_user(self, nodename, user, password):
+        async with self.max_jobs:
+            if user not in self.known_users:
+                print(f"Adding user {user} ...")
+                await self.run(f"rabbitmqctl add_user {user} {password}")
+                await self.run(
+                    f"rabbitmqctl set_permissions -p /sensu {user} "
+                    f"'((?!keepalives|results).)*' "
+                    f"'^(keepalives|results|{nodename}.*)$' "
+                    f"'((?!keepalives|results).)*'"
+                )
+            print(f"Updating user {user}")
+            await self.run(f"rabbitmqctl change_password {user} {password}")
+
+    async def run(self, cmd):
+        proc = await asyncio.create_subprocess_shell(
+            cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.STDOUT,
+        )
+        stdout, _ = await proc.communicate()
+        stdout = stdout.decode("utf-8", errors="replace")
+        if proc.returncode:
+            self.errors.append((cmd, proc.returncode, stdout))
+            print(f"`{cmd}` exited with error: {proc.returncode}")
+            print(textwrap.indent(stdout, "   > ", lambda line: True))
+            raise RuntimeError(stdout)
+
+
+if __name__ == "__main__":
+    config = UserConfig(Path(sys.argv[1]))
+    sys.exit(asyncio.run(config.configure()))

nixos/services/sensu/server.nix

@@ -103,7 +103,7 @@ in {
       description = "Prepare rabbitmq for sensu-server.";
       wantedBy = [ "multi-user.target" ];
       requires = ["rabbitmq.service" ];
-      after = ["rabbitmq.service" ];
+      after = [ "rabbitmq.service" "fc-rabbitmq-settings.service"];
       path = [ config.services.rabbitmq.package ];
       serviceConfig = {
         Type = "oneshot";
@@ -115,25 +115,8 @@ in {
       script = let
         # Permission settings required for sensu
         # see https://docs.sensu.io/sensu-core/1.7/guides/securing-rabbitmq
-        clients = (lib.concatMapStrings (
-          client:
-            let
-              inherit (client) node password;
-              name = builtins.head (lib.splitString "." node);
-              permissions = [
-                "((?!keepalives|results).)*"
-                "^(keepalives|results|${name}.*)$"
-                "((?!keepalives|results).)*"
-              ];
-            in ''
-              # Configure user and permissions for ${node}:
-              grep ^${node} $known_users > /dev/null || (
-                echo "Adding user ${node} ..."
-                rabbitmqctl add_user ${node} ${password} ;
-                rabbitmqctl change_password ${client.node} ${password} ;
-                rabbitmqctl set_permissions -p /sensu ${node} ${lib.concatMapStringsSep " " (p: "'${p}'") permissions}
-              )
-            '')
+        clients = (lib.concatMapStrings
+          (client: "${builtins.head (lib.splitString "." client.node)}:${client.node}:${client.password}\n")
           sensuClients);
       in
       ''
@@ -156,8 +139,13 @@ in {
         rabbitmqctl set_permissions -p /sensu sensu-server ".*" ".*" ".*"
 
         echo "Ensuring client users ..."
+        touch /var/lib/rabbitmq/sensu-clients
+        chmod o-r /var/lib/rabbitmq/sensu-clients
+        cat > /var/lib/rabbitmq/sensu-clients <<__EOF__
         ${clients}
+        __EOF__
 
+        ${pkgs.python38Full}/bin/python -u ${./configure-sensu-clients.py} $known_users
         echo "All done"
       '';
     };

tests/sensu.nix

@@ -150,6 +150,10 @@ in {
   in ''
     start_all()
     sensu.wait_for_unit("rabbitmq.service")
+    sensu.wait_for_unit("prepare-rabbitmq-for-sensu.service")
+
+    print(sensu.succeed("journalctl -u prepare-rabbitmq-for-sensu"))
+
     sensu.wait_until_succeeds("${amqpPortCheck}")
     sensu.wait_for_unit("sensu-server")
     sensu.wait_for_unit("sensu-api")