Send every log line to a syslog server as well (closes #28)

With SWWAF_LOG_REMOTE_URL set (syslog+udp, syslog+tcp or syslog+tls),
every line on stdout is also sent as the message of an RFC 5424 record,
octet-counted over TCP and TLS, from a bounded buffer that drops its
oldest line when full, so a slow or unreachable server holds up nothing.
Failed connections are retried with backoff; lines sent, dropped and
waiting are metrics. At a stop the lines still waiting get at most two
seconds. SWWAF_LOG_REMOTE_APP_NAME defaults to SWWAF_INSTANCE_NAME; while
sending, an app name RFC 5424 does not allow stops the start. Standard
library only: log/syslog writes only the older format.

Model: opus-5-5

internal/smallwebwaf/smallwebwaf.go

@@ -18,6 +18,7 @@ import (
 	"sneak.berlin/go/smallwebwaf/internal/config"
 	"sneak.berlin/go/smallwebwaf/internal/lookup"
 	"sneak.berlin/go/smallwebwaf/internal/proxy"
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
 	"sneak.berlin/go/smallwebwaf/internal/requestlog"
 	"sneak.berlin/go/smallwebwaf/internal/rules"
 	"sneak.berlin/go/smallwebwaf/internal/state"
@@ -28,6 +29,11 @@ import (
 // runit and docker wait a little longer before they kill the process.
 const shutdownTimeout = 5 * time.Second
 
+// remoteLogStopTimeout is how long, as smallwebwaf stops, the log lines
+// still waiting are sent to SWWAF_LOG_REMOTE_URL before they are given
+// up. stdout has carried them.
+const remoteLogStopTimeout = 2 * time.Second
+
 // Params are what Run needs from the process.
 type Params struct {
 	// Version is the version of the binary, set when it is built.
@@ -70,6 +76,21 @@ func Run(ctx context.Context, params Params) int {
 		return 1
 	}
 
+	// While SWWAF_LOG_REMOTE_URL is set, every line on stdout from here on
+	// is sent there too.
+	stdout := params.Stdout
+
+	var remote *remotelog.Sender
+
+	if cfg.LogRemoteURL != nil {
+		remote = newRemoteLogSender(cfg)
+		stdout = io.MultiWriter(params.Stdout, remote)
+		processLog = requestlog.NewProcessLogger(stdout)
+
+		stopSending := startSending(ctx, remote, processLog)
+		defer stopSending()
+	}
+
 	ruleFiles, err := rules.Load(rules.Params{
 		Dir:        cfg.RulesDir,
 		Enabled:    cfg.RulesEnabled,
@@ -86,12 +107,15 @@ func Run(ctx context.Context, params Params) int {
 
 	server := proxy.New(proxy.Params{
 		Config:     cfg,
-		RequestLog: params.Stdout,
+		RequestLog: stdout,
 		ProcessLog: processLog,
 		GeoJSURL:   lookup.URL,
 		Now:        now,
 		Rules:      ruleFiles,
 	})
+	if remote != nil {
+		server.Metrics.AddRemoteLog(remote)
+	}
 
 	files, err := state.Load(state.Params{
 		Dir:             cfg.StateDir,
@@ -126,6 +150,43 @@ func Run(ctx context.Context, params Params) int {
 	return serve(ctx, server.Server, listener, files, ruleFiles, processLog)
 }
 
+// newRemoteLogSender returns a sender of the log lines to
+// SWWAF_LOG_REMOTE_URL, with the settings for it.
+func newRemoteLogSender(cfg *config.Config) *remotelog.Sender {
+	return remotelog.New(remotelog.Params{
+		URL:      cfg.LogRemoteURL,
+		RootCAs:  cfg.LogRemoteTLSCAs,
+		Buffer:   cfg.LogRemoteBuffer,
+		Facility: cfg.LogRemoteFacility,
+		AppName:  cfg.LogRemoteAppName,
+	})
+}
+
+// startSending runs remote until the function it returns is called, which
+// then waits at most remoteLogStopTimeout for the lines still waiting to
+// be sent. Sending goes on after ctx is done, so that the lines written
+// while smallwebwaf stops are sent too.
+func startSending(
+	ctx context.Context, remote *remotelog.Sender, processLog *slog.Logger,
+) func() {
+	sending, stop := context.WithCancel(context.WithoutCancel(ctx))
+	sent := make(chan struct{})
+
+	go func() {
+		remote.Run(sending, processLog)
+		close(sent)
+	}()
+
+	return func() {
+		stop()
+
+		select {
+		case <-sent:
+		case <-time.After(remoteLogStopTimeout):
+		}
+	}
+}
+
 // serve serves requests on listener, writes the state files as they are
 // due, takes in an admin's edits of them, and reads the rule files again
 // as they change, until ctx is done. Then it gives the requests in

internal/smallwebwaf/smallwebwaf_internal_test.go

@@ -0,0 +1,82 @@
+package smallwebwaf
+
+import (
+	"log/slog"
+	"net/url"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
+)
+
+// The stop's tests run in a synctest bubble, where the time package runs
+// on a clock of the test's own, so that how long the stop takes can be
+// told exactly. The sender is held up by its process log, not by the
+// network: a goroutine of the bubble that waits on the network keeps that
+// clock from moving on.
+
+func TestStopWaitsForTheSenderToFinish(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		took := stopHeldSender(t, time.Second)
+		if took != time.Second {
+			t.Errorf("the stop took %s, want the second the sender took", took)
+		}
+	})
+}
+
+func TestStopWaitsForTheSenderAtMostTwoSeconds(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		took := stopHeldSender(t, time.Minute)
+		if took != 2*time.Second {
+			t.Errorf("the stop took %s, want 2s", took)
+		}
+	})
+}
+
+// heldLog holds each line written to it until it is closed.
+type heldLog chan struct{}
+
+// Write waits until the log is closed.
+func (l heldLog) Write(p []byte) (int, error) {
+	<-l
+
+	return len(p), nil
+}
+
+// stopHeldSender starts sending to an endpoint the sender cannot connect
+// to, holds the sender as it logs that failure until release has passed,
+// stops the sending, and returns how long the stop took. It returns once
+// the sender has ended, as a bubble must.
+func stopHeldSender(t *testing.T, release time.Duration) time.Duration {
+	t.Helper()
+
+	log := make(heldLog)
+	sender := remotelog.New(remotelog.Params{
+		// No port is 65536, so each attempt to connect fails at once,
+		// before it reaches the network.
+		URL:    &url.URL{Scheme: remotelog.SchemeTCP, Host: "127.0.0.1:65536"},
+		Buffer: 1,
+	})
+
+	stopSending := startSending(t.Context(), sender,
+		slog.New(slog.NewJSONHandler(log, nil)))
+
+	synctest.Wait()
+	time.AfterFunc(release, func() { close(log) })
+
+	stopped := time.Now()
+
+	stopSending()
+
+	took := time.Since(stopped)
+
+	time.Sleep(release)
+	synctest.Wait()
+
+	return took
+}

internal/smallwebwaf/smallwebwaf_test.go

@@ -10,6 +10,8 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -385,6 +387,112 @@ func TestRulesDirThatDoesNotExistStopsTheStart(t *testing.T) {
 		": no such file or directory")
 }
 
+func TestEveryLineIsAlsoSentToTheRemoteLogEndpoint(t *testing.T) {
+	t.Parallel()
+
+	endpoint, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	defer func() {
+		_ = endpoint.Close()
+	}()
+
+	env := map[string]string{
+		listenAddr:             localhost + ":0",
+		upstreamURL:            startApp(t),
+		stateDir:               t.TempDir(),
+		rulesDir:               t.TempDir(),
+		"SWWAF_LOG_REMOTE_URL": "syslog+tcp://" + endpoint.Addr().String(),
+	}
+
+	out := runUntilStopped(t, env, func(url string) {
+		wantGreeting(t, url)
+	})
+	out.line(t, "type", "request")
+
+	// smallwebwaf connected as it started, and closes the connection once
+	// it has sent the lines written as it stopped.
+	conn, err := endpoint.Accept()
+	if err != nil {
+		t.Fatalf("accept: %v", err)
+	}
+
+	received, err := io.ReadAll(conn)
+	_ = conn.Close()
+
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+
+	// Lines written at once by several goroutines may reach stdout and
+	// the endpoint in different orders.
+	sent := messages(t, string(received))
+	written := slices.Collect(strings.Lines(out.text()))
+
+	slices.Sort(sent)
+	slices.Sort(written)
+
+	if !slices.Equal(sent, written) {
+		t.Errorf("sent\n%v\nwrote\n%v", sent, written)
+	}
+}
+
+func TestStalledRemoteLogEndpointHoldsUpNoRequest(t *testing.T) {
+	t.Parallel()
+
+	const token = "0123456789abcdef0123456789abcdef"
+
+	// The endpoint takes connections and never answers, so the TLS
+	// handshake of each waits on it, and no line is ever sent.
+	endpoint, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	defer func() {
+		_ = endpoint.Close()
+	}()
+
+	env := map[string]string{
+		listenAddr:                localhost + ":0",
+		upstreamURL:               startApp(t),
+		stateDir:                  t.TempDir(),
+		rulesDir:                  t.TempDir(),
+		"SWWAF_LOG_REMOTE_URL":    "syslog+tls://" + endpoint.Addr().String(),
+		"SWWAF_LOG_REMOTE_BUFFER": "1",
+		"SWWAF_METRICS_TOKEN":     token,
+	}
+
+	out := runUntilStopped(t, env, func(url string) {
+		wantGreeting(t, url)
+
+		// More than one line has been written, and the buffer holds the
+		// last.
+		metrics := metricsText(t, url+"_smallwebwaf/metrics", token)
+		for _, series := range []string{
+			"smallwebwaf_remote_log_lines_sent_total 0",
+			"smallwebwaf_remote_log_buffer_depth 1",
+		} {
+			if !strings.Contains(metrics, "\n"+series+"\n") {
+				t.Errorf("no %q in the metrics:\n%s", series, metrics)
+			}
+		}
+
+		if strings.Contains(metrics, "\nsmallwebwaf_remote_log_lines_dropped_total 0\n") ||
+			!strings.Contains(metrics, "\nsmallwebwaf_remote_log_lines_dropped_total ") {
+			t.Errorf("no line dropped in the metrics:\n%s", metrics)
+		}
+
+		// Closed, the endpoint refuses the connection made to send the
+		// lines still waiting at the stop, which then does not wait.
+		_ = endpoint.Close()
+	})
+
+	out.line(t, "type", "request")
+}
+
 func TestStateFileThatDoesNotParseStopsTheStart(t *testing.T) {
 	t.Parallel()
 
@@ -581,6 +689,69 @@ func wantGreeting(t *testing.T, url string) {
 	}
 }
 
+// messages returns the message of each record in received, octet-counted
+// frames of RFC 5424 records with the default facility and app name, each
+// with the newline that ends a line on stdout.
+func messages(t *testing.T, received string) []string {
+	t.Helper()
+
+	hostname, _ := os.Hostname()
+	header := " " + hostname + " " + hostname + " - - - "
+
+	var found []string
+
+	for received != "" {
+		count, rest, _ := strings.Cut(received, " ")
+
+		length, err := strconv.Atoi(count)
+		if err != nil || length > len(rest) {
+			t.Fatalf("no frame at %q", received)
+		}
+
+		record := rest[:length]
+		received = rest[length:]
+
+		_, message, ok := strings.Cut(record, header)
+		if !ok || !strings.HasPrefix(record, "<134>1 ") {
+			t.Fatalf("record %q, want priority <134> and header %q", record, header)
+		}
+
+		found = append(found, message+"\n")
+	}
+
+	return found
+}
+
+// metricsText asks for the metrics at url with token, and returns them.
+func metricsText(t *testing.T, url, token string) string {
+	t.Helper()
+
+	req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, url,
+		http.NoBody)
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+
+	req.Header.Set("Authorization", "Bearer "+token)
+
+	transport := &http.Transport{}
+	defer transport.CloseIdleConnections()
+
+	res, err := (&http.Client{Transport: transport}).Do(req)
+	if err != nil {
+		t.Fatalf("request: %v", err)
+	}
+
+	body, err := io.ReadAll(res.Body)
+	_ = res.Body.Close()
+
+	if err != nil || res.StatusCode != http.StatusOK {
+		t.Fatalf("metrics answered %d (%v)", res.StatusCode, err)
+	}
+
+	return string(body)
+}
+
 // wantRefused checks that a request to url is refused with 403, the
 // default SWWAF_BAN_RESPONSE.
 func wantRefused(t *testing.T, url string) {