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/config/config.go

@@ -4,6 +4,7 @@
 package config
 
 import (
+	"crypto/x509"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -19,6 +20,8 @@ import (
 	"strings"
 	"time"
 	"unicode/utf8"
+
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
 )
 
 // Config is smallwebwaf's settings. A timeout, size or rate limit of zero
@@ -135,6 +138,21 @@ type Config struct {
 	// unless RulesEnabled is false (SWWAF_RULES_ENABLED).
 	RulesDir     string
 	RulesEnabled bool
+	// LogRemoteURL is where every line on stdout is also sent
+	// (SWWAF_LOG_REMOTE_URL), nil while it is unset and nothing is sent.
+	// LogRemoteTLSCAs are the certificates a syslog+tls endpoint's
+	// certificate must chain to (SWWAF_LOG_REMOTE_TLS_CA_FILE), nil while
+	// it is unset and the host's own are used. LogRemoteBuffer is the most
+	// lines held while they wait to be sent (SWWAF_LOG_REMOTE_BUFFER).
+	// LogRemoteFacility is the number of the syslog facility
+	// (SWWAF_LOG_REMOTE_FACILITY), and LogRemoteAppName the APP-NAME
+	// (SWWAF_LOG_REMOTE_APP_NAME, by default InstanceName), of the records
+	// the lines are sent in.
+	LogRemoteURL      *url.URL
+	LogRemoteTLSCAs   *x509.CertPool
+	LogRemoteBuffer   int
+	LogRemoteFacility int
+	LogRemoteAppName  string
 
 	// settings are the values read, as given or by default, for the
 	// log line at start.
@@ -195,7 +213,14 @@ var (
 	errNotMode       = errors.New("is not enforce or observe")
 	errNotPathPrefix = errors.New(
 		"is not a path prefix starting with /, such as /assets/")
-	errNotBoolean = errors.New("is not true or false")
+	errNotBoolean      = errors.New("is not true or false")
+	errNotLogRemoteURL = errors.New(
+		"is not syslog+udp, syslog+tcp or syslog+tls with a host and a port, " +
+			"and nothing more, such as syslog+tls://logs.example:6514")
+	errNoCertificate = errors.New("holds no PEM certificate")
+	errNotFacility   = errors.New("is not a syslog facility such as local0 or daemon")
+	errNotAppName    = errors.New(
+		"is not 1 to 48 printable ASCII characters without a space, such as gitea")
 )
 
 // FromEnvironment reads the settings with lookupEnv, normally
@@ -241,12 +266,19 @@ func FromEnvironment(lookupEnv func(string) (string, bool)) (*Config, error) {
 		StateCounterInterval: env.durationNotOff("SWWAF_STATE_COUNTER_INTERVAL", "15m"),
 		LogRequestHeaders: env.headerNames("SWWAF_LOG_REQUEST_HEADERS",
 			"accept,accept-language,accept-encoding,content-type,origin,range"),
-		MetricsToken: env.token("SWWAF_METRICS_TOKEN"),
-		MetricsTopN:  env.numberNotOff("SWWAF_METRICS_TOP_N", "50"),
-		RulesDir:     env.value("SWWAF_RULES_DIR", "/etc/smallwebwaf/rules.d"),
-		RulesEnabled: env.boolean("SWWAF_RULES_ENABLED", "true"),
+		MetricsToken:      env.token("SWWAF_METRICS_TOKEN"),
+		MetricsTopN:       env.numberNotOff("SWWAF_METRICS_TOP_N", "50"),
+		RulesDir:          env.value("SWWAF_RULES_DIR", "/etc/smallwebwaf/rules.d"),
+		RulesEnabled:      env.boolean("SWWAF_RULES_ENABLED", "true"),
+		LogRemoteURL:      env.logRemoteURL("SWWAF_LOG_REMOTE_URL"),
+		LogRemoteTLSCAs:   env.certificates("SWWAF_LOG_REMOTE_TLS_CA_FILE"),
+		LogRemoteBuffer:   env.numberNotOff("SWWAF_LOG_REMOTE_BUFFER", "10000"),
+		LogRemoteFacility: env.facility("SWWAF_LOG_REMOTE_FACILITY", "local0"),
 	}
 
+	cfg.LogRemoteAppName = env.appName("SWWAF_LOG_REMOTE_APP_NAME",
+		cfg.InstanceName, cfg.LogRemoteURL != nil)
+
 	for _, country := range cfg.ExclusivelyAllowedCountries {
 		if slices.Contains(cfg.DeniedCountries, country) {
 			env.check("SWWAF_EXCLUSIVELY_ALLOWED_COUNTRIES",
@@ -468,6 +500,74 @@ func (e *environment) token(name string) string {
 	return value
 }
 
+// logRemoteURL reads the setting that is where every log line is also
+// sent. Unset or empty, it is nil, and nothing is sent.
+func (e *environment) logRemoteURL(name string) *url.URL {
+	value := e.value(name, "")
+	if value == "" {
+		return nil
+	}
+
+	remote, err := parseLogRemoteURL(value)
+	e.check(name, err)
+
+	return remote
+}
+
+// certificates reads a setting that is the path of a file of PEM
+// certificates. Unset or empty, it is nil.
+func (e *environment) certificates(name string) *x509.CertPool {
+	path := e.value(name, "")
+	if path == "" {
+		return nil
+	}
+
+	pem, err := os.ReadFile(path) //nolint:gosec // a file the admin names
+	if err != nil {
+		e.check(name, fmt.Errorf("cannot be read: %w", err))
+
+		return nil
+	}
+
+	pool := x509.NewCertPool()
+	if !pool.AppendCertsFromPEM(pem) {
+		e.check(name, fmt.Errorf("%q %w", path, errNoCertificate))
+
+		return nil
+	}
+
+	return pool
+}
+
+// facility reads a setting that is a syslog facility, and returns its
+// number.
+func (e *environment) facility(name, defaultValue string) int {
+	number, err := parseFacility(e.value(name, defaultValue))
+	e.check(name, err)
+
+	return number
+}
+
+// appName reads the setting that is the APP-NAME of the records the log
+// lines are sent in, by default the instance name. Its value is checked
+// when it is set, and, while lines are sent, when it is the instance name.
+func (e *environment) appName(name, instanceName string, sending bool) string {
+	_, set := e.lookupEnv(name)
+
+	value := e.value(name, instanceName)
+
+	switch {
+	case isAppName(value):
+	case set:
+		e.check(name, fmt.Errorf("%q %w", value, errNotAppName))
+	case sending:
+		e.check(name, fmt.Errorf("is unset, and SWWAF_INSTANCE_NAME %q, its default, %w",
+			value, errNotAppName))
+	}
+
+	return value
+}
+
 // parseDuration reads a duration in Go's syntax, such as 90s or 15m, a
 // whole number of days such as 7d, or off.
 func parseDuration(value string) (time.Duration, error) {
@@ -841,3 +941,64 @@ func parseUpstreamURL(value string) (*url.URL, error) {
 
 	return upstream, nil
 }
+
+// parseLogRemoteURL reads where every log line is also sent:
+// syslog+udp, syslog+tcp or syslog+tls, a host and a port from 1 to
+// 65535, and nothing else.
+func parseLogRemoteURL(value string) (*url.URL, error) {
+	remote, err := url.Parse(value)
+	if err != nil {
+		return nil, fmt.Errorf("%q %w", value, errNotLogRemoteURL)
+	}
+
+	schemes := []string{remotelog.SchemeUDP, remotelog.SchemeTCP, remotelog.SchemeTLS}
+	port, err := strconv.ParseUint(remote.Port(), 10, 16)
+
+	onlySchemeHostAndPort := slices.Contains(schemes, remote.Scheme) &&
+		remote.Hostname() != "" && err == nil && port != 0 &&
+		remote.User == nil && remote.Opaque == "" &&
+		(remote.Path == "" || remote.Path == "/") &&
+		remote.RawQuery == "" && remote.Fragment == ""
+	if !onlySchemeHostAndPort {
+		return nil, fmt.Errorf("%q %w", value, errNotLogRemoteURL)
+	}
+
+	return remote, nil
+}
+
+// parseFacility reads the name of a syslog facility, and returns its
+// number, as RFC 5424 numbers them.
+func parseFacility(value string) (int, error) {
+	//nolint:mnd // the facilities' numbers in RFC 5424
+	number, known := map[string]int{
+		"kern": 0, "user": 1, "mail": 2, "daemon": 3, "auth": 4, "syslog": 5,
+		"lpr": 6, "news": 7, "uucp": 8, "cron": 9, "authpriv": 10, "ftp": 11,
+		"local0": 16, "local1": 17, "local2": 18, "local3": 19,
+		"local4": 20, "local5": 21, "local6": 22, "local7": 23,
+	}[value]
+	if !known {
+		return 0, fmt.Errorf("%q %w", value, errNotFacility)
+	}
+
+	return number, nil
+}
+
+// appNameMaxLength is the most characters RFC 5424 allows in an
+// APP-NAME.
+const appNameMaxLength = 48
+
+// isAppName reports whether value can be an APP-NAME: 1 to
+// appNameMaxLength printable ASCII characters, none of them a space.
+func isAppName(value string) bool {
+	if value == "" || len(value) > appNameMaxLength {
+		return false
+	}
+
+	for _, char := range []byte(value) {
+		if char < '!' || char > '~' {
+			return false
+		}
+	}
+
+	return true
+}

internal/config/config_test.go

@@ -2,11 +2,13 @@ package config_test
 
 import (
 	"bytes"
+	"crypto/x509"
 	"encoding/json"
 	"log/slog"
 	"maps"
 	"net/netip"
 	"os"
+	"path/filepath"
 	"slices"
 	"strings"
 	"testing"
@@ -54,15 +56,41 @@ const (
 	logRequestHeaders       = "SWWAF_LOG_REQUEST_HEADERS"
 	rulesDir                = "SWWAF_RULES_DIR"
 	rulesEnabled            = "SWWAF_RULES_ENABLED"
+	logRemoteURL            = "SWWAF_LOG_REMOTE_URL"
+	logRemoteTLSCAFile      = "SWWAF_LOG_REMOTE_TLS_CA_FILE"
+	logRemoteBuffer         = "SWWAF_LOG_REMOTE_BUFFER"
+	logRemoteFacility       = "SWWAF_LOG_REMOTE_FACILITY"
+	logRemoteAppName        = "SWWAF_LOG_REMOTE_APP_NAME"
 )
 
 // defaultLogRequestHeaders is the default of SWWAF_LOG_REQUEST_HEADERS.
 const defaultLogRequestHeaders = "accept,accept-language,accept-encoding," +
 	"content-type,origin,range"
 
+// testCA is a CA certificate, of which only that it reads matters here.
+const testCA = `-----BEGIN CERTIFICATE-----
+MIIBkzCCATmgAwIBAgIUeySaE27dnr6A2HijrMB13gTLUKIwCgYIKoZIzj0EAwIw
+HjEcMBoGA1UEAwwTc21hbGx3ZWJ3YWYgdGVzdCBDQTAgFw0yNjEwMDYxNDI2MTNa
+GA8yMTI2MDkxMjE0MjYxM1owHjEcMBoGA1UEAwwTc21hbGx3ZWJ3YWYgdGVzdCBD
+QTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABKDEhcWKKhet2KgSdME+iEPxyEyn
+2sd9IdElbt8DM2SfCdB2JsXo0C07UNZaywMPMfn/n8LNI/PKwu+N2uX7gfSjUzBR
+MB0GA1UdDgQWBBRZo3BPLv0KbV4drw6JI1JIUKmoRzAfBgNVHSMEGDAWgBRZo3BP
+Lv0KbV4drw6JI1JIUKmoRzAPBgNVHRMBAf8EBTADAQH/MAoGCCqGSM49BAMCA0gA
+MEUCICC5k+76UpWoSwVbZA+atu5WcALEOJGqwOUWua3zemhcAiEA8Hfdxgwp0z2v
+rlG9y/jrJb6ORy3kTLWo2EA0BA67vuI=
+-----END CERTIFICATE-----
+`
+
 // token is a token of 32 characters, the shortest allowed.
 const token = "0123456789abcdef0123456789abcdef"
 
+// instance is an SWWAF_INSTANCE_NAME that is a valid app name too, and
+// remoteURL an SWWAF_LOG_REMOTE_URL, for the tests that send the lines.
+const (
+	instance  = "fsn1app1/gitea"
+	remoteURL = "syslog+udp://192.0.2.1:514"
+)
+
 // off switches a timeout, a size limit or a rate limit off.
 const off = "off"
 
@@ -273,6 +301,177 @@ func TestInstanceNameAndLoggedHeadersAsSet(t *testing.T) {
 	}
 }
 
+func TestRemoteLogSettingsDefaults(t *testing.T) {
+	t.Parallel()
+
+	cfg := fromEnvironment(t, environment{instanceName: instance})
+
+	if cfg.LogRemoteURL != nil || cfg.LogRemoteTLSCAs != nil ||
+		cfg.LogRemoteBuffer != 10000 || cfg.LogRemoteFacility != 16 ||
+		cfg.LogRemoteAppName != instance {
+		t.Errorf("remote log settings %v, %v, %d, %d and %q, want no URL, no "+
+			"certificates, 10000, 16 and %s's %s", cfg.LogRemoteURL,
+			cfg.LogRemoteTLSCAs, cfg.LogRemoteBuffer, cfg.LogRemoteFacility,
+			cfg.LogRemoteAppName, instanceName, instance)
+	}
+}
+
+func TestRemoteLogSettingsAsSet(t *testing.T) {
+	t.Parallel()
+
+	caFile := filepath.Join(t.TempDir(), "ca.pem")
+
+	err := os.WriteFile(caFile, []byte(testCA), 0o600)
+	if err != nil {
+		t.Fatalf("write %s: %v", caFile, err)
+	}
+
+	cfg := fromEnvironment(t, environment{
+		logRemoteURL:       "syslog+tls://logs.example:6514",
+		logRemoteTLSCAFile: caFile,
+		logRemoteBuffer:    "500",
+		logRemoteFacility:  "daemon",
+		logRemoteAppName:   instance,
+	})
+
+	roots := x509.NewCertPool()
+	roots.AppendCertsFromPEM([]byte(testCA))
+
+	if cfg.LogRemoteURL.String() != "syslog+tls://logs.example:6514" ||
+		!roots.Equal(cfg.LogRemoteTLSCAs) || cfg.LogRemoteBuffer != 500 ||
+		cfg.LogRemoteFacility != 3 || cfg.LogRemoteAppName != instance {
+		t.Errorf("remote log settings %v, %v, %d, %d and %q", cfg.LogRemoteURL,
+			cfg.LogRemoteTLSCAs, cfg.LogRemoteBuffer, cfg.LogRemoteFacility,
+			cfg.LogRemoteAppName)
+	}
+}
+
+func TestRemoteLogURLForms(t *testing.T) {
+	t.Parallel()
+
+	for _, value := range []string{
+		"syslog+udp://192.0.2.1:514",
+		"syslog+tcp://[2001:db8::1]:514",
+		"syslog+tls://logs.example:6514/",
+	} {
+		cfg := fromEnvironment(t, environment{logRemoteURL: value})
+		if cfg.LogRemoteURL.String() != value {
+			t.Errorf("%s read as %v", value, cfg.LogRemoteURL)
+		}
+	}
+
+	cfg := fromEnvironment(t, environment{logRemoteURL: ""})
+	if cfg.LogRemoteURL != nil {
+		t.Errorf("set but empty, %s read as %v", logRemoteURL, cfg.LogRemoteURL)
+	}
+}
+
+func TestRemoteLogFacilitiesByNumber(t *testing.T) {
+	t.Parallel()
+
+	for name, number := range map[string]int{
+		"kern": 0, "user": 1, "auth": 4, "authpriv": 10, "ftp": 11,
+		"local0": 16, "local5": 21, "local7": 23,
+	} {
+		cfg := fromEnvironment(t, environment{logRemoteFacility: name})
+		if cfg.LogRemoteFacility != number {
+			t.Errorf("%s read as %d, want %d", name, cfg.LogRemoteFacility, number)
+		}
+	}
+}
+
+func TestInvalidRemoteLogSettingStopsTheStart(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct{ name, value string }{
+		{logRemoteURL, "logs.example:514"},
+		{logRemoteURL, "syslog://logs.example:514"},
+		{logRemoteURL, "http://logs.example:514"},
+		{logRemoteURL, "syslog+udp://logs.example"},
+		{logRemoteURL, "syslog+tcp://:514"},
+		{logRemoteURL, "syslog+tcp://logs.example:0"},
+		{logRemoteURL, "syslog+tls://logs.example:65536"},
+		{logRemoteURL, "syslog+tls://user@logs.example:6514"},
+		{logRemoteURL, "syslog+tcp://logs.example:514/app"},
+		{logRemoteURL, "syslog+tcp://logs.example:514?tls=1"},
+		{logRemoteTLSCAFile, "/nonexistent/ca.pem"},
+		{logRemoteBuffer, off}, {logRemoteBuffer, "0"}, {logRemoteBuffer, "10K"},
+		{logRemoteFacility, "local8"}, {logRemoteFacility, "LOCAL0"},
+		{logRemoteFacility, "16"}, {logRemoteFacility, ""},
+		{logRemoteAppName, ""}, {logRemoteAppName, "my app"},
+		{logRemoteAppName, "gitéa"}, {logRemoteAppName, strings.Repeat("a", 49)},
+	} {
+		_, err := config.FromEnvironment(environment{tc.name: tc.value}.lookupEnv)
+		if err == nil || !strings.HasPrefix(err.Error(), tc.name+": ") {
+			t.Errorf("%s=%q: error %v, want one naming it", tc.name, tc.value, err)
+		}
+	}
+}
+
+func TestRemoteLogCAFileWithoutCertificateStopsTheStart(t *testing.T) {
+	t.Parallel()
+
+	caFile := filepath.Join(t.TempDir(), "ca.pem")
+
+	err := os.WriteFile(caFile, []byte("not a certificate\n"), 0o600)
+	if err != nil {
+		t.Fatalf("write %s: %v", caFile, err)
+	}
+
+	_, err = config.FromEnvironment(environment{logRemoteTLSCAFile: caFile}.lookupEnv)
+
+	want := logRemoteTLSCAFile + `: "` + caFile + `" holds no PEM certificate`
+	if err == nil || err.Error() != want {
+		t.Errorf("error %v, want %s", err, want)
+	}
+}
+
+func TestInstanceNameNotAnAppNameStopsTheStartOnlyWhileSending(t *testing.T) {
+	t.Parallel()
+
+	const spaced = "fsn1 app1"
+
+	sending := environment{logRemoteURL: remoteURL, instanceName: spaced}
+
+	_, err := config.FromEnvironment(sending.lookupEnv)
+
+	want := logRemoteAppName + `: is unset, and ` + instanceName +
+		` "fsn1 app1", its default, is not 1 to 48 printable ASCII characters ` +
+		`without a space, such as gitea`
+	if err == nil || err.Error() != want {
+		t.Errorf("error %v, want %s", err, want)
+	}
+
+	cfg := fromEnvironment(t, environment{instanceName: spaced})
+	if cfg.LogRemoteAppName != spaced {
+		t.Errorf("not sending, %s is %q", logRemoteAppName, cfg.LogRemoteAppName)
+	}
+
+	sending[logRemoteAppName] = instance
+
+	cfg = fromEnvironment(t, sending)
+	if cfg.LogRemoteAppName != instance {
+		t.Errorf("set to %s, %s is %q", instance, logRemoteAppName,
+			cfg.LogRemoteAppName)
+	}
+}
+
+func TestAppNameSetStopsTheStartWhileSending(t *testing.T) {
+	t.Parallel()
+
+	_, err := config.FromEnvironment(environment{
+		logRemoteURL:     remoteURL,
+		instanceName:     instance,
+		logRemoteAppName: "my app",
+	}.lookupEnv)
+
+	want := logRemoteAppName + `: "my app" is not 1 to 48 printable ASCII ` +
+		`characters without a space, such as gitea`
+	if err == nil || err.Error() != want {
+		t.Errorf("error %v, want %s", err, want)
+	}
+}
+
 func TestCodeOnBothCountryListsStopsTheStart(t *testing.T) {
 	t.Parallel()
 
@@ -581,6 +780,11 @@ func TestLogsEachSettingWithItsValue(t *testing.T) {
 		logRequestHeaders:       defaultLogRequestHeaders,
 		rulesDir:                "/etc/smallwebwaf/rules.d",
 		rulesEnabled:            "true",
+		logRemoteURL:            "",
+		logRemoteTLSCAFile:      "",
+		logRemoteBuffer:         "10000",
+		logRemoteFacility:       "local0",
+		logRemoteAppName:        hostname,
 	}
 	if !maps.Equal(line.Settings, want) {
 		t.Errorf("logged settings\n%v\nwant\n%v", line.Settings, want)

internal/metrics/metrics.go

@@ -13,6 +13,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"sneak.berlin/go/smallwebwaf/internal/bans"
 	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
 	"sneak.berlin/go/smallwebwaf/internal/requestlog"
 	"sneak.berlin/go/smallwebwaf/internal/rules"
 )
@@ -197,6 +198,33 @@ func (m *Metrics) AddRules(ruleFiles *rules.Files) {
 		}))
 }
 
+// AddRemoteLog adds the metrics of sending the log lines to
+// SWWAF_LOG_REMOTE_URL, read from remote as the metrics are asked for: the
+// lines sent, those dropped, and those waiting in the buffer.
+func (m *Metrics) AddRemoteLog(remote *remotelog.Sender) {
+	m.registry.MustRegister(
+		prometheus.NewCounterFunc(prometheus.CounterOpts{
+			Name: "smallwebwaf_remote_log_lines_sent_total",
+			Help: "Log lines sent to SWWAF_LOG_REMOTE_URL.",
+		}, func() float64 {
+			return float64(remote.Sent())
+		}),
+		prometheus.NewCounterFunc(prometheus.CounterOpts{
+			Name: "smallwebwaf_remote_log_lines_dropped_total",
+			Help: "Log lines dropped: the oldest in a full buffer, and those " +
+				"whose sending failed.",
+		}, func() float64 {
+			return float64(remote.Dropped())
+		}),
+		prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+			Name: "smallwebwaf_remote_log_buffer_depth",
+			Help: "Log lines in the buffer, waiting to be sent.",
+		}, func() float64 {
+			return float64(remote.Depth())
+		}),
+	)
+}
+
 // ServeHTTP answers with the metrics in the Prometheus text format.
 func (m *Metrics) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	m.handler.ServeHTTP(w, r)

internal/remotelog/remotelog.go

@@ -0,0 +1,310 @@
+// Package remotelog sends the lines smallwebwaf writes on stdout to the
+// remote log endpoint, SWWAF_LOG_REMOTE_URL, as the "Request log" section
+// of SPEC.md describes: each line as the message of an RFC 5424 syslog
+// record, over UDP, TCP or TLS. Lines wait in a bounded buffer, so a slow
+// or unreachable endpoint never holds up a request or stdout.
+package remotelog
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/url"
+	"os"
+	"strconv"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+)
+
+// The forms of SWWAF_LOG_REMOTE_URL, by its scheme.
+const (
+	SchemeUDP = "syslog+udp"
+	SchemeTCP = "syslog+tcp"
+	SchemeTLS = "syslog+tls"
+)
+
+// A record's priority is the number of its facility times the number of
+// severities there are, plus the number of its severity. Every record's
+// severity is informational.
+const (
+	severities    = 8
+	informational = 6
+)
+
+const (
+	// dialTimeout bounds connecting to the endpoint, the TLS handshake
+	// included.
+	dialTimeout = 10 * time.Second
+	// After a failed attempt to connect, or a connection on which a record
+	// fails, the next attempt to connect is made a second later, and
+	// retryDelayFactor times as long after each further failure in a row,
+	// up to a minute. A connection that fails after it has stayed up for
+	// resetRetryDelayAfter ends the row.
+	firstRetryDelay      = time.Second
+	retryDelayFactor     = 2
+	maxRetryDelay        = time.Minute
+	resetRetryDelayAfter = time.Minute
+)
+
+// Params are what New needs.
+type Params struct {
+	// URL is the endpoint (SWWAF_LOG_REMOTE_URL): SchemeUDP, SchemeTCP or
+	// SchemeTLS, a host and a port.
+	URL *url.URL
+	// RootCAs are the certificates a SchemeTLS endpoint's certificate
+	// must chain to (SWWAF_LOG_REMOTE_TLS_CA_FILE), nil for the host's.
+	RootCAs *x509.CertPool
+	// Buffer is the most lines held while they wait to be sent
+	// (SWWAF_LOG_REMOTE_BUFFER).
+	Buffer int
+	// Facility is the number of the records' syslog facility
+	// (SWWAF_LOG_REMOTE_FACILITY), and AppName their APP-NAME
+	// (SWWAF_LOG_REMOTE_APP_NAME).
+	Facility int
+	AppName  string
+}
+
+// Sender sends lines to the endpoint. Write puts them in its buffer, and
+// Run sends them from there.
+type Sender struct {
+	url       *url.URL
+	tlsConfig *tls.Config
+	// beforeTime and afterTime are the parts of every record's header
+	// before and after its time, as RFC 5424 lays the header out.
+	beforeTime string
+	afterTime  string
+	// records is the buffer: each line's record, framed to be sent.
+	records chan []byte
+	sent    atomic.Int64
+	dropped atomic.Int64
+}
+
+// New returns a Sender for the endpoint params.URL.
+func New(params Params) *Sender {
+	hostname, err := os.Hostname()
+	if err != nil || hostname == "" {
+		hostname = "-" // RFC 5424's value for a field that has none
+	}
+
+	priority := params.Facility*severities + informational
+
+	return &Sender{
+		url: params.URL,
+		tlsConfig: &tls.Config{
+			RootCAs:    params.RootCAs,
+			MinVersion: tls.VersionTLS12,
+		},
+		// The 1 is the version of the format. The process id, the message
+		// id and the structured data have no value.
+		beforeTime: "<" + strconv.Itoa(priority) + ">1 ",
+		afterTime:  " " + hostname + " " + params.AppName + " - - - ",
+		records:    make(chan []byte, params.Buffer),
+	}
+}
+
+// Write puts each line in p in the buffer, as the message of a record of
+// its own, and never waits: when the buffer is full, the oldest record in
+// it is dropped to make room. It is safe for concurrent use.
+func (s *Sender) Write(p []byte) (int, error) {
+	at := requestlog.FormatTime(time.Now())
+
+	for line := range bytes.Lines(p) {
+		line = bytes.TrimSuffix(line, []byte("\n"))
+		if len(line) > 0 {
+			s.put(s.record(at, line))
+		}
+	}
+
+	return len(p), nil
+}
+
+// Sent is how many records have been sent.
+func (s *Sender) Sent() int64 {
+	return s.sent.Load()
+}
+
+// Dropped is how many records were dropped: the oldest in a full buffer,
+// and those whose sending failed.
+func (s *Sender) Dropped() int64 {
+	return s.dropped.Load()
+}
+
+// Depth is how many records are in the buffer.
+func (s *Sender) Depth() int {
+	return len(s.records)
+}
+
+// Run connects to the endpoint and sends each record as it comes into the
+// buffer, until ctx is done. Then it sends the records still in the buffer,
+// on the connection open at that time or, if there is none, on a new one,
+// until none is left or one fails, and returns. How long it may take over
+// that is for the caller to bound.
+//
+// A connection on which a record fails is closed and the record dropped.
+// That failure, like a failed attempt to connect, is logged to processLog
+// and followed by the next attempt after firstRetryDelay, retryDelayFactor
+// times as long after each further failure in a row up to maxRetryDelay,
+// and firstRetryDelay again after a connection that stayed up for
+// resetRetryDelayAfter. Meanwhile the records wait in the buffer.
+func (s *Sender) Run(ctx context.Context, processLog *slog.Logger) {
+	conn := s.send(ctx, processLog)
+	if conn == nil && len(s.records) > 0 {
+		conn, _ = s.dial(context.WithoutCancel(ctx))
+	}
+
+	if conn == nil {
+		return
+	}
+
+	defer func() {
+		_ = conn.Close()
+	}()
+
+	for {
+		select {
+		case record := <-s.records:
+			if s.write(conn, record) != nil {
+				return
+			}
+		default:
+			return
+		}
+	}
+}
+
+// record returns line as an RFC 5424 record made at the time at, framed
+// for the endpoint: on its own over UDP, since each datagram holds one,
+// and over TCP and TLS after its length in bytes and a space, the
+// octet-counted framing of RFC 6587 and RFC 5425.
+func (s *Sender) record(at string, line []byte) []byte {
+	record := make([]byte, 0, len(s.beforeTime)+len(at)+len(s.afterTime)+len(line))
+	record = append(record, s.beforeTime...)
+	record = append(record, at...)
+	record = append(record, s.afterTime...)
+	record = append(record, line...)
+
+	if s.url.Scheme == SchemeUDP {
+		return record
+	}
+
+	return append([]byte(strconv.Itoa(len(record))+" "), record...)
+}
+
+// put adds record to the buffer, first dropping the oldest record in it
+// while it is full.
+func (s *Sender) put(record []byte) {
+	for {
+		select {
+		case s.records <- record:
+			return
+		default:
+		}
+
+		select {
+		case <-s.records:
+			s.dropped.Add(1)
+		default:
+		}
+	}
+}
+
+// send connects to the endpoint and sends each record as it comes into
+// the buffer, until ctx is done, and returns the connection then open, or
+// nil.
+func (s *Sender) send(ctx context.Context, processLog *slog.Logger) net.Conn {
+	delay := firstRetryDelay
+
+	for {
+		conn, err := s.dial(ctx)
+		if ctx.Err() != nil {
+			return conn
+		}
+
+		if err == nil {
+			connected := time.Now()
+
+			err = s.sendOn(ctx, conn)
+			if err == nil {
+				return conn
+			}
+
+			_ = conn.Close()
+
+			if time.Since(connected) >= resetRetryDelayAfter {
+				delay = firstRetryDelay
+			}
+		}
+
+		processLog.Warn("sending to SWWAF_LOG_REMOTE_URL failed",
+			"error", err.Error(), "connecting_again_in", delay.String())
+
+		select {
+		case <-time.After(delay):
+		case <-ctx.Done():
+			return nil
+		}
+
+		delay = min(retryDelayFactor*delay, maxRetryDelay)
+	}
+}
+
+// sendOn sends each record on conn as it comes into the buffer, until one
+// fails, whose error it returns, or ctx is done.
+func (s *Sender) sendOn(ctx context.Context, conn net.Conn) error {
+	for {
+		select {
+		case record := <-s.records:
+			err := s.write(conn, record)
+			if err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+// write sends record on conn, and counts it as sent or, if that fails,
+// as dropped. A record too long for one UDP datagram is dropped without
+// an error, since the connection has not failed: a long request must not
+// hold up the lines after it.
+func (s *Sender) write(conn net.Conn, record []byte) error {
+	_, err := conn.Write(record)
+	if err != nil {
+		s.dropped.Add(1)
+
+		if errors.Is(err, syscall.EMSGSIZE) {
+			return nil
+		}
+
+		return fmt.Errorf("send a record: %w", err)
+	}
+
+	s.sent.Add(1)
+
+	return nil
+}
+
+// dial connects to the endpoint.
+func (s *Sender) dial(ctx context.Context) (net.Conn, error) {
+	dialer := &net.Dialer{Timeout: dialTimeout}
+
+	switch s.url.Scheme {
+	case SchemeUDP:
+		return dialer.DialContext(ctx, "udp", s.url.Host)
+	case SchemeTLS:
+		tlsDialer := &tls.Dialer{NetDialer: dialer, Config: s.tlsConfig}
+
+		return tlsDialer.DialContext(ctx, "tcp", s.url.Host)
+	default:
+		return dialer.DialContext(ctx, "tcp", s.url.Host)
+	}
+}

internal/remotelog/remotelog_test.go

@@ -0,0 +1,640 @@
+package remotelog_test
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"math/big"
+	"net"
+	"net/url"
+	"os"
+	"slices"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/remotelog"
+)
+
+// The tests run in a synctest bubble, where the time package runs on a
+// clock of the test's own, which starts at 2000-01-01T00:00:00Z: a wait
+// lasts exactly as long as it should, however slowly the test process
+// runs, and synctest.Wait returns once the sender has done all it can
+// before time passes. The endpoint is a listener on the loopback address.
+// A test reads from it only once the records are on their way, and checks
+// the sender's counts first, since a goroutine of the bubble that waits on
+// the network keeps that clock from moving on. For the same reason the
+// endpoint that refuses connections, a tlsEndpoint, runs outside the
+// bubble: a sender connecting over TLS waits on the endpoint's answer.
+
+const (
+	// started is the time a record made as a test starts gives.
+	started = "2000-01-01T00:00:00.000Z"
+	appName = "fsn1app1/gitea"
+	// local0 is the number of the default facility, and local0Info the
+	// priority of its records.
+	local0     = 16
+	local0Info = "<134>"
+	// loopback is where the endpoints listen.
+	loopback = "127.0.0.1:0"
+)
+
+func TestRecordsOverUDPGoOnePerDatagram(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		endpoint, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp", loopback)
+		if err != nil {
+			t.Fatalf("listen: %v", err)
+		}
+
+		t.Cleanup(func() { _ = endpoint.Close() })
+
+		sender, _, _ := run(t, params(remotelog.SchemeUDP, endpoint.LocalAddr()))
+
+		_, _ = sender.Write([]byte(`{"type":"request"}` + "\n" + `{"type":"process"}` + "\n"))
+
+		synctest.Wait()
+		wantCounts(t, sender, 2, 0, 0)
+
+		for _, line := range []string{`{"type":"request"}`, `{"type":"process"}`} {
+			datagram := make([]byte, 1024)
+
+			n, _, err := endpoint.ReadFrom(datagram)
+			if err != nil {
+				t.Fatalf("read: %v", err)
+			}
+
+			want := record(t, local0Info, appName, line)
+			if string(datagram[:n]) != want {
+				t.Errorf("datagram %q, want %q", datagram[:n], want)
+			}
+		}
+	})
+}
+
+func TestRecordsOverTCPAreOctetCountedWithTheirFacility(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		endpoint := listen(t)
+		endpointParams := params(remotelog.SchemeTCP, endpoint.Addr())
+		endpointParams.Facility = 19 // local3
+		endpointParams.AppName = "gitea"
+		sender, _, _ := run(t, endpointParams)
+
+		_, _ = sender.Write([]byte("first\nsecond\n"))
+
+		synctest.Wait()
+		wantCounts(t, sender, 2, 0, 0)
+
+		frames := bufio.NewReader(accept(t, endpoint))
+		wantFrame(t, frames, record(t, "<158>", "gitea", "first"))
+		wantFrame(t, frames, record(t, "<158>", "gitea", "second"))
+	})
+}
+
+func TestStalledEndpointHoldsUpNoWriteAndOldestRecordsAreDropped(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		certificate, roots := testCertificate(t)
+		endpoint := listen(t)
+		endpointParams := params(remotelog.SchemeTLS, endpoint.Addr())
+		endpointParams.RootCAs = roots
+		endpointParams.Buffer = 3
+		sender, _, _ := run(t, endpointParams)
+
+		// The sender connects, and its TLS handshake waits for an answer
+		// the endpoint does not give yet.
+		conn := accept(t, endpoint)
+
+		var stdout bytes.Buffer
+
+		out := io.MultiWriter(&stdout, sender)
+		for i := range 5 {
+			_, _ = fmt.Fprintf(out, "line %d\n", i+1)
+		}
+
+		if stdout.String() != "line 1\nline 2\nline 3\nline 4\nline 5\n" {
+			t.Errorf("stdout has %q", stdout.String())
+		}
+
+		wantCounts(t, sender, 0, 2, 3)
+
+		// Once the endpoint answers, the three newest records are sent.
+		server := tls.Server(conn, &tls.Config{
+			Certificates: []tls.Certificate{certificate},
+			MinVersion:   tls.VersionTLS12,
+		})
+
+		err := server.HandshakeContext(t.Context())
+		if err != nil {
+			t.Fatalf("handshake: %v", err)
+		}
+
+		synctest.Wait()
+		wantCounts(t, sender, 3, 2, 0)
+
+		frames := bufio.NewReader(server)
+		for _, line := range []string{"line 3", "line 4", "line 5"} {
+			wantFrame(t, frames, record(t, local0Info, appName, line))
+		}
+	})
+}
+
+func TestReconnectsWithBackoffAfterTheEndpointGoesAway(t *testing.T) {
+	t.Parallel()
+
+	certificate, roots := testCertificate(t)
+	endpoint := startTLSEndpoint(t, certificate)
+
+	synctest.Test(t, func(t *testing.T) {
+		endpointParams := params(remotelog.SchemeTLS, endpoint.addr)
+		endpointParams.RootCAs = roots
+		sender, logged, _ := run(t, endpointParams)
+
+		_, _ = sender.Write([]byte("one\n"))
+
+		synctest.Wait()
+		wantCounts(t, sender, 1, 0, 0)
+
+		conn := endpoint.next(t)
+		wantFrame(t, bufio.NewReader(conn), record(t, local0Info, appName, "one"))
+
+		// The endpoint goes away: it closes the connection, and refuses the
+		// next ones. The sender notices when a record fails, and tries to
+		// connect again a second later, then two seconds after that.
+		endpoint.refusing.Store(true)
+
+		_ = conn.Close()
+
+		writeUntilDropped(t, sender, 1)
+		sent := sender.Sent()
+
+		_, _ = sender.Write([]byte("two\n"))
+
+		time.Sleep(time.Second)
+		synctest.Wait()
+
+		endpoint.refusing.Store(false)
+
+		time.Sleep(2*time.Second - time.Nanosecond)
+		synctest.Wait()
+		wantCounts(t, sender, sent, 1, 1)
+
+		// The endpoint is back, and the record waiting is sent.
+		time.Sleep(time.Nanosecond)
+		synctest.Wait()
+		wantCounts(t, sender, sent+1, 1, 0)
+
+		conn = endpoint.next(t)
+		wantFrame(t, bufio.NewReader(conn), record(t, local0Info, appName, "two"))
+		wantRetries(t, logged, "1s", "2s")
+	})
+}
+
+func TestAConnectionClosedAtOnceIsMadeAgainAfterAGrowingDelay(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		endpoint := listen(t)
+		sender, logged, _ := run(t, params(remotelog.SchemeTCP, endpoint.Addr()))
+
+		// The endpoint closes each connection as soon as it takes it. The
+		// sender notices when a record fails, and connects again a second
+		// later, then two seconds after that, then four.
+		delays := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}
+		for i, delay := range delays {
+			_ = accept(t, endpoint).Close()
+
+			writeUntilDropped(t, sender, int64(i+1))
+			wantConnectedAgainAfter(t, sender, delay)
+		}
+
+		wantRetries(t, logged, "1s", "2s", "4s")
+	})
+}
+
+func TestTheDelayStartsAgainAfterAConnectionThatStayedUpAMinute(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		endpoint := listen(t)
+		sender, logged, _ := run(t, params(remotelog.SchemeTCP, endpoint.Addr()))
+
+		_ = accept(t, endpoint).Close()
+
+		writeUntilDropped(t, sender, 1)
+		wantConnectedAgainAfter(t, sender, time.Second)
+
+		// A connection that fails just short of a minute after it was made
+		// leaves the delay growing.
+		conn := accept(t, endpoint)
+
+		time.Sleep(time.Minute - time.Nanosecond)
+
+		_ = conn.Close()
+
+		writeUntilDropped(t, sender, 2)
+		wantConnectedAgainAfter(t, sender, 2*time.Second)
+
+		// One that fails a minute after it was made starts it again from a
+		// second.
+		conn = accept(t, endpoint)
+
+		time.Sleep(time.Minute)
+
+		_ = conn.Close()
+
+		writeUntilDropped(t, sender, 3)
+		wantConnectedAgainAfter(t, sender, time.Second)
+		wantRetries(t, logged, "1s", "2s", "1s")
+	})
+}
+
+func TestALineTooLongForADatagramIsDroppedAlone(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		endpoint, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp", loopback)
+		if err != nil {
+			t.Fatalf("listen: %v", err)
+		}
+
+		t.Cleanup(func() { _ = endpoint.Close() })
+
+		sender, logged, _ := run(t, params(remotelog.SchemeUDP, endpoint.LocalAddr()))
+
+		// With its header, the first line's record is longer than the 65507
+		// bytes a UDP datagram over IPv4 holds. It is dropped, nothing is
+		// logged, and the next line is sent at once.
+		_, _ = sender.Write([]byte(strings.Repeat("x", 65507) + "\nnext\n"))
+
+		synctest.Wait()
+		wantCounts(t, sender, 1, 1, 0)
+		wantRetries(t, logged)
+
+		datagram := make([]byte, 1024)
+
+		n, _, err := endpoint.ReadFrom(datagram)
+		if err != nil {
+			t.Fatalf("read: %v", err)
+		}
+
+		want := record(t, local0Info, appName, "next")
+		if string(datagram[:n]) != want {
+			t.Errorf("datagram %q, want %q", datagram[:n], want)
+		}
+	})
+}
+
+func TestRecordsWaitingAtTheStopAreSent(t *testing.T) {
+	t.Parallel()
+
+	certificate, roots := testCertificate(t)
+	endpoint := startTLSEndpoint(t, certificate)
+
+	synctest.Test(t, func(t *testing.T) {
+		// The endpoint refuses the sender's first connection: it fails to
+		// connect, and waits a second to try again.
+		endpoint.refusing.Store(true)
+
+		endpointParams := params(remotelog.SchemeTLS, endpoint.addr)
+		endpointParams.RootCAs = roots
+		sender, logged, stop := run(t, endpointParams)
+
+		synctest.Wait()
+		wantRetries(t, logged, "1s")
+
+		_, _ = sender.Write([]byte("one\ntwo\n"))
+
+		endpoint.refusing.Store(false)
+
+		// Stopped before that second is over, it connects to send them.
+		stop()
+		wantCounts(t, sender, 2, 0, 0)
+
+		frames := bufio.NewReader(endpoint.next(t))
+		wantFrame(t, frames, record(t, local0Info, appName, "one"))
+		wantFrame(t, frames, record(t, local0Info, appName, "two"))
+	})
+}
+
+// output collects what the sender logs.
+type output struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+// Write adds lines the sender logs.
+func (o *output) Write(p []byte) (int, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	return o.buf.Write(p)
+}
+
+// text returns everything logged so far.
+func (o *output) text() string {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	return o.buf.String()
+}
+
+// params returns the settings of a Sender for the endpoint at addr, in
+// the form scheme names: room for ten lines, the default facility, and
+// appName.
+func params(scheme string, addr net.Addr) remotelog.Params {
+	return remotelog.Params{
+		URL:      &url.URL{Scheme: scheme, Host: addr.String()},
+		Buffer:   10,
+		Facility: local0,
+		AppName:  appName,
+	}
+}
+
+// run runs a Sender with settings until the test ends or the function
+// it returns is called, which waits for Run to return. It returns the
+// Sender, and what it logs.
+func run(t *testing.T, settings remotelog.Params) (*remotelog.Sender, *output, func()) {
+	t.Helper()
+
+	sender := remotelog.New(settings)
+	logged := &output{}
+	ctx, cancel := context.WithCancel(t.Context())
+	ran := make(chan struct{})
+
+	go func() {
+		sender.Run(ctx, slog.New(slog.NewJSONHandler(logged, nil)))
+		close(ran)
+	}()
+
+	stop := func() {
+		cancel()
+		<-ran
+	}
+	t.Cleanup(stop)
+
+	return sender, logged, stop
+}
+
+// listen returns a TCP listener on the loopback address, closed when the
+// test ends.
+func listen(t *testing.T) net.Listener {
+	t.Helper()
+
+	listener, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", loopback)
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	t.Cleanup(func() { _ = listener.Close() })
+
+	return listener
+}
+
+// accept returns the next connection to listener, closed when the test
+// ends.
+func accept(t *testing.T, listener net.Listener) net.Conn {
+	t.Helper()
+
+	conn, err := listener.Accept()
+	if err != nil {
+		t.Fatalf("accept: %v", err)
+	}
+
+	t.Cleanup(func() { _ = conn.Close() })
+
+	return conn
+}
+
+// tlsEndpoint is a syslog+tls endpoint on the loopback address, which a
+// test starts outside its bubble. It keeps its listener until the test
+// ends, and either takes each connection or refuses it.
+type tlsEndpoint struct {
+	addr net.Addr
+	// refusing is set while the endpoint closes each connection before the
+	// TLS handshake, which fails the sender's attempt to connect.
+	refusing atomic.Bool
+	// conns are the connections it has taken, after the handshake.
+	conns chan net.Conn
+}
+
+// startTLSEndpoint starts a tlsEndpoint with certificate, which takes
+// connections until it is told to refuse them.
+func startTLSEndpoint(t *testing.T, certificate tls.Certificate) *tlsEndpoint {
+	t.Helper()
+
+	listener := listen(t)
+	endpoint := &tlsEndpoint{addr: listener.Addr(), conns: make(chan net.Conn, 10)}
+	config := &tls.Config{
+		Certificates: []tls.Certificate{certificate},
+		MinVersion:   tls.VersionTLS12,
+	}
+
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				return
+			}
+
+			server := tls.Server(conn, config)
+			if endpoint.refusing.Load() || server.HandshakeContext(t.Context()) != nil {
+				_ = conn.Close()
+
+				continue
+			}
+
+			endpoint.conns <- server
+		}
+	}()
+
+	return endpoint
+}
+
+// next returns the next connection the endpoint has taken, closed when
+// the test ends.
+func (e *tlsEndpoint) next(t *testing.T) net.Conn {
+	t.Helper()
+
+	conn := <-e.conns
+
+	t.Cleanup(func() { _ = conn.Close() })
+
+	return conn
+}
+
+// record returns the record of line made as the test started, with the
+// priority and the app name given.
+func record(t *testing.T, priority, app, line string) string {
+	t.Helper()
+
+	hostname, err := os.Hostname()
+	if err != nil || hostname == "" {
+		hostname = "-"
+	}
+
+	return priority + "1 " + started + " " + hostname + " " + app + " - - - " + line
+}
+
+// wantFrame reads the next octet-counted frame from frames, and checks
+// that it holds want.
+func wantFrame(t *testing.T, frames *bufio.Reader, want string) {
+	t.Helper()
+
+	count, err := frames.ReadString(' ')
+	if err != nil {
+		t.Fatalf("read a frame's length: %v", err)
+	}
+
+	length, err := strconv.Atoi(strings.TrimSuffix(count, " "))
+	if err != nil {
+		t.Fatalf("frame starts %q, not with its length", count)
+	}
+
+	got := make([]byte, length)
+
+	_, err = io.ReadFull(frames, got)
+	if err != nil {
+		t.Fatalf("read a frame: %v", err)
+	}
+
+	if string(got) != want {
+		t.Errorf("frame %q, want %q", got, want)
+	}
+}
+
+// wantCounts checks the records sender has sent, dropped and holds in
+// its buffer.
+func wantCounts(
+	t *testing.T, sender *remotelog.Sender, sent, dropped int64, depth int,
+) {
+	t.Helper()
+
+	if sender.Sent() != sent || sender.Dropped() != dropped || sender.Depth() != depth {
+		t.Fatalf("sent %d, dropped %d, %d in the buffer; want %d, %d and %d",
+			sender.Sent(), sender.Dropped(), sender.Depth(), sent, dropped, depth)
+	}
+}
+
+// writeUntilDropped writes a line at a time until the count of records
+// sender has dropped reaches dropped. The records it sends on a
+// connection the endpoint has closed are lost before one fails; how many
+// depends on when the endpoint's host answers that the connection is
+// gone.
+func writeUntilDropped(t *testing.T, sender *remotelog.Sender, dropped int64) {
+	t.Helper()
+
+	for sender.Dropped() < dropped {
+		_, _ = sender.Write([]byte("lost\n"))
+
+		synctest.Wait()
+	}
+}
+
+// wantConnectedAgainAfter writes a line while the sender waits to connect
+// again, and checks that it connects, and takes the line from the buffer,
+// only once delay is over.
+func wantConnectedAgainAfter(
+	t *testing.T, sender *remotelog.Sender, delay time.Duration,
+) {
+	t.Helper()
+
+	_, _ = sender.Write([]byte("waiting\n"))
+
+	time.Sleep(delay - time.Nanosecond)
+	synctest.Wait()
+
+	if sender.Depth() != 1 {
+		t.Fatalf("connected again before %v", delay)
+	}
+
+	time.Sleep(time.Nanosecond)
+	synctest.Wait()
+
+	if sender.Depth() != 0 {
+		t.Fatalf("not connected again after %v", delay)
+	}
+}
+
+// wantRetries checks that the sender logged a failure, of an attempt to
+// connect or of a connection, for each of delays, the time until the next
+// attempt, in order, and logged nothing else.
+func wantRetries(t *testing.T, logged *output, delays ...string) {
+	t.Helper()
+
+	var got []string
+
+	for line := range strings.Lines(logged.text()) {
+		var fields map[string]any
+
+		err := json.Unmarshal([]byte(line), &fields)
+		if err != nil || fields["msg"] != "sending to SWWAF_LOG_REMOTE_URL failed" {
+			t.Fatalf("logged %q", line)
+		}
+
+		delay, _ := fields["connecting_again_in"].(string)
+		got = append(got, delay)
+	}
+
+	if !slices.Equal(got, delays) {
+		t.Errorf("logged failures to connect again in %v, want %v", got, delays)
+	}
+}
+
+// testCertificate returns a certificate for 127.0.0.1 that is its own
+// CA, and a pool that holds it. It is valid on the bubble's clock, which
+// starts at 2000-01-01T00:00:00Z.
+func testCertificate(t *testing.T) (tls.Certificate, *x509.CertPool) {
+	t.Helper()
+
+	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatalf("generate a key: %v", err)
+	}
+
+	template := &x509.Certificate{
+		SerialNumber:          big.NewInt(1),
+		Subject:               pkix.Name{CommonName: "smallwebwaf test CA"},
+		NotBefore:             time.Date(1999, 12, 31, 0, 0, 0, 0, time.UTC),
+		NotAfter:              time.Date(2000, 1, 2, 0, 0, 0, 0, time.UTC),
+		IsCA:                  true,
+		BasicConstraintsValid: true,
+		KeyUsage:              x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+		IPAddresses:           []net.IP{net.IPv4(127, 0, 0, 1)},
+	}
+
+	der, err := x509.CreateCertificate(rand.Reader, template, template,
+		&key.PublicKey, key)
+	if err != nil {
+		t.Fatalf("create a certificate: %v", err)
+	}
+
+	certificate, err := x509.ParseCertificate(der)
+	if err != nil {
+		t.Fatalf("parse the certificate: %v", err)
+	}
+
+	roots := x509.NewCertPool()
+	roots.AddCert(certificate)
+
+	return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}, roots
+}

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) {