Files
go/pubsub_commands.go
T
ee6e9dbf09 feat(otel): Add OpenTelemetry Native Metrics Support (#3637)
* First draft. One metric - command duration.

Signed-off-by: Elena Kolevska <[email protected]>

* Adds connection state metrics

Signed-off-by: Elena Kolevska <[email protected]>

* wip

Signed-off-by: Elena Kolevska <[email protected]>

* Add phase 1 metrics

* Add client api configurations

* remove redundent code

* refactor: move otel-metrics example to separate module

* refactor: move otel-metrics example to separate module

* remove internal test

* Resolve merge conflicts with upstream/master

* Fix linting issues

* Fix pool idle connection counter bug

* Add OTel performance benchmarking suite

* Improve perf compatison script

* Add phase 2,3 metrics

* Remove local benchmark tests

* Add new fields to tests

* Change db.client.connection.count and db.client.connection.pending_request metric type from UpAndDownCounter to Async gauge.

* fix golangci lint

* Add piplined commands metrics support

* Redesign otel init method

* Address comments

* Fix golangci-lint errors

* Fix passing nil pointer

* Add metric prefix to callbacks and atomic batch registration

* Add unique name for pool.name attr

* Fix unique pool name for pubsub

* Refactor all hardcoded OTel attribute strings to use constants

* Refactor all hardcoded OTel attribute strings to use constants

* extract metric names and pool names as constants, improve error handling

---------

Signed-off-by: Elena Kolevska <[email protected]>
Co-authored-by: Elena Kolevska <[email protected]>
2026-02-13 18:08:10 +02:00

89 lines
2.5 KiB
Go

package redis
import (
"context"
"github.com/redis/go-redis/v9/internal/otel"
)
type PubSubCmdable interface {
Publish(ctx context.Context, channel string, message interface{}) *IntCmd
SPublish(ctx context.Context, channel string, message interface{}) *IntCmd
PubSubChannels(ctx context.Context, pattern string) *StringSliceCmd
PubSubNumSub(ctx context.Context, channels ...string) *MapStringIntCmd
PubSubNumPat(ctx context.Context) *IntCmd
PubSubShardChannels(ctx context.Context, pattern string) *StringSliceCmd
PubSubShardNumSub(ctx context.Context, channels ...string) *MapStringIntCmd
}
// Publish posts the message to the channel.
func (c cmdable) Publish(ctx context.Context, channel string, message interface{}) *IntCmd {
cmd := NewIntCmd(ctx, "publish", channel, message)
_ = c(ctx, cmd)
// Record PubSub message sent (if command succeeded)
if cmd.Err() == nil {
otel.RecordPubSubMessage(ctx, nil, "sent", channel, false)
}
return cmd
}
func (c cmdable) SPublish(ctx context.Context, channel string, message interface{}) *IntCmd {
cmd := NewIntCmd(ctx, "spublish", channel, message)
_ = c(ctx, cmd)
// Record PubSub message sent (if command succeeded)
if cmd.Err() == nil {
otel.RecordPubSubMessage(ctx, nil, "sent", channel, true)
}
return cmd
}
func (c cmdable) PubSubChannels(ctx context.Context, pattern string) *StringSliceCmd {
args := []interface{}{"pubsub", "channels"}
if pattern != "*" {
args = append(args, pattern)
}
cmd := NewStringSliceCmd(ctx, args...)
_ = c(ctx, cmd)
return cmd
}
func (c cmdable) PubSubNumSub(ctx context.Context, channels ...string) *MapStringIntCmd {
args := make([]interface{}, 2+len(channels))
args[0] = "pubsub"
args[1] = "numsub"
for i, channel := range channels {
args[2+i] = channel
}
cmd := NewMapStringIntCmd(ctx, args...)
_ = c(ctx, cmd)
return cmd
}
func (c cmdable) PubSubShardChannels(ctx context.Context, pattern string) *StringSliceCmd {
args := []interface{}{"pubsub", "shardchannels"}
if pattern != "*" {
args = append(args, pattern)
}
cmd := NewStringSliceCmd(ctx, args...)
_ = c(ctx, cmd)
return cmd
}
func (c cmdable) PubSubShardNumSub(ctx context.Context, channels ...string) *MapStringIntCmd {
args := make([]interface{}, 2+len(channels))
args[0] = "pubsub"
args[1] = "shardnumsub"
for i, channel := range channels {
args[2+i] = channel
}
cmd := NewMapStringIntCmd(ctx, args...)
_ = c(ctx, cmd)
return cmd
}
func (c cmdable) PubSubNumPat(ctx context.Context) *IntCmd {
cmd := NewIntCmd(ctx, "pubsub", "numpat")
_ = c(ctx, cmd)
return cmd
}