this repo has no description
replace 'zap' with 'slog' (#514)
+118
-157
+13
-19
cmd/querycheck/main.go
+13
-19
cmd/querycheck/main.go
···
4
4
"context"
5
5
"fmt"
6
6
"log"
7
+
"log/slog"
7
8
"os"
8
9
"os/signal"
9
10
"sync"
···
21
22
22
23
"github.com/prometheus/client_golang/prometheus/promhttp"
23
24
"go.opentelemetry.io/otel/trace"
24
-
"go.uber.org/zap"
25
25
26
26
"github.com/carlmjohnson/versioninfo"
27
27
"github.com/urfave/cli/v2"
···
75
75
signals := make(chan os.Signal, 1)
76
76
signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
77
77
78
-
rawlog, err := zap.NewDevelopment()
79
-
if err != nil {
80
-
log.Fatalf("failed to create logger: %+v\n", err)
81
-
}
78
+
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
79
+
Level: slog.LevelInfo,
80
+
}))
82
81
defer func() {
83
-
log.Printf("main function teardown\n")
84
-
err := rawlog.Sync()
85
-
if err != nil {
86
-
log.Printf("failed to sync logger on teardown: %+v", err.Error())
87
-
}
82
+
logger.Info("main function teardown")
88
83
}()
89
84
90
-
log := rawlog.Sugar().With("source", "querycheck_main")
91
-
92
-
log.Info("starting querycheck")
85
+
logger = logger.With("source", "querycheck_main")
86
+
logger.Info("starting querycheck")
93
87
94
88
// Registers a tracer Provider globally if the exporter endpoint is set
95
89
if os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") != "" {
96
-
log.Info("initializing tracer...")
90
+
logger.Info("initializing tracer")
97
91
shutdown, err := tracing.InstallExportPipeline(ctx, "Querycheck", 1)
98
92
if err != nil {
99
93
log.Fatal(err)
···
153
147
// Start the metrics server
154
148
wg.Add(1)
155
149
go func() {
156
-
log.Infof("starting metrics server on port %d", cctx.Int("port"))
150
+
logger.Info("starting metrics serverd", "port", cctx.Int("port"))
157
151
if err := e.Start(fmt.Sprintf(":%d", cctx.Int("port"))); err != nil {
158
-
log.Errorf("failed to start metrics server: %+v\n", err)
152
+
logger.Error("failed to start metrics server", "err", err)
159
153
}
160
154
wg.Done()
161
155
}()
···
168
162
fmt.Println("shutting down on context done")
169
163
}
170
164
171
-
log.Info("shutting down, waiting for workers to clean up...")
165
+
logger.Info("shutting down, waiting for workers to clean up")
172
166
173
167
if err := e.Shutdown(ctx); err != nil {
174
-
log.Errorf("failed to shut down metrics server: %+v\n", err)
168
+
logger.Error("failed to shut down metrics server", "err", err)
175
169
wg.Done()
176
170
}
177
171
178
172
querychecker.Stop()
179
173
180
174
wg.Wait()
181
-
log.Info("shut down successfully")
175
+
logger.Info("shut down successfully")
182
176
183
177
return nil
184
178
}
+26
-45
cmd/sonar/main.go
+26
-45
cmd/sonar/main.go
···
4
4
"context"
5
5
"fmt"
6
6
"log"
7
+
"log/slog"
7
8
"net/http"
8
9
"net/url"
9
10
"os"
···
18
19
"github.com/gorilla/websocket"
19
20
"github.com/prometheus/client_golang/prometheus/promhttp"
20
21
_ "go.uber.org/automaxprocs"
21
-
"go.uber.org/zap"
22
22
23
23
"github.com/carlmjohnson/versioninfo"
24
24
"github.com/urfave/cli/v2"
···
80
80
// Trap SIGINT to trigger a shutdown.
81
81
signals := make(chan os.Signal, 1)
82
82
signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
83
-
84
-
rawlog, err := zap.NewProduction()
85
-
if err != nil {
86
-
log.Fatalf("failed to create logger: %+v", err)
87
-
}
83
+
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
84
+
Level: slog.LevelInfo,
85
+
}))
88
86
defer func() {
89
-
log.Printf("main function teardown")
90
-
err := rawlog.Sync()
91
-
if err != nil {
92
-
log.Printf("failed to sync logger on teardown: %+v", err.Error())
93
-
}
87
+
logger.Info("main function teardown")
94
88
}()
95
89
96
-
log := rawlog.Sugar().With("source", "sonar_main")
97
-
98
-
log.Info("starting sonar")
90
+
logger = logger.With("source", "sonar_main")
91
+
logger.Info("starting sonar")
99
92
100
93
u, err := url.Parse(cctx.String("ws-url"))
101
94
if err != nil {
102
95
log.Fatalf("failed to parse ws-url: %+v", err)
103
96
}
104
97
105
-
s, err := sonar.NewSonar(log, cctx.String("cursor-file"), u.String())
98
+
s, err := sonar.NewSonar(logger, cctx.String("cursor-file"), u.String())
106
99
if err != nil {
107
100
log.Fatalf("failed to create sonar: %+v", err)
108
101
}
···
120
113
wg.Add(1)
121
114
defer wg.Done()
122
115
ticker := time.NewTicker(5 * time.Second)
123
-
rawlog, err := zap.NewProduction()
124
-
if err != nil {
125
-
log.Fatalf("failed to create logger: %+v", err)
126
-
}
127
-
log := rawlog.Sugar().With("source", "cursor_file_manager")
116
+
logger := logger.With("source", "cursor_file_manager")
128
117
129
118
for {
130
119
select {
131
120
case <-ctx.Done():
132
-
log.Info("shutting down cursor file manager")
121
+
logger.Info("shutting down cursor file manager")
133
122
err := s.WriteCursorFile()
134
123
if err != nil {
135
-
log.Errorf("failed to write cursor file: %+v", err)
124
+
logger.Error("failed to write cursor file", "err", err)
136
125
}
137
-
log.Info("cursor file manager shut down successfully")
126
+
logger.Info("cursor file manager shut down successfully")
138
127
return
139
128
case <-ticker.C:
140
129
err := s.WriteCursorFile()
141
130
if err != nil {
142
-
log.Errorf("failed to write cursor file: %+v", err)
131
+
logger.Error("failed to write cursor file", "err", err)
143
132
}
144
133
}
145
134
}
···
152
141
ticker := time.NewTicker(15 * time.Second)
153
142
lastSeq := int64(0)
154
143
155
-
rawlog, err := zap.NewProduction()
156
-
if err != nil {
157
-
log.Fatalf("failed to create logger: %+v", err)
158
-
}
159
-
log := rawlog.Sugar().With("source", "liveness_checker")
144
+
logger = logger.With("source", "liveness_checker")
160
145
161
146
for {
162
147
select {
163
148
case <-ctx.Done():
164
-
log.Info("shutting down liveness checker")
149
+
logger.Info("shutting down liveness checker")
165
150
return
166
151
case <-ticker.C:
167
152
s.ProgMux.Lock()
168
153
seq := s.Progress.LastSeq
169
154
s.ProgMux.Unlock()
170
155
if seq <= lastSeq {
171
-
log.Errorf("no new events in last 15 seconds, shutting down for docker to restart me")
156
+
logger.Error("no new events in last 15 seconds, shutting down for docker to restart me")
172
157
cancel()
173
158
} else {
174
-
log.Infof("last event sequence: %d", seq)
159
+
logger.Info("last event sequence", "seq", seq)
175
160
lastSeq = seq
176
161
}
177
162
}
···
190
175
go func() {
191
176
wg.Add(1)
192
177
defer wg.Done()
193
-
rawlog, err := zap.NewProduction()
194
-
if err != nil {
195
-
log.Fatalf("failed to create logger: %+v", err)
196
-
}
197
-
log := rawlog.Sugar().With("source", "metrics_server")
178
+
logger = logger.With("source", "metrics_server")
198
179
199
-
log.Infof("metrics server listening on port %d", cctx.Int("port"))
180
+
logger.Info("metrics server listening", "port", cctx.Int("port"))
200
181
201
182
if err := metricServer.ListenAndServe(); err != http.ErrServerClosed {
202
183
log.Fatalf("failed to start metrics server: %+v", err)
203
184
}
204
-
log.Info("metrics server shut down successfully")
185
+
logger.Info("metrics server shut down successfully")
205
186
}()
206
187
207
188
if s.Progress.LastSeq >= 0 {
208
189
u.RawQuery = fmt.Sprintf("cursor=%d", s.Progress.LastSeq)
209
190
}
210
191
211
-
log.Infof("connecting to WebSocket at: %s", u.String())
192
+
logger.Info("connecting to WebSocket", "url", u.String())
212
193
c, _, err := websocket.DefaultDialer.Dial(u.String(), http.Header{
213
194
"User-Agent": []string{"sonar/1.0"},
214
195
})
215
196
if err != nil {
216
-
log.Infof("failed to connect to websocket: %v", err)
197
+
logger.Info("failed to connect to websocket", "err", err)
217
198
return err
218
199
}
219
200
defer c.Close()
···
222
203
wg.Add(1)
223
204
defer wg.Done()
224
205
err = events.HandleRepoStream(ctx, c, pool)
225
-
log.Infof("HandleRepoStream returned unexpectedly: %+v...", err)
206
+
logger.Info("HandleRepoStream returned unexpectedly", "err", err)
226
207
cancel()
227
208
}()
228
209
···
234
215
fmt.Println("shutting down on context done")
235
216
}
236
217
237
-
log.Info("shutting down, waiting for workers to clean up...")
218
+
logger.Info("shutting down, waiting for workers to clean up")
238
219
239
220
if err := metricServer.Shutdown(ctx); err != nil {
240
-
log.Errorf("failed to shut down metrics server: %+v", err)
221
+
logger.Error("failed to shut down metrics server", "err", err)
241
222
wg.Done()
242
223
}
243
224
244
225
wg.Wait()
245
-
log.Info("shut down successfully")
226
+
logger.Info("shut down successfully")
246
227
247
228
return nil
248
229
}
+45
-55
cmd/supercollider/main.go
+45
-55
cmd/supercollider/main.go
···
5
5
"crypto/rand"
6
6
"fmt"
7
7
"log"
8
+
"log/slog"
8
9
"net"
9
10
"os"
10
11
"os/signal"
···
49
50
50
51
"github.com/carlmjohnson/versioninfo"
51
52
cbg "github.com/whyrusleeping/cbor-gen"
52
-
"go.uber.org/zap"
53
53
)
54
54
55
55
var eventsGeneratedCounter = promauto.NewCounter(prometheus.CounterOpts{
···
67
67
Dids []string
68
68
Host string
69
69
EnableSSL bool
70
-
Logger *zap.SugaredLogger
70
+
Logger *slog.Logger
71
71
EventControl chan string
72
72
MultibaseKey string
73
73
RepoManager *repomgr.RepoManager
···
192
192
}
193
193
}()
194
194
195
-
rawlog, err := zap.NewDevelopment()
196
-
if err != nil {
197
-
log.Fatalf("failed to create logger: %+v\n", err)
198
-
}
195
+
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
196
+
Level: slog.LevelInfo,
197
+
}))
199
198
defer func() {
200
-
log.Printf("main function teardown\n")
201
-
err := rawlog.Sync()
202
-
if err != nil {
203
-
log.Printf("failed to sync logger on teardown: %+v", err.Error())
204
-
}
199
+
logger.Info("main function teardown")
205
200
}()
206
201
207
-
log := rawlog.Sugar().With("source", "supercollider_main")
202
+
logger = logger.With("source", "supercollider_main")
208
203
209
-
log.Info("Starting Supercollider in Reload Mode")
210
-
log.Infof("Generating %d total events and writing them to %s",
211
-
cctx.Int("total-events"), cctx.String("output-file"))
204
+
logger.Info("Starting Supercollider in Reload Mode")
205
+
logger.Info(fmt.Sprintf("Generating %d total events and writing them to %s",
206
+
cctx.Int("total-events"), cctx.String("output-file")))
212
207
213
208
em := events.NewEventManager(events.NewYoloPersister())
214
209
215
210
// Try to read the key from disk
216
211
keyBytes, err := os.ReadFile(cctx.String("key-file"))
217
212
if err != nil {
218
-
log.Warnf("failed to read key from disk, creating new key: %s", err.Error())
213
+
logger.Warn("failed to read key from disk, creating new key", "err", err.Error())
219
214
}
220
215
221
216
var privkey *godid.PrivKey
···
259
254
260
255
// Instantiate Server
261
256
s := &Server{
262
-
Logger: log,
257
+
Logger: logger,
263
258
EnableSSL: cctx.Bool("use-ssl"),
264
259
Host: cctx.String("hostname"),
265
260
···
316
311
}
317
312
defer cancel()
318
313
319
-
log.Infof("writing events to %s", outFile)
314
+
logger.Info("writing events", "path", outFile)
320
315
321
316
header := events.EventHeader{Op: events.EvtKindMessage}
322
317
for {
323
318
select {
324
319
case <-ctx.Done():
325
-
log.Info("shutting down file writer")
320
+
logger.Info("shutting down file writer")
326
321
err = f.Sync()
327
322
if err != nil {
328
-
log.Errorf("failed to sync file: %+v\n", err)
323
+
logger.Error("failed to sync file", "err", err)
329
324
}
330
-
log.Info("file writer shutdown complete")
325
+
logger.Info("file writer shutdown complete")
331
326
return
332
327
case evt := <-evts:
333
328
if evt.Error != nil {
334
-
log.Errorf("error in event stream: %+v\n", evt.Error)
329
+
logger.Error("error in event stream", "err", evt.Error)
335
330
continue
336
331
}
337
332
var obj lexutil.CBOR
···
355
350
header.MsgType = "#tombstone"
356
351
obj = evt.RepoTombstone
357
352
default:
358
-
log.Errorf("unrecognized event kind")
353
+
logger.Error("unrecognized event kind")
359
354
continue
360
355
}
361
356
362
357
if err := header.MarshalCBOR(f); err != nil {
363
-
log.Errorf("failed to write header: %+v\n", err)
358
+
logger.Error("failed to write header", "err", err)
364
359
}
365
360
366
361
if err := obj.MarshalCBOR(f); err != nil {
367
-
log.Errorf("failed to write event: %+v\n", err)
362
+
logger.Error("failed to write event", "err", err)
368
363
}
369
364
}
370
365
}
···
384
379
err = e.Start(listenAddress)
385
380
}
386
381
if err != nil {
387
-
log.Errorf("failed to start server: %+v\n", err)
382
+
logger.Error("failed to start server", "err", err)
388
383
}
389
384
}()
390
385
<-ctx.Done()
391
-
log.Info("shutting down server...")
386
+
logger.Info("shutting down server...")
392
387
wg.Wait()
393
-
log.Info("server shutdown complete")
388
+
logger.Info("server shutdown complete")
394
389
return nil
395
390
}
396
391
···
416
411
}
417
412
}()
418
413
419
-
rawlog, err := zap.NewDevelopment()
420
-
if err != nil {
421
-
log.Fatalf("failed to create logger: %+v\n", err)
422
-
}
414
+
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
415
+
Level: slog.LevelInfo,
416
+
}))
417
+
423
418
defer func() {
424
-
log.Printf("main function teardown\n")
425
-
err := rawlog.Sync()
426
-
if err != nil {
427
-
log.Printf("failed to sync logger on teardown: %+v", err.Error())
428
-
}
419
+
logger.Info("main function teardown")
429
420
}()
430
421
431
-
log := rawlog.Sugar().With("source", "supercollider_main")
432
-
433
-
log.Info("Starting Supercollider in Fire Mode")
422
+
logger = logger.With("source", "supercollider_main")
423
+
logger.Info("Starting Supercollider in Fire Mode")
434
424
435
425
// Try to read the key from disk
436
426
keyBytes, err := os.ReadFile(cctx.String("key-file"))
437
427
if err != nil {
438
-
log.Warnf("failed to read key from disk, creating new key: %s", err.Error())
428
+
logger.Warn("failed to read key from disk, creating new key", "err", err.Error())
439
429
}
440
430
441
431
var privkey *godid.PrivKey
···
466
456
467
457
// Instantiate Server
468
458
s := &Server{
469
-
Logger: log,
459
+
Logger: logger,
470
460
EnableSSL: cctx.Bool("use-ssl"),
471
461
Host: cctx.String("hostname"),
472
462
MultibaseKey: *vMethod.PublicKeyMultibase,
···
489
479
if err2 := ctx.JSON(err.Code, map[string]any{
490
480
"error": err.Message,
491
481
}); err2 != nil {
492
-
log.Errorf("Failed to write http error: %s", err2)
482
+
logger.Error("Failed to write http error", "err", err2)
493
483
}
494
484
default:
495
485
sendHeader := true
···
497
487
sendHeader = false
498
488
}
499
489
500
-
log.Warnf("HANDLER ERROR: (%s) %s", ctx.Path(), err)
490
+
logger.Warn("HANDLER ERROR", "path", ctx.Path(), "err", err)
501
491
502
492
if sendHeader {
503
493
ctx.Response().WriteHeader(500)
···
532
522
err = e.Start(listenAddress)
533
523
}
534
524
if err != nil {
535
-
log.Errorf("failed to start server: %+v\n", err)
525
+
logger.Error("failed to start server", "err", err)
536
526
}
537
527
}()
538
528
<-ctx.Done()
539
-
log.Info("shutting down server")
529
+
logger.Info("shutting down server")
540
530
return nil
541
531
}
542
532
···
620
610
},
621
611
PrivUid: evt.User,
622
612
}); err != nil {
623
-
s.Logger.Errorf("failed to add event: %+v\n", err)
613
+
s.Logger.Error("failed to add event", "err", err)
624
614
}
625
615
}
626
616
627
617
// EventGenerationLoop is the main loop for generating events
628
618
func (s *Server) EventGenerationLoop(ctx context.Context, cancel context.CancelFunc) {
629
619
defer cancel()
630
-
s.Logger.Infof("starting event generation for %d events", s.TotalDesiredEvents)
620
+
s.Logger.Info(fmt.Sprintf("starting event generation for %d events", s.TotalDesiredEvents))
631
621
632
-
s.Logger.Infof("initializing %d fake users", len(s.Dids))
622
+
s.Logger.Info(fmt.Sprintf("initializing %d fake users", len(s.Dids)))
633
623
for i, did := range s.Dids {
634
624
uid := models.Uid(i + 1)
635
625
if err := s.RepoManager.InitNewActor(ctx, uid, strings.TrimPrefix(did, "did:web:"), did, "catdog", "", ""); err != nil {
···
637
627
}
638
628
}
639
629
640
-
s.Logger.Infof("generating %d events", s.TotalDesiredEvents)
630
+
s.Logger.Info("generating events", "count", s.TotalDesiredEvents)
641
631
642
632
for i := 0; i < s.TotalDesiredEvents; i++ {
643
633
text := fake.SentencesN(3)
···
650
640
Text: text,
651
641
})
652
642
if err != nil {
653
-
s.Logger.Errorf("failed to create record: %+v\n", err)
643
+
s.Logger.Error("failed to create record", "err", err)
654
644
} else {
655
645
eventsGeneratedCounter.Inc()
656
646
}
657
647
select {
658
648
case <-ctx.Done():
659
-
s.Logger.Infof("shutting down event generation loop on context done")
649
+
s.Logger.Info("shutting down event generation loop on context done")
660
650
return
661
651
default:
662
652
}
663
653
}
664
654
665
-
s.Logger.Infof("event generation complete, shutting down")
655
+
s.Logger.Info("event generation complete, shutting down")
666
656
return
667
657
}
668
658
···
712
702
713
703
// HandleSubscribeRepos opens and manages a websocket connection for subscribing to repo events
714
704
func (s *Server) HandleSubscribeRepos(c echo.Context) error {
715
-
s.Logger.Infof("new repo subscription from %s\n", c.Request().RemoteAddr)
705
+
s.Logger.Info("new repo subscription", "remote", c.Request().RemoteAddr)
716
706
conn, err := websocket.Upgrade(c.Response().Writer, c.Request(), c.Response().Header(), 1<<10, 1<<10)
717
707
if err != nil {
718
708
return err
···
725
715
726
716
f, err := os.Open(s.PlaybackFile)
727
717
if err != nil {
728
-
s.Logger.Errorf("failed to open playback file: %+v\n", err)
718
+
s.Logger.Error("failed to open playback file", "err", err)
729
719
return err
730
720
}
731
721
defer f.Close()
+2
-2
go.mod
+2
-2
go.mod
···
36
36
github.com/joho/godotenv v1.5.1
37
37
github.com/labstack/echo-contrib v0.15.0
38
38
github.com/labstack/echo/v4 v4.11.3
39
-
github.com/labstack/gommon v0.4.1
40
39
github.com/lestrrat-go/jwx/v2 v2.0.12
41
40
github.com/lib/pq v1.10.9
42
41
github.com/minio/sha256-simd v1.0.1
···
64
63
go.opentelemetry.io/otel/sdk v1.21.0
65
64
go.opentelemetry.io/otel/trace v1.21.0
66
65
go.uber.org/automaxprocs v1.5.3
67
-
go.uber.org/zap v1.26.0
68
66
golang.org/x/crypto v0.17.0
69
67
golang.org/x/sync v0.5.0
70
68
golang.org/x/text v0.14.0
···
82
80
github.com/hashicorp/golang-lru v1.0.2 // indirect
83
81
github.com/jackc/puddle/v2 v2.2.1 // indirect
84
82
github.com/klauspost/compress v1.17.3 // indirect
83
+
github.com/labstack/gommon v0.4.1 // indirect
85
84
github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
86
85
github.com/petar/GoLLRB v0.0.0-20210522233825-ae3b015fd3e9 // indirect
87
86
github.com/vmihailenco/go-tinylfu v0.2.2 // indirect
88
87
github.com/vmihailenco/msgpack/v5 v5.4.1 // indirect
89
88
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
90
89
github.com/whyrusleeping/cbor v0.0.0-20171005072247-63513f603b11 // indirect
90
+
go.uber.org/zap v1.26.0 // indirect
91
91
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa // indirect
92
92
)
93
93
+18
-21
querycheck/check.go
+18
-21
querycheck/check.go
···
2
2
3
3
import (
4
4
"context"
5
-
"log"
5
+
"fmt"
6
+
"log/slog"
6
7
"math"
8
+
"os"
7
9
"sync"
8
10
"time"
9
11
10
12
"github.com/jackc/pgx/v5"
11
13
"go.opentelemetry.io/otel"
12
14
"go.opentelemetry.io/otel/attribute"
13
-
"go.uber.org/zap"
14
15
)
15
16
16
17
var tracer = otel.Tracer("querycheck")
···
33
34
// Querychecker is a query checker meta object
34
35
type Querychecker struct {
35
36
Queries []*Query
36
-
Logger *zap.SugaredLogger
37
+
Logger *slog.Logger
37
38
38
39
connectionURL string
39
40
lk sync.RWMutex
···
41
42
42
43
// NewQuerychecker creates a new querychecker
43
44
func NewQuerychecker(ctx context.Context, connectionURL string) (*Querychecker, error) {
44
-
logger, err := zap.NewDevelopment()
45
-
if err != nil {
46
-
return nil, err
47
-
}
48
-
l := logger.Sugar().With("source", "querychecker_manager")
45
+
logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
46
+
Level: slog.LevelInfo,
47
+
}))
48
+
logger = logger.With("source", "querychecker_manager")
49
49
50
50
return &Querychecker{
51
51
connectionURL: connectionURL,
52
-
Logger: l,
52
+
Logger: logger,
53
53
Queries: []*Query{},
54
54
}, nil
55
55
}
···
167
167
168
168
for _, qu := range q.Queries {
169
169
go func(query *Query) {
170
-
rawlog, err := zap.NewDevelopment()
171
-
if err != nil {
172
-
log.Fatalf("failed to create logger: %+v\n", err)
173
-
}
174
-
log := rawlog.Sugar().With("source", "query_checker_routine", "query", query.Name)
170
+
log := q.Logger.With("source", "query_checker_routine", "query", query.Name)
175
171
176
-
log.Infof("query checker routine started for query: %s\n", query.Name)
177
-
log.Infof("Query: \n%s\n", query.Query)
172
+
log.Info("query checker routine started for query", "query", query.Name)
173
+
log.Info(fmt.Sprintf("Query: \n%s\n", query.Query))
178
174
179
175
// Check the query plan every CheckEvery duration
180
176
ticker := time.NewTicker(query.CheckEvery)
181
177
defer ticker.Stop()
182
178
179
+
var err error
183
180
query.LatestPlan, err = q.CheckQueryPlan(ctx, query.Query)
184
181
if err != nil {
185
-
log.Errorf("failed to check query plan: %+v\n", err)
182
+
log.Error("failed to check query plan", "err", err)
186
183
}
187
184
188
185
if query.LatestPlan != nil {
189
-
log.Infof("Initial plan:\n%+v\n", query.LatestPlan.String())
186
+
log.Info(fmt.Sprintf("Initial plan:\n%+v\n", query.LatestPlan.String()))
190
187
query.RecordPlanMetrics(*query.LatestPlan)
191
188
query.LastChecked = time.Now()
192
189
}
···
211
208
212
209
if err != nil || qp == nil {
213
210
if qp == nil {
214
-
log.Errorf("query plan is nil")
211
+
log.Error("query plan is nil")
215
212
}
216
-
log.Errorf("failed to check query plan: %+v\n", err)
213
+
log.Error("failed to check query plan", "err", err)
217
214
errorCounter.WithLabelValues(query.Name).Inc()
218
215
continue
219
216
}
···
231
228
sign = "-"
232
229
}
233
230
234
-
log.Infof("query plan has changed (%s%.03fms): \n%+v\n", sign, diff, qp.String())
231
+
log.Info("query plan has changed", "diff", fmt.Sprintf("%s%.03fms", sign, diff), "query_plan", qp.String())
235
232
236
233
query.lk.Lock()
237
234
query.PreviousPlan = query.LatestPlan
+14
-15
sonar/sonar.go
+14
-15
sonar/sonar.go
···
4
4
"bytes"
5
5
"context"
6
6
"fmt"
7
+
"log/slog"
7
8
"os"
8
9
"strings"
9
10
"sync"
···
14
15
"github.com/bluesky-social/indigo/api/bsky"
15
16
lexutil "github.com/bluesky-social/indigo/lex/util"
16
17
"github.com/goccy/go-json"
17
-
"github.com/labstack/gommon/log"
18
18
19
19
"github.com/bluesky-social/indigo/events"
20
20
"github.com/bluesky-social/indigo/repo"
21
21
"github.com/bluesky-social/indigo/repomgr"
22
22
"go.opentelemetry.io/otel"
23
-
"go.uber.org/zap"
24
23
)
25
24
26
25
type Sonar struct {
27
26
SocketURL string
28
27
Progress *Progress
29
28
ProgMux sync.Mutex
30
-
Logger *zap.SugaredLogger
29
+
Logger *slog.Logger
31
30
CursorFile string
32
31
}
33
32
···
72
71
return nil
73
72
}
74
73
75
-
func NewSonar(logger *zap.SugaredLogger, cursorFile string, socketURL string) (*Sonar, error) {
74
+
func NewSonar(logger *slog.Logger, cursorFile string, socketURL string) (*Sonar, error) {
76
75
s := Sonar{
77
76
SocketURL: socketURL,
78
77
Progress: &Progress{
···
85
84
86
85
// Check to see if the cursor file exists
87
86
if _, err := os.Stat(cursorFile); os.IsNotExist(err) {
88
-
logger.Infof("cursor file does not exist, creating %s", cursorFile)
87
+
logger.Info("cursor file does not exist, creating", "path", cursorFile)
89
88
// Create the cursor file
90
89
err := s.WriteCursorFile()
91
90
if err != nil {
···
95
94
// Read the cursor file
96
95
err := s.ReadCursorFile()
97
96
if err != nil {
98
-
logger.Errorf("read cursor file, will start drinking from live: %+v", err.Error())
97
+
logger.Error("read cursor file, will start drinking from live", "err", err.Error())
99
98
}
100
99
}
101
100
···
120
119
// Parse time from the event time string
121
120
t, err := time.Parse(time.RFC3339, xe.RepoHandle.Time)
122
121
if err != nil {
123
-
log.Errorf("error parsing time: %+v", err)
122
+
s.Logger.Error("error parsing time", "err", err)
124
123
return nil
125
124
}
126
125
lastEvtCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
···
139
138
// Parse time from the event time string
140
139
t, err := time.Parse(time.RFC3339, xe.RepoMigrate.Time)
141
140
if err != nil {
142
-
log.Errorf("error parsing time: %+v", err)
141
+
s.Logger.Error("error parsing time", "err", err)
143
142
return nil
144
143
}
145
144
lastEvtCreatedAtGauge.WithLabelValues(s.SocketURL).Set(float64(t.UnixNano()))
···
175
174
176
175
rr, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks))
177
176
if err != nil {
178
-
log.Errorf("failed to read repo from car: %+v", err)
177
+
s.Logger.Error("failed to read repo from car", "err", err)
179
178
return nil
180
179
}
181
180
···
187
186
// Parse time from the event time string
188
187
evtCreatedAt, err := time.Parse(time.RFC3339, evt.Time)
189
188
if err != nil {
190
-
log.Errorf("error parsing time: %+v", err)
189
+
s.Logger.Error("error parsing time", "err", err)
191
190
return nil
192
191
}
193
192
···
209
208
rc, rec, err := rr.GetRecord(ctx, op.Path)
210
209
if err != nil {
211
210
e := fmt.Errorf("getting record %s (%s) within seq %d for %s: %w", op.Path, *op.Cid, evt.Seq, evt.Repo, err)
212
-
log.Errorf("failed to get a record from the event: %+v", e)
211
+
s.Logger.Error("failed to get a record from the event", "err", e)
213
212
break
214
213
}
215
214
216
215
// Verify that the record cid matches the cid in the event
217
216
if lexutil.LexLink(rc) != *op.Cid {
218
217
e := fmt.Errorf("mismatch in record and op cid: %s != %s", rc, *op.Cid)
219
-
log.Errorf("failed to LexLink the record in the event: %+v", e)
218
+
s.Logger.Error("failed to LexLink the record in the event", "err", e)
220
219
break
221
220
}
222
221
···
255
254
recordsProcessedCounter.WithLabelValues("graph_listitem", s.SocketURL).Inc()
256
255
recCreatedAt, parseError = dateparse.ParseAny(rec.CreatedAt)
257
256
default:
258
-
log.Warnf("unknown record type: %+v", rec)
257
+
log.Warn("unknown record type", "rec", rec)
259
258
}
260
259
if parseError != nil {
261
-
log.Errorf("error parsing time: %+v", parseError)
260
+
s.Logger.Error("error parsing time", "err", parseError)
262
261
continue
263
262
}
264
263
if !recCreatedAt.IsZero() {
···
269
268
270
269
case repomgr.EvtKindDeleteRecord:
271
270
default:
272
-
log.Warnf("unknown event kind from op action: %+v", op.Action)
271
+
s.Logger.Warn("unknown event kind from op action", "action", op.Action)
273
272
}
274
273
}
275
274