this repo has no description
Track socket connection URL in Sonar metrics (#221)
This should make it easier to filter dashboards between environments and
eventually allow for sonar to connect to multiple sockets at a time and
emit distinct metrics from each socket.
+60
-58
+1
-1
cmd/sonar/main.go
+1
-1
cmd/sonar/main.go
···
100
100
log.Fatalf("failed to parse ws-url: %+v\n", err)
101
101
}
102
102
103
-
s, err := sonar.NewSonar(log, cctx.String("cursor-file"))
103
+
s, err := sonar.NewSonar(log, cctx.String("cursor-file"), u.String())
104
104
if err != nil {
105
105
log.Fatalf("failed to create sonar: %+v\n", err)
106
106
}
+17
-17
sonar/metrics.go
+17
-17
sonar/metrics.go
···
9
9
var eventsProcessedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
10
10
Name: "sonar_events_processed_total",
11
11
Help: "The total number of firehose events processed by Sonar",
12
-
}, []string{"event_type"})
12
+
}, []string{"event_type", "socket_url"})
13
13
14
-
var rebasesProcessedCounter = promauto.NewCounter(prometheus.CounterOpts{
14
+
var rebasesProcessedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
15
15
Name: "sonar_rebases_processed_total",
16
16
Help: "The total number of rebase operations processed by Sonar",
17
-
})
17
+
}, []string{"socket_url"})
18
18
19
19
var recordsProcessedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
20
20
Name: "sonar_records_processed_total",
21
21
Help: "The total number of records processed by Sonar",
22
-
}, []string{"record_type"})
22
+
}, []string{"record_type", "socket_url"})
23
23
24
-
var quoteRepostsProcessedCounter = promauto.NewCounter(prometheus.CounterOpts{
24
+
var quoteRepostsProcessedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
25
25
Name: "sonar_quote_reposts_processed_total",
26
26
Help: "The total number quote repost operations processed by Sonar",
27
-
})
27
+
}, []string{"socket_url"})
28
28
29
29
var opsProcessedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
30
30
Name: "sonar_ops_processed_total",
31
31
Help: "The total number of repo operations processed by Sonar",
32
-
}, []string{"kind", "op_path"})
32
+
}, []string{"kind", "op_path", "socket_url"})
33
33
34
34
// Initialize Prometheus metrics for duration of processing events
35
-
var eventProcessingDurationHistogram = promauto.NewHistogram(prometheus.HistogramOpts{
35
+
var eventProcessingDurationHistogram = promauto.NewHistogramVec(prometheus.HistogramOpts{
36
36
Name: "sonar_event_processing_duration_seconds",
37
37
Help: "The amount of time it takes to process a firehose event",
38
38
Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
39
-
})
39
+
}, []string{"socket_url"})
40
40
41
-
var lastSeqGauge = promauto.NewGauge(prometheus.GaugeOpts{
41
+
var lastSeqGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
42
42
Name: "sonar_last_seq",
43
43
Help: "The last sequence number processed",
44
-
})
44
+
}, []string{"socket_url"})
45
45
46
-
var lastSeqProcessedAtGauge = promauto.NewGauge(prometheus.GaugeOpts{
46
+
var lastSeqProcessedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
47
47
Name: "sonar_last_seq_processed_at",
48
48
Help: "The timestamp of the last sequence number processed",
49
-
})
49
+
}, []string{"socket_url"})
50
50
51
-
var lastSeqCreatedAtGauge = promauto.NewGauge(prometheus.GaugeOpts{
51
+
var lastSeqCreatedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
52
52
Name: "sonar_last_seq_created_at",
53
53
Help: "The timestamp of the last sequence number created",
54
-
})
54
+
}, []string{"socket_url"})
55
55
56
-
var lastSeqCommittedAtGauge = promauto.NewGauge(prometheus.GaugeOpts{
56
+
var lastSeqCommittedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
57
57
Name: "sonar_last_seq_committed_at",
58
58
Help: "The commit timestamp of the last sequence number processed",
59
-
})
59
+
}, []string{"socket_url"})
+42
-40
sonar/sonar.go
+42
-40
sonar/sonar.go
···
24
24
)
25
25
26
26
type Sonar struct {
27
+
SocketURL string
27
28
Progress *Progress
28
29
ProgMux sync.Mutex
29
30
Logger *zap.SugaredLogger
···
71
72
return nil
72
73
}
73
74
74
-
func NewSonar(logger *zap.SugaredLogger, cursorFile string) (*Sonar, error) {
75
+
func NewSonar(logger *zap.SugaredLogger, cursorFile string, socketURL string) (*Sonar, error) {
75
76
s := Sonar{
77
+
SocketURL: socketURL,
76
78
Progress: &Progress{
77
79
LastSeq: -1,
78
80
},
···
106
108
107
109
switch {
108
110
case xe.RepoCommit != nil:
109
-
eventsProcessedCounter.WithLabelValues("repo_commit").Inc()
111
+
eventsProcessedCounter.WithLabelValues("repo_commit", s.SocketURL).Inc()
110
112
return s.HandleRepoCommit(ctx, xe.RepoCommit)
111
113
case xe.RepoHandle != nil:
112
-
eventsProcessedCounter.WithLabelValues("repo_handle").Inc()
114
+
eventsProcessedCounter.WithLabelValues("repo_handle", s.SocketURL).Inc()
113
115
now := time.Now()
114
116
s.ProgMux.Lock()
115
117
s.Progress.LastSeq = xe.RepoHandle.Seq
···
121
123
log.Errorf("error parsing time: %+v", err)
122
124
return nil
123
125
}
124
-
lastSeqCommittedAtGauge.Set(float64(t.UnixNano()))
125
-
lastSeqProcessedAtGauge.Set(float64(now.UnixNano()))
126
-
lastSeqGauge.Set(float64(xe.RepoHandle.Seq))
126
+
lastSeqCommittedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
127
+
lastSeqProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(now.UnixNano()))
128
+
lastSeqGauge.WithLabelValues(s.SocketURL).Set(float64(xe.RepoHandle.Seq))
127
129
case xe.RepoInfo != nil:
128
-
eventsProcessedCounter.WithLabelValues("repo_info").Inc()
130
+
eventsProcessedCounter.WithLabelValues("repo_info", s.SocketURL).Inc()
129
131
case xe.RepoMigrate != nil:
130
-
eventsProcessedCounter.WithLabelValues("repo_migrate").Inc()
132
+
eventsProcessedCounter.WithLabelValues("repo_migrate", s.SocketURL).Inc()
131
133
now := time.Now()
132
134
s.ProgMux.Lock()
133
135
s.Progress.LastSeq = xe.RepoMigrate.Seq
···
139
141
log.Errorf("error parsing time: %+v", err)
140
142
return nil
141
143
}
142
-
lastSeqCommittedAtGauge.Set(float64(t.UnixNano()))
143
-
lastSeqProcessedAtGauge.Set(float64(now.UnixNano()))
144
-
lastSeqGauge.Set(float64(xe.RepoHandle.Seq))
144
+
lastSeqCommittedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
145
+
lastSeqProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(now.UnixNano()))
146
+
lastSeqGauge.WithLabelValues(s.SocketURL).Set(float64(xe.RepoHandle.Seq))
145
147
case xe.RepoTombstone != nil:
146
-
eventsProcessedCounter.WithLabelValues("repo_tombstone").Inc()
148
+
eventsProcessedCounter.WithLabelValues("repo_tombstone", s.SocketURL).Inc()
147
149
case xe.LabelInfo != nil:
148
-
eventsProcessedCounter.WithLabelValues("label_info").Inc()
150
+
eventsProcessedCounter.WithLabelValues("label_info", s.SocketURL).Inc()
149
151
case xe.LabelLabels != nil:
150
-
eventsProcessedCounter.WithLabelValues("label_labels").Inc()
152
+
eventsProcessedCounter.WithLabelValues("label_labels", s.SocketURL).Inc()
151
153
case xe.Error != nil:
152
-
eventsProcessedCounter.WithLabelValues("error").Inc()
154
+
eventsProcessedCounter.WithLabelValues("error", s.SocketURL).Inc()
153
155
}
154
156
return nil
155
157
}
···
165
167
s.Progress.LastSeqProcessedAt = start
166
168
s.ProgMux.Unlock()
167
169
168
-
lastSeqGauge.Set(float64(evt.Seq))
170
+
lastSeqGauge.WithLabelValues(s.SocketURL).Set(float64(evt.Seq))
169
171
170
172
log := s.Logger.With("repo", evt.Repo, "seq", evt.Seq, "commit", evt.Commit)
171
173
···
177
179
178
180
if evt.Rebase {
179
181
log.Debug("rebase")
180
-
rebasesProcessedCounter.Inc()
182
+
rebasesProcessedCounter.WithLabelValues(s.SocketURL).Inc()
181
183
}
182
184
183
185
// Parse time from the event time string
···
187
189
return nil
188
190
}
189
191
190
-
lastSeqCommittedAtGauge.Set(float64(t.UnixNano()))
191
-
lastSeqProcessedAtGauge.Set(float64(start.UnixNano()))
192
+
lastSeqCommittedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
193
+
lastSeqProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(start.UnixNano()))
192
194
193
195
for _, op := range evt.Ops {
194
196
collection := strings.Split(op.Path, "/")[0]
···
196
198
ek := repomgr.EventKind(op.Action)
197
199
log = log.With("action", op.Action, "collection", collection)
198
200
199
-
opsProcessedCounter.WithLabelValues(op.Action, collection).Inc()
201
+
opsProcessedCounter.WithLabelValues(op.Action, collection, s.SocketURL).Inc()
200
202
201
203
switch ek {
202
204
case repomgr.EvtKindCreateRecord, repomgr.EvtKindUpdateRecord:
···
218
220
// Unpack the record and process it
219
221
switch rec := rec.(type) {
220
222
case *bsky.FeedPost:
221
-
recordsProcessedCounter.WithLabelValues("feed_post").Inc()
223
+
recordsProcessedCounter.WithLabelValues("feed_post", s.SocketURL).Inc()
222
224
if rec.Embed != nil && rec.Embed.EmbedRecord != nil && rec.Embed.EmbedRecord.Record != nil {
223
-
quoteRepostsProcessedCounter.Inc()
225
+
quoteRepostsProcessedCounter.WithLabelValues(s.SocketURL).Inc()
224
226
}
225
227
// Parse time from the event time string
226
228
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
···
228
230
log.Errorf("error parsing time: %+v", err)
229
231
continue
230
232
}
231
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
233
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
232
234
case *bsky.FeedLike:
233
-
recordsProcessedCounter.WithLabelValues("feed_like").Inc()
235
+
recordsProcessedCounter.WithLabelValues("feed_like", s.SocketURL).Inc()
234
236
// Parse time from the event time string
235
237
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
236
238
if err != nil {
237
239
log.Errorf("error parsing time: %+v", err)
238
240
continue
239
241
}
240
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
242
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
241
243
case *bsky.FeedRepost:
242
-
recordsProcessedCounter.WithLabelValues("feed_repost").Inc()
244
+
recordsProcessedCounter.WithLabelValues("feed_repost", s.SocketURL).Inc()
243
245
// Parse time from the event time string
244
246
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
245
247
if err != nil {
246
248
log.Errorf("error parsing time: %+v", err)
247
249
continue
248
250
}
249
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
251
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
250
252
case *bsky.GraphBlock:
251
-
recordsProcessedCounter.WithLabelValues("graph_block").Inc()
253
+
recordsProcessedCounter.WithLabelValues("graph_block", s.SocketURL).Inc()
252
254
// Parse time from the event time string
253
255
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
254
256
if err != nil {
255
257
log.Errorf("error parsing time: %+v", err)
256
258
continue
257
259
}
258
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
260
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
259
261
case *bsky.GraphFollow:
260
-
recordsProcessedCounter.WithLabelValues("graph_follow").Inc()
262
+
recordsProcessedCounter.WithLabelValues("graph_follow", s.SocketURL).Inc()
261
263
// Parse time from the event time string
262
264
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
263
265
if err != nil {
264
266
log.Errorf("error parsing time: %+v", err)
265
267
continue
266
268
}
267
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
269
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
268
270
case *bsky.ActorProfile:
269
-
recordsProcessedCounter.WithLabelValues("actor_profile").Inc()
271
+
recordsProcessedCounter.WithLabelValues("actor_profile", s.SocketURL).Inc()
270
272
// Parse time from the event time string
271
273
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
272
274
if err != nil {
273
275
log.Errorf("error parsing time: %+v", err)
274
276
continue
275
277
}
276
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
278
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
277
279
case *bsky.FeedGenerator:
278
-
recordsProcessedCounter.WithLabelValues("feed_generator").Inc()
280
+
recordsProcessedCounter.WithLabelValues("feed_generator", s.SocketURL).Inc()
279
281
// Parse time from the event time string
280
282
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
281
283
if err != nil {
282
284
log.Errorf("error parsing time: %+v", err)
283
285
continue
284
286
}
285
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
287
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
286
288
case *bsky.GraphList:
287
-
recordsProcessedCounter.WithLabelValues("graph_list").Inc()
289
+
recordsProcessedCounter.WithLabelValues("graph_list", s.SocketURL).Inc()
288
290
// Parse time from the event time string
289
291
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
290
292
if err != nil {
291
293
log.Errorf("error parsing time: %+v", err)
292
294
continue
293
295
}
294
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
296
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
295
297
case *bsky.GraphListitem:
296
-
recordsProcessedCounter.WithLabelValues("graph_listitem").Inc()
298
+
recordsProcessedCounter.WithLabelValues("graph_listitem", s.SocketURL).Inc()
297
299
// Parse time from the event time string
298
300
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
299
301
if err != nil {
300
302
log.Errorf("error parsing time: %+v", err)
301
303
continue
302
304
}
303
-
lastSeqCreatedAtGauge.Set(float64(recCreatedAt.UnixNano()))
305
+
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
304
306
default:
305
307
log.Warnf("unknown record type: %+v", rec)
306
308
}
···
311
313
}
312
314
}
313
315
314
-
eventProcessingDurationHistogram.Observe(time.Since(start).Seconds())
316
+
eventProcessingDurationHistogram.WithLabelValues(s.SocketURL).Observe(time.Since(start).Seconds())
315
317
return nil
316
318
}