Coffee journaling on ATProto (alpha)
alpha.arabica.social
coffee
feat: otel tracing for startup backfill
+24
-5
+24
-5
cmd/server/main.go
+24
-5
cmd/server/main.go
···
29
29
"arabica/internal/routing"
30
30
"arabica/internal/tracing"
31
31
32
+
"go.opentelemetry.io/otel/attribute"
32
33
"github.com/prometheus/client_golang/prometheus/promhttp"
33
34
"github.com/rs/zerolog"
34
35
"github.com/rs/zerolog/log"
···
271
272
go func() {
272
273
time.Sleep(5 * time.Second) // Wait for initial connection
273
274
275
+
// Create a root span so all backfill work is grouped under one trace
276
+
backfillCtx, backfillSpan := tracing.HandlerSpan(ctx, "backfill.startup")
277
+
defer backfillSpan.End()
278
+
274
279
// Collect all DIDs to backfill
280
+
collectCtx, collectSpan := tracing.HandlerSpan(backfillCtx, "backfill.collect_dids")
275
281
didsToBackfill := make(map[string]struct{})
276
282
277
283
// Add registered users
···
295
301
Msg("Loaded known DIDs from file")
296
302
}
297
303
}
304
+
collectSpan.SetAttributes(attribute.Int("backfill.total_dids", len(didsToBackfill)))
305
+
collectSpan.End()
298
306
299
307
// Pre-load already-backfilled DIDs to avoid N+1 SELECT per DID
300
-
alreadyBackfilled, err := firehoseConsumer.BackfilledDIDs(ctx)
308
+
filterCtx, filterSpan := tracing.HandlerSpan(collectCtx, "backfill.filter_backfilled")
309
+
alreadyBackfilled, err := firehoseConsumer.BackfilledDIDs(filterCtx)
301
310
if err != nil {
302
311
log.Warn().Err(err).Msg("Failed to load backfilled DIDs, will check individually")
303
312
alreadyBackfilled = make(map[string]struct{})
···
307
316
for did := range alreadyBackfilled {
308
317
delete(didsToBackfill, did)
309
318
}
310
-
311
-
// Create a root span so all backfill PDS calls are grouped under one trace
312
-
backfillCtx, backfillSpan := tracing.HandlerSpan(ctx, "backfill.startup")
319
+
filterSpan.SetAttributes(
320
+
attribute.Int("backfill.already_backfilled", len(alreadyBackfilled)),
321
+
attribute.Int("backfill.remaining", len(didsToBackfill)),
322
+
)
323
+
filterSpan.End()
313
324
314
325
// Backfill remaining DIDs
326
+
_, execSpan := tracing.HandlerSpan(filterCtx, "backfill.execute",
327
+
attribute.Int("backfill.count", len(didsToBackfill)),
328
+
)
315
329
successCount := 0
316
330
for did := range didsToBackfill {
317
331
if err := firehoseConsumer.BackfillDID(backfillCtx, did); err != nil {
···
320
334
successCount++
321
335
}
322
336
}
323
-
backfillSpan.End()
337
+
execSpan.SetAttributes(
338
+
attribute.Int("backfill.success", successCount),
339
+
attribute.Int("backfill.failed", len(didsToBackfill)-successCount),
340
+
)
341
+
execSpan.End()
342
+
324
343
log.Info().
325
344
Int("skipped", len(alreadyBackfilled)).
326
345
Int("backfilled", successCount).