Skip to content

Commit 98e2fca

Browse files
serprexserprex
committed
replace github.com/siddontang/go-log with log/slog
1 parent cf91430 commit 98e2fca

20 files changed

+100
-148
lines changed

‎README.md

+1-21
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,6 @@ package main
147147

148148
import (
149149
"github.com/go-mysql-org/go-mysql/canal"
150-
"github.com/siddontang/go-log/log"
151150
)
152151

153152
type MyEventHandler struct {
@@ -537,26 +536,7 @@ We pass all tests in https://github.com/bradfitz/go-sql-test using go-mysql driv
537536

538537
## Logging
539538

540-
Logging by default is send to stdout.
541-
542-
To disable logging completely:
543-
```go
544-
import "github.com/siddontang/go-log/log"
545-
...
546-
nullHandler, _ := log.NewNullHandler()
547-
cfg.Logger = log.NewDefault(nullHandler)
548-
```
549-
550-
To write logging to any [`io.Writer`](https://pkg.go.dev/io#Writer):
551-
```go
552-
import "github.com/siddontang/go-log/log"
553-
...
554-
w := ...
555-
streamHandler, _ := log.NewStreamHandler(w)
556-
cfg.Logger = log.NewDefault(streamHandler)
557-
```
558-
559-
Or you can implement your own [`log.Handler`](https://pkg.go.dev/github.com/siddontang/go-log/log#Handler).
539+
Logging uses log/slog.
560540

561541
## How to migrate to this repo
562542
To change the used package in your repo it's enough to add this `replace` directive to your `go.mod`:

‎canal/canal.go

+6-7
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"context"
55
"fmt"
66
"io"
7+
"log/slog"
78
"net"
89
"os"
910
"regexp"
@@ -21,7 +22,6 @@ import (
2122
"github.com/go-mysql-org/go-mysql/utils"
2223
"github.com/pingcap/errors"
2324
"github.com/pingcap/tidb/pkg/parser"
24-
"github.com/siddontang/go-log/log"
2525
)
2626

2727
// Canal can sync your MySQL data into everywhere, like Elasticsearch, Redis, etc...
@@ -64,8 +64,7 @@ var ErrExcludedTable = errors.New("excluded table meta")
6464
func NewCanal(cfg *Config) (*Canal, error) {
6565
c := new(Canal)
6666
if cfg.Logger == nil {
67-
streamHandler, _ := log.NewStreamHandler(os.Stdout)
68-
cfg.Logger = log.NewDefault(streamHandler)
67+
cfg.Logger = slog.Default()
6968
}
7069
if cfg.Dialer == nil {
7170
dialer := &net.Dialer{}
@@ -241,14 +240,14 @@ func (c *Canal) run() error {
241240
close(c.dumpDoneCh)
242241

243242
if err != nil {
244-
c.cfg.Logger.Errorf("canal dump mysql err: %v", err)
243+
c.cfg.Logger.Error("canal dump mysql err", slog.Any("error", err))
245244
return errors.Trace(err)
246245
}
247246
}
248247

249248
if err := c.runSyncBinlog(); err != nil {
250249
if errors.Cause(err) != context.Canceled {
251-
c.cfg.Logger.Errorf("canal start sync binlog err: %v", err)
250+
c.cfg.Logger.Error("canal start sync binlog err", slog.Any("error", err))
252251
return errors.Trace(err)
253252
}
254253
}
@@ -257,7 +256,7 @@ func (c *Canal) run() error {
257256
}
258257

259258
func (c *Canal) Close() {
260-
c.cfg.Logger.Infof("closing canal")
259+
c.cfg.Logger.Info("closing canal")
261260
c.m.Lock()
262261
defer c.m.Unlock()
263262

@@ -377,7 +376,7 @@ func (c *Canal) GetTable(db string, table string) (*schema.Table, error) {
377376
c.errorTablesGetTime[key] = utils.Now()
378377
c.tableLock.Unlock()
379378
// log error and return ErrMissingTableMeta
380-
c.cfg.Logger.Errorf("canal get table meta err: %v", errors.Trace(err))
379+
c.cfg.Logger.Error("canal get table meta err", slog.Any("error", errors.Trace(err)))
381380
return nil, schema.ErrMissingTableMeta
382381
}
383382
return nil, err

‎canal/canal_test.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"time"
77

88
"github.com/pingcap/tidb/pkg/parser"
9-
"github.com/siddontang/go-log/log"
109
"github.com/stretchr/testify/require"
1110
"github.com/stretchr/testify/suite"
1211

@@ -99,7 +98,7 @@ func (s *canalTestSuite) SetupSuite() {
9998

10099
s.execute("SET GLOBAL binlog_format = 'ROW'")
101100

102-
s.c.SetEventHandler(&testEventHandler{})
101+
s.c.SetEventHandler(&testEventHandler{T: s.T()})
103102
go func() {
104103
set, _ := mysql.ParseGTIDSet("mysql", "")
105104
err = s.c.StartFromGTID(set)
@@ -126,10 +125,11 @@ func (s *canalTestSuite) execute(query string, args ...interface{}) *mysql.Resul
126125

127126
type testEventHandler struct {
128127
DummyEventHandler
128+
T *testing.T
129129
}
130130

131131
func (h *testEventHandler) OnRow(e *RowsEvent) error {
132-
log.Infof("OnRow %s %v\n", e.Action, e.Rows)
132+
h.T.Log("OnRow", e.Action, e.Rows)
133133
umi, ok := e.Rows[0][4].(uint32) // 4th col is umi. mysqldump gives uint64 instead of uint32
134134
if ok && (umi != umiA && umi != umiB && umi != umiC) {
135135
return fmt.Errorf("invalid unsigned medium int %d", umi)

‎canal/config.go

+3-5
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,14 @@ package canal
22

33
import (
44
"crypto/tls"
5+
"log/slog"
56
"math/rand"
67
"net"
78
"os"
89
"time"
910

1011
"github.com/BurntSushi/toml"
1112
"github.com/pingcap/errors"
12-
"github.com/siddontang/go-log/log"
13-
"github.com/siddontang/go-log/loggers"
1413

1514
"github.com/go-mysql-org/go-mysql/client"
1615
"github.com/go-mysql-org/go-mysql/mysql"
@@ -101,7 +100,7 @@ type Config struct {
101100
TLSConfig *tls.Config
102101

103102
// Set Logger
104-
Logger loggers.Advanced
103+
Logger *slog.Logger
105104

106105
// Set Dialer
107106
Dialer client.Dialer
@@ -150,8 +149,7 @@ func NewDefaultConfig() *Config {
150149
c.Dump.DiscardErr = true
151150
c.Dump.SkipMasterData = false
152151

153-
streamHandler, _ := log.NewStreamHandler(os.Stdout)
154-
c.Logger = log.NewDefault(streamHandler)
152+
c.Logger = slog.Default()
155153

156154
dialer := &net.Dialer{}
157155
c.Dialer = dialer.DialContext

‎canal/dump.go

+6-5
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package canal
33
import (
44
"encoding/hex"
55
"fmt"
6+
"log/slog"
67
"strconv"
78
"strings"
89
"time"
@@ -49,7 +50,7 @@ func (h *dumpParseHandler) Data(db string, table string, values []string) error
4950
e == schema.ErrMissingTableMeta {
5051
return nil
5152
}
52-
h.c.cfg.Logger.Errorf("get %s.%s information err: %v", db, table, err)
53+
h.c.cfg.Logger.Error("error getting table information", slog.String("databse", db), slog.String("table", table), slog.Any("error", err))
5354
return errors.Trace(err)
5455
}
5556

@@ -163,7 +164,7 @@ func (c *Canal) dump() error {
163164
if err != nil {
164165
return errors.Trace(err)
165166
}
166-
c.cfg.Logger.Infof("skip master data, get current binlog position %v", pos)
167+
c.cfg.Logger.Info("skip master data, get current binlog position", slog.Any("pos", pos))
167168
h.name = pos.Name
168169
h.pos = uint64(pos.Pos)
169170
}
@@ -185,8 +186,8 @@ func (c *Canal) dump() error {
185186
c.master.UpdateGTIDSet(h.gset)
186187
startPos = h.gset
187188
}
188-
c.cfg.Logger.Infof("dump MySQL and parse OK, use %0.2f seconds, start binlog replication at %s",
189-
time.Since(start).Seconds(), startPos)
189+
c.cfg.Logger.Info(fmt.Sprintf("dump MySQL and parse OK, use %0.2f seconds, start binlog replication at %s",
190+
time.Since(start).Seconds(), startPos))
190191
return nil
191192
}
192193

@@ -196,7 +197,7 @@ func (c *Canal) tryDump() error {
196197
if (len(pos.Name) > 0 && pos.Pos > 0) ||
197198
(gset != nil && gset.String() != "") {
198199
// we will sync with binlog name and position
199-
c.cfg.Logger.Infof("skip dump, use last binlog replication pos %s or GTID set %v", pos, gset)
200+
c.cfg.Logger.Info(fmt.Sprintf("skip dump, use last binlog replication pos %s or GTID set %v", pos, gset))
200201
return nil
201202
}
202203

‎canal/master.go

+5-5
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
package canal
22

33
import (
4+
"log/slog"
45
"sync"
56

67
"github.com/go-mysql-org/go-mysql/mysql"
7-
"github.com/siddontang/go-log/loggers"
88
)
99

1010
type masterInfo struct {
@@ -16,27 +16,27 @@ type masterInfo struct {
1616

1717
timestamp uint32
1818

19-
logger loggers.Advanced
19+
logger *slog.Logger
2020
}
2121

2222
func (m *masterInfo) Update(pos mysql.Position) {
23-
m.logger.Debugf("update master position %s", pos)
23+
m.logger.Debug("update master position", slog.Any("pos", pos))
2424

2525
m.Lock()
2626
m.pos = pos
2727
m.Unlock()
2828
}
2929

3030
func (m *masterInfo) UpdateTimestamp(ts uint32) {
31-
m.logger.Debugf("update master timestamp %d", ts)
31+
m.logger.Debug("update master timestamp", slog.Int64("ts", int64(ts)))
3232

3333
m.Lock()
3434
m.timestamp = ts
3535
m.Unlock()
3636
}
3737

3838
func (m *masterInfo) UpdateGTIDSet(gset mysql.GTIDSet) {
39-
m.logger.Debugf("update master gtid set %s", gset)
39+
m.logger.Debug("update master gtid set", slog.Any("gset", gset))
4040

4141
m.Lock()
4242
m.gset = gset

‎canal/sync.go

+11-9
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package canal
22

33
import (
4+
"fmt"
5+
"log/slog"
46
"sync/atomic"
57
"time"
68

@@ -20,15 +22,15 @@ func (c *Canal) startSyncer() (*replication.BinlogStreamer, error) {
2022
if err != nil {
2123
return nil, errors.Errorf("start sync replication at binlog %v error %v", pos, err)
2224
}
23-
c.cfg.Logger.Infof("start sync binlog at binlog file %v", pos)
25+
c.cfg.Logger.Info("start sync binlog at binlog file", slog.Any("pos", pos))
2426
return s, nil
2527
} else {
2628
gsetClone := gset.Clone()
2729
s, err := c.syncer.StartSyncGTID(gset)
2830
if err != nil {
2931
return nil, errors.Errorf("start sync replication at GTID set %v error %v", gset, err)
3032
}
31-
c.cfg.Logger.Infof("start sync binlog at GTID set %v", gsetClone)
33+
c.cfg.Logger.Info("start sync binlog at GTID set", slog.Any("gset", gsetClone))
3234
return s, nil
3335
}
3436
}
@@ -57,7 +59,7 @@ func (c *Canal) runSyncBinlog() error {
5759
// and https://github.com/mysql/mysql-server/blob/8cc757da3d87bf4a1f07dcfb2d3c96fed3806870/sql/rpl_binlog_sender.cc#L899
5860
if ev.Header.Timestamp == 0 {
5961
fakeRotateLogName := string(e.NextLogName)
60-
c.cfg.Logger.Infof("received fake rotate event, next log name is %s", e.NextLogName)
62+
c.cfg.Logger.Info("received fake rotate event", slog.String("nextLogName", string(e.NextLogName)))
6163

6264
if fakeRotateLogName != c.master.Position().Name {
6365
c.cfg.Logger.Info("log name changed, the fake rotate event will be handled as a real rotate event")
@@ -93,7 +95,7 @@ func (c *Canal) handleEvent(ev *replication.BinlogEvent) error {
9395
case *replication.RotateEvent:
9496
pos.Name = string(e.NextLogName)
9597
pos.Pos = uint32(e.Position)
96-
c.cfg.Logger.Infof("rotate binlog to %s", pos)
98+
c.cfg.Logger.Info("rotate binlog", slog.Any("pos", pos))
9799
savePos = true
98100
force = true
99101
if err = c.eventHandler.OnRotate(ev.Header, e); err != nil {
@@ -103,7 +105,7 @@ func (c *Canal) handleEvent(ev *replication.BinlogEvent) error {
103105
// we only focus row based event
104106
err = c.handleRowsEvent(ev)
105107
if err != nil {
106-
c.cfg.Logger.Errorf("handle rows event at (%s, %d) error %v", pos.Name, curPos, err)
108+
c.cfg.Logger.Error(fmt.Sprintf("handle rows event at (%s, %d)", pos.Name, curPos), slog.Any("error", err))
107109
return errors.Trace(err)
108110
}
109111
return nil
@@ -113,7 +115,7 @@ func (c *Canal) handleEvent(ev *replication.BinlogEvent) error {
113115
for _, subEvent := range ev.Events {
114116
err = c.handleEvent(subEvent)
115117
if err != nil {
116-
c.cfg.Logger.Errorf("handle transaction payload subevent at (%s, %d) error %v", pos.Name, curPos, err)
118+
c.cfg.Logger.Error(fmt.Sprintf("handle transaction payload subevent at (%s, %d)", pos.Name, curPos), slog.Any("error", err))
117119
return errors.Trace(err)
118120
}
119121
}
@@ -144,7 +146,7 @@ func (c *Canal) handleEvent(ev *replication.BinlogEvent) error {
144146
if err != nil {
145147
// The parser does not understand all syntax.
146148
// For example, it won't parse [CREATE|DROP] TRIGGER statements.
147-
c.cfg.Logger.Errorf("parse query(%s) err %v, will skip this event", e.Query, err)
149+
c.cfg.Logger.Error("error parsing query, will skip this event", slog.String("query", string(e.Query)), slog.Any("error", err))
148150
return nil
149151
}
150152
if len(stmts) > 0 {
@@ -246,7 +248,7 @@ func parseStmt(stmt ast.StmtNode) (ns []*node) {
246248

247249
func (c *Canal) updateTable(header *replication.EventHeader, db, table string) (err error) {
248250
c.ClearTableCache([]byte(db), []byte(table))
249-
c.cfg.Logger.Infof("table structure changed, clear table cache: %s.%s\n", db, table)
251+
c.cfg.Logger.Info("table structure changed, clear table cache", slog.String("database", db), slog.String("table", table))
250252
if err = c.eventHandler.OnTableChanged(header, db, table); err != nil && errors.Cause(err) != schema.ErrTableNotExist {
251253
return errors.Trace(err)
252254
}
@@ -315,7 +317,7 @@ func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error {
315317
if curPos.Compare(pos) >= 0 {
316318
return nil
317319
} else {
318-
c.cfg.Logger.Debugf("master pos is %v, wait catching %v", curPos, pos)
320+
c.cfg.Logger.Debug(fmt.Sprintf("master pos is %v, wait catching %v", curPos, pos))
319321
time.Sleep(100 * time.Millisecond)
320322
}
321323
}

0 commit comments

Comments
 (0)