diff --git a/README.md b/README.md index 057c8e3..f3311fa 100644 --- a/README.md +++ b/README.md @@ -1,33 +1,3 @@ # journal -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") -} -``` +Native Go output to the systemd journal socket. \ No newline at end of file diff --git a/conn.go b/conn.go index 2c43f0c..eb43b00 100644 --- a/conn.go +++ b/conn.go @@ -101,9 +101,3 @@ 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() -} diff --git a/conn_test.go b/conn_test.go index b5f5a63..6c89a6a 100644 --- a/conn_test.go +++ b/conn_test.go @@ -1,124 +1,187 @@ package journal import ( + "bytes" + "encoding/binary" + "encoding/hex" + "errors" + "fmt" + "io" + "net" "path/filepath" + "strconv" "testing" - - "src.lwithers.me.uk/go/journal/testsink" ) -// 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) +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", + }) if err != nil { - t.Fatal(err) - } - defer sink.Stop() - - conn, err := Connect(sockpath) - if err != nil { - t.Fatal(err) - } - defer conn.Close() - - conn.Entry(PriInfo, "hello, world", nil) - - msg, err := sink.Message(0) - if err != nil { - t.Fatal(err) + t.Fatalf("testConnector: ListenUnix: %v", 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) + 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) } } -// 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() +// 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() - - expAttrs := []Attr{ - Attr{ - Key: MustAttrKey("SHORT"), - Value: []byte("short val"), + // add some common attributes + c.Common = make([]Attr, 0, 10) + c.Common = append(c.Common, Attr{ + Key: AttrKey{ + key: "COMMON_ATTR", }, - Attr{ - Key: MustAttrKey("BINARY"), - Value: []byte("string with\n=embedded newline\nrequires binary protocol\n"), + Value: []byte("abc123\n"), + }) + + // attributes which will be common to all EntryErr calls + attr := make([]Attr, 0, 10) + attr = append(attr, Attr{ + Key: AttrKey{ + key: "HELLO", }, - Attr{ - Key: MustAttrKey("LAST"), - Value: []byte("last\n"), - }, - } - conn.Entry(PriDebug, "hello, binary world", expAttrs) + Value: []byte("world"), + }) - 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) + 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) } } } diff --git a/global.go b/global.go index 0434077..e9e45c8 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 := GetGlobalConn() + _, err := getGlobal() return err } @@ -39,12 +39,7 @@ func SetGlobalConn(c *Conn) { global = c } -// 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) { +func getGlobal() (*Conn, error) { globalLock.Lock() defer globalLock.Unlock() @@ -124,7 +119,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, _ := GetGlobalConn() + c, _ := getGlobal() if c == nil { return } diff --git a/journslog/handler_option.go b/journslog/handler_option.go deleted file mode 100644 index 38478c2..0000000 --- a/journslog/handler_option.go +++ /dev/null @@ -1,91 +0,0 @@ -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 - } -} - -// 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 deleted file mode 100644 index ca71d9c..0000000 --- a/journslog/journslog.go +++ /dev/null @@ -1,304 +0,0 @@ -/* -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 - - // 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. - 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)...) - } - 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) - 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 deleted file mode 100644 index 33c342b..0000000 --- a/journslog/journslog_test.go +++ /dev/null @@ -1,182 +0,0 @@ -package journslog - -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" -) - -// 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)) -} - -// 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) -} diff --git a/journslog/key_map.go b/journslog/key_map.go deleted file mode 100644 index f23689b..0000000 --- a/journslog/key_map.go +++ /dev/null @@ -1,97 +0,0 @@ -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 deleted file mode 100644 index 1d61b07..0000000 --- a/journslog/level_map.go +++ /dev/null @@ -1,57 +0,0 @@ -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 deleted file mode 100644 index ca6ff67..0000000 --- a/journslog/pools.go +++ /dev/null @@ -1,40 +0,0 @@ -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) -} diff --git a/testsink/message.go b/testsink/message.go deleted file mode 100644 index 99cae43..0000000 --- a/testsink/message.go +++ /dev/null @@ -1,126 +0,0 @@ -/* -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 deleted file mode 100644 index 427a775..0000000 --- a/testsink/testsink.go +++ /dev/null @@ -1,82 +0,0 @@ -/* -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 - } - } -}