Add log_filter: pluggable per-unit output filter subprocess

Each unit can specify log_filter: <path> pointing to any executable
that reads stdin and writes filtered output to stdout. The daemon
spawns it once at startup (and on SIGHUP), pipes raw log data through
it before ring-buffering and broadcasting to attached clients.

filter.c handles spawn (fork/exec with pipes), per-chunk apply
(write + poll with 250ms timeout, pass-through on silence/timeout),
and stop (SIGTERM + fd cleanup). Filters are stopped and restarted
cleanly on SIGHUP alongside log_tail.

Bundled filters in filters/:
  source.py    — TF2, GMod: strips server_cvar/stuck/path_goal spam,
                 strips the L MM/DD/YYYY - HH:MM:SS: prefix
  minecraft.py — vanilla/Paper/Spigot: strips keepAlive, autosave,
                 internal class logs, strips [HH:MM:SS] [thread] prefix
  terraria.py  — vanilla/tModLoader: strips blank lines and mod
                 loading noise during startup

Any executable reading stdin/writing stdout works as a custom filter.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

1 files changed, 122 insertions, 0 deletions

diff --git a/src/filter.c b/src/filter.c
new file mode 100644
index 0000000..7bc2a68
--- /dev/null
+++ b/src/filter.c
@@ -0,0 +1,122 @@
+#include "filter.h"
+#include "log.h"
+
+#include <assert.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <poll.h>
+#include <signal.h>
+#include <string.h>
+#include <unistd.h>
+
+#define FILTER_TIMEOUT_MS 250
+
+/* ── filter_start ────────────────────────────────────────────────────────── */
+
+void filter_start(Unit *u) {
+    assert(u != NULL);
+    assert(u->filter_in_fd == -1);
+
+    if (!u->log_filter[0]) return;
+
+    int to_filter[2], from_filter[2];
+    if (pipe(to_filter) < 0) {
+        log_warn("filter: pipe() failed for %s: %s", u->name, strerror(errno));
+        return;
+    }
+    if (pipe(from_filter) < 0) {
+        log_warn("filter: pipe() failed for %s: %s", u->name, strerror(errno));
+        close(to_filter[0]); close(to_filter[1]);
+        return;
+    }
+
+    pid_t pid = fork();
+    if (pid < 0) {
+        log_warn("filter: fork() failed for %s: %s", u->name, strerror(errno));
+        close(to_filter[0]);  close(to_filter[1]);
+        close(from_filter[0]); close(from_filter[1]);
+        return;
+    }
+
+    if (pid == 0) {
+        /* Child: wire pipes and exec the filter */
+        if (dup2(to_filter[0],   STDIN_FILENO)  < 0) _exit(1);
+        if (dup2(from_filter[1], STDOUT_FILENO) < 0) _exit(1);
+        close(to_filter[0]);  close(to_filter[1]);
+        close(from_filter[0]); close(from_filter[1]);
+        execl(u->log_filter, u->log_filter, (char *)NULL);
+        _exit(1);
+    }
+
+    /* Parent: keep write end of to_filter, read end of from_filter */
+    close(to_filter[0]);
+    close(from_filter[1]);
+    if (fcntl(from_filter[0], F_SETFL, O_NONBLOCK) < 0)
+        log_warn("filter: fcntl O_NONBLOCK failed for %s", u->name);
+
+    u->filter_pid    = pid;
+    u->filter_in_fd  = to_filter[1];
+    u->filter_out_fd = from_filter[0];
+    log_info("filter: started '%s' (pid %d) for unit %s",
+             u->log_filter, (int)pid, u->name);
+}
+
+/* ── filter_apply ────────────────────────────────────────────────────────── */
+
+void filter_apply(Unit *u, char *buf, int bufsize, int *len) {
+    assert(u != NULL);
+    assert(buf != NULL);
+    assert(len != NULL);
+
+    if (u->filter_in_fd < 0) return;
+
+    ssize_t w = write(u->filter_in_fd, buf, (size_t)*len);
+    if (w < 0) {
+        log_warn("filter: write failed for %s: %s — stopping filter",
+                 u->name, strerror(errno));
+        filter_stop(u);
+        return;
+    }
+
+    struct pollfd pfd = { u->filter_out_fd, POLLIN, 0 };
+    int r = poll(&pfd, 1, FILTER_TIMEOUT_MS);
+    if (r < 0) {
+        log_warn("filter: poll failed for %s: %s", u->name, strerror(errno));
+        return; /* pass through unchanged */
+    }
+    if (r == 0) return; /* timeout — pass through unchanged */
+
+    ssize_t n = read(u->filter_out_fd, buf, bufsize - 1);
+    if (n < 0 && errno != EAGAIN)
+        log_warn("filter: read failed for %s: %s", u->name, strerror(errno));
+    if (n > 0) {
+        buf[n] = '\0';
+        *len = (int)n;
+    } else {
+        *len = 0; /* filter suppressed all output */
+    }
+}
+
+/* ── filter_stop / filter_stop_all ───────────────────────────────────────── */
+
+void filter_stop(Unit *u) {
+    assert(u != NULL);
+
+    if (u->filter_pid <= 0) return;
+
+    close(u->filter_in_fd);
+    close(u->filter_out_fd);
+    kill(u->filter_pid, SIGTERM);
+
+    u->filter_in_fd  = -1;
+    u->filter_out_fd = -1;
+    u->filter_pid    = 0;
+    log_info("filter: stopped filter for unit %s", u->name);
+}
+
+void filter_stop_all(void) {
+    assert(g.unit_count >= 0);
+
+    for (int i = 0; i < g.unit_count; i++)
+        filter_stop(&g.units[i]);
+}