unistack-org/micro-broker-kgo
4
0
Fork 2
Code Issues 3 Pull Requests 2 Actions 8 Packages Projects Releases 44 Wiki Activity

meter: write metrics inside broker implementation #123

Merged
vtolstov merged 5 commits from improve into master 2023-04-29 21:32:47 +03:00
Conversation 0 Commits 5 Files Changed 10 +592 -516
10 changed files with 592 additions and 516 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
event.go Normal file
View File

@ -0,0 +1,42 @@
package kgo
import (
"sync"
"go.unistack.org/micro/v4/broker"
)
type event struct {
topic string
err error
sync.RWMutex
msg *broker.Message
ack bool
}
func (p *event) Topic() string {
return p.topic
}
func (p *event) Message() *broker.Message {
return p.msg
}
func (p *event) Ack() error {
p.ack = true
return nil
}
func (p *event) Error() error {
return p.err
}
func (p *event) SetError(err error) {
p.err = err
}
var eventPool = sync.Pool{
New: func() interface{} {
return &event{msg: &broker.Message{}}
},
}

6
go.mod
View File

@ -1,14 +1,14 @@
module go.unistack.org/micro-broker-kgo/v3
module go.unistack.org/micro-broker-kgo/v4
go 1.19
require (
github.com/twmb/franz-go v1.11.5
github.com/twmb/franz-go/pkg/kmsg v1.3.0
go.unistack.org/micro/v3 v3.10.19
go.unistack.org/micro/v4 v4.0.1
)
require (
github.com/klauspost/compress v1.15.9 // indirect
github.com/pierrec/lz4/v4 v4.1.15 // indirect
github.com/twmb/franz-go/pkg/kmsg v1.3.0 // indirect
)

4
go.sum
View File

@ -6,8 +6,8 @@ github.com/twmb/franz-go v1.11.5 h1:TTv5lVJd+87XkmP9dWN9Jgpf7IUUr7a7jee+byR8LBE=
github.com/twmb/franz-go v1.11.5/go.mod h1:FvaHNlpT6woVYIl6LAuIeL7yHol1Fp6Gv2Dn21AvH78=
github.com/twmb/franz-go/pkg/kmsg v1.3.0 h1:ouBETB7nTqRxiO5E8/pySoFZtVEW2VWw55z3/bsUzTw=
github.com/twmb/franz-go/pkg/kmsg v1.3.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=
go.unistack.org/micro/v3 v3.10.19 h1:JtPacglETxHyb+GiuMdfxqGtnWSTYXPQm6PlDQfNIl8=
go.unistack.org/micro/v3 v3.10.19/go.mod h1:XIArw29f0b3uvF4cq96X/nQt2f0J2OGnjh8J+DBbC0s=
go.unistack.org/micro/v4 v4.0.1 h1:xo1IxbVfgh8i0eY0VeYa3cbb13u5n/Mxnp3FOgWD4Jo=
go.unistack.org/micro/v4 v4.0.1/go.mod h1:p/J5UcSJjfHsWGT31uKoghQ5rUQZzQJBAFy+Z4+ZVMs=
golang.org/x/crypto v0.0.0-20220817201139-bc19a97f63c8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

407
kgo.go
View File

@ -1,109 +1,93 @@
// Package kgo provides a kafka broker using kgo
package kgo // import "go.unistack.org/micro-broker-kgo/v3"
package kgo // import "go.unistack.org/micro-broker-kgo/v4"
import (
"context"
"errors"
"fmt"
"math/rand"
"strings"
"sync"
"time"
"github.com/twmb/franz-go/pkg/kerr"
kgo "github.com/twmb/franz-go/pkg/kgo"
"github.com/twmb/franz-go/pkg/kmsg"
"github.com/twmb/franz-go/pkg/kversion"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
"go.unistack.org/micro/v3/util/id"
mrand "go.unistack.org/micro/v3/util/rand"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/metadata"
id "go.unistack.org/micro/v4/util/id"
mrand "go.unistack.org/micro/v4/util/rand"
)
var _ broker.Broker = &kBroker{}
var _ broker.Broker = &Broker{}
type kBroker struct {
writer *kgo.Client // used only to push messages
var ErrLostMessage = errors.New("message not marked for offsets commit and will be lost in next iteration")
var DefaultRetryBackoffFn = func() func(int) time.Duration {
var rngMu sync.Mutex
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
return func(fails int) time.Duration {
const (
min = 100 * time.Millisecond
max = time.Second
)
if fails <= 0 {
return min
}
if fails > 10 {
return max
}
backoff := min * time.Duration(1<<(fails-1))
rngMu.Lock()
jitter := 0.8 + 0.4*rng.Float64()
rngMu.Unlock()
backoff = time.Duration(float64(backoff) * jitter)
if backoff > max {
return max
}
return backoff
}
}()
type Broker struct {
c *kgo.Client
kopts []kgo.Opt
connected bool
init bool
sync.RWMutex
opts broker.Options
subs []*subscriber
}
type subscriber struct {
reader *kgo.Client // used only to pull messages
topic string
opts broker.SubscribeOptions
kopts broker.Options
handler broker.Handler
batchhandler broker.BatchHandler
closed bool
done chan struct{}
consumers map[string]map[int32]worker
sync.RWMutex
}
type publication struct {
topic string
err error
sync.RWMutex
msg *broker.Message
ack bool
}
func (p *publication) Topic() string {
return p.topic
}
func (p *publication) Message() *broker.Message {
return p.msg
}
func (p *publication) Ack() error {
p.ack = true
return nil
}
func (p *publication) Error() error {
return p.err
}
func (p *publication) SetError(err error) {
p.err = err
}
func (s *subscriber) Options() broker.SubscribeOptions {
return s.opts
}
func (s *subscriber) Topic() string {
return s.topic
}
func (s *subscriber) Unsubscribe(ctx context.Context) error {
if s.closed {
return nil
}
select {
case <-ctx.Done():
return ctx.Err()
default:
close(s.done)
s.closed = true
}
return nil
}
func (k *kBroker) Address() string {
func (k *Broker) Address() string {
return strings.Join(k.opts.Addrs, ",")
}
func (k *kBroker) Name() string {
func (k *Broker) Name() string {
return k.opts.Name
}
func (k *kBroker) Connect(ctx context.Context) error {
func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, error) {
var c *kgo.Client
var err error
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
c, err = kgo.NewClient(opts...)
if err == nil {
err = c.Ping(ctx) // check connectivity to cluster
}
if err != nil {
return nil, err
}
}
return c, nil
}
func (k *Broker) Connect(ctx context.Context) error {
k.RLock()
if k.connected {
k.RUnlock()
@ -116,51 +100,20 @@ func (k *kBroker) Connect(ctx context.Context) error {
nctx = ctx
}
kaddrs := k.opts.Addrs
// shuffle addrs
var rng mrand.Rand
rng.Shuffle(len(kaddrs), func(i, j int) {
kaddrs[i], kaddrs[j] = kaddrs[j], kaddrs[i]
})
kopts := append(k.kopts, kgo.SeedBrokers(kaddrs...))
select {
case <-nctx.Done():
return nctx.Err()
default:
c, err := kgo.NewClient(kopts...)
if err != nil {
return err
}
// Request versions in order to guess Kafka Cluster version
versionsReq := kmsg.NewApiVersionsRequest()
versionsRes, err := versionsReq.RequestWith(ctx, c)
if err != nil {
return fmt.Errorf("failed to request api versions: %w", err)
}
err = kerr.ErrorForCode(versionsRes.ErrorCode)
if err != nil {
return fmt.Errorf("failed to request api versions. Inner kafka error: %w", err)
}
versions := kversion.FromApiVersionsResponse(versionsRes)
if k.opts.Logger.V(logger.InfoLevel) {
logger.Infof(ctx, "[kgo] connected to to kafka cluster version %v", versions.VersionGuess())
}
k.Lock()
k.connected = true
k.writer = c
k.Unlock()
c, err := k.connect(nctx, k.kopts...)
if err != nil {
return err
}
k.Lock()
k.c = c
k.connected = true
k.Unlock()
return nil
}
func (k *kBroker) Disconnect(ctx context.Context) error {
func (k *Broker) Disconnect(ctx context.Context) error {
k.RLock()
if !k.connected {
k.RUnlock()
@ -168,14 +121,13 @@ func (k *kBroker) Disconnect(ctx context.Context) error {
}
k.RUnlock()
k.Lock()
defer k.Unlock()
nctx := k.opts.Context
if ctx != nil {
nctx = ctx
}
k.Lock()
defer k.Unlock()
select {
case <-nctx.Done():
return nctx.Err()
@ -185,20 +137,20 @@ func (k *kBroker) Disconnect(ctx context.Context) error {
return err
}
}
k.writer.Close()
if k.c != nil {
k.c.CloseAllowingRebalance()
// k.c.Close()
}
}
k.connected = false
return nil
}
func (k *kBroker) Init(opts ...broker.Option) error {
func (k *Broker) Init(opts ...broker.Option) error {
k.Lock()
defer k.Unlock()
if len(opts) == 0 && k.init {
return nil
}
for _, o := range opts {
o(&k.opts)
}
@ -222,55 +174,65 @@ func (k *kBroker) Init(opts ...broker.Option) error {
}
}
// kgo.RecordPartitioner(),
k.init = true
return nil
}
func (k *kBroker) Options() broker.Options {
func (k *Broker) Options() broker.Options {
return k.opts
}
func (k *kBroker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
func (k *Broker) BatchPublish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
return k.publish(ctx, msgs, opts...)
}
func (k *kBroker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
func (k *Broker) Publish(ctx context.Context, topic string, msg *broker.Message, opts ...broker.PublishOption) error {
msg.Header.Set(metadata.HeaderTopic, topic)
return k.publish(ctx, []*broker.Message{msg}, opts...)
}
func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...broker.PublishOption) error {
k.RLock()
if !k.connected {
k.RUnlock()
return broker.ErrNotConnected
k.Lock()
c, err := k.connect(ctx, k.kopts...)
if err != nil {
k.Unlock()
return err
}
k.c = c
k.connected = true
k.Unlock()
}
k.RUnlock()
options := broker.NewPublishOptions(opts...)
records := make([]*kgo.Record, 0, len(msgs))
var errs []string
var err error
var key []byte
var promise func(*kgo.Record, error)
if options.Context != nil {
if k, ok := options.Context.Value(publishKey{}).([]byte); ok && k != nil {
key = k
}
if p, ok := options.Context.Value(publishPromiseKey{}).(func(*kgo.Record, error)); ok && p != nil {
promise = p
}
}
for _, msg := range msgs {
rec := &kgo.Record{Context: ctx, Key: key}
rec.Topic, _ = msg.Header.Get(metadata.HeaderTopic)
if k.opts.Codec.String() == "noop" {
k.opts.Meter.Counter(broker.PublishMessageInflight, "endpoint", rec.Topic).Inc()
if options.BodyOnly {
rec.Value = msg.Body
} else if k.opts.Codec.String() == "noop" {
rec.Value = msg.Body
for k, v := range msg.Header {
rec.Headers = append(rec.Headers, kgo.RecordHeader{Key: k, Value: []byte(v)})
}
} else if options.BodyOnly {
rec.Value = msg.Body
} else {
rec.Value, err = k.opts.Codec.Marshal(msg)
if err != nil {
@ -280,10 +242,36 @@ func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...b
records = append(records, rec)
}
results := k.writer.ProduceSync(ctx, records...)
if promise != nil {
ts := time.Now()
for _, rec := range records {
k.c.Produce(ctx, rec, func(r *kgo.Record, err error) {
te := time.Since(ts)
k.opts.Meter.Counter(broker.PublishMessageInflight, "endpoint", rec.Topic).Dec()
k.opts.Meter.Summary(broker.PublishMessageLatencyMicroseconds, "endpoint", r.Topic).Update(te.Seconds())
k.opts.Meter.Histogram(broker.PublishMessageDurationSeconds, "endpoint", r.Topic).Update(te.Seconds())
if err != nil {
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", r.Topic, "status", "failure").Inc()
} else {
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", r.Topic, "status", "success").Inc()
}
promise(r, err)
})
}
return nil
}
ts := time.Now()
results := k.c.ProduceSync(ctx, records...)
te := time.Since(ts)
for _, result := range results {
k.opts.Meter.Summary(broker.PublishMessageLatencyMicroseconds, "endpoint", result.Record.Topic).Update(te.Seconds())
k.opts.Meter.Histogram(broker.PublishMessageDurationSeconds, "endpoint", result.Record.Topic).Update(te.Seconds())
k.opts.Meter.Counter(broker.PublishMessageInflight, "endpoint", result.Record.Topic).Dec()
if result.Err != nil {
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", result.Record.Topic, "status", "failure").Inc()
errs = append(errs, result.Err.Error())
} else {
k.opts.Meter.Counter(broker.PublishMessageTotal, "endpoint", result.Record.Topic, "status", "success").Inc()
}
}
@ -294,53 +282,11 @@ func (k *kBroker) publish(ctx context.Context, msgs []*broker.Message, opts ...b
return nil
}
type mlogger struct {
l logger.Logger
ctx context.Context
}
func (l *mlogger) Log(lvl kgo.LogLevel, msg string, args ...interface{}) {
var mlvl logger.Level
switch lvl {
case kgo.LogLevelNone:
return
case kgo.LogLevelError:
mlvl = logger.ErrorLevel
case kgo.LogLevelWarn:
mlvl = logger.WarnLevel
case kgo.LogLevelInfo:
mlvl = logger.InfoLevel
case kgo.LogLevelDebug:
mlvl = logger.DebugLevel
default:
return
}
fields := make(map[string]interface{}, int(len(args)/2))
for i := 0; i < len(args)/2; i += 2 {
fields[fmt.Sprintf("%v", args[i])] = args[i+1]
}
l.l.Fields(fields).Log(l.ctx, mlvl, msg)
}
func (l *mlogger) Level() kgo.LogLevel {
switch l.l.Options().Level {
case logger.ErrorLevel:
return kgo.LogLevelError
case logger.WarnLevel:
return kgo.LogLevelWarn
case logger.InfoLevel:
return kgo.LogLevelInfo
case logger.DebugLevel, logger.TraceLevel:
return kgo.LogLevelDebug
}
return kgo.LogLevelNone
}
func (k *kBroker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
func (k *Broker) BatchSubscribe(ctx context.Context, topic string, handler broker.BatchHandler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
return nil, nil
}
func (k *kBroker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
options := broker.NewSubscribeOptions(opts...)
if options.Group == "" {
@ -351,54 +297,47 @@ func (k *kBroker) Subscribe(ctx context.Context, topic string, handler broker.Ha
options.Group = uid
}
kaddrs := k.opts.Addrs
// shuffle addrs
var rng mrand.Rand
rng.Shuffle(len(kaddrs), func(i, j int) {
kaddrs[i], kaddrs[j] = kaddrs[j], kaddrs[i]
})
td := DefaultCommitInterval
commitInterval := DefaultCommitInterval
if k.opts.Context != nil {
if v, ok := k.opts.Context.Value(commitIntervalKey{}).(time.Duration); ok && v > 0 {
td = v
commitInterval = v
}
}
sub := &subscriber{
topic: topic,
done: make(chan struct{}),
opts: options,
handler: handler,
kopts: k.opts,
consumers: make(map[string]map[int32]worker),
consumers: make(map[tp]*consumer),
done: make(chan struct{}),
}
kopts := append(k.kopts,
kgo.SeedBrokers(kaddrs...),
kgo.ConsumerGroup(options.Group),
kgo.ConsumeTopics(topic),
kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
kgo.FetchMaxWait(1*time.Second),
// kgo.KeepControlRecords(),
kgo.Balancers(kgo.CooperativeStickyBalancer(), kgo.StickyBalancer()),
kgo.FetchIsolationLevel(kgo.ReadUncommitted()),
kgo.WithHooks(&metrics{meter: k.opts.Meter}),
kgo.AutoCommitMarks(),
kgo.AutoCommitInterval(td),
kgo.AutoCommitInterval(commitInterval),
kgo.OnPartitionsAssigned(sub.assigned),
kgo.OnPartitionsRevoked(sub.revoked),
kgo.OnPartitionsLost(sub.revoked),
kgo.OnPartitionsLost(sub.lost),
kgo.AutoCommitMarks(),
)
reader, err := kgo.NewClient(kopts...)
if options.Context != nil {
if v, ok := options.Context.Value(optionsKey{}).([]kgo.Opt); ok && len(v) > 0 {
kopts = append(kopts, v...)
}
}
c, err := k.connect(ctx, kopts...)
if err != nil {
return nil, err
}
sub.reader = reader
go sub.run(ctx)
sub.c = c
go sub.poll(ctx)
k.Lock()
k.subs = append(k.subs, sub)
@ -406,45 +345,31 @@ func (k *kBroker) Subscribe(ctx context.Context, topic string, handler broker.Ha
return sub, nil
}
func (k *kBroker) String() string {
func (k *Broker) String() string {
return "kgo"
}
func NewBroker(opts ...broker.Option) *kBroker {
func NewBroker(opts ...broker.Option) *Broker {
options := broker.NewOptions(opts...)
if options.Codec.String() != "noop" {
options.Logger.Infof(options.Context, "broker codec not noop, disable plain kafka headers usage")
}
kaddrs := options.Addrs
// shuffle addrs
var rng mrand.Rand
rng.Shuffle(len(kaddrs), func(i, j int) {
kaddrs[i], kaddrs[j] = kaddrs[j], kaddrs[i]
})
kopts := []kgo.Opt{
kgo.DialTimeout(3 * time.Second),
kgo.DisableIdempotentWrite(),
kgo.ProducerBatchCompression(kgo.NoCompression()),
kgo.WithLogger(&mlogger{l: options.Logger, ctx: options.Context}),
kgo.RetryBackoffFn(
func() func(int) time.Duration {
var rng mrand.Rand
return func(fails int) time.Duration {
const (
min = 250 * time.Millisecond
max = 2 * time.Second
)
if fails <= 0 {
return min
}
if fails > 10 {
return max
}
backoff := min * time.Duration(1<<(fails-1))
jitter := 0.8 + 0.4*rng.Float64()
backoff = time.Duration(float64(backoff) * jitter)
if backoff > max {
return max
}
return backoff
}
}(),
),
// kgo.WithLogger(kgo.BasicLogger(os.Stderr, kgo.LogLevelDebug, func() string { return time.Now().Format(time.StampMilli) })),
kgo.WithHooks(&metrics{meter: options.Meter}),
kgo.SeedBrokers(kaddrs...),
kgo.RetryBackoffFn(DefaultRetryBackoffFn),
kgo.BlockRebalanceOnPoll(),
kgo.Balancers(kgo.CooperativeStickyBalancer()),
kgo.FetchIsolationLevel(kgo.ReadUncommitted()),
}
if options.Context != nil {
@ -453,7 +378,7 @@ func NewBroker(opts ...broker.Option) *kBroker {
}
}
return &kBroker{
return &Broker{
opts: options,
kopts: kopts,
}

25
kgo_test.go
View File

@ -10,10 +10,10 @@ import (
"time"
kg "github.com/twmb/franz-go/pkg/kgo"
kgo "go.unistack.org/micro-broker-kgo/v3"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
kgo "go.unistack.org/micro-broker-kgo/v4"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
)
var (
@ -28,6 +28,23 @@ var bm = &broker.Message{
Body: []byte(`"body"`),
}
func TestConnect(t *testing.T) {
var addrs []string
ctx := context.TODO()
b := kgo.NewBroker(
broker.Addrs(addrs...),
kgo.CommitInterval(5*time.Second),
kgo.Options(kg.ClientID("test"), kg.FetchMaxBytes(10*1024*1024)),
)
if err := b.Init(); err != nil {
t.Fatal(err)
}
if err := b.Connect(ctx); err != nil {
t.Fatal(err)
}
}
func TestPubSub(t *testing.T) {
if tr := os.Getenv("INTEGRATION_TESTS"); len(tr) > 0 {
t.Skip()

55
logger.go Normal file
View File

@ -0,0 +1,55 @@
package kgo
import (
"context"
"fmt"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v4/logger"
)
type mlogger struct {
l logger.Logger
ctx context.Context
}
func (l *mlogger) Log(lvl kgo.LogLevel, msg string, args ...interface{}) {
var mlvl logger.Level
switch lvl {
case kgo.LogLevelNone:
return
case kgo.LogLevelError:
mlvl = logger.ErrorLevel
case kgo.LogLevelWarn:
mlvl = logger.WarnLevel
case kgo.LogLevelInfo:
mlvl = logger.InfoLevel
case kgo.LogLevelDebug:
mlvl = logger.DebugLevel
default:
return
}
if len(args) > 0 {
fields := make(map[string]interface{}, int(len(args)/2))
for i := 0; i <= len(args)/2; i += 2 {
fields[fmt.Sprintf("%v", args[i])] = args[i+1]
}
l.l.Fields(fields).Log(l.ctx, mlvl, msg)
} else {
l.l.Log(l.ctx, mlvl, msg)
}
}
func (l *mlogger) Level() kgo.LogLevel {
switch l.l.Options().Level {
case logger.ErrorLevel:
return kgo.LogLevelError
case logger.WarnLevel:
return kgo.LogLevelWarn
case logger.InfoLevel:
return kgo.LogLevelInfo
case logger.DebugLevel, logger.TraceLevel:
return kgo.LogLevelDebug
}
return kgo.LogLevelNone
}

37
metrics.go
View File

@ -6,9 +6,44 @@ import (
"time"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v3/meter"
"go.unistack.org/micro/v4/meter"
)
/*
func NewServerSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
handler := &wrapper{
opts: NewOptions(opts...),
}
return handler.SubscriberFunc
}
func (w *wrapper) SubscriberFunc(fn server.SubscriberFunc) server.SubscriberFunc {
return func(ctx context.Context, msg server.Message) error {
endpoint := msg.Topic()
labels := make([]string, 0, 4)
labels = append(labels, labelEndpoint, endpoint)
w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Inc()
ts := time.Now()
err := fn(ctx, msg)
te := time.Since(ts)
w.opts.Meter.Counter(SubscribeMessageInflight, labels...).Dec()
w.opts.Meter.Summary(SubscribeMessageLatencyMicroseconds, labels...).Update(te.Seconds())
w.opts.Meter.Histogram(SubscribeMessageDurationSeconds, labels...).Update(te.Seconds())
if err == nil {
labels = append(labels, labelStatus, labelSuccess)
} else {
labels = append(labels, labelStatus, labelFailure)
}
w.opts.Meter.Counter(SubscribeMessageTotal, labels...).Inc()
return err
}
}
*/
type metrics struct {
meter meter.Meter
}

38
options.go
View File

@ -4,10 +4,9 @@ import (
"context"
"time"
kgo "github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/client"
"go.unistack.org/micro/v3/server"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/client"
)
// DefaultCommitInterval specifies how fast send commit offsets to kafka
@ -49,7 +48,7 @@ func Options(opts ...kgo.Opt) broker.Option {
}
}
// SubscribeOptions pass additional options to broker
// SubscribeOptions pass additional options to broker in Subscribe
func SubscribeOptions(opts ...kgo.Opt) broker.SubscribeOption {
return func(o *broker.SubscribeOptions) {
if o.Context == nil {
@ -64,21 +63,6 @@ func SubscribeOptions(opts ...kgo.Opt) broker.SubscribeOption {
}
}
// SubscriberOptions pass additional options to broker
func SubscriberOptions(opts ...kgo.Opt) server.SubscriberOption {
return func(o *server.SubscriberOptions) {
if o.Context == nil {
o.Context = context.Background()
}
options, ok := o.Context.Value(optionsKey{}).([]kgo.Opt)
if !ok {
options = make([]kgo.Opt, 0, len(opts))
}
options = append(options, opts...)
o.Context = context.WithValue(o.Context, optionsKey{}, options)
}
}
type commitIntervalKey struct{}
// CommitInterval specifies interval to send commits
@ -86,7 +70,7 @@ func CommitInterval(td time.Duration) broker.Option {
return broker.SetOption(commitIntervalKey{}, td)
}
var DefaultSubscribeMaxInflight = 1000
var DefaultSubscribeMaxInflight = 10
type subscribeMaxInflightKey struct{}
@ -94,3 +78,15 @@ type subscribeMaxInflightKey struct{}
func SubscribeMaxInFlight(n int) broker.SubscribeOption {
return broker.SetSubscribeOption(subscribeMaxInflightKey{}, n)
}
type publishPromiseKey struct{}
// PublishPromise set the kafka promise func for Produce
func PublishPromise(fn func(*kgo.Record, error)) broker.PublishOption {
return broker.SetPublishOption(publishPromiseKey{}, fn)
}
// ClientPublishKey set the kafka message key (client option)
func ClientPublishPromise(fn func(*kgo.Record, error)) client.PublishOption {
return client.SetPublishOption(publishPromiseKey{}, fn)
}

250
subscriber.go Normal file
View File

@ -0,0 +1,250 @@
package kgo
import (
"context"
"sync"
"time"
"github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v4/broker"
"go.unistack.org/micro/v4/logger"
"go.unistack.org/micro/v4/metadata"
)
type tp struct {
t string
p int32
}
type consumer struct {
c *kgo.Client
topic string
partition int32
opts broker.SubscribeOptions
kopts broker.Options
handler broker.Handler
quit chan struct{}
done chan struct{}
recs chan kgo.FetchTopicPartition
}
type subscriber struct {
c *kgo.Client
topic string
opts broker.SubscribeOptions
kopts broker.Options
handler broker.Handler
closed bool
done chan struct{}
consumers map[tp]*consumer
sync.RWMutex
}
func (s *subscriber) Options() broker.SubscribeOptions {
return s.opts
}
func (s *subscriber) Topic() string {
return s.topic
}
func (s *subscriber) Unsubscribe(ctx context.Context) error {
if s.closed {
return nil
}
select {
case <-ctx.Done():
return ctx.Err()
default:
close(s.done)
s.closed = true
}
return nil
}
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
}
}
for {
select {
case <-ctx.Done():
s.c.Close()
return
case <-s.done:
s.c.Close()
return
default:
fetches := s.c.PollRecords(ctx, maxInflight)
if fetches.IsClientClosed() {
s.kopts.Logger.Errorf(ctx, "[kgo] client closed")
s.closed = true
return
}
fetches.EachError(func(t string, p int32, err error) {
s.kopts.Logger.Fatalf(ctx, "[kgo] fetch topic %s partition %d err: %v", t, p, err)
})
fetches.EachPartition(func(p kgo.FetchTopicPartition) {
tp := tp{p.Topic, p.Partition}
s.consumers[tp].recs <- p
})
s.c.AllowRebalance()
}
}
}
func (s *subscriber) killConsumers(ctx context.Context, lost map[string][]int32) {
var wg sync.WaitGroup
defer wg.Wait()
for topic, partitions := range lost {
for _, partition := range partitions {
tp := tp{topic, partition}
pc := s.consumers[tp]
delete(s.consumers, tp)
close(pc.quit)
s.kopts.Logger.Debugf(ctx, "[kgo] waiting for work to finish topic %s partition %d", topic, partition)
wg.Add(1)
go func() { <-pc.done; wg.Done() }()
}
}
}
func (s *subscriber) lost(ctx context.Context, _ *kgo.Client, lost map[string][]int32) {
s.kopts.Logger.Debugf(ctx, "[kgo] lost %#+v", lost)
s.killConsumers(ctx, lost)
}
func (s *subscriber) revoked(ctx context.Context, c *kgo.Client, revoked map[string][]int32) {
s.kopts.Logger.Debugf(ctx, "[kgo] revoked %#+v", revoked)
s.killConsumers(ctx, revoked)
if err := c.CommitMarkedOffsets(ctx); err != nil {
s.kopts.Logger.Errorf(ctx, "[kgo] revoked CommitMarkedOffsets err: %v", err)
}
}
func (s *subscriber) assigned(_ context.Context, c *kgo.Client, assigned map[string][]int32) {
for topic, partitions := range assigned {
for _, partition := range partitions {
pc := &consumer{
c: c,
topic: topic,
partition: partition,
quit: make(chan struct{}),
done: make(chan struct{}),
recs: make(chan kgo.FetchTopicPartition, 4),
handler: s.handler,
kopts: s.kopts,
opts: s.opts,
}
s.consumers[tp{topic, partition}] = pc
go pc.consume()
}
}
}
func (pc *consumer) consume() {
defer close(pc.done)
pc.kopts.Logger.Debugf(pc.kopts.Context, "starting, topic %s partition %d", pc.topic, pc.partition)
defer pc.kopts.Logger.Debugf(pc.kopts.Context, "killing, topic %s partition %d", pc.topic, pc.partition)
eh := pc.kopts.ErrorHandler
if pc.opts.ErrorHandler != nil {
eh = pc.opts.ErrorHandler
}
for {
select {
case <-pc.quit:
return
case p := <-pc.recs:
for _, record := range p.Records {
ts := time.Now()
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Inc()
p := eventPool.Get().(*event)
p.msg.Header = nil
p.msg.Body = nil
p.topic = record.Topic
p.err = nil
p.ack = false
if pc.kopts.Codec.String() == "noop" {
p.msg.Header = metadata.New(len(record.Headers))
for _, hdr := range record.Headers {
p.msg.Header.Set(hdr.Key, string(hdr.Value))
}
p.msg.Body = record.Value
} else if pc.opts.BodyOnly {
p.msg.Body = record.Value
} else {
if err := pc.kopts.Codec.Unmarshal(record.Value, p.msg); err != nil {
pc.kopts.Meter.Counter(broker.SubscribeMessageTotal, "endpoint", record.Topic, "status", "failure").Inc()
p.err = err
p.msg.Body = record.Value
if eh != nil {
_ = eh(p)
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Dec()
if p.ack {
pc.c.MarkCommitRecords(record)
} else {
eventPool.Put(p)
pc.kopts.Logger.Fatalf(pc.kopts.Context, "[kgo] ErrLostMessage wtf?")
return
}
eventPool.Put(p)
te := time.Since(ts)
pc.kopts.Meter.Summary(broker.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic).Update(te.Seconds())
pc.kopts.Meter.Histogram(broker.SubscribeMessageDurationSeconds, "endpoint", record.Topic).Update(te.Seconds())
continue
} else {
if pc.kopts.Logger.V(logger.ErrorLevel) {
pc.kopts.Logger.Errorf(pc.kopts.Context, "[kgo]: failed to unmarshal: %v", err)
}
}
te := time.Since(ts)
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Dec()
pc.kopts.Meter.Summary(broker.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic).Update(te.Seconds())
pc.kopts.Meter.Histogram(broker.SubscribeMessageDurationSeconds, "endpoint", record.Topic).Update(te.Seconds())
eventPool.Put(p)
pc.kopts.Logger.Fatalf(pc.kopts.Context, "[kgo] Unmarshal err not handled wtf?")
return
}
}
err := pc.handler(p)
if err == nil {
pc.kopts.Meter.Counter(broker.SubscribeMessageTotal, "endpoint", record.Topic, "status", "success").Inc()
} else {
pc.kopts.Meter.Counter(broker.SubscribeMessageTotal, "endpoint", record.Topic, "status", "failure").Inc()
}
pc.kopts.Meter.Counter(broker.SubscribeMessageInflight, "endpoint", record.Topic).Dec()
if err == nil && pc.opts.AutoAck {
p.ack = true
} else if err != nil {
p.err = err
if eh != nil {
_ = eh(p)
} else {
if pc.kopts.Logger.V(logger.ErrorLevel) {
pc.kopts.Logger.Errorf(pc.kopts.Context, "[kgo]: subscriber error: %v", err)
}
}
}
te := time.Since(ts)
pc.kopts.Meter.Summary(broker.SubscribeMessageLatencyMicroseconds, "endpoint", record.Topic).Update(te.Seconds())
pc.kopts.Meter.Histogram(broker.SubscribeMessageDurationSeconds, "endpoint", record.Topic).Update(te.Seconds())
if p.ack {
eventPool.Put(p)
pc.c.MarkCommitRecords(record)
} else {
eventPool.Put(p)
pc.kopts.Logger.Fatalf(pc.kopts.Context, "[kgo] ErrLostMessage wtf?")
return
}
}
}
}
}

244
util.go
View File

@ -1,244 +0,0 @@
package kgo
import (
"context"
"errors"
"sync"
kgo "github.com/twmb/franz-go/pkg/kgo"
"go.unistack.org/micro/v3/broker"
"go.unistack.org/micro/v3/logger"
"go.unistack.org/micro/v3/metadata"
)
var ErrLostMessage = errors.New("message not marked for offsets commit and will be lost in next iteration")
var pPool = sync.Pool{
New: func() interface{} {
return &publication{msg: broker.NewMessage("")}
},
}
type worker struct {
done chan struct{}
recs chan []*kgo.Record
cherr chan error
handler broker.Handler
batchHandler broker.BatchHandler
opts broker.SubscribeOptions
kopts broker.Options
tpmap map[string][]int32
maxInflight int
reader *kgo.Client
ctx context.Context
}
func (s *subscriber) run(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case <-s.kopts.Context.Done():
return
default:
fetches := s.reader.PollFetches(ctx)
if fetches.IsClientClosed() {
// TODO: fatal ?
return
}
if len(fetches.Errors()) > 0 {
for _, err := range fetches.Errors() {
s.kopts.Logger.Fatalf(ctx, "fetch err topic %s partition %d: %v", err.Topic, err.Partition, err.Err)
}
// TODO: fatal ?
return
}
fetches.EachPartition(func(p kgo.FetchTopicPartition) {
s.Lock()
consumers := s.consumers[p.Topic]
s.Unlock()
if consumers == nil {
return
}
w, ok := consumers[p.Partition]
if !ok {
return
}
select {
case err := <-w.cherr:
s.kopts.Logger.Fatalf(ctx, "handle err: %v", err)
return
case w.recs <- p.Records:
case <-w.done:
}
})
}
}
}
func (s *subscriber) assigned(ctx context.Context, _ *kgo.Client, assigned map[string][]int32) {
maxInflight := DefaultSubscribeMaxInflight
if s.opts.Context != nil {
if n, ok := s.opts.Context.Value(subscribeMaxInflightKey{}).(int); n > 0 && ok {
maxInflight = n
}
}
s.Lock()
for topic, partitions := range assigned {
if s.consumers[topic] == nil {
s.consumers[topic] = make(map[int32]worker)
}
for _, partition := range partitions {
w := worker{
done: make(chan struct{}),
recs: make(chan []*kgo.Record),
cherr: make(chan error),
kopts: s.kopts,
opts: s.opts,
ctx: ctx,
tpmap: map[string][]int32{topic: []int32{partition}},
reader: s.reader,
handler: s.handler,
batchHandler: s.batchhandler,
maxInflight: maxInflight,
}
s.consumers[topic][partition] = w
go w.handle()
}
}
s.Unlock()
}
func (s *subscriber) revoked(_ context.Context, _ *kgo.Client, revoked map[string][]int32) {
s.Lock()
for topic, partitions := range revoked {
ptopics := s.consumers[topic]
for _, partition := range partitions {
w := ptopics[partition]
delete(ptopics, partition)
if len(ptopics) == 0 {
delete(s.consumers, topic)
}
close(w.done)
}
}
s.Unlock()
}
func (w *worker) handle() {
var err error
eh := w.kopts.ErrorHandler
if w.opts.ErrorHandler != nil {
eh = w.opts.ErrorHandler
}
paused := false
for {
select {
case <-w.ctx.Done():
w.cherr <- w.ctx.Err()
return
case <-w.done:
return
case recs := <-w.recs:
if len(recs) >= w.maxInflight {
paused = true
w.reader.PauseFetchPartitions(w.tpmap)
}
for _, record := range recs {
p := pPool.Get().(*publication)
p.msg.Header = nil
p.msg.Body = nil
p.topic = record.Topic
p.err = nil
p.ack = false
if w.opts.BodyOnly {
p.msg.Body = record.Value
if l := len(record.Headers); l > 0 {
if p.msg.Header == nil {
p.msg.Header = metadata.New(l)
}
for _, h := range record.Headers {
p.msg.Header.Set(h.Key, string(h.Value))
}
}
} else if w.kopts.Codec.String() == "noop" {
p.msg.Body = record.Value
p.msg.Header = metadata.New(len(record.Headers))
for _, h := range record.Headers {
p.msg.Header.Set(h.Key, string(h.Value))
}
} else {
if err := w.kopts.Codec.Unmarshal(record.Value, p.msg); err != nil {
p.err = err
p.msg.Body = record.Value
if l := len(record.Headers); l > 0 {
if p.msg.Header == nil {
p.msg.Header = metadata.New(l)
}
for _, h := range record.Headers {
p.msg.Header.Set(h.Key, string(h.Value))
}
}
if eh != nil {
_ = eh(p)
if p.ack {
w.reader.MarkCommitRecords(record)
} else {
w.cherr <- ErrLostMessage
pPool.Put(p)
return
}
pPool.Put(p)
continue
} else {
if w.kopts.Logger.V(logger.ErrorLevel) {
w.kopts.Logger.Errorf(w.kopts.Context, "[kgo]: failed to unmarshal: %v", err)
}
}
pPool.Put(p)
w.cherr <- err
return
}
if l := len(record.Headers); l > 0 {
if p.msg.Header == nil {
p.msg.Header = metadata.New(l)
}
for _, h := range record.Headers {
p.msg.Header.Set(h.Key, string(h.Value))
}
}
}
err = w.handler(p)
if err == nil && w.opts.AutoAck {
p.ack = true
} else if err != nil {
p.err = err
if eh != nil {
_ = eh(p)
} else {
if w.kopts.Logger.V(logger.ErrorLevel) {
w.kopts.Logger.Errorf(w.kopts.Context, "[kgo]: subscriber error: %v", err)
}
}
}
if p.ack {
pPool.Put(p)
w.reader.MarkCommitRecords(record)
} else {
pPool.Put(p)
w.cherr <- ErrLostMessage
return
}
}
if paused {
paused = false
w.reader.ResumeFetchPartitions(w.tpmap)
}
}
}
}