From ffb40d78439fa7b6aae3a59ed7a136802c9943f2 Mon Sep 17 00:00:00 2001 From: Laurence Withers Date: Sun, 18 Jan 2026 12:48:32 +0000 Subject: [PATCH 1/5] Add GetGlobalConn --- global.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/global.go b/global.go index e9e45c8..0434077 100644 --- a/global.go +++ b/global.go @@ -26,7 +26,7 @@ const ( // error otherwise. It will attempt to establish a connection to the journal // if there is not already one in progress. func CheckConnection() error { - _, err := getGlobal() + _, err := GetGlobalConn() return err } @@ -39,7 +39,12 @@ func SetGlobalConn(c *Conn) { global = c } -func getGlobal() (*Conn, error) { +// GetGlobalConn returns the global (default) connection to the journal. It may +// return an error if no connection could be established. If no connection is +// yet established, it will attempt to establish one. This is rate limited to +// avoid excessive connection attempts should there be a persistent problem +// connecting to the journal socket. +func GetGlobalConn() (*Conn, error) { globalLock.Lock() defer globalLock.Unlock() @@ -119,7 +124,7 @@ func SetGlobalErrHandler(errHandler func(err error)) { // Note: to avoid allocation / garbage, ensure attrs has capacity for an extra // 2+len(GlobalCommonAttr) values. func Entry(pri Priority, msg string, attrs []Attr) { - c, _ := getGlobal() + c, _ := GetGlobalConn() if c == nil { return } From 8f364a74df89ca48c6c94e658b2db1d2ad801f9c Mon Sep 17 00:00:00 2001 From: Laurence Withers Date: Sun, 18 Jan 2026 12:48:46 +0000 Subject: [PATCH 2/5] Add Conn.Close --- conn.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/conn.go b/conn.go index eb43b00..2c43f0c 100644 --- a/conn.go +++ b/conn.go @@ -101,3 +101,9 @@ func (c *Conn) WriteAttrs(attrs []Attr) error { c.bufs.Put(buf) return err } + +// Close the underlying connection. After this, all attempts to write onto the +// connection will return an error. +func (c *Conn) Close() error { + return c.s.Close() +} From ecd852eedeedfc4c8ad8dd306c14b25611b7389c Mon Sep 17 00:00:00 2001 From: Laurence Withers Date: Sun, 18 Jan 2026 12:49:25 +0000 Subject: [PATCH 3/5] Add testsink package, and unit tests for Conn / entries --- conn_test.go | 277 +++++++++++++++++-------------------------- testsink/message.go | 126 ++++++++++++++++++++ testsink/testsink.go | 82 +++++++++++++ 3 files changed, 315 insertions(+), 170 deletions(-) create mode 100644 testsink/message.go create mode 100644 testsink/testsink.go diff --git a/conn_test.go b/conn_test.go index 6c89a6a..b5f5a63 100644 --- a/conn_test.go +++ b/conn_test.go @@ -1,187 +1,124 @@ package journal import ( - "bytes" - "encoding/binary" - "encoding/hex" - "errors" - "fmt" - "io" - "net" "path/filepath" - "strconv" "testing" + + "src.lwithers.me.uk/go/journal/testsink" ) -type testingCommon interface { - TempDir() string - Fatalf(string, ...any) -} - -// testConnector spawns a Conn with a local Unix datagram socket that checks -// incoming datagrams for well-formedness but otherwise discards them. -func testConnector(t testingCommon) *Conn { - sockPath := filepath.Join(t.TempDir(), "test-socket") - sock, err := net.ListenUnixgram("unixgram", &net.UnixAddr{ - Name: sockPath, - Net: "unixgram", - }) +// TestConn opens a connection to a test sink and writes one message, +// ensuring it is received. +func TestConn(t *testing.T) { + sockpath := filepath.Join(t.TempDir(), "socket") + sink, err := testsink.New(sockpath) if err != nil { - t.Fatalf("testConnector: ListenUnix: %v", err) - } - - go func() { - buf := make([]byte, 1<<16 /*enough for our tests */) - for { - n, err := sock.Read(buf) - switch err { - case nil: - case io.EOF: - return - default: - t.Fatalf("testConnector: Read: %v", err) - } - ok, pos, desc := checkWellFormedProto(buf[:n]) - if !ok { - t.Fatalf("received malformed data at pos 0x%x: %s\n%s", pos, desc, hex.Dump(buf[:n])) - } - } - }() - - conn, err := Connect(sockPath) - if err != nil { - t.Fatalf("testConnector: DialUnix: %v", err) - } - conn.ErrHandler = func(err error) { - t.Fatalf("testConnector: connection error: %v", err) - } - return conn -} - -func checkWellFormedProto(buf []byte) (ok bool, pos int, desc string) { - for pos < len(buf) { - // grab attribute name up to '=' or '\n' - off := bytes.IndexAny(buf[pos:], "=\n") - if off == -1 { - return false, pos, "unterminated key" - } - key := string(buf[pos : pos+off]) - if err := AttrKeyValid(key); err != nil { - return false, pos, err.Error() - } - pos += off - - // for KEY=VALUE, the value is terminated by a newline - if buf[pos] == '=' { - pos++ - off = bytes.IndexByte(buf[pos:], '\n') - if off == -1 { - return false, pos, "unterminated value" - } - pos += off // consume value - pos++ // consume trailing newline - continue - } - - // otherwise, expect an 8-bit little-endian length - pos++ // consume newline after key - if pos+8 > len(buf) { - return false, pos, "value length too short" - } - vlen := binary.LittleEndian.Uint64(buf[pos:]) - pos += 8 - if vlen > uint64(len(buf)) /* protect against overflow */ || - uint64(pos)+vlen+1 > uint64(len(buf)) { - return false, pos, "value length too long" - } - pos += int(vlen) - if buf[pos] != '\n' { - return false, pos, "value not terminated by newline" - } - pos++ - } - return true, pos, "" -} - -// TestConcurrentEntries is best run with the race detector, and tries to pick -// up any faults that might occur when concurrent goroutines write into the same -// Conn. -func TestConcurrentEntries(t *testing.T) { - c := testConnector(t) - const ( - numGoroutines = 16 - numEntries = 100 - ) - - // attributes which will be common to all EntryErr calls - attr := make([]Attr, 0, 10 /* enough capacity to avoid realloc on append; might trigger data races */) - attr = append(attr, Attr{ - Key: AttrKey{ - key: "HELLO", - }, - Value: []byte("world"), - }) - - // spawn goroutines - start := make(chan struct{}) - result := make(chan error, numGoroutines) - for i := range numGoroutines { - go func() { - var err error - <-start - for j := range numEntries { - err = c.EntryErr(PriDebug, "message "+strconv.Itoa(i)+"."+strconv.Itoa(j), attr) - if err != nil { - err = fmt.Errorf("message %d.%d error: %w", i, j, err) - break - } - } - result <- err - }() - } - - // try to get all the goroutines to start at roughly the same time - close(start) - - // collect results - var errs []error - for range numGoroutines { - if err := <-result; err != nil { - errs = append(errs, err) - } - } - if err := errors.Join(errs...); err != nil { t.Fatal(err) } -} + defer sink.Stop() -// BenchmarkEntry is a benchmark for the common Entry function. -func BenchmarkEntry(b *testing.B) { - c := testConnector(b) + conn, err := Connect(sockpath) + if err != nil { + t.Fatal(err) + } + defer conn.Close() - // add some common attributes - c.Common = make([]Attr, 0, 10) - c.Common = append(c.Common, Attr{ - Key: AttrKey{ - key: "COMMON_ATTR", - }, - Value: []byte("abc123\n"), - }) + conn.Entry(PriInfo, "hello, world", nil) - // attributes which will be common to all EntryErr calls - attr := make([]Attr, 0, 10) - attr = append(attr, Attr{ - Key: AttrKey{ - key: "HELLO", - }, - Value: []byte("world"), - }) + msg, err := sink.Message(0) + if err != nil { + t.Fatal(err) + } - b.ResetTimer() - for i := 0; i < b.N; i++ { - err := c.EntryErr(PriDebug, "hello world!", attr) - if err != nil { - b.Fatalf("message %d: error %v", i, err) + msgText, attrs, err := msg.Decode() + if err != nil { + t.Error(err) + } + if msgText != "hello, world" { + t.Errorf("unexpected message text %q", msgText) + } + val, ok := testsink.GetAttr(attrs, "PRIORITY") + switch { + case !ok: + t.Error("missing PRIORITY attribute") + case val != "6": + t.Error("unexpected PRIORITY value") + } + + if t.Failed() { + for i := range attrs { + t.Errorf("attr %q=%q", attrs[i].Key, attrs[i].Val) + } + } +} + +// TestEntryBinary ensures that we can write a message with an attribute encoded +// as a binary field. +func TestEntryBinary(t *testing.T) { + sockpath := filepath.Join(t.TempDir(), "socket") + sink, err := testsink.New(sockpath) + if err != nil { + t.Fatal(err) + } + defer sink.Stop() + + conn, err := Connect(sockpath) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + expAttrs := []Attr{ + Attr{ + Key: MustAttrKey("SHORT"), + Value: []byte("short val"), + }, + Attr{ + Key: MustAttrKey("BINARY"), + Value: []byte("string with\n=embedded newline\nrequires binary protocol\n"), + }, + Attr{ + Key: MustAttrKey("LAST"), + Value: []byte("last\n"), + }, + } + conn.Entry(PriDebug, "hello, binary world", expAttrs) + + msg, err := sink.Message(0) + if err != nil { + t.Fatal(err) + } + + msgText, attrs, err := msg.Decode() + if err != nil { + t.Error(err) + } + if msgText != "hello, binary world" { + t.Errorf("unexpected message text %q", msgText) + } + + val, ok := testsink.GetAttr(attrs, "PRIORITY") + switch { + case !ok: + t.Error("missing PRIORITY attribute") + case val != "7": + t.Error("unexpected PRIORITY value") + } + + for i := range expAttrs { + key, expVal := expAttrs[i].Key.Key(), string(expAttrs[i].Value) + val, ok = testsink.GetAttr(attrs, key) + switch { + case !ok: + t.Errorf("missing %s attribute", key) + case val != expVal: + t.Errorf("unexpected %s value", key) + } + } + + if t.Failed() { + for i := range attrs { + t.Errorf("attr %q=%q", attrs[i].Key, attrs[i].Val) } } } diff --git a/testsink/message.go b/testsink/message.go new file mode 100644 index 0000000..99cae43 --- /dev/null +++ b/testsink/message.go @@ -0,0 +1,126 @@ +/* +Package testsink provides a partial implementation of the systemd-journald +Unix socket. Datagrams received on its socket are decoded and stored for unit +tests to examine. +*/ +package testsink + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" +) + +// Message returns the Nth message received. It waits for the message to arrive. +func (sink *Sink) Message(N int) (Message, error) { + sink.lock.Lock() + defer sink.lock.Unlock() + + for len(sink.msgs) <= N { + sink.mcond.Wait() + if sink.err != nil { + return Message{}, sink.err + } + } + return sink.msgs[N], nil +} + +// Message is recorded for each datagram received. +type Message struct { + Raw []byte +} + +type DecodedAttr struct { + Key, Val string +} + +func (m *Message) Decode() (msg string, attr []DecodedAttr, err error) { + raw := m.Raw + var errs []error + +DecodeLoop: + for len(raw) > 0 { + n, key := decodeAttrKey(raw) + raw = raw[n:] + switch { + case len(raw) == 0: + errs = append(errs, fmt.Errorf("unterminated attribute name %q", key)) + break DecodeLoop + case key == "": + errs = append(errs, errors.New("empty attribute name")) + } + + var val string + switch raw[0] { + case '=': + n, val = decodeAttrValText(raw[1:]) + case '\n': + var err error + n, val, err = decodeAttrValLen(raw[1:]) + if err != nil { + errs = append(errs, err) + } + } + raw = raw[1+n:] + if len(raw) == 0 { + errs = append(errs, fmt.Errorf("unterminated value for attribute %q", key)) + break DecodeLoop + } + + if raw[0] != '\n' { + errs = append(errs, errors.New("incorrectly terminated attribute value")) + } + raw = raw[1:] + + switch key { + case "MESSAGE": + msg = val + default: + attr = append(attr, DecodedAttr{Key: key, Val: val}) + } + } + + return msg, attr, errors.Join(errs...) +} + +func decodeAttrKey(raw []byte) (n int, key string) { + for i := range raw { + switch raw[i] { + case '\n', '=': + return i, string(raw[:i]) + } + } + return len(raw), string(raw) +} + +func decodeAttrValText(raw []byte) (n int, val string) { + term := bytes.IndexByte(raw, '\n') + if term == -1 { + term = len(raw) + } + return term, string(raw[:term]) +} + +func decodeAttrValLen(raw []byte) (n int, val string, err error) { + if len(raw) < 8 { + return len(raw), "", errors.New("not enough bytes for binary attribute value length") + } + amt := binary.LittleEndian.Uint64(raw) + raw = raw[8:] + if uint64(len(raw)) < amt { + return 8 + len(raw), string(raw), errors.New("not enough bytes for binary attribute value") + } + return int(8 + amt), string(raw[:amt]), nil +} + +// GetAttr returns the value of the attribute whose key name matches, and a +// boolean to indicate if it found a match. +func GetAttr(attr []DecodedAttr, key string) (value string, ok bool) { + for i := range attr { + if attr[i].Key == key { + return attr[i].Val, true + } + } + return "", false +} diff --git a/testsink/testsink.go b/testsink/testsink.go new file mode 100644 index 0000000..427a775 --- /dev/null +++ b/testsink/testsink.go @@ -0,0 +1,82 @@ +/* +Package testsink provides a partial implementation of the systemd-journald +Unix socket. Datagrams received on its socket are decoded and stored for unit +tests to examine. +*/ +package testsink + +import ( + "net" + "slices" + "sync" +) + +// Sink provides a Unix socket and captures messages sent to it using the +// systemd-journald wire protocol. +type Sink struct { + sock *net.UnixConn + stop chan struct{} + + lock sync.Mutex + mcond *sync.Cond + msgs []Message + err error +} + +// New returns a new Sink that is listening on the given path. +func New(sockpath string) (*Sink, error) { + sock, err := net.ListenUnixgram("unixgram", &net.UnixAddr{ + Name: sockpath, + Net: "unixgram", + }) + if err != nil { + return nil, err + } + sink := &Sink{ + sock: sock, + stop: make(chan struct{}, 1), + } + sink.mcond = sync.NewCond(&sink.lock) + go sink.stopper() + go sink.recv() + return sink, nil +} + +// Stop listening and close the socket. +func (sink *Sink) Stop() { + // non-blocking write onto channel; we only need to read from it once in + // order to stop the receiver, but using this rather than close ensures + // Stop() can be called multiple times without negative side effects + select { + case sink.stop <- struct{}{}: + default: + } +} + +func (sink *Sink) stopper() { + <-sink.stop + sink.sock.Close() +} + +func (sink *Sink) recv() { + buf := make([]byte, 131072) + for { + n, err := sink.sock.Read(buf) + + sink.lock.Lock() + if n > 0 { + sink.msgs = append(sink.msgs, Message{ + Raw: slices.Clone(buf[:n]), + }) + } + if err != nil { + sink.err = err + } + sink.lock.Unlock() + + sink.mcond.Broadcast() + if err != nil { + return + } + } +} From df6107a67e75b26fdc6c9ab911c6e4318eab90e0 Mon Sep 17 00:00:00 2001 From: Laurence Withers Date: Sun, 18 Jan 2026 13:47:50 +0000 Subject: [PATCH 4/5] Added log/slog adapter package --- README.md | 32 +++- journslog/handler_option.go | 83 ++++++++++ journslog/journslog.go | 293 ++++++++++++++++++++++++++++++++++++ journslog/journslog_test.go | 84 +++++++++++ journslog/key_map.go | 97 ++++++++++++ journslog/level_map.go | 57 +++++++ journslog/pools.go | 40 +++++ 7 files changed, 685 insertions(+), 1 deletion(-) create mode 100644 journslog/handler_option.go create mode 100644 journslog/journslog.go create mode 100644 journslog/journslog_test.go create mode 100644 journslog/key_map.go create mode 100644 journslog/level_map.go create mode 100644 journslog/pools.go diff --git a/README.md b/README.md index f3311fa..057c8e3 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,33 @@ # journal -Native Go output to the systemd journal socket. \ No newline at end of file +Native Go output to the [systemd journal socket](https://systemd.io/JOURNAL_NATIVE_PROTOCOL/). +See package `journslog` for a [`log/slog`](https://pkg.go.dev/log/slog) adapter. + +Quick start: + +```go +import "src.lwithers.me.uk/go/journal" + +func main() { + journal.Entry(journal.PriInfo, "hello, world", nil) +} +``` + +The `log/slog` adapter: + +```go +import ( + "log/slog" + + "src.lwithers.me.uk/go/journal/journslog" +) + +func main() { + h, err := journslog.NewHandler() + if err != nil { + panic(err) + } + slog.SetDefault(slog.New(h)) + slog.Info("hello, world", "attr", "value") +} +``` diff --git a/journslog/handler_option.go b/journslog/handler_option.go new file mode 100644 index 0000000..f52ec05 --- /dev/null +++ b/journslog/handler_option.go @@ -0,0 +1,83 @@ +package journslog + +import ( + "context" + "log/slog" + "time" + + "src.lwithers.me.uk/go/journal" +) + +// HandlerOption is passed to [NewHandler] to control the behaviour of the +// newly-created [JournalHandler]. +type HandlerOption func(jh *JournalHandler) + +// WithHandlerOptions applies an [slog.HandlerOptions] to a newly-created +// [JournalHandler]. +// +// If AddSource is set, then the widley-recognised journal attributes +// CODE_FILE, CODE_LINE and CODE_FUNC will be added to each journal entry. +func WithHandlerOptions(opts slog.HandlerOptions) HandlerOption { + return func(jh *JournalHandler) { + jh.addSource = opts.AddSource + if opts.Level == nil { + jh.minLevel = slog.LevelInfo + } else { + jh.minLevel = opts.Level + } + jh.replaceAttr = opts.ReplaceAttr + } +} + +// WithLevelMapper allows control over how [slog.Level] values are mapped to +// [journal.Priority] values. +func WithLevelMapper(levelMapper LevelMapper) HandlerOption { + return func(jh *JournalHandler) { + jh.levelMapper = levelMapper + } +} + +// WithKeyMapper allows control over how attribute string keys are mapped to +// journal keys. +func WithKeyMapper(keyMapper KeyMapper) HandlerOption { + return func(jh *JournalHandler) { + jh.keyMapper = keyMapper + } +} + +// WithConn uses the given [journal.Conn] connection object, rather than the +// global default. +func WithConn(conn *journal.Conn) HandlerOption { + return func(jh *JournalHandler) { + jh.conn = conn + } +} + +// WithLevelFromContext supplies a function which can optionally extract a +// more-specific minimum log level to use for a given message from a +// [context.Context]. If the supplied levelFromContext returns true in its +// second argument, then its first argument is used as the minimum level. +// Otherwise, the handler's default minimum level will be used. +func WithLevelFromContext(levelFromContext func(context.Context) (slog.Level, bool)) HandlerOption { + return func(jh *JournalHandler) { + jh.levelFromContext = levelFromContext + } +} + +// WithAttrsFromContext supplies a function which can optionally extract +// attributes from a given [context.Context] value. It may return nil or a +// zero-length slice. +func WithAttrsFromContext(attrsFromContext func(context.Context) []slog.Attr) HandlerOption { + return func(jh *JournalHandler) { + jh.attrsFromContext = attrsFromContext + } +} + +// WithAttrsFromTime supplies a function which can optionally map the Time +// field from an [slog.Record] onto zero or more attributes. The supplied +// function is never called with a zero [time.Time]. +func WithAttrsFromTime(attrsFromTime func(time.Time) []slog.Attr) HandlerOption { + return func(jh *JournalHandler) { + jh.attrsFromTime = attrsFromTime + } +} diff --git a/journslog/journslog.go b/journslog/journslog.go new file mode 100644 index 0000000..9185489 --- /dev/null +++ b/journslog/journslog.go @@ -0,0 +1,293 @@ +/* +Package journslog implements an [slog.Handler] which writes output to the +[systemd journal]. + +[systemd journal]: https://www.freedesktop.org/software/systemd/man/latest/systemd-journald.service.html +*/ +package journslog + +import ( + "context" + "log/slog" + "runtime" + "slices" + "strconv" + "strings" + "time" + + "src.lwithers.me.uk/go/journal" +) + +// JournalHandler is an [slog.Handler] which writes output to the +// [systemd journal]. +// +// Note that the handler discards the Time field from [slog.Record] by default, +// as there is no standardised way to override the timestamp recorded by the +// systemd-journal upon reception of a message. It is possible to emit a field +// using [WithAttrsFromTime]. +// +// [systemd journal]: https://www.freedesktop.org/software/systemd/man/latest/systemd-journald.service.html +type JournalHandler struct { + // addSource causes CODE_FILE, CODE_LINE and CODE_FUNC attrs to be added + // to each entry. + addSource bool + + // minLevel is the minimum level of messages we emit. Lower levels are + // not emitted. Note it can be overridden by levelFromContext. + minLevel slog.Leveler + + // replaceAttr is an optional function (may be nil) which the user can + // use to map (resolved) [slog.Attr] values into other values, including + // empty to skip. + replaceAttr func(groups []string, a slog.Attr) slog.Attr + + // levelMapper turns [slog.Level] values into [journal.Priority] values. + levelMapper LevelMapper + + // keyMapper turns [slog.Attr] key strings into [journal.AttrKey] values. + keyMapper KeyMapper + + // conn is the underlying journal connection to use. + conn *journal.Conn + + // levelFromContext allows overriding the minimum log level for a logger + // based on the [context.Context] value. It may be nil. + levelFromContext func(context.Context) (slog.Level, bool) + + // attrsFromContext may supply additional attributes to the logger from + // the [context.Context] value. It may be nil. + attrsFromContext func(context.Context) []slog.Attr + + // attrsFromTime may return zero or more attributes to represent the + // Time from an [slog.Record]. It may be nil. + attrsFromTime func(time.Time) []slog.Attr + + // groups are the ordered group names added by WithGroup. We need to + // maintain the full list, not just a pre-rendered prefix string, + // because of the API of replaceAttr. + groups []string + + // group is the pre-rendered prefix string corresponding to groups. + group groupPrefix + + // jattrs are the pre-rendered attributes added by WithAttrs. + jattrs []journal.Attr +} + +// NewHandler returns a new [JournalHandler]. It returns an error if a +// connection to the journal socket could not be established. +func NewHandler(opts ...HandlerOption) (*JournalHandler, error) { + jh := &JournalHandler{ + minLevel: slog.LevelInfo, + levelMapper: DefaultLevelMapper(), + keyMapper: DefaultKeyMapper(), + } + for _, opt := range opts { + opt(jh) + } + + // For fields which require allocation of resources, initialise them + // after processing options in case the user already supplied a non-nil + // value. In particular this means we don't cause the global connection + // to the journal to be opened if the user supplied something more + // specific. + if jh.levelMapper == nil { + jh.levelMapper = DefaultLevelMapper() + } + if jh.keyMapper == nil { + jh.keyMapper = DefaultKeyMapper() + } + if jh.conn == nil { + var err error + if jh.conn, err = journal.GetGlobalConn(); err != nil { + return nil, err + } + } + + return jh, nil +} + +// WithAttrs returns an [slog.Handler] containing some fixed attributes, which +// will be emitted on all future calls to Handle. The attribute values are +// resolved within this function call. +func (jh *JournalHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + if len(attrs) == 0 { + return jh + } + + jattrs := make([]journal.Attr, 0, len(attrs)) + for _, attr := range attrs { + jattrs = jh.renderAttribute(jattrs, jh.groups, jh.group, attr) + } + + if len(jattrs) == 0 { + return jh + } + + nh := new(JournalHandler) + *nh = *jh + nh.jattrs = slices.Concat(jh.jattrs, jattrs) + return nh +} + +// WithGroup returns an [slog.Handler] where are future attributes are assigned +// to the given group. +func (jh *JournalHandler) WithGroup(group string) slog.Handler { + nh := new(JournalHandler) + *nh = *jh + nh.groups = append(slices.Clone(jh.groups), group) + nh.group = jh.group.Append(group) + return nh +} + +// Enabled returns true if the given log level is enabled, false otherwise. +func (jh *JournalHandler) Enabled(ctx context.Context, level slog.Level) bool { + var ( + minLevel slog.Level + ok bool + ) + // avoid call to jh.minLevel.Level() if we have a more specific value + // from the context + if jh.levelFromContext != nil { + minLevel, ok = jh.levelFromContext(ctx) + } + if !ok { + minLevel = jh.minLevel.Level() + } + return level >= minLevel +} + +// Handle a log record, writing it to the journal similar to [journal.Entry]. +func (jh *JournalHandler) Handle(ctx context.Context, rec slog.Record) error { + pri := jh.levelMapper.LevelMap(rec.Level) + + var addAttrs []slog.Attr + if jh.attrsFromTime != nil && !rec.Time.IsZero() { + addAttrs = attrPoolGet() + defer attrPoolPut(addAttrs) + addAttrs = append(addAttrs, jh.attrsFromTime(rec.Time)...) + } + if jh.attrsFromContext != nil { + if addAttrs == nil { + addAttrs = attrPoolGet() + defer attrPoolPut(addAttrs) + } + addAttrs = append(addAttrs, jh.attrsFromContext(ctx)...) + } + + // 2 for journal.Conn.Entry (MESSAGE and PRIORITY), + // 3 for SRC_* (optional but we always allocate space) + numAttrsEstimate := 2 + len(jh.conn.Common) + 3 + len(addAttrs) + len(jh.jattrs) + rec.NumAttrs() + jattrs := slices.Grow(jattrPoolGet(), numAttrsEstimate) + defer jattrPoolPut(jattrs) + + // add SRC_* if requested + if jh.addSource && rec.PC != 0 { + jattrs = append(jattrs, jh.sourceAttrs(rec.PC)...) + } + + // add time / context attributes, if any + for i := range addAttrs { + jattrs = jh.renderAttribute(jattrs, nil, "", addAttrs[i]) + } + + // add pre-rendered attributes from WithAttrs + jattrs = append(jattrs, jh.jattrs...) + + // map [slog.Attr] values to [journal.Attr] values + rec.Attrs(func(a slog.Attr) bool { + jattrs = jh.renderAttribute(jattrs, jh.groups, jh.group, a) + return true + }) + + return jh.conn.EntryErr(pri, rec.Message, jattrs) +} + +// groupPrefix enables efficient building of attribute keys prepended by group +// names, separated by '.'. Invariant: always n × (group-name '.'), n ≥ 0. +type groupPrefix string + +func (gp groupPrefix) Prepend(key string) string { + if gp == "" { // avoid allocation in frequent path + return key + } + return string(gp) + key +} + +func (gp groupPrefix) Append(group string) groupPrefix { + var b strings.Builder + b.WriteString(string(gp)) + b.WriteString(group) + b.WriteRune('.') + return groupPrefix(b.String()) +} + +// renderAttribute turns an [slog.Attr] into some number of resolved, +// fully-qualified [journal.Attr] values. It calls replaceAttr if defined. +// New attributes are appended to the slice jattrs, and the new slice header +// is returned. +// +// If the attribute (after replacement) is a zero [slog.Attr] value, or it is +// of kind [slog.KindGroup] with zero child elements, no change is made to the +// jattrs slice. +// +// If the attribute (after replacement) is of kind [slog.Group], the function +// recurses, potentially adding multiple elements to the jattrs slice. +// list:[]journal.Attr{...}}. +// +// Otherwise, the function appends a single element to the jattrs slice. +func (jh *JournalHandler) renderAttribute(jattrs []journal.Attr, groups []string, gp groupPrefix, attr slog.Attr) []journal.Attr { + val := attr.Value.Resolve() + + if jh.replaceAttr != nil && val.Kind() != slog.KindGroup { + attr = jh.replaceAttr(groups, attr) + val = attr.Value.Resolve() + } + + if attr.Equal(slog.Attr{}) { + return jattrs + } + + // handle slog.KindGroup by recursing + if val.Kind() == slog.KindGroup { + groups = append(groups, attr.Key) + gp = gp.Append(attr.Key) + for _, attr := range val.Group() { + jattrs = jh.renderAttribute(jattrs, groups, gp, attr) + } + return jattrs + } + + key := gp.Prepend(attr.Key) + return append(jattrs, journal.Attr{ + Key: jh.keyMapper.KeyMap(key), + Value: []byte(val.String()), + }) +} + +var ( + srcAttrCodeFile = journal.MustAttrKey("CODE_FILE") + srcAttrCodeLine = journal.MustAttrKey("CODE_LINE") + srcAttrCodeFunc = journal.MustAttrKey("CODE_FUNC") +) + +// sourceAttrs returns the attributes we emit for source location. +func (jh *JournalHandler) sourceAttrs(PC uintptr) []journal.Attr { + fs := runtime.CallersFrames([]uintptr{PC}) + f, _ := fs.Next() + + return []journal.Attr{ + journal.Attr{ + Key: srcAttrCodeFile, + Value: []byte(f.File), + }, + journal.Attr{ + Key: srcAttrCodeLine, + Value: []byte(strconv.Itoa(f.Line)), + }, + journal.Attr{ + Key: srcAttrCodeFunc, + Value: []byte(f.Function), + }, + } +} diff --git a/journslog/journslog_test.go b/journslog/journslog_test.go new file mode 100644 index 0000000..53129b8 --- /dev/null +++ b/journslog/journslog_test.go @@ -0,0 +1,84 @@ +package journslog + +import ( + "log/slog" + "net/http" + "path/filepath" + "testing" + + "src.lwithers.me.uk/go/journal" + "src.lwithers.me.uk/go/journal/testsink" +) + +// TestHandler emits a log message with various attributes / groups and ensures +// the result is logged successfully. +func TestHandler(t *testing.T) { + sockpath := filepath.Join(t.TempDir(), "socket") + sink, err := testsink.New(sockpath) + if err != nil { + t.Fatal(err) + } + defer sink.Stop() + + conn, err := journal.Connect(sockpath) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + jh, err := NewHandler(WithConn(conn)) + if err != nil { + t.Fatal(err) + } + + httpGroup := slog.Group("http", "path", "/foo", "status", http.StatusOK) + logger := slog.New(jh) + logger = logger.With("global_attr_key", "global_attr_val", httpGroup).WithGroup("app") + logger.Info("hello, journal", "slowkey", slowValue("slowval"), slog.Group("g1", "attr1", "val1", "attr2", 123)) + + msg, err := sink.Message(0) + if err != nil { + t.Fatal(err) + } + + msgText, attrs, err := msg.Decode() + if err != nil { + t.Error(err) + } + if msgText != "hello, journal" { + t.Errorf("unexpected message text %q", msgText) + } + + expAttrs := []testsink.DecodedAttr{ + testsink.DecodedAttr{Key: "PRIORITY", Val: "6"}, + testsink.DecodedAttr{Key: "GLOBAL_ATTR_KEY", Val: "global_attr_val"}, + testsink.DecodedAttr{Key: "HTTP_PATH", Val: "/foo"}, + testsink.DecodedAttr{Key: "HTTP_STATUS", Val: "200"}, + testsink.DecodedAttr{Key: "APP_SLOWKEY", Val: "slowval"}, + testsink.DecodedAttr{Key: "APP_G1_ATTR1", Val: "val1"}, + testsink.DecodedAttr{Key: "APP_G1_ATTR2", Val: "123"}, + } + + for i := range expAttrs { + key := expAttrs[i].Key + val, ok := testsink.GetAttr(attrs, key) + switch { + case !ok: + t.Errorf("missing %s attribute", key) + case val != expAttrs[i].Val: + t.Errorf("unexpected %s value", key) + } + } + + if t.Failed() { + for i := range attrs { + t.Errorf("attr %q=%q", attrs[i].Key, attrs[i].Val) + } + } +} + +type slowValue string + +func (s slowValue) LogValue() slog.Value { + return slog.StringValue(string(s)) +} diff --git a/journslog/key_map.go b/journslog/key_map.go new file mode 100644 index 0000000..f23689b --- /dev/null +++ b/journslog/key_map.go @@ -0,0 +1,97 @@ +package journslog + +import ( + _ "log/slog" + "strings" + "sync" + + "src.lwithers.me.uk/go/journal" +) + +// KeyMapper maps [slog.Attr] keys onto [journal.AttrKey] objects. Such a mapper +// is necessary because systemd-journald is very restrictive about suitable +// attribute keys. +type KeyMapper interface { + KeyMap(slogAttrKey string) journal.AttrKey +} + +// DefaultKeyMapper returns a [KeyMapper] with the package's default behaviour. +// The returned mapper contains an internal, persistent map and a [sync.RWMutex] +// to cache map results. +// +// The default behaviour is to strip leading underscores; to turn lowercase +// ASCII ‘a’–‘z’ chars into uppercase ‘A’–‘Z’ (uppercase is passed through +// verbatim), to accept underscores, and to map all other chars to underscores. +// If the result is an empty string, it instead returns the key "BAD_ATTR_KEY". +func DefaultKeyMapper() KeyMapper { + return &defaultKeyMapper{ + badKey: journal.MustAttrKey("BAD_ATTR_KEY"), + keys: make(map[string]journal.AttrKey), + } +} + +type defaultKeyMapper struct { + badKey journal.AttrKey + keys map[string]journal.AttrKey + lock sync.RWMutex +} + +func (dkm *defaultKeyMapper) KeyMap(slogAttrKey string) journal.AttrKey { + if key, ok := dkm.mapFast(slogAttrKey); ok { + return key + } + return dkm.mapSlow(slogAttrKey) +} + +func (dkm *defaultKeyMapper) mapFast(slogAttrKey string) (journal.AttrKey, bool) { + dkm.lock.RLock() + defer dkm.lock.RUnlock() + key, ok := dkm.keys[slogAttrKey] + return key, ok +} + +func (dkm *defaultKeyMapper) mapSlow(slogAttrKey string) journal.AttrKey { + key := dkm.mapFunc(slogAttrKey) + + // it's OK if another goroutine in the slow path just wrote the key, we + // will overwrite its result but the result will still be correct + dkm.lock.Lock() + defer dkm.lock.Unlock() + dkm.keys[slogAttrKey] = key + return key +} + +func (dkm *defaultKeyMapper) mapFunc(slogAttrKey string) journal.AttrKey { + key, err := journal.NewAttrKey(slogAttrKey) + if err == nil { + return key + } + + var s strings.Builder + for _, r := range slogAttrKey { + switch { + case r >= 'a' && r <= 'z': + s.WriteRune(r - 'a' + 'A') + case r >= '0' && r <= '9', + r >= 'A' && r <= 'Z': + s.WriteRune(r) + default: + if s.Len() > 0 { + s.WriteRune('_') + } + } + } + if s.Len() == 0 { + return dkm.badKey + } + + skey := s.String() + if len(skey) > 255 { + skey = skey[:255] // known to only contain ASCII chars + } + + if key, err = journal.NewAttrKey(skey); err != nil { + return dkm.badKey + } + return key +} diff --git a/journslog/level_map.go b/journslog/level_map.go new file mode 100644 index 0000000..1d61b07 --- /dev/null +++ b/journslog/level_map.go @@ -0,0 +1,57 @@ +package journslog + +import ( + "log/slog" + + "src.lwithers.me.uk/go/journal" +) + +// LevelMapper maps [slog.Level] values onto [journal.Priority] values. +type LevelMapper interface { + LevelMap(slogLevel slog.Level) journal.Priority +} + +// DefaultLevelMapper returns a [LevelMapper] with the package's default +// behaviour. +// +// The mapping for the four defined [slog.Level] constants is as follows: +// +// slog.Level journal.Priority +// ---------- ---------------- +// slog.LevelDebug (-4) journal.PriDebug +// slog.LevelInfo (0) journal.PriInfo +// slog.LevelWarn (4) journal.PriWarning +// slog.LevelError (8) journal.PriErr +// +// Other integer values are mapped as follows: +// +// integer slog.Level journal.Priority +// ------------------ ---------------- +// -∞ ... -4 journal.PriDebug +// -3 ... 0 journal.PriInfo +// 1 ... 3 journal.PriNotice +// 4 ... 7 journal.PriWarning +// 8 journal.PriErr +// 9 ... ∞ journal.PriCrit +func DefaultLevelMapper() LevelMapper { + return &defaultLevelMapper{} +} + +type defaultLevelMapper struct{} + +func (*defaultLevelMapper) LevelMap(slogLevel slog.Level) journal.Priority { + switch { + case slogLevel <= slog.LevelDebug: + return journal.PriDebug + case slogLevel <= slog.LevelInfo: + return journal.PriInfo + case slogLevel < slog.LevelWarn: + return journal.PriNotice + case slogLevel < slog.LevelError: + return journal.PriWarning + case slogLevel == slog.LevelError: + return journal.PriErr + default: + return journal.PriCrit + } +} diff --git a/journslog/pools.go b/journslog/pools.go new file mode 100644 index 0000000..ca6ff67 --- /dev/null +++ b/journslog/pools.go @@ -0,0 +1,40 @@ +package journslog + +import ( + "log/slog" + "sync" + + "src.lwithers.me.uk/go/journal" +) + +var attrPool = sync.Pool{ + New: func() any { + return make([]slog.Attr, 0, 16) + }, +} + +func attrPoolGet() []slog.Attr { + slice := attrPool.Get().([]slog.Attr) + return slice[:0] +} + +func attrPoolPut(slice []slog.Attr) { + clear(slice) + attrPool.Put(slice) +} + +var jattrPool = sync.Pool{ + New: func() any { + return make([]journal.Attr, 0, 16) + }, +} + +func jattrPoolGet() []journal.Attr { + slice := jattrPool.Get().([]journal.Attr) + return slice[:0] +} + +func jattrPoolPut(slice []journal.Attr) { + clear(slice) + jattrPool.Put(slice) +} From f4e1589ec33ca4ee8f8ad2f756dd73918aa9d554 Mon Sep 17 00:00:00 2001 From: Laurence Withers Date: Tue, 23 Jun 2026 13:04:50 +0100 Subject: [PATCH 5/5] journslog: integrate with testing/slogtest --- journslog/handler_option.go | 8 +++ journslog/journslog.go | 11 +++++ journslog/journslog_test.go | 98 +++++++++++++++++++++++++++++++++++++ 3 files changed, 117 insertions(+) diff --git a/journslog/handler_option.go b/journslog/handler_option.go index f52ec05..38478c2 100644 --- a/journslog/handler_option.go +++ b/journslog/handler_option.go @@ -81,3 +81,11 @@ func WithAttrsFromTime(attrsFromTime func(time.Time) []slog.Attr) HandlerOption jh.attrsFromTime = attrsFromTime } } + +// WithAttrsFromLevel supplies a function which can optionally map the given +// log level onto zero or more attributes. +func WithAttrsFromLevel(attrsFromLevel func(slog.Level) []slog.Attr) HandlerOption { + return func(jh *JournalHandler) { + jh.attrsFromLevel = attrsFromLevel + } +} diff --git a/journslog/journslog.go b/journslog/journslog.go index 9185489..ca71d9c 100644 --- a/journslog/journslog.go +++ b/journslog/journslog.go @@ -62,6 +62,10 @@ type JournalHandler struct { // Time from an [slog.Record]. It may be nil. attrsFromTime func(time.Time) []slog.Attr + // attrsFromLevel may return zero or more attributes to represent the + // log level. It may be nil. + attrsFromLevel func(slog.Level) []slog.Attr + // groups are the ordered group names added by WithGroup. We need to // maintain the full list, not just a pre-rendered prefix string, // because of the API of replaceAttr. @@ -174,6 +178,13 @@ func (jh *JournalHandler) Handle(ctx context.Context, rec slog.Record) error { } addAttrs = append(addAttrs, jh.attrsFromContext(ctx)...) } + if jh.attrsFromLevel != nil { + if addAttrs == nil { + addAttrs = attrPoolGet() + defer attrPoolPut(addAttrs) + } + addAttrs = append(addAttrs, jh.attrsFromLevel(rec.Level)...) + } // 2 for journal.Conn.Entry (MESSAGE and PRIORITY), // 3 for SRC_* (optional but we always allocate space) diff --git a/journslog/journslog_test.go b/journslog/journslog_test.go index 53129b8..33c342b 100644 --- a/journslog/journslog_test.go +++ b/journslog/journslog_test.go @@ -4,7 +4,11 @@ import ( "log/slog" "net/http" "path/filepath" + "strings" "testing" + "testing/slogtest" + "time" + "uuid" "src.lwithers.me.uk/go/journal" "src.lwithers.me.uk/go/journal/testsink" @@ -82,3 +86,97 @@ type slowValue string func (s slowValue) LogValue() slog.Value { return slog.StringValue(string(s)) } + +// TestSlog integrates with [testing/slogtest]. +func TestSlog(t *testing.T) { + tmpdir := t.TempDir() + + attrsFromTime := func(t time.Time) []slog.Attr { + return []slog.Attr{ + { + Key: "TEST_TIME", + Value: slog.StringValue(t.Format(time.RFC3339Nano)), + }, + } + } + + attrsFromLevel := func(l slog.Level) []slog.Attr { + return []slog.Attr{ + { + Key: "TEST_LEVEL", + Value: slog.StringValue(l.String()), + }, + } + } + + // XXX: this relies on [slogtest.Run] executing its test cases + // completely sequentially + var curSink *testsink.Sink + + newHandler := func(t *testing.T) slog.Handler { + sockpath := filepath.Join(tmpdir, uuid.New().String()) + sink, err := testsink.New(sockpath) + if err != nil { + t.Fatalf("failed to create test sink: %v", err) + } + + conn, err := journal.Connect(sockpath) + if err != nil { + sink.Stop() + t.Fatalf("failed to connect to test sink socket: %v", err) + } + + h, err := NewHandler(WithConn(conn), WithAttrsFromTime(attrsFromTime), WithAttrsFromLevel(attrsFromLevel), WithHandlerOptions(slog.HandlerOptions{ + Level: slog.LevelInfo, + })) + if err != nil { + sink.Stop() + t.Fatalf("failed to create handler: %v", err) + } + + curSink = sink + return h + } + + result := func(t *testing.T) map[string]any { + defer curSink.Stop() + + m, err := curSink.Message(0) + if err != nil { + t.Fatalf("failed to fetch message from test sink: %v", err) + } + + msg, attrs, err := m.Decode() + if err != nil { + t.Errorf("failed to decode message from test sink: %v", err) + } + + out := make(map[string]any) + out[slog.MessageKey] = msg + for i := range attrs { + k, v := attrs[i].Key, attrs[i].Val + switch k { + case "TEST_TIME": + out[slog.TimeKey] = v + case "TEST_LEVEL": + out[slog.LevelKey] = v + default: + parts := strings.Split(k, "_") + Ng := len(parts) - 1 // number of groups + g := out + for i := range Ng { + next, _ := g[parts[i]].(map[string]any) + if next == nil { + next = make(map[string]any) + g[parts[i]] = next + } + g = next + } + g[strings.ToLower(parts[Ng])] = v + } + } + return out + } + + slogtest.Run(t, newHandler, result) +}