-
Notifications
You must be signed in to change notification settings - Fork 1
/
triggeredloghandler.go
146 lines (135 loc) · 4.18 KB
/
triggeredloghandler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
package triggeredloghandler
import (
"context"
"log/slog"
"sync"
)
// TriggeredLogStreamIDKey is the name of the attribute used to
// identify logs that are part of a group of triggere logs
const TriggeredLogStreamIDKey = "log_stream_id"
// TriggeredLogHandler is a slog.Handler that tracks message streams
// and only logs them if a message is logged that exceeds the trigger
// threshold. It does not process messages directly, but delegates
// processing to a target handler. TriggeredLogHandler is only
// responsible for tracking messages and determining when logging
// is triggered.
type TriggeredLogHandler struct {
target slog.Handler
base *baseTriggeredLogHandler
}
// baseTriggeredLogHandler is data common to all TriggeredLogHandlers
// in a tree
type baseTriggeredLogHandler struct {
triggerLevel slog.Level
triggered bool
backlog []backlogRecord
mu sync.Mutex
}
// backlogRecord is all the data needed to log a log record
type backlogRecord struct {
ctx context.Context
target slog.Handler
record slog.Record
}
// NewTriggeredLogHandler returns a TriggeredLogHandler at the root
// of a TriggeredLogHandler tree. It is initialized as not triggered.
// target is a handler where messages will be sent once triggered.
// The streamID is added as a value to the target handler to provide
// a consistent value across all messages using TriggeredLogStreamIDKey
// as the attribute key.
func NewTriggeredLogHandler(
target slog.Handler,
streamID string,
triggerLevel slog.Level,
) *TriggeredLogHandler {
target = target.WithAttrs([]slog.Attr{{
Key: TriggeredLogStreamIDKey,
Value: slog.StringValue(streamID),
}})
return &TriggeredLogHandler{
target: target,
base: &baseTriggeredLogHandler{
triggerLevel: triggerLevel,
},
}
}
// Enabled always returns true because the nature of the
// handler is that it processes all messages and if we
// ever returned false, slog won't send a message at all
func (tlh *TriggeredLogHandler) Enabled(_ context.Context, _ slog.Level) bool {
return true
}
// Handle stores the record if this handler hasn't been triggered
// yet, or sends any backlog including this record if it's been
// triggered
func (tlh *TriggeredLogHandler) Handle(ctx context.Context, record slog.Record) error {
tlh.base.mu.Lock()
defer tlh.base.mu.Unlock()
if record.Level >= tlh.base.triggerLevel {
tlh.base.triggered = true
}
if tlh.base.triggered {
if len(tlh.base.backlog) > 0 {
err := tlh.forwardBacklog()
if err != nil {
tlh.base.backlog = append(
tlh.base.backlog,
backlogRecord{
ctx: ctx,
target: tlh.target,
record: record,
},
)
return err
}
}
return tlh.target.Handle(ctx, record)
}
tlh.base.backlog = append(
tlh.base.backlog,
backlogRecord{
ctx: ctx,
target: tlh.target,
record: record,
},
)
return nil
}
// forwardBacklog sends the entire message backlog to the
// target log handler then clears the backlog.
// If an error is encountered when sending messages, unsent
// messages are preserved and an error is returned to indicate
// a retry on the next message submission
func (tlh *TriggeredLogHandler) forwardBacklog() error {
for idx, record := range tlh.base.backlog {
err := record.target.Handle(record.ctx, record.record)
if err != nil {
tlh.base.backlog = tlh.base.backlog[idx:]
return err
}
}
tlh.base.backlog = nil
return nil
}
// WithAttrs returns a new TriggeredLogHandler with the provided
// attributes. The new handler shares the message backlog with
// it's parent, so triggering any handler in the tree will cause
// the entire backlog to be processed
func (tlh *TriggeredLogHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
newTarget := tlh.target.WithAttrs(attrs)
return &TriggeredLogHandler{
target: newTarget,
base: tlh.base,
}
}
// WithGroup returns a new TriggeredLogHandler with the provided
// group. The new handler shares the message backlog with
// it's parent, so triggering any handler in the tree will cause
// the entire backlog to be processed
func (tlh *TriggeredLogHandler) WithGroup(name string) slog.Handler {
newTarget := tlh.target.WithGroup(name)
return &TriggeredLogHandler{
target: newTarget,
base: tlh.base,
}
}