512 lines
16 KiB
Go
512 lines
16 KiB
Go
package prometheus
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/unistack-org/micro/v3/client"
|
|
"github.com/unistack-org/micro/v3/logger"
|
|
"github.com/unistack-org/micro/v3/server"
|
|
)
|
|
|
|
var (
|
|
// default metric prefix
|
|
DefaultMetricPrefix = "micro_"
|
|
// default label prefix
|
|
DefaultLabelPrefix = "micro_"
|
|
|
|
clientOpsCounter *prometheus.CounterVec
|
|
clientTimeCounterSummary *prometheus.SummaryVec
|
|
clientTimeCounterHistogram *prometheus.HistogramVec
|
|
|
|
serverOpsCounter *prometheus.CounterVec
|
|
serverTimeCounterSummary *prometheus.SummaryVec
|
|
serverTimeCounterHistogram *prometheus.HistogramVec
|
|
|
|
publishOpsCounter *prometheus.CounterVec
|
|
publishTimeCounterSummary *prometheus.SummaryVec
|
|
publishTimeCounterHistogram *prometheus.HistogramVec
|
|
|
|
subscribeOpsCounter *prometheus.CounterVec
|
|
subscribeTimeCounterSummary *prometheus.SummaryVec
|
|
subscribeTimeCounterHistogram *prometheus.HistogramVec
|
|
|
|
mu sync.Mutex
|
|
)
|
|
|
|
type Options struct {
|
|
Name string
|
|
Version string
|
|
ID string
|
|
Context context.Context
|
|
}
|
|
|
|
type Option func(*Options)
|
|
|
|
func Context(ctx context.Context) Option {
|
|
return func(o *Options) {
|
|
o.Context = ctx
|
|
}
|
|
}
|
|
|
|
func ServiceName(name string) Option {
|
|
return func(o *Options) {
|
|
o.Name = name
|
|
}
|
|
}
|
|
|
|
func ServiceVersion(version string) Option {
|
|
return func(o *Options) {
|
|
o.Version = version
|
|
}
|
|
}
|
|
|
|
func ServiceID(id string) Option {
|
|
return func(o *Options) {
|
|
o.ID = id
|
|
}
|
|
}
|
|
|
|
func registerServerMetrics(ctx context.Context) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
|
|
if serverOpsCounter == nil {
|
|
serverOpsCounter = prometheus.NewCounterVec(
|
|
prometheus.CounterOpts{
|
|
Name: fmt.Sprintf("%sserver_request_total", DefaultMetricPrefix),
|
|
Help: "Requests processed, partitioned by endpoint and status",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "status"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if serverTimeCounterSummary == nil {
|
|
serverTimeCounterSummary = prometheus.NewSummaryVec(
|
|
prometheus.SummaryOpts{
|
|
Name: fmt.Sprintf("%sserver_latency_microseconds", DefaultMetricPrefix),
|
|
Help: "Request latencies in microseconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if serverTimeCounterHistogram == nil {
|
|
serverTimeCounterHistogram = prometheus.NewHistogramVec(
|
|
prometheus.HistogramOpts{
|
|
Name: fmt.Sprintf("%sserver_request_duration_seconds", DefaultMetricPrefix),
|
|
Help: "Request time in seconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
for _, collector := range []prometheus.Collector{serverOpsCounter, serverTimeCounterSummary, serverTimeCounterHistogram} {
|
|
if err := prometheus.DefaultRegisterer.Register(collector); err != nil {
|
|
// if already registered, skip fatal
|
|
if _, ok := err.(prometheus.AlreadyRegisteredError); !ok {
|
|
logger.Fatal(ctx, err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
func registerPublishMetrics(ctx context.Context) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
|
|
if publishOpsCounter == nil {
|
|
publishOpsCounter = prometheus.NewCounterVec(
|
|
prometheus.CounterOpts{
|
|
Name: fmt.Sprintf("%spublish_message_total", DefaultMetricPrefix),
|
|
Help: "Messages sent, partitioned by endpoint and status",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "status"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if publishTimeCounterSummary == nil {
|
|
publishTimeCounterSummary = prometheus.NewSummaryVec(
|
|
prometheus.SummaryOpts{
|
|
Name: fmt.Sprintf("%spublish_message_latency_microseconds", DefaultMetricPrefix),
|
|
Help: "Message latencies in microseconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if publishTimeCounterHistogram == nil {
|
|
publishTimeCounterHistogram = prometheus.NewHistogramVec(
|
|
prometheus.HistogramOpts{
|
|
Name: fmt.Sprintf("%spublish_message_duration_seconds", DefaultMetricPrefix),
|
|
Help: "Message publish time in seconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
for _, collector := range []prometheus.Collector{publishOpsCounter, publishTimeCounterSummary, publishTimeCounterHistogram} {
|
|
if err := prometheus.DefaultRegisterer.Register(collector); err != nil {
|
|
// if already registered, skip fatal
|
|
if _, ok := err.(prometheus.AlreadyRegisteredError); !ok {
|
|
logger.Fatal(ctx, err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
func registerSubscribeMetrics(ctx context.Context) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
|
|
if subscribeOpsCounter == nil {
|
|
subscribeOpsCounter = prometheus.NewCounterVec(
|
|
prometheus.CounterOpts{
|
|
Name: fmt.Sprintf("%ssubscribe_message_total", DefaultMetricPrefix),
|
|
Help: "Messages processed, partitioned by endpoint and status",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "status"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if subscribeTimeCounterSummary == nil {
|
|
subscribeTimeCounterSummary = prometheus.NewSummaryVec(
|
|
prometheus.SummaryOpts{
|
|
Name: fmt.Sprintf("%ssubscribe_message_latency_microseconds", DefaultMetricPrefix),
|
|
Help: "Message processing latencies in microseconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if subscribeTimeCounterHistogram == nil {
|
|
subscribeTimeCounterHistogram = prometheus.NewHistogramVec(
|
|
prometheus.HistogramOpts{
|
|
Name: fmt.Sprintf("%ssubscribe_message_duration_seconds", DefaultMetricPrefix),
|
|
Help: "Request time in seconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
for _, collector := range []prometheus.Collector{subscribeOpsCounter, subscribeTimeCounterSummary, subscribeTimeCounterHistogram} {
|
|
if err := prometheus.DefaultRegisterer.Register(collector); err != nil {
|
|
// if already registered, skip fatal
|
|
if _, ok := err.(prometheus.AlreadyRegisteredError); !ok {
|
|
logger.Fatal(ctx, err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
func registerClientMetrics(ctx context.Context) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
|
|
if clientOpsCounter == nil {
|
|
clientOpsCounter = prometheus.NewCounterVec(
|
|
prometheus.CounterOpts{
|
|
Name: fmt.Sprintf("%srequest_total", DefaultMetricPrefix),
|
|
Help: "Requests processed, partitioned by endpoint and status",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "status"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if clientTimeCounterSummary == nil {
|
|
clientTimeCounterSummary = prometheus.NewSummaryVec(
|
|
prometheus.SummaryOpts{
|
|
Name: fmt.Sprintf("%slatency_microseconds", DefaultMetricPrefix),
|
|
Help: "Request latencies in microseconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
if clientTimeCounterHistogram == nil {
|
|
clientTimeCounterHistogram = prometheus.NewHistogramVec(
|
|
prometheus.HistogramOpts{
|
|
Name: fmt.Sprintf("%srequest_duration_seconds", DefaultMetricPrefix),
|
|
Help: "Request time in seconds, partitioned by endpoint",
|
|
},
|
|
[]string{
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "name"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "version"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "id"),
|
|
fmt.Sprintf("%s%s", DefaultLabelPrefix, "endpoint"),
|
|
},
|
|
)
|
|
}
|
|
|
|
for _, collector := range []prometheus.Collector{clientOpsCounter, clientTimeCounterSummary, clientTimeCounterHistogram} {
|
|
if err := prometheus.DefaultRegisterer.Register(collector); err != nil {
|
|
// if already registered, skip fatal
|
|
if _, ok := err.(prometheus.AlreadyRegisteredError); !ok {
|
|
logger.Fatal(ctx, err.Error())
|
|
}
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
type wrapper struct {
|
|
options Options
|
|
callFunc client.CallFunc
|
|
client.Client
|
|
}
|
|
|
|
func NewClientWrapper(opts ...Option) client.Wrapper {
|
|
options := Options{Context: context.Background()}
|
|
for _, o := range opts {
|
|
o(&options)
|
|
}
|
|
|
|
registerClientMetrics(options.Context)
|
|
registerPublishMetrics(options.Context)
|
|
|
|
return func(c client.Client) client.Client {
|
|
handler := &wrapper{
|
|
options: options,
|
|
Client: c,
|
|
}
|
|
|
|
return handler
|
|
}
|
|
}
|
|
|
|
func NewCallWrapper(opts ...Option) client.CallWrapper {
|
|
options := Options{Context: context.Background()}
|
|
for _, o := range opts {
|
|
o(&options)
|
|
}
|
|
|
|
registerClientMetrics(options.Context)
|
|
|
|
return func(fn client.CallFunc) client.CallFunc {
|
|
handler := &wrapper{
|
|
options: options,
|
|
callFunc: fn,
|
|
}
|
|
|
|
return handler.CallFunc
|
|
}
|
|
}
|
|
|
|
func (w *wrapper) CallFunc(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
|
|
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
|
|
|
|
timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
|
|
us := v * 1000000 // make microseconds
|
|
clientTimeCounterSummary.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(us)
|
|
clientTimeCounterHistogram.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(v)
|
|
}))
|
|
defer timer.ObserveDuration()
|
|
|
|
err := w.callFunc(ctx, addr, req, rsp, opts)
|
|
if err == nil {
|
|
clientOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "success").Inc()
|
|
} else {
|
|
clientOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "failure").Inc()
|
|
}
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
func (w *wrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
|
|
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
|
|
|
|
timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
|
|
us := v * 1000000 // make microseconds
|
|
clientTimeCounterSummary.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(us)
|
|
clientTimeCounterHistogram.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(v)
|
|
}))
|
|
defer timer.ObserveDuration()
|
|
|
|
err := w.Client.Call(ctx, req, rsp, opts...)
|
|
if err == nil {
|
|
clientOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "success").Inc()
|
|
} else {
|
|
clientOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "failure").Inc()
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func (w *wrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
|
|
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
|
|
|
|
timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
|
|
us := v * 1000000 // make microseconds
|
|
clientTimeCounterSummary.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(us)
|
|
clientTimeCounterHistogram.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(v)
|
|
}))
|
|
defer timer.ObserveDuration()
|
|
|
|
stream, err := w.Client.Stream(ctx, req, opts...)
|
|
if err == nil {
|
|
clientOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "success").Inc()
|
|
} else {
|
|
clientOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "failure").Inc()
|
|
}
|
|
|
|
return stream, err
|
|
}
|
|
|
|
func (w *wrapper) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
|
|
endpoint := p.Topic()
|
|
|
|
timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
|
|
us := v * 1000000 // make microseconds
|
|
publishTimeCounterSummary.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(us)
|
|
publishTimeCounterHistogram.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(v)
|
|
}))
|
|
defer timer.ObserveDuration()
|
|
|
|
err := w.Client.Publish(ctx, p, opts...)
|
|
if err == nil {
|
|
publishOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "success").Inc()
|
|
} else {
|
|
publishOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "failure").Inc()
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func NewHandlerWrapper(opts ...Option) server.HandlerWrapper {
|
|
options := Options{Context: context.Background()}
|
|
for _, o := range opts {
|
|
o(&options)
|
|
}
|
|
registerServerMetrics(options.Context)
|
|
|
|
handler := &wrapper{
|
|
options: options,
|
|
}
|
|
|
|
return handler.HandlerFunc
|
|
}
|
|
|
|
func (w *wrapper) HandlerFunc(fn server.HandlerFunc) server.HandlerFunc {
|
|
return func(ctx context.Context, req server.Request, rsp interface{}) error {
|
|
endpoint := req.Endpoint()
|
|
|
|
timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
|
|
us := v * 1000000 // make microseconds
|
|
serverTimeCounterSummary.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(us)
|
|
serverTimeCounterHistogram.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(v)
|
|
}))
|
|
defer timer.ObserveDuration()
|
|
|
|
err := fn(ctx, req, rsp)
|
|
if err == nil {
|
|
serverOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "success").Inc()
|
|
} else {
|
|
serverOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "failure").Inc()
|
|
}
|
|
|
|
return err
|
|
}
|
|
}
|
|
|
|
func NewSubscriberWrapper(opts ...Option) server.SubscriberWrapper {
|
|
options := Options{Context: context.Background()}
|
|
for _, o := range opts {
|
|
o(&options)
|
|
}
|
|
|
|
registerSubscribeMetrics(options.Context)
|
|
|
|
handler := &wrapper{
|
|
options: options,
|
|
}
|
|
|
|
return handler.SubscriberFunc
|
|
}
|
|
|
|
func (w *wrapper) SubscriberFunc(fn server.SubscriberFunc) server.SubscriberFunc {
|
|
return func(ctx context.Context, msg server.Message) error {
|
|
endpoint := msg.Topic()
|
|
|
|
timer := prometheus.NewTimer(prometheus.ObserverFunc(func(v float64) {
|
|
us := v * 1000000 // make microseconds
|
|
subscribeTimeCounterSummary.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(us)
|
|
subscribeTimeCounterHistogram.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint).Observe(v)
|
|
}))
|
|
defer timer.ObserveDuration()
|
|
|
|
err := fn(ctx, msg)
|
|
if err == nil {
|
|
subscribeOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "success").Inc()
|
|
} else {
|
|
subscribeOpsCounter.WithLabelValues(w.options.Name, w.options.Version, w.options.ID, endpoint, "failure").Inc()
|
|
}
|
|
|
|
return err
|
|
}
|
|
}
|