this repo has no description
Use record timestamps for record timestamp tracker in sonar (#243)
+82
-99
+14
-12
cmd/sonar/main.go
+14
-12
cmd/sonar/main.go
···
81
81
82
82
rawlog, err := zap.NewProduction()
83
83
if err != nil {
84
-
log.Fatalf("failed to create logger: %+v\n", err)
84
+
log.Fatalf("failed to create logger: %+v", err)
85
85
}
86
86
defer func() {
87
-
log.Printf("main function teardown\n")
87
+
log.Printf("main function teardown")
88
88
err := rawlog.Sync()
89
89
if err != nil {
90
90
log.Printf("failed to sync logger on teardown: %+v", err.Error())
···
97
97
98
98
u, err := url.Parse(cctx.String("ws-url"))
99
99
if err != nil {
100
-
log.Fatalf("failed to parse ws-url: %+v\n", err)
100
+
log.Fatalf("failed to parse ws-url: %+v", err)
101
101
}
102
102
103
103
s, err := sonar.NewSonar(log, cctx.String("cursor-file"), u.String())
104
104
if err != nil {
105
-
log.Fatalf("failed to create sonar: %+v\n", err)
105
+
log.Fatalf("failed to create sonar: %+v", err)
106
106
}
107
107
108
108
wg := sync.WaitGroup{}
···
116
116
ticker := time.NewTicker(5 * time.Second)
117
117
rawlog, err := zap.NewProduction()
118
118
if err != nil {
119
-
log.Fatalf("failed to create logger: %+v\n", err)
119
+
log.Fatalf("failed to create logger: %+v", err)
120
120
}
121
121
log := rawlog.Sugar().With("source", "cursor_file_manager")
122
122
···
126
126
log.Info("shutting down cursor file manager")
127
127
err := s.WriteCursorFile()
128
128
if err != nil {
129
-
log.Errorf("failed to write cursor file: %+v\n", err)
129
+
log.Errorf("failed to write cursor file: %+v", err)
130
130
}
131
131
log.Info("cursor file manager shut down successfully")
132
132
return
133
133
case <-ticker.C:
134
134
err := s.WriteCursorFile()
135
135
if err != nil {
136
-
log.Errorf("failed to write cursor file: %+v\n", err)
136
+
log.Errorf("failed to write cursor file: %+v", err)
137
137
}
138
138
}
139
139
}
···
148
148
149
149
rawlog, err := zap.NewProduction()
150
150
if err != nil {
151
-
log.Fatalf("failed to create logger: %+v\n", err)
151
+
log.Fatalf("failed to create logger: %+v", err)
152
152
}
153
153
log := rawlog.Sugar().With("source", "liveness_checker")
154
154
···
186
186
defer wg.Done()
187
187
rawlog, err := zap.NewProduction()
188
188
if err != nil {
189
-
log.Fatalf("failed to create logger: %+v\n", err)
189
+
log.Fatalf("failed to create logger: %+v", err)
190
190
}
191
191
log := rawlog.Sugar().With("source", "metrics_server")
192
192
193
193
log.Infof("metrics server listening on port %d", cctx.Int("port"))
194
194
195
195
if err := metricServer.ListenAndServe(); err != http.ErrServerClosed {
196
-
log.Fatalf("failed to start metrics server: %+v\n", err)
196
+
log.Fatalf("failed to start metrics server: %+v", err)
197
197
}
198
198
log.Info("metrics server shut down successfully")
199
199
}()
···
203
203
}
204
204
205
205
log.Infof("connecting to WebSocket at: %s", u.String())
206
-
c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
206
+
c, _, err := websocket.DefaultDialer.Dial(u.String(), http.Header{
207
+
"User-Agent": []string{"sonar/1.0"},
208
+
})
207
209
if err != nil {
208
210
log.Infof("failed to connect to websocket: %v", err)
209
211
return err
···
229
231
log.Info("shutting down, waiting for workers to clean up...")
230
232
231
233
if err := metricServer.Shutdown(ctx); err != nil {
232
-
log.Errorf("failed to shut down metrics server: %+v\n", err)
234
+
log.Errorf("failed to shut down metrics server: %+v", err)
233
235
wg.Done()
234
236
}
235
237
+1
go.mod
+1
go.mod
···
4
4
5
5
require (
6
6
contrib.go.opencensus.io/exporter/prometheus v0.4.2
7
+
github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de
7
8
github.com/brianvoe/gofakeit/v6 v6.20.2
8
9
github.com/dustinkirkland/golang-petname v0.0.0-20230626224747-e794b9370d49
9
10
github.com/goccy/go-json v0.10.2
+5
go.sum
+5
go.sum
···
45
45
github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5 h1:iW0a5ljuFxkLGPNem5Ui+KBjFJzKg4Fv2fnxe4dvzpM=
46
46
github.com/alexbrainman/goissue34681 v0.0.0-20191006012335-3fc7a47baff5/go.mod h1:Y2QMoi1vgtOIfc+6DhrMOGkLoGzqSV2rKp4Sm+opsyA=
47
47
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
48
+
github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de h1:FxWPpzIjnTlhPwqqXc4/vE0f7GvRjuAsbW+HOIe8KnA=
49
+
github.com/araddon/dateparse v0.0.0-20210429162001-6b43995a97de/go.mod h1:DCaWoUhZrYW9p1lxo/cm8EmUOOzAPSEZNGF2DK1dJgw=
48
50
github.com/aws/aws-sdk-go v1.44.180/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI=
49
51
github.com/aws/aws-sdk-go-v2 v1.17.3/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
50
52
github.com/aws/aws-sdk-go-v2/config v1.18.8/go.mod h1:5XCmmyutmzzgkpk/6NYTjeWb6lgo9N170m1j6pQkIBs=
···
391
393
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
392
394
github.com/mattn/go-isatty v0.0.17 h1:BTarxUcIeDqL27Mc+vyvdWYSL28zpIhv3RoTdsLMPng=
393
395
github.com/mattn/go-isatty v0.0.17/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
396
+
github.com/mattn/go-runewidth v0.0.10/go.mod h1:RAqKPSqVFrSLVXbA8x7dzmKdmGzieGRCM46jaSJTDAk=
394
397
github.com/mattn/go-sqlite3 v1.14.15 h1:vfoHhTN1af61xCRSWzFIWzx2YskyMTwHLrExkBOjvxI=
395
398
github.com/mattn/go-sqlite3 v1.14.15/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
396
399
github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
···
493
496
github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB/chUwxUZY=
494
497
github.com/prometheus/statsd_exporter v0.22.7 h1:7Pji/i2GuhK6Lu7DHrtTkFmNBCudCPT1pX2CziuyQR0=
495
498
github.com/prometheus/statsd_exporter v0.22.7/go.mod h1:N/TevpjkIh9ccs6nuzY3jQn9dFqnUakOjnEuMPJJJnI=
499
+
github.com/rivo/uniseg v0.1.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
496
500
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
497
501
github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
498
502
github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
···
500
504
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
501
505
github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
502
506
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
507
+
github.com/scylladb/termtables v0.0.0-20191203121021-c4c0b6d42ff4/go.mod h1:C1a7PQSMz9NShzorzCiG2fk9+xuCgLkPeCvMHYR2OWg=
503
508
github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
504
509
github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
505
510
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
+25
-10
sonar/metrics.go
+25
-10
sonar/metrics.go
···
40
40
41
41
var lastSeqGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
42
42
Name: "sonar_last_seq",
43
-
Help: "The last sequence number processed",
43
+
Help: "The sequence number of the last event processed",
44
+
}, []string{"socket_url"})
45
+
46
+
var lastEvtProcessedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
47
+
Name: "sonar_last_evt_processed_at",
48
+
Help: "The timestamp of the last event processed",
44
49
}, []string{"socket_url"})
45
50
46
-
var lastSeqProcessedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
47
-
Name: "sonar_last_seq_processed_at",
48
-
Help: "The timestamp of the last sequence number processed",
51
+
var lastEvtCreatedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
52
+
Name: "sonar_last_evt_created_at",
53
+
Help: "The timestamp of the last event created",
54
+
}, []string{"socket_url"})
55
+
56
+
var lastRecordCreatedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
57
+
Name: "sonar_last_record_created_at",
58
+
Help: "The timestamp of the last record processed",
59
+
}, []string{"socket_url"})
60
+
61
+
var lastEvtCreatedRecordCreatedGapGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
62
+
Name: "sonar_last_evt_created_record_created_gap",
63
+
Help: "The gap between the last event's event timestamp and it's record timestamp",
49
64
}, []string{"socket_url"})
50
65
51
-
var lastSeqCreatedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
52
-
Name: "sonar_last_seq_created_at",
53
-
Help: "The timestamp of the last sequence number created",
66
+
var lastEvtCreatedEvtProcessedGapGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
67
+
Name: "sonar_last_evt_created_evt_processed_gap",
68
+
Help: "The gap between the last event's event timestamp and when it was processed by sonar",
54
69
}, []string{"socket_url"})
55
70
56
-
var lastSeqCommittedAtGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
57
-
Name: "sonar_last_seq_committed_at",
58
-
Help: "The commit timestamp of the last sequence number processed",
71
+
var lastRecordCreatedEvtProcessedGapGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
72
+
Name: "sonar_last_record_created_evt_processed_gap",
73
+
Help: "The gap between the last record's record timestamp and when it was processed by sonar",
59
74
}, []string{"socket_url"})
+37
-77
sonar/sonar.go
+37
-77
sonar/sonar.go
···
9
9
"sync"
10
10
"time"
11
11
12
+
"github.com/araddon/dateparse"
12
13
comatproto "github.com/bluesky-social/indigo/api/atproto"
13
14
"github.com/bluesky-social/indigo/api/bsky"
14
15
lexutil "github.com/bluesky-social/indigo/lex/util"
15
-
"github.com/bluesky-social/indigo/util"
16
16
"github.com/goccy/go-json"
17
17
"github.com/labstack/gommon/log"
18
18
···
123
123
log.Errorf("error parsing time: %+v", err)
124
124
return nil
125
125
}
126
-
lastSeqCommittedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
127
-
lastSeqProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(now.UnixNano()))
126
+
lastEvtCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
127
+
lastEvtProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(now.UnixNano()))
128
+
lastEvtCreatedEvtProcessedGapGauge.WithLabelValues(s.SocketURL).Set(float64(now.Sub(t).Seconds()))
128
129
lastSeqGauge.WithLabelValues(s.SocketURL).Set(float64(xe.RepoHandle.Seq))
129
130
case xe.RepoInfo != nil:
130
131
eventsProcessedCounter.WithLabelValues("repo_info", s.SocketURL).Inc()
···
141
142
log.Errorf("error parsing time: %+v", err)
142
143
return nil
143
144
}
144
-
lastSeqCommittedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
145
-
lastSeqProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(now.UnixNano()))
145
+
lastEvtCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
146
+
lastEvtProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(now.UnixNano()))
147
+
lastEvtCreatedEvtProcessedGapGauge.WithLabelValues(s.SocketURL).Set(float64(now.Sub(t).Seconds()))
146
148
lastSeqGauge.WithLabelValues(s.SocketURL).Set(float64(xe.RepoHandle.Seq))
147
149
case xe.RepoTombstone != nil:
148
150
eventsProcessedCounter.WithLabelValues("repo_tombstone", s.SocketURL).Inc()
···
160
162
ctx, span := otel.Tracer("sonar").Start(ctx, "HandleRepoCommit")
161
163
defer span.End()
162
164
163
-
start := time.Now()
165
+
processedAt := time.Now()
164
166
165
167
s.ProgMux.Lock()
166
168
s.Progress.LastSeq = evt.Seq
167
-
s.Progress.LastSeqProcessedAt = start
169
+
s.Progress.LastSeqProcessedAt = processedAt
168
170
s.ProgMux.Unlock()
169
171
170
172
lastSeqGauge.WithLabelValues(s.SocketURL).Set(float64(evt.Seq))
···
173
175
174
176
rr, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks))
175
177
if err != nil {
176
-
log.Errorf("failed to read repo from car: %+v\n", err)
178
+
log.Errorf("failed to read repo from car: %+v", err)
177
179
return nil
178
180
}
179
181
···
183
185
}
184
186
185
187
// Parse time from the event time string
186
-
t, err := time.Parse(time.RFC3339, evt.Time)
188
+
evtCreatedAt, err := time.Parse(time.RFC3339, evt.Time)
187
189
if err != nil {
188
190
log.Errorf("error parsing time: %+v", err)
189
191
return nil
190
192
}
191
193
192
-
lastSeqCommittedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
193
-
lastSeqProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(start.UnixNano()))
194
+
lastEvtCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(evtCreatedAt.UnixNano()))
195
+
lastEvtProcessedAtGauge.WithLabelValues(s.SocketURL).Set(float64(processedAt.UnixNano()))
196
+
lastEvtCreatedEvtProcessedGapGauge.WithLabelValues(s.SocketURL).Set(float64(processedAt.Sub(evtCreatedAt).Seconds()))
194
197
195
198
for _, op := range evt.Ops {
196
199
collection := strings.Split(op.Path, "/")[0]
···
206
209
rc, rec, err := rr.GetRecord(ctx, op.Path)
207
210
if err != nil {
208
211
e := fmt.Errorf("getting record %s (%s) within seq %d for %s: %w", op.Path, *op.Cid, evt.Seq, evt.Repo, err)
209
-
log.Errorf("failed to get a record from the event: %+v\n", e)
212
+
log.Errorf("failed to get a record from the event: %+v", e)
210
213
break
211
214
}
212
215
213
216
// Verify that the record cid matches the cid in the event
214
217
if lexutil.LexLink(rc) != *op.Cid {
215
218
e := fmt.Errorf("mismatch in record and op cid: %s != %s", rc, *op.Cid)
216
-
log.Errorf("failed to LexLink the record in the event: %+v\n", e)
219
+
log.Errorf("failed to LexLink the record in the event: %+v", e)
217
220
break
218
221
}
219
222
223
+
var recCreatedAt time.Time
224
+
var parseError error
225
+
220
226
// Unpack the record and process it
221
227
switch rec := rec.(type) {
222
228
case *bsky.FeedPost:
···
224
230
if rec.Embed != nil && rec.Embed.EmbedRecord != nil && rec.Embed.EmbedRecord.Record != nil {
225
231
quoteRepostsProcessedCounter.WithLabelValues(s.SocketURL).Inc()
226
232
}
227
-
// Parse time from the event time string
228
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
229
-
if err != nil {
230
-
log.Errorf("error parsing time: %+v", err)
231
-
continue
232
-
}
233
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
233
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
234
234
case *bsky.FeedLike:
235
235
recordsProcessedCounter.WithLabelValues("feed_like", s.SocketURL).Inc()
236
-
// Parse time from the event time string
237
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
238
-
if err != nil {
239
-
log.Errorf("error parsing time: %+v", err)
240
-
continue
241
-
}
242
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
236
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
243
237
case *bsky.FeedRepost:
244
238
recordsProcessedCounter.WithLabelValues("feed_repost", s.SocketURL).Inc()
245
-
// Parse time from the event time string
246
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
247
-
if err != nil {
248
-
log.Errorf("error parsing time: %+v", err)
249
-
continue
250
-
}
251
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
239
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
252
240
case *bsky.GraphBlock:
253
241
recordsProcessedCounter.WithLabelValues("graph_block", s.SocketURL).Inc()
254
-
// Parse time from the event time string
255
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
256
-
if err != nil {
257
-
log.Errorf("error parsing time: %+v", err)
258
-
continue
259
-
}
260
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
242
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
261
243
case *bsky.GraphFollow:
262
244
recordsProcessedCounter.WithLabelValues("graph_follow", s.SocketURL).Inc()
263
-
// Parse time from the event time string
264
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
265
-
if err != nil {
266
-
log.Errorf("error parsing time: %+v", err)
267
-
continue
268
-
}
269
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
245
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
270
246
case *bsky.ActorProfile:
271
247
recordsProcessedCounter.WithLabelValues("actor_profile", s.SocketURL).Inc()
272
-
// Parse time from the event time string
273
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
274
-
if err != nil {
275
-
log.Errorf("error parsing time: %+v", err)
276
-
continue
277
-
}
278
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
279
248
case *bsky.FeedGenerator:
280
249
recordsProcessedCounter.WithLabelValues("feed_generator", s.SocketURL).Inc()
281
-
// Parse time from the event time string
282
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
283
-
if err != nil {
284
-
log.Errorf("error parsing time: %+v", err)
285
-
continue
286
-
}
287
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
250
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
288
251
case *bsky.GraphList:
289
252
recordsProcessedCounter.WithLabelValues("graph_list", s.SocketURL).Inc()
290
-
// Parse time from the event time string
291
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
292
-
if err != nil {
293
-
log.Errorf("error parsing time: %+v", err)
294
-
continue
295
-
}
296
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
253
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
297
254
case *bsky.GraphListitem:
298
255
recordsProcessedCounter.WithLabelValues("graph_listitem", s.SocketURL).Inc()
299
-
// Parse time from the event time string
300
-
recCreatedAt, err := time.Parse(util.ISO8601, evt.Time)
301
-
if err != nil {
302
-
log.Errorf("error parsing time: %+v", err)
303
-
continue
304
-
}
305
-
lastSeqCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
256
+
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
306
257
default:
307
258
log.Warnf("unknown record type: %+v", rec)
308
259
}
260
+
if parseError != nil {
261
+
log.Errorf("error parsing time: %+v", parseError)
262
+
continue
263
+
}
264
+
if !recCreatedAt.IsZero() {
265
+
lastEvtCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(recCreatedAt.UnixNano()))
266
+
lastEvtCreatedRecordCreatedGapGauge.WithLabelValues(s.SocketURL).Set(float64(evtCreatedAt.Sub(recCreatedAt).Seconds()))
267
+
lastRecordCreatedEvtProcessedGapGauge.WithLabelValues(s.SocketURL).Set(float64(processedAt.Sub(recCreatedAt).Seconds()))
268
+
}
309
269
310
270
case repomgr.EvtKindDeleteRecord:
311
271
default:
···
313
273
}
314
274
}
315
275
316
-
eventProcessingDurationHistogram.WithLabelValues(s.SocketURL).Observe(time.Since(start).Seconds())
276
+
eventProcessingDurationHistogram.WithLabelValues(s.SocketURL).Observe(time.Since(processedAt).Seconds())
317
277
return nil
318
278
}