this repo has no description
automod: prometheus metrics (#558)
+182
-1
+8
automod/engine/blobs.go
+8
automod/engine/blobs.go
···
5
5
"io"
6
6
"net/http"
7
7
"strings"
8
+
"time"
8
9
9
10
appbsky "github.com/bluesky-social/indigo/api/bsky"
10
11
lexutil "github.com/bluesky-social/indigo/lex/util"
···
94
95
95
96
func (c *RecordContext) fetchBlob(blob lexutil.LexBlob) ([]byte, error) {
96
97
98
+
start := time.Now()
99
+
defer func() {
100
+
duration := time.Since(start)
101
+
blobDownloadDuration.Observe(duration.Seconds())
102
+
}()
103
+
97
104
var blobBytes []byte
98
105
99
106
// TODO: better way to do this, eg a shared client?
···
121
128
}
122
129
defer resp.Body.Close()
123
130
131
+
blobDownloadCount.WithLabelValues(fmt.Sprint(resp.StatusCode)).Inc()
124
132
if resp.StatusCode != 200 {
125
133
return nil, fmt.Errorf("failed to fetch blob from PDS. did=%s cid=%s statusCode=%d", c.Account.Identity.DID, blob.Ref, resp.StatusCode)
126
134
}
+44
automod/engine/engine.go
+44
automod/engine/engine.go
···
44
44
//
45
45
// This method can be called concurrently, though cached state may end up inconsistent if multiple events for the same account (DID) are processed in parallel.
46
46
func (eng *Engine) ProcessIdentityEvent(ctx context.Context, typ string, did syntax.DID) error {
47
+
eventProcessCount.WithLabelValues("identity").Inc()
48
+
start := time.Now()
49
+
defer func() {
50
+
duration := time.Since(start)
51
+
eventProcessDuration.WithLabelValues("identity").Observe(duration.Seconds())
52
+
}()
53
+
47
54
// similar to an HTTP server, we want to recover any panics from rule execution
48
55
defer func() {
49
56
if r := recover(); r != nil {
50
57
eng.Logger.Error("automod event execution exception", "err", r, "did", did, "type", typ)
58
+
eventErrorCount.WithLabelValues("identity").Inc()
51
59
}
52
60
}()
53
61
ctx, cancel := context.WithTimeout(ctx, identityEventTimeout)
···
59
67
}
60
68
ident, err := eng.Directory.LookupDID(ctx, did)
61
69
if err != nil {
70
+
eventErrorCount.WithLabelValues("identity").Inc()
62
71
return fmt.Errorf("resolving identity: %w", err)
63
72
}
64
73
if ident == nil {
74
+
eventErrorCount.WithLabelValues("identity").Inc()
65
75
return fmt.Errorf("identity not found for DID: %s", did.String())
66
76
}
67
77
68
78
am, err := eng.GetAccountMeta(ctx, ident)
69
79
if err != nil {
80
+
eventErrorCount.WithLabelValues("identity").Inc()
70
81
return fmt.Errorf("failed to fetch account metadata: %w", err)
71
82
}
72
83
ac := NewAccountContext(ctx, eng, *am)
73
84
if err := eng.Rules.CallIdentityRules(&ac); err != nil {
85
+
eventErrorCount.WithLabelValues("identity").Inc()
74
86
return fmt.Errorf("rule execution failed: %w", err)
75
87
}
76
88
eng.CanonicalLogLineAccount(&ac)
77
89
if err := eng.persistAccountModActions(&ac); err != nil {
90
+
eventErrorCount.WithLabelValues("identity").Inc()
78
91
return fmt.Errorf("failed to persist actions for identity event: %w", err)
79
92
}
80
93
if err := eng.persistCounters(ctx, ac.effects); err != nil {
94
+
eventErrorCount.WithLabelValues("identity").Inc()
81
95
return fmt.Errorf("failed to persist counters for identity event: %w", err)
82
96
}
83
97
return nil
···
87
101
//
88
102
// This method can be called concurrently, though cached state may end up inconsistent if multiple events for the same account (DID) are processed in parallel.
89
103
func (eng *Engine) ProcessRecordOp(ctx context.Context, op RecordOp) error {
104
+
eventProcessCount.WithLabelValues("record").Inc()
105
+
start := time.Now()
106
+
defer func() {
107
+
duration := time.Since(start)
108
+
eventProcessDuration.WithLabelValues("record").Observe(duration.Seconds())
109
+
}()
110
+
90
111
// similar to an HTTP server, we want to recover any panics from rule execution
91
112
defer func() {
92
113
if r := recover(); r != nil {
···
97
118
defer cancel()
98
119
99
120
if err := op.Validate(); err != nil {
121
+
eventErrorCount.WithLabelValues("record").Inc()
100
122
return fmt.Errorf("bad record op: %w", err)
101
123
}
102
124
ident, err := eng.Directory.LookupDID(ctx, op.DID)
103
125
if err != nil {
126
+
eventErrorCount.WithLabelValues("record").Inc()
104
127
return fmt.Errorf("resolving identity: %w", err)
105
128
}
106
129
if ident == nil {
130
+
eventErrorCount.WithLabelValues("record").Inc()
107
131
return fmt.Errorf("identity not found for DID: %s", op.DID)
108
132
}
109
133
110
134
am, err := eng.GetAccountMeta(ctx, ident)
111
135
if err != nil {
136
+
eventErrorCount.WithLabelValues("record").Inc()
112
137
return fmt.Errorf("failed to fetch account metadata: %w", err)
113
138
}
114
139
rc := NewRecordContext(ctx, eng, *am, op)
···
116
141
switch op.Action {
117
142
case CreateOp, UpdateOp:
118
143
if err := eng.Rules.CallRecordRules(&rc); err != nil {
144
+
eventErrorCount.WithLabelValues("record").Inc()
119
145
return fmt.Errorf("rule execution failed: %w", err)
120
146
}
121
147
case DeleteOp:
122
148
if err := eng.Rules.CallRecordDeleteRules(&rc); err != nil {
149
+
eventErrorCount.WithLabelValues("record").Inc()
123
150
return fmt.Errorf("rule execution failed: %w", err)
124
151
}
125
152
default:
153
+
eventErrorCount.WithLabelValues("record").Inc()
126
154
return fmt.Errorf("unexpected op action: %s", op.Action)
127
155
}
128
156
eng.CanonicalLogLineRecord(&rc)
···
133
161
}
134
162
}
135
163
if err := eng.persistRecordModActions(&rc); err != nil {
164
+
eventErrorCount.WithLabelValues("record").Inc()
136
165
return fmt.Errorf("failed to persist actions for record event: %w", err)
137
166
}
138
167
if err := eng.persistCounters(ctx, rc.effects); err != nil {
168
+
eventErrorCount.WithLabelValues("record").Inc()
139
169
return fmt.Errorf("failed to persist counts for record event: %w", err)
140
170
}
141
171
return nil
···
143
173
144
174
// returns a boolean indicating "block the event"
145
175
func (eng *Engine) ProcessNotificationEvent(ctx context.Context, senderDID, recipientDID syntax.DID, reason string, subject syntax.ATURI) (bool, error) {
176
+
eventProcessCount.WithLabelValues("notif").Inc()
177
+
start := time.Now()
178
+
defer func() {
179
+
duration := time.Since(start)
180
+
eventProcessDuration.WithLabelValues("notif").Observe(duration.Seconds())
181
+
}()
182
+
146
183
// similar to an HTTP server, we want to recover any panics from rule execution
147
184
defer func() {
148
185
if r := recover(); r != nil {
···
154
191
155
192
senderIdent, err := eng.Directory.LookupDID(ctx, senderDID)
156
193
if err != nil {
194
+
eventErrorCount.WithLabelValues("notif").Inc()
157
195
return false, fmt.Errorf("resolving identity: %w", err)
158
196
}
159
197
if senderIdent == nil {
198
+
eventErrorCount.WithLabelValues("notif").Inc()
160
199
return false, fmt.Errorf("identity not found for sender DID: %s", senderDID.String())
161
200
}
162
201
163
202
recipientIdent, err := eng.Directory.LookupDID(ctx, recipientDID)
164
203
if err != nil {
204
+
eventErrorCount.WithLabelValues("notif").Inc()
165
205
return false, fmt.Errorf("resolving identity: %w", err)
166
206
}
167
207
if recipientIdent == nil {
208
+
eventErrorCount.WithLabelValues("notif").Inc()
168
209
return false, fmt.Errorf("identity not found for sender DID: %s", recipientDID.String())
169
210
}
170
211
171
212
senderMeta, err := eng.GetAccountMeta(ctx, senderIdent)
172
213
if err != nil {
214
+
eventErrorCount.WithLabelValues("notif").Inc()
173
215
return false, fmt.Errorf("failed to fetch account metadata: %w", err)
174
216
}
175
217
recipientMeta, err := eng.GetAccountMeta(ctx, recipientIdent)
176
218
if err != nil {
219
+
eventErrorCount.WithLabelValues("notif").Inc()
177
220
return false, fmt.Errorf("failed to fetch account metadata: %w", err)
178
221
}
179
222
180
223
nc := NewNotificationContext(ctx, eng, *senderMeta, *recipientMeta, reason, subject)
181
224
if err := eng.Rules.CallNotificationRules(&nc); err != nil {
225
+
eventErrorCount.WithLabelValues("notif").Inc()
182
226
return false, fmt.Errorf("rule execution failed: %w", err)
183
227
}
184
228
eng.CanonicalLogLineNotification(&nc)
+3
automod/engine/fetch_account_meta.go
+3
automod/engine/fetch_account_meta.go
+1
automod/engine/fetch_relationship.go
+1
automod/engine/fetch_relationship.go
···
45
45
}
46
46
47
47
// fetch account relationship from AppView
48
+
accountRelationshipFetches.Inc()
48
49
resp, err := appbsky.GraphGetRelationships(ctx, eng.BskyClient, primary.String(), []string{other.String()})
49
50
if err != nil || len(resp.Relationships) != 1 {
50
51
logger.Warn("account relationship lookup failed", "err", err)
+61
automod/engine/metrics.go
+61
automod/engine/metrics.go
···
1
+
package engine
2
+
3
+
import (
4
+
"github.com/prometheus/client_golang/prometheus"
5
+
"github.com/prometheus/client_golang/prometheus/promauto"
6
+
)
7
+
8
+
var eventProcessDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
9
+
Name: "automod_event_duration_sec",
10
+
Help: "Total duration of automod event processing",
11
+
}, []string{"type"})
12
+
13
+
var eventProcessCount = promauto.NewCounterVec(prometheus.CounterOpts{
14
+
Name: "automod_event_processed",
15
+
Help: "Number of events processed",
16
+
}, []string{"type"})
17
+
18
+
var eventErrorCount = promauto.NewCounterVec(prometheus.CounterOpts{
19
+
Name: "automod_event_errors",
20
+
Help: "Number of events which failed processing",
21
+
}, []string{"type"})
22
+
23
+
var actionNewLabelCount = promauto.NewCounterVec(prometheus.CounterOpts{
24
+
Name: "automod_new_action_labels",
25
+
Help: "Number of new labels persisted",
26
+
}, []string{"type", "val"})
27
+
28
+
var actionNewFlagCount = promauto.NewCounterVec(prometheus.CounterOpts{
29
+
Name: "automod_new_action_flags",
30
+
Help: "Number of new flags persisted",
31
+
}, []string{"type", "val"})
32
+
33
+
var actionNewReportCount = promauto.NewCounterVec(prometheus.CounterOpts{
34
+
Name: "automod_new_action_reports",
35
+
Help: "Number of new flags persisted",
36
+
}, []string{"type"})
37
+
38
+
var actionNewTakedownCount = promauto.NewCounterVec(prometheus.CounterOpts{
39
+
Name: "automod_new_action_takedowns",
40
+
Help: "Number of new flags persisted",
41
+
}, []string{"type"})
42
+
43
+
var accountMetaFetches = promauto.NewCounter(prometheus.CounterOpts{
44
+
Name: "automod_account_meta_fetches",
45
+
Help: "Number of account metadata reads (API calls)",
46
+
})
47
+
48
+
var accountRelationshipFetches = promauto.NewCounter(prometheus.CounterOpts{
49
+
Name: "automod_account_relationship_fetches",
50
+
Help: "Number of account relationship reads (API calls)",
51
+
})
52
+
53
+
var blobDownloadCount = promauto.NewCounterVec(prometheus.CounterOpts{
54
+
Name: "automod_blob_downloads",
55
+
Help: "Number of blobs downloaded, by HTTP status code",
56
+
}, []string{"status"})
57
+
58
+
var blobDownloadDuration = promauto.NewHistogram(prometheus.HistogramOpts{
59
+
Name: "automod_blob_download_duration_sec",
60
+
Help: "Duration of blob download attempts",
61
+
})
+18
automod/engine/persist.go
+18
automod/engine/persist.go
···
68
68
69
69
// flags don't require admin auth
70
70
if len(newFlags) > 0 {
71
+
for _, val := range newFlags {
72
+
// note: WithLabelValues is a prometheus label, not an atproto label
73
+
actionNewFlagCount.WithLabelValues("record", val).Inc()
74
+
}
71
75
eng.Flags.Add(ctx, c.Account.Identity.DID.String(), newFlags)
72
76
}
73
77
···
83
87
84
88
if len(newLabels) > 0 {
85
89
c.Logger.Info("labeling record", "newLabels", newLabels)
90
+
for _, val := range newLabels {
91
+
// note: WithLabelValues is a prometheus label, not an atproto label
92
+
actionNewLabelCount.WithLabelValues("account", val).Inc()
93
+
}
86
94
comment := "[automod]: auto-labeling account"
87
95
_, err := comatproto.AdminEmitModerationEvent(ctx, xrpcc, &comatproto.AdminEmitModerationEvent_Input{
88
96
CreatedBy: xrpcc.Auth.Did,
···
118
126
119
127
if newTakedown {
120
128
c.Logger.Warn("account-takedown")
129
+
actionNewTakedownCount.WithLabelValues("account").Inc()
121
130
comment := "[automod]: auto account-takedown"
122
131
_, err := comatproto.AdminEmitModerationEvent(ctx, xrpcc, &comatproto.AdminEmitModerationEvent_Input{
123
132
CreatedBy: xrpcc.Auth.Did,
···
212
221
213
222
// flags don't require admin auth
214
223
if len(newFlags) > 0 {
224
+
for _, val := range newFlags {
225
+
// note: WithLabelValues is a prometheus label, not an atproto label
226
+
actionNewFlagCount.WithLabelValues("record", val).Inc()
227
+
}
215
228
eng.Flags.Add(ctx, atURI, newFlags)
216
229
}
217
230
···
238
251
xrpcc := eng.AdminClient
239
252
if len(newLabels) > 0 {
240
253
c.Logger.Info("labeling record", "newLabels", newLabels)
254
+
for _, val := range newLabels {
255
+
// note: WithLabelValues is a prometheus label, not an atproto label
256
+
actionNewLabelCount.WithLabelValues("record", val).Inc()
257
+
}
241
258
comment := "[automod]: auto-labeling record"
242
259
_, err := comatproto.AdminEmitModerationEvent(ctx, xrpcc, &comatproto.AdminEmitModerationEvent_Input{
243
260
CreatedBy: xrpcc.Auth.Did,
···
266
283
267
284
if newTakedown {
268
285
c.Logger.Warn("record-takedown")
286
+
actionNewTakedownCount.WithLabelValues("record").Inc()
269
287
comment := "[automod]: automated record-takedown"
270
288
_, err := comatproto.AdminEmitModerationEvent(ctx, xrpcc, &comatproto.AdminEmitModerationEvent_Input{
271
289
CreatedBy: xrpcc.Auth.Did,
+3
-1
automod/engine/persisthelpers.go
+3
-1
automod/engine/persisthelpers.go
···
142
142
}
143
143
144
144
eng.Logger.Info("reporting account", "reasonType", mr.ReasonType, "comment", mr.Comment)
145
+
actionNewReportCount.WithLabelValues("account").Inc()
145
146
comment := "[automod] " + mr.Comment
146
147
_, err = comatproto.ModerationCreateReport(ctx, xrpcc, &comatproto.ModerationCreateReport_Input{
147
148
ReasonType: &mr.ReasonType,
···
191
192
return false, nil
192
193
}
193
194
194
-
eng.Logger.Info("reporting account", "reasonType", mr.ReasonType, "comment", mr.Comment)
195
+
eng.Logger.Info("reporting record", "reasonType", mr.ReasonType, "comment", mr.Comment)
196
+
actionNewReportCount.WithLabelValues("record").Inc()
195
197
comment := "[automod] " + mr.Comment
196
198
_, err = comatproto.ModerationCreateReport(ctx, xrpcc, &comatproto.ModerationCreateReport_Input{
197
199
ReasonType: &mr.ReasonType,
+9
automod/visual/abyss_client.go
+9
automod/visual/abyss_client.go
···
8
8
"io"
9
9
"log/slog"
10
10
"net/http"
11
+
"time"
11
12
12
13
lexutil "github.com/bluesky-social/indigo/lex/util"
13
14
"github.com/bluesky-social/indigo/util"
···
56
57
req.Header.Set("x-ratelimit-bypass", ac.RatelimitBypass)
57
58
}
58
59
60
+
start := time.Now()
61
+
defer func() {
62
+
duration := time.Since(start)
63
+
abyssAPIDuration.Observe(duration.Seconds())
64
+
}()
65
+
59
66
req = req.WithContext(ctx)
60
67
res, err := ac.Client.Do(req)
61
68
if err != nil {
62
69
return nil, fmt.Errorf("abyss request failed: %v", err)
63
70
}
64
71
defer res.Body.Close()
72
+
73
+
abyssAPICount.WithLabelValues(fmt.Sprint(res.StatusCode)).Inc()
65
74
if res.StatusCode != 200 {
66
75
return nil, fmt.Errorf("abyss request failed statusCode=%d", res.StatusCode)
67
76
}
+9
automod/visual/hiveai_client.go
+9
automod/visual/hiveai_client.go
···
9
9
"log/slog"
10
10
"mime/multipart"
11
11
"net/http"
12
+
"time"
12
13
13
14
lexutil "github.com/bluesky-social/indigo/lex/util"
14
15
"github.com/bluesky-social/indigo/util"
···
182
183
return nil, err
183
184
}
184
185
186
+
start := time.Now()
187
+
defer func() {
188
+
duration := time.Since(start)
189
+
hiveAPIDuration.Observe(duration.Seconds())
190
+
}()
191
+
185
192
req.Header.Set("Authorization", fmt.Sprintf("Token %s", hal.ApiToken))
186
193
req.Header.Add("Content-Type", writer.FormDataContentType())
187
194
req.Header.Set("Accept", "application/json")
···
193
200
return nil, fmt.Errorf("HiveAI request failed: %v", err)
194
201
}
195
202
defer res.Body.Close()
203
+
204
+
hiveAPICount.WithLabelValues(fmt.Sprint(res.StatusCode)).Inc()
196
205
if res.StatusCode != 200 {
197
206
return nil, fmt.Errorf("HiveAI request failed statusCode=%d", res.StatusCode)
198
207
}
+26
automod/visual/metrics.go
+26
automod/visual/metrics.go
···
1
+
package visual
2
+
3
+
import (
4
+
"github.com/prometheus/client_golang/prometheus"
5
+
"github.com/prometheus/client_golang/prometheus/promauto"
6
+
)
7
+
8
+
var hiveAPIDuration = promauto.NewHistogram(prometheus.HistogramOpts{
9
+
Name: "automod_hive_api_duration_sec",
10
+
Help: "Duration of Hive image auto-labeling API calls",
11
+
})
12
+
13
+
var hiveAPICount = promauto.NewCounterVec(prometheus.CounterOpts{
14
+
Name: "automod_hive_api_count",
15
+
Help: "Number of Hive image auto-labeling API calls, by HTTP status code",
16
+
}, []string{"status"})
17
+
18
+
var abyssAPIDuration = promauto.NewHistogram(prometheus.HistogramOpts{
19
+
Name: "automod_abyss_api_duration_sec",
20
+
Help: "Duration of abyss image scanning API call",
21
+
})
22
+
23
+
var abyssAPICount = promauto.NewCounterVec(prometheus.CounterOpts{
24
+
Name: "automod_abyss_api_count",
25
+
Help: "Number of abyss image scanning API calls, by HTTP status code",
26
+
}, []string{"status"})