meter and tracing

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

go.mod

@@ -3,15 +3,14 @@ module go.unistack.org/micro-client-grpc/v3
 go 1.20
 
 require (
-	go.unistack.org/micro/v3 v3.10.28
+	go.unistack.org/micro/v3 v3.10.64
-	google.golang.org/grpc v1.59.0
+	google.golang.org/grpc v1.63.2
 )
 
 require (
-	github.com/golang/protobuf v1.5.3 // indirect
+	golang.org/x/net v0.24.0 // indirect
-	golang.org/x/net v0.17.0 // indirect
+	golang.org/x/sys v0.19.0 // indirect
-	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
-	golang.org/x/text v0.13.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
-	google.golang.org/protobuf v1.31.0 // indirect
 )

go.sum

@@ -1,22 +1,15 @@
-github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
-github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg=
+go.unistack.org/micro/v3 v3.10.64 h1:hTRW/4krLjdGb1gDwN8UDub0RmUqrn/oYclL4RXZTOw=
-github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+go.unistack.org/micro/v3 v3.10.64/go.mod h1:erMgt3Bl7vQQ0e9UpQyR5NlLiZ9pKeEJ9+1tfYFaqUg=
-github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+golang.org/x/net v0.24.0 h1:1PcaxkF854Fu3+lvBIx5SYn9wRlBzzcnHZSiaFFAb0w=
-github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+golang.org/x/net v0.24.0/go.mod h1:2Q7sJY5mzlzWjKtYUEXSlBWCdyaioyXzRB2RtU8KVE8=
-go.unistack.org/micro/v3 v3.10.28 h1:/87lGekrmi0/66pioy+Nh8lVUBBYnVqKoHiNYX5OmMI=
+golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
-go.unistack.org/micro/v3 v3.10.28/go.mod h1:eUgtvbtiiz6te93m0ZdmoecbitWwjdBmmr84srmEIKA=
+golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
-golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be h1:LG9vZxsWGOmUKieR8wPAUR3u3MpnYFQZROPIMaXh7/A=
-golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be/go.mod h1:WtryC6hu0hhx87FDGxWCDptyssuo68sk10vYjF+T9fY=
-golang.org/x/text v0.13.0 h1:ablQoSUd0tRdKxZewP80B+BaqeKJuVhuRxj/dkrun3k=
+google.golang.org/grpc v1.63.2 h1:MUeiw1B2maTVZthpU5xvASfTh3LDbxHd6IJ6QQVU+xM=
-golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
+google.golang.org/grpc v1.63.2/go.mod h1:WAX/8DgncnokcFUldAxq7GeB5DXHDbMF+lLvDomNkRA=
-golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b h1:ZlWIi1wSK56/8hn4QcBp/j9M7Gt3U/3hZw3mC7vDICo=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b/go.mod h1:swOH3j0KzcDDgGUWr+SNpyTen5YrXjS3eyPzFYKc6lc=
-google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
-google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
-google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
-google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=

grpc.go

@@ -1,5 +1,5 @@
-// Package grpc provides a gRPC client
+// Package grpc provides a gRPC client for micro framework
-package grpc // import "go.unistack.org/micro-client-grpc/v3"
+package grpc
 
 import (
 	"context"
@@ -8,6 +8,7 @@ import (
 	"net"
 	"os"
 	"reflect"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -17,12 +18,17 @@ import (
 	"go.unistack.org/micro/v3/codec"
 	"go.unistack.org/micro/v3/errors"
 	"go.unistack.org/micro/v3/metadata"
+	"go.unistack.org/micro/v3/options"
 	"go.unistack.org/micro/v3/selector"
+	"go.unistack.org/micro/v3/semconv"
+	"go.unistack.org/micro/v3/tracer"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding"
 	gmetadata "google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 )
 
 const (
@@ -30,6 +36,10 @@ const (
 )
 
 type grpcClient struct {
+	funcPublish      client.FuncPublish
+	funcBatchPublish client.FuncBatchPublish
+	funcCall         client.FuncCall
+	funcStream       client.FuncStream
 	pool             *ConnPool
 	opts             client.Options
 	sync.RWMutex
@@ -406,26 +416,23 @@ func (g *grpcClient) Init(opts ...client.Option) error {
 		g.pool.Unlock()
 	}
 
-	if err := g.opts.Broker.Init(); err != nil {
+	g.funcCall = g.fnCall
-		return err
+	g.funcStream = g.fnStream
-	}
+	g.funcPublish = g.fnPublish
-	if err := g.opts.Tracer.Init(); err != nil {
+	g.funcBatchPublish = g.fnBatchPublish
-		return err
-	}
-	if err := g.opts.Router.Init(); err != nil {
-		return err
-	}
-	if err := g.opts.Logger.Init(); err != nil {
-		return err
-	}
-	if err := g.opts.Meter.Init(); err != nil {
-		return err
-	}
-	if err := g.opts.Transport.Init(); err != nil {
-		return err
-	}
 
-	g.init = true
+	g.opts.Hooks.EachNext(func(hook options.Hook) {
+		switch h := hook.(type) {
+		case client.HookCall:
+			g.funcCall = h(g.funcCall)
+		case client.HookStream:
+			g.funcStream = h(g.funcStream)
+		case client.HookPublish:
+			g.funcPublish = h(g.funcPublish)
+		case client.HookBatchPublish:
+			g.funcBatchPublish = h(g.funcBatchPublish)
+		}
+	})
 
 	return nil
 }
@@ -448,6 +455,32 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 	} else if rsp == nil {
 		return errors.InternalServerError("go.micro.client", "rsp is nil")
 	}
+
+	ts := time.Now()
+	g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
+	var sp tracer.Span
+	ctx, sp = g.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels("endpoint", req.Endpoint()),
+	)
+	err := g.funcCall(ctx, req, rsp, opts...)
+	g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
+	te := time.Since(ts)
+	g.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+	g.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+
+	if me := errors.FromError(err); me == nil {
+		sp.Finish()
+		g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(200))).Inc()
+	} else {
+		sp.SetStatus(tracer.SpanStatusError, err.Error())
+		g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code))).Inc()
+	}
+
+	return err
+}
+
+func (g *grpcClient) fnCall(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
 	// make a copy of call opts
 	callOpts := g.opts.CallOptions
 
@@ -479,11 +512,6 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 	// make copy of call method
 	gcall := g.call
 
-	// wrap the call in reverse
-	for i := len(callOpts.CallWrappers); i > 0; i-- {
-		gcall = callOpts.CallWrappers[i-1](gcall)
-	}
-
 	// use the router passed as a call option, or fallback to the rpc clients router
 	if callOpts.Router == nil {
 		callOpts.Router = g.opts.Router
@@ -583,6 +611,31 @@ func (g *grpcClient) Call(ctx context.Context, req client.Request, rsp interface
 }
 
 func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
+	ts := time.Now()
+	g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Inc()
+	var sp tracer.Span
+	ctx, sp = g.opts.Tracer.Start(ctx, req.Endpoint()+" rpc-client",
+		tracer.WithSpanKind(tracer.SpanKindClient),
+		tracer.WithSpanLabels("endpoint", req.Endpoint()),
+	)
+	stream, err := g.funcStream(ctx, req, opts...)
+	g.opts.Meter.Counter(semconv.ClientRequestInflight, "endpoint", req.Endpoint()).Dec()
+	te := time.Since(ts)
+	g.opts.Meter.Summary(semconv.ClientRequestLatencyMicroseconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+	g.opts.Meter.Histogram(semconv.ClientRequestDurationSeconds, "endpoint", req.Endpoint()).Update(te.Seconds())
+
+	if me := status.Convert(err); me == nil {
+		sp.Finish()
+		g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "success", "code", strconv.Itoa(int(codes.OK))).Inc()
+	} else {
+		sp.SetStatus(tracer.SpanStatusError, err.Error())
+		g.opts.Meter.Counter(semconv.ClientRequestTotal, "endpoint", req.Endpoint(), "status", "failure", "code", strconv.Itoa(int(me.Code()))).Inc()
+	}
+
+	return stream, err
+}
+
+func (g *grpcClient) fnStream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
 	// make a copy of call opts
 	callOpts := g.opts.CallOptions
 	for _, opt := range opts {
@@ -601,11 +654,6 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 	// make a copy of stream
 	gstream := g.stream
 
-	// wrap the call in reverse
-	for i := len(callOpts.CallWrappers); i > 0; i-- {
-		gstream = callOpts.CallWrappers[i-1](gstream)
-	}
-
 	// use the router passed as a call option, or fallback to the rpc clients router
 	if callOpts.Router == nil {
 		callOpts.Router = g.opts.Router
@@ -716,10 +764,18 @@ func (g *grpcClient) Stream(ctx context.Context, req client.Request, opts ...cli
 }
 
 func (g *grpcClient) BatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
+	return g.funcBatchPublish(ctx, ps, opts...)
+}
+
+func (g *grpcClient) fnBatchPublish(ctx context.Context, ps []client.Message, opts ...client.PublishOption) error {
 	return g.publish(ctx, ps, opts...)
 }
 
 func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
+	return g.funcPublish(ctx, p, opts...)
+}
+
+func (g *grpcClient) fnPublish(ctx context.Context, p client.Message, opts ...client.PublishOption) error {
 	return g.publish(ctx, []client.Message{p}, opts...)
 }
 
@@ -843,22 +899,16 @@ func NewClient(opts ...client.Option) client.Client {
 		options.ContentType = DefaultContentType
 	}
 
-	rc := &grpcClient{
+	c := &grpcClient{
 		opts: options,
 	}
 
-	rc.pool = NewConnPool(options.PoolSize, options.PoolTTL, rc.poolMaxIdle(), rc.poolMaxStreams())
+	c.pool = NewConnPool(options.PoolSize, options.PoolTTL, c.poolMaxIdle(), c.poolMaxStreams())
-	c := client.Client(rc)
 
-	// wrap in reverse
+	if c.opts.Context != nil {
-	for i := len(options.Wrappers); i > 0; i-- {
+		if codecs, ok := c.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
-		c = options.Wrappers[i-1](c)
-	}
-
-	if rc.opts.Context != nil {
-		if codecs, ok := rc.opts.Context.Value(codecsKey{}).(map[string]encoding.Codec); ok && codecs != nil {
 			for k, v := range codecs {
-				rc.opts.Codecs[k] = &wrapGrpcCodec{v}
+				c.opts.Codecs[k] = &wrapGrpcCodec{v}
 			}
 		}
 	}
@@ -867,5 +917,10 @@ func NewClient(opts ...client.Option) client.Client {
 		encoding.RegisterCodec(&wrapMicroCodec{k})
 	}
 
+	c.funcCall = c.fnCall
+	c.funcStream = c.fnStream
+	c.funcPublish = c.fnPublish
+	c.funcBatchPublish = c.fnBatchPublish
+
 	return c
 }