diff --git a/go.mod b/go.mod index 858b071..29ac541 100644 --- a/go.mod +++ b/go.mod @@ -1,15 +1,15 @@ module go.unistack.org/micro-broker-kgo/v4 -go 1.23.8 +go 1.24.0 require ( - github.com/stretchr/testify v1.10.0 + github.com/stretchr/testify v1.11.1 github.com/twmb/franz-go v1.19.5 - github.com/twmb/franz-go/pkg/kadm v1.16.0 + github.com/twmb/franz-go/pkg/kadm v1.16.1 github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3 github.com/twmb/franz-go/pkg/kmsg v1.11.2 - go.opentelemetry.io/otel v1.36.0 - go.unistack.org/micro/v4 v4.1.17 + go.opentelemetry.io/otel v1.38.0 + go.unistack.org/micro/v4 v4.1.21 ) require ( @@ -19,9 +19,9 @@ require ( github.com/matoous/go-nanoid v1.5.1 // indirect github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect - github.com/spf13/cast v1.9.2 // indirect + github.com/spf13/cast v1.10.0 // indirect go.unistack.org/micro-proto/v4 v4.1.0 // indirect - golang.org/x/crypto v0.39.0 // indirect - google.golang.org/protobuf v1.36.6 // indirect + golang.org/x/crypto v0.43.0 // indirect + google.golang.org/protobuf v1.36.10 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 999e454..84edc68 100644 --- a/go.sum +++ b/go.sum @@ -20,28 +20,28 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= -github.com/spf13/cast v1.9.2 h1:SsGfm7M8QOFtEzumm7UZrZdLLquNdzFYfIbEXntcFbE= -github.com/spf13/cast v1.9.2/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= -github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= +github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/twmb/franz-go v1.19.5 h1:W7+o8D0RsQsedqib71OVlLeZ0zI6CbFra7yTYhZTs5Y= github.com/twmb/franz-go v1.19.5/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM= -github.com/twmb/franz-go/pkg/kadm v1.16.0 h1:STMs1t5lYR5mR974PSiwNzE5TvsosByTp+rKXLOhAjE= -github.com/twmb/franz-go/pkg/kadm v1.16.0/go.mod h1:MUdcUtnf9ph4SFBLLA/XxE29rvLhWYLM9Ygb8dfSCvw= +github.com/twmb/franz-go/pkg/kadm v1.16.1 h1:IEkrhTljgLHJ0/hT/InhXGjPdmWfFvxp7o/MR7vJ8cw= +github.com/twmb/franz-go/pkg/kadm v1.16.1/go.mod h1:Ue/ye1cc9ipsQFg7udFbbGiFNzQMqiH73fGC2y0rwyc= github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3 h1:p24opKWPySAy8xSl8NqRgOv7Q+bX7kdrQirBVRJzQfo= github.com/twmb/franz-go/pkg/kfake v0.0.0-20250508175730-72e1646135e3/go.mod h1:7uQs3Ae6HkWT1Y9elMbqtAcNFCI0y6+iS+Phw49L49U= github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg= github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE= -go.opentelemetry.io/otel v1.36.0 h1:UumtzIklRBY6cI/lllNZlALOF5nNIzJVb16APdvgTXg= -go.opentelemetry.io/otel v1.36.0/go.mod h1:/TcFMXYjyRNh8khOAO9ybYkqaDBb/70aVwkNML4pP8E= +go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8= +go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM= go.unistack.org/micro-proto/v4 v4.1.0 h1:qPwL2n/oqh9RE3RTTDgt28XK3QzV597VugQPaw9lKUk= go.unistack.org/micro-proto/v4 v4.1.0/go.mod h1:ArmK7o+uFvxSY3dbJhKBBX4Pm1rhWdLEFf3LxBrMtec= -go.unistack.org/micro/v4 v4.1.17 h1:26QDtRSYVpozYuassyvLP4sEQRo3dxgD3sVILRXmIPo= -go.unistack.org/micro/v4 v4.1.17/go.mod h1:xleO2M5Yxh4s6I+RUcLrEpUjobefh+71ctrdIfn7TUs= -golang.org/x/crypto v0.39.0 h1:SHs+kF4LP+f+p14esP5jAoDpHU8Gu/v9lFRK6IT5imM= -golang.org/x/crypto v0.39.0/go.mod h1:L+Xg3Wf6HoL4Bn4238Z6ft6KfEpN0tJGo53AAPC632U= -google.golang.org/protobuf v1.36.6 h1:z1NpPI8ku2WgiWnf+t9wTPsn6eP1L7ksHUlkfLvd9xY= -google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= +go.unistack.org/micro/v4 v4.1.21 h1:F9PrbI1BhXSDS0FopwcO5wWrT2Xh38w9VhVZZBHjfg8= +go.unistack.org/micro/v4 v4.1.21/go.mod h1:nlBXTbx0rQrSZX4HPp2m57PHmpuGPWUd0O+jpUIiPto= +golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= +golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/kgo.go b/kgo.go index bed4ca4..9babf9a 100644 --- a/kgo.go +++ b/kgo.go @@ -6,11 +6,13 @@ import ( "errors" "fmt" "math/rand/v2" + "strconv" "strings" "sync" "sync/atomic" "time" + "github.com/twmb/franz-go/pkg/kadm" "github.com/twmb/franz-go/pkg/kgo" "github.com/twmb/franz-go/pkg/kmsg" "go.unistack.org/micro/v4/broker" @@ -245,6 +247,68 @@ func (k *Broker) Connect(ctx context.Context) error { k.connected.Store(1) k.mu.Unlock() + exposeLag := false + if k.opts.Context != nil { + if v, ok := k.opts.Context.Value(exposeLagKey{}).(bool); ok && v { + exposeLag = v + } + } + + if exposeLag { + var mu sync.Mutex + var lastUpdate time.Time + type pl struct { + p string + l float64 + } + + lag := make(map[string]map[string]pl) // topic => group => partition => lag + ac := kadm.NewClient(k.c) + + updateStats := func() { + mu.Lock() + if time.Since(lastUpdate) < DefaultStatsInterval { + return + } + mu.Unlock() + + k.mu.Lock() + groups := make([]string, 0, len(k.subs)) + for _, g := range k.subs { + groups = append(groups, g.opts.Group) + } + k.mu.Unlock() + + dgls, err := ac.Lag(ctx, groups...) + if err != nil || !dgls.Ok() { + k.opts.Logger.Error(k.opts.Context, "kgo describe group lag error", err) + return + } + + for gn, dgl := range dgls { + for tn, lmap := range dgl.Lag { + if _, ok := lag[tn]; !ok { + lag[tn] = make(map[string]pl) + } + for p, l := range lmap { + lag[tn][gn] = pl{p: strconv.Itoa(int(p)), l: float64(l.Lag)} + } + } + } + } + + for tn, dg := range lag { + for gn, gl := range dg { + k.opts.Meter.Gauge(semconv.BrokerGroupLag, + func() float64 { updateStats(); return gl.l }, + "topic", tn, + "group", gn, + "partition", gl.p) + } + } + + } + return nil } diff --git a/options.go b/options.go index 4f7eae3..457c680 100644 --- a/options.go +++ b/options.go @@ -14,7 +14,7 @@ var ( DefaultCommitInterval = 5 * time.Second // DefaultStatsInterval specifies how fast check consumer lag - DefaultStatsInterval = 5 * time.Second + DefaultStatsInterval = 30 * time.Second // DefaultSubscribeMaxInflight specifies how much messages keep inflight DefaultSubscribeMaxInflight = 100 @@ -116,3 +116,10 @@ type subscribeMessagePoolKey struct{} func SubscribeMessagePool(b bool) broker.SubscribeOption { return broker.SetSubscribeOption(subscribeMessagePoolKey{}, b) } + +type exposeLagKey struct{} + +// ExposeLag enabled subscriber lag via [meter.Meter] +func ExposeLag(b bool) broker.Option { + return broker.SetOption(exposeLagKey{}, b) +} diff --git a/subscriber.go b/subscriber.go index cde2f68..5a7b300 100644 --- a/subscriber.go +++ b/subscriber.go @@ -8,7 +8,6 @@ import ( "sync/atomic" "time" - "github.com/twmb/franz-go/pkg/kadm" "github.com/twmb/franz-go/pkg/kgo" "github.com/twmb/franz-go/pkg/kmsg" "go.unistack.org/micro/v4/broker" @@ -87,46 +86,13 @@ func (s *Subscriber) Unsubscribe(ctx context.Context) error { func (s *Subscriber) poll(ctx context.Context) { maxInflight := DefaultSubscribeMaxInflight + if s.opts.Context != nil { if n, ok := s.opts.Context.Value(subscribeMaxInflightKey{}).(int); n > 0 && ok { maxInflight = n } } - go func() { - ac := kadm.NewClient(s.c) - ticker := time.NewTicker(DefaultStatsInterval) - - for { - select { - case <-ctx.Done(): - ticker.Stop() - return - case <-ticker.C: - dgls, err := ac.Lag(ctx, s.opts.Group) - if err != nil || !dgls.Ok() { - continue - } - - dgl, ok := dgls[s.opts.Group] - if !ok { - continue - } - lmap, ok := dgl.Lag[s.topic] - if !ok { - continue - } - - s.mu.Lock() - for p, l := range lmap { - s.kopts.Meter.Counter(semconv.BrokerGroupLag, "topic", s.topic, "group", s.opts.Group, "partition", strconv.Itoa(int(p))).Set(uint64(l.Lag)) - } - s.mu.Unlock() - - } - } - }() - for { select { case <-ctx.Done(): @@ -141,7 +107,11 @@ func (s *Subscriber) poll(ctx context.Context) { return } fetches.EachError(func(t string, p int32, err error) { - s.kopts.Logger.Fatal(ctx, fmt.Sprintf("[kgo] fetch topic %s partition %d error", t, p), err) + if kgo.IsRetryableBrokerErr(err) { + s.kopts.Logger.Error(ctx, fmt.Sprintf("[kgo] fetch topic %s partition %d error", t, p), err) + } else { + s.kopts.Logger.Fatal(ctx, fmt.Sprintf("[kgo] fetch topic %s partition %d error", t, p), err) + } }) fetches.EachPartition(func(p kgo.FetchTopicPartition) {