add ExposeLag option, to be able to disable lag exporting
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
16
go.mod
16
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
|
||||
)
|
||||
|
||||
28
go.sum
28
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=
|
||||
|
||||
64
kgo.go
64
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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user