Compare commits

...

5 Commits

Author SHA1 Message Date
ed61cad961 meter/handler: regen
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-26 15:45:09 +03:00
040fc4548f client: add TLSConfig option
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-26 15:44:34 +03:00
6189a1b980 add SkipEndpoints for wrappers
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-25 23:30:38 +03:00
eb2a450a7b meter/handler: fix func signature
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-25 12:36:48 +03:00
Renovate Bot
d2a30a5da1 fix(deps): update golang.org/x/net commit hash to d1beb07 2021-03-24 23:28:38 +00:00
10 changed files with 142 additions and 21 deletions

View File

@@ -2,6 +2,7 @@ package client
import (
"context"
"crypto/tls"
"time"
"github.com/unistack-org/micro/v3/broker"
@@ -52,6 +53,8 @@ type Options struct {
PoolSize int
// PoolTTL connection pool ttl
PoolTTL time.Duration
// TLSConfig specifies tls.Config for secure connection
TLSConfig *tls.Config
}
// NewCallOptions creates new call options struct
@@ -312,6 +315,22 @@ func Lookup(l LookupFunc) Option {
}
}
// TLSConfig specifies a *tls.Config
func TLSConfig(t *tls.Config) Option {
return func(o *Options) {
// set the internal tls
o.TLSConfig = t
// set the default transport if one is not
// already set. Required for Init call below.
// set the transport tls
o.Transport.Init(
transport.TLSConfig(t),
)
}
}
// Retries sets the retry count when making the request.
func Retries(i int) Option {
return func(o *Options) {

2
go.mod
View File

@@ -9,5 +9,5 @@ require (
github.com/imdario/mergo v0.3.12
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/silas/dag v0.0.0-20210121180416-41cf55125c34
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4
golang.org/x/net v0.0.0-20210324205630-d1beb07c2056
)

6
go.sum
View File

@@ -10,10 +10,10 @@ github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaR
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/silas/dag v0.0.0-20210121180416-41cf55125c34 h1:vBfVmA5mZhsQa2jr1FOL9nfA37N/jnbBmi5XUfviVTI=
github.com/silas/dag v0.0.0-20210121180416-41cf55125c34/go.mod h1:7RTUFBdIRC9nZ7/3RyRNH1bdqIShrDejd1YbLwgPS+I=
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4 h1:b0LrWgu8+q7z4J+0Y3Umo5q1dL7NXBkKBWkaVkAq17E=
golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
golang.org/x/net v0.0.0-20210324205630-d1beb07c2056 h1:sANdAef76Ioam9aQUUdcAqricwY/WUaMc4+7LY4eGg8=
golang.org/x/net v0.0.0-20210324205630-d1beb07c2056/go.mod h1:uSPa2vr4CLtc/ILN5odXGNXS6mhrKVzTaCXzk9m6W3k=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210324051608-47abb6519492/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

View File

@@ -3,6 +3,7 @@ package wrapper
import (
"context"
"fmt"
"github.com/unistack-org/micro/v3/client"
"github.com/unistack-org/micro/v3/logger"
@@ -57,6 +58,8 @@ var (
}
return labels
}
DefaultSkipEndpoints = []string{"Meter.Metrics"}
)
type lWrapper struct {
@@ -94,6 +97,8 @@ type Options struct {
ServerHandlerObservers []ServerHandlerObserver
// ServerSubscriberObservers funcs
ServerSubscriberObservers []ServerSubscriberObserver
// SkipEndpoints
SkipEndpoints []string
}
// Option func signature
@@ -110,6 +115,7 @@ func NewOptions(opts ...Option) Options {
ClientCallFuncObservers: []ClientCallFuncObserver{DefaultClientCallFuncObserver},
ServerHandlerObservers: []ServerHandlerObserver{DefaultServerHandlerObserver},
ServerSubscriberObservers: []ServerSubscriberObserver{DefaultServerSubscriberObserver},
SkipEndpoints: DefaultSkipEndpoints,
}
for _, o := range opts {
@@ -182,9 +188,23 @@ func WithServerSubscriberObservers(ob ...ServerSubscriberObserver) Option {
}
}
// SkipEndpoins
func SkipEndpoints(eps ...string) Option {
return func(o *Options) {
o.SkipEndpoints = append(o.SkipEndpoints, eps...)
}
}
func (l *lWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
err := l.Client.Call(ctx, req, rsp, opts...)
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
for _, ep := range l.opts.SkipEndpoints {
if ep == endpoint {
return err
}
}
if !l.opts.Enabled {
return err
}
@@ -205,6 +225,13 @@ func (l *lWrapper) Call(ctx context.Context, req client.Request, rsp interface{}
func (l *lWrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
stream, err := l.Client.Stream(ctx, req, opts...)
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
for _, ep := range l.opts.SkipEndpoints {
if ep == endpoint {
return stream, err
}
}
if !l.opts.Enabled {
return stream, err
}
@@ -225,6 +252,13 @@ func (l *lWrapper) Stream(ctx context.Context, req client.Request, opts ...clien
func (l *lWrapper) Publish(ctx context.Context, msg client.Message, opts ...client.PublishOption) error {
err := l.Client.Publish(ctx, msg, opts...)
endpoint := msg.Topic()
for _, ep := range l.opts.SkipEndpoints {
if ep == endpoint {
return err
}
}
if !l.opts.Enabled {
return err
}
@@ -245,6 +279,13 @@ func (l *lWrapper) Publish(ctx context.Context, msg client.Message, opts ...clie
func (l *lWrapper) ServerHandler(ctx context.Context, req server.Request, rsp interface{}) error {
err := l.serverHandler(ctx, req, rsp)
endpoint := req.Endpoint()
for _, ep := range l.opts.SkipEndpoints {
if ep == endpoint {
return err
}
}
if !l.opts.Enabled {
return err
}
@@ -265,6 +306,13 @@ func (l *lWrapper) ServerHandler(ctx context.Context, req server.Request, rsp in
func (l *lWrapper) ServerSubscriber(ctx context.Context, msg server.Message) error {
err := l.serverSubscriber(ctx, msg)
endpoint := msg.Topic()
for _, ep := range l.opts.SkipEndpoints {
if ep == endpoint {
return err
}
}
if !l.opts.Enabled {
return err
}
@@ -309,6 +357,13 @@ func NewClientCallWrapper(opts ...Option) client.CallWrapper {
func (l *lWrapper) ClientCallFunc(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
err := l.clientCallFunc(ctx, addr, req, rsp, opts)
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
for _, ep := range l.opts.SkipEndpoints {
if ep == endpoint {
return err
}
}
if !l.opts.Enabled {
return err
}

View File

@@ -24,7 +24,7 @@ func NewHandler(meter meter.Meter, opts ...meter.Option) *handler {
return &handler{meter: meter, opts: opts}
}
func (h *handler) Metrics(ctx context.Context, req *Empty, rsp *codec.Frame) error {
func (h *handler) Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
buf := bytes.NewBuffer(nil)
if err := h.meter.Write(buf, h.opts...); err != nil {
return err

View File

@@ -8,7 +8,7 @@ import "openapiv2/annotations.proto";
import "codec/frame.proto";
service Meter {
rpc Metrics(Empty) returns (micro.codec.Frame) {
rpc Metrics(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv2.openapiv2_operation) = {
operation_id: "Metrics";
responses: {
@@ -26,5 +26,3 @@ service Meter {
option (micro.api.http) = { get: "/metrics"; };
};
};
message Empty{};

View File

@@ -20,5 +20,5 @@ func NewMeterEndpoints() []*api.Endpoint {
}
type MeterServer interface {
Metrics(ctx context.Context, req *Empty, rsp *codec.Frame) error
Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}

View File

@@ -13,20 +13,21 @@ type meterServer struct {
MeterServer
}
func (h *meterServer) Metrics(ctx context.Context, req *Empty, rsp *codec.Frame) error {
func (h *meterServer) Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.MeterServer.Metrics(ctx, req, rsp)
}
func RegisterMeterServer(s server.Server, sh MeterServer, opts ...server.HandlerOption) error {
type meter interface {
Metrics(ctx context.Context, req *Empty, rsp *codec.Frame) error
Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}
type Meter struct {
meter
}
h := &meterServer{sh}
var nopts []server.HandlerOption
for _, endpoint := range NewMeterEndpoints() {
opts = append(opts, api.WithEndpoint(endpoint))
nopts = append(nopts, api.WithEndpoint(endpoint))
}
return s.Handle(s.NewHandler(&Meter{h}, opts...))
return s.Handle(s.NewHandler(&Meter{h}, append(nopts, opts...)...))
}

View File

@@ -34,16 +34,18 @@ var (
)
type Options struct {
Meter meter.Meter
lopts []meter.Option
Meter meter.Meter
lopts []meter.Option
SkipEndpoints []string
}
type Option func(*Options)
func NewOptions(opts ...Option) Options {
options := Options{
Meter: meter.DefaultMeter,
lopts: make([]meter.Option, 0, 5),
Meter: meter.DefaultMeter,
lopts: make([]meter.Option, 0, 5),
SkipEndpoints: DefaultSkipEndpoints,
}
for _, o := range opts {
o(&options)
@@ -75,6 +77,12 @@ func Meter(m meter.Meter) Option {
}
}
func SkipEndoints(eps ...string) Option {
return func(o *Options) {
o.SkipEndpoints = append(o.SkipEndpoints, eps...)
}
}
type wrapper struct {
client.Client
callFunc client.CallFunc
@@ -103,7 +111,7 @@ func NewCallWrapper(opts ...Option) client.CallWrapper {
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())
for _, ep := range DefaultSkipEndpoints {
for _, ep := range w.opts.SkipEndpoints {
if ep == endpoint {
return w.callFunc(ctx, addr, req, rsp, opts)
}
@@ -130,7 +138,7 @@ func (w *wrapper) CallFunc(ctx context.Context, addr string, req client.Request,
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())
for _, ep := range DefaultSkipEndpoints {
for _, ep := range w.opts.SkipEndpoints {
if ep == endpoint {
return w.Client.Call(ctx, req, rsp, opts...)
}
@@ -158,7 +166,7 @@ func (w *wrapper) Call(ctx context.Context, req client.Request, rsp interface{},
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())
for _, ep := range DefaultSkipEndpoints {
for _, ep := range w.opts.SkipEndpoints {
if ep == endpoint {
return w.Client.Stream(ctx, req, opts...)
}
@@ -217,7 +225,7 @@ func NewHandlerWrapper(opts ...Option) server.HandlerWrapper {
func (w *wrapper) HandlerFunc(fn server.HandlerFunc) server.HandlerFunc {
return func(ctx context.Context, req server.Request, rsp interface{}) error {
endpoint := req.Endpoint()
for _, ep := range DefaultSkipEndpoints {
for _, ep := range w.opts.SkipEndpoints {
if ep == endpoint {
return fn(ctx, req, rsp)
}

View File

@@ -101,6 +101,8 @@ var (
}
sp.SetLabels(labels...)
}
DefaultSkipEndpoints = []string{"Meter.Metrics"}
)
type tWrapper struct {
@@ -134,6 +136,8 @@ type Options struct {
ServerHandlerObservers []ServerHandlerObserver
// ServerSubscriberObservers funcs
ServerSubscriberObservers []ServerSubscriberObserver
// SkipEndpoints
SkipEndpoints []string
}
// Option func signature
@@ -149,6 +153,7 @@ func NewOptions(opts ...Option) Options {
ClientCallFuncObservers: []ClientCallFuncObserver{DefaultClientCallFuncObserver},
ServerHandlerObservers: []ServerHandlerObserver{DefaultServerHandlerObserver},
ServerSubscriberObservers: []ServerSubscriberObserver{DefaultServerSubscriberObserver},
SkipEndpoints: DefaultSkipEndpoints,
}
for _, o := range opts {
@@ -165,6 +170,13 @@ func WithTracer(t tracer.Tracer) Option {
}
}
// SkipEndponts
func SkipEndpoins(eps ...string) Option {
return func(o *Options) {
o.SkipEndpoints = append(o.SkipEndpoints, eps...)
}
}
// WithClientCallObservers funcs
func WithClientCallObservers(ob ...ClientCallObserver) Option {
return func(o *Options) {
@@ -208,6 +220,13 @@ func WithServerSubscriberObservers(ob ...ServerSubscriberObserver) Option {
}
func (ot *tWrapper) Call(ctx context.Context, req client.Request, rsp interface{}, opts ...client.CallOption) error {
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
for _, ep := range ot.opts.SkipEndpoints {
if ep == endpoint {
return ot.Client.Call(ctx, req, rsp, opts...)
}
}
sp := tracer.SpanFromContext(ctx)
defer sp.Finish()
@@ -221,6 +240,13 @@ func (ot *tWrapper) Call(ctx context.Context, req client.Request, rsp interface{
}
func (ot *tWrapper) Stream(ctx context.Context, req client.Request, opts ...client.CallOption) (client.Stream, error) {
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
for _, ep := range ot.opts.SkipEndpoints {
if ep == endpoint {
return ot.Client.Stream(ctx, req, opts...)
}
}
sp := tracer.SpanFromContext(ctx)
defer sp.Finish()
@@ -247,6 +273,13 @@ func (ot *tWrapper) Publish(ctx context.Context, msg client.Message, opts ...cli
}
func (ot *tWrapper) ServerHandler(ctx context.Context, req server.Request, rsp interface{}) error {
endpoint := req.Endpoint()
for _, ep := range ot.opts.SkipEndpoints {
if ep == endpoint {
return ot.serverHandler(ctx, req, rsp)
}
}
sp := tracer.SpanFromContext(ctx)
defer sp.Finish()
@@ -297,6 +330,13 @@ func NewClientCallWrapper(opts ...Option) client.CallWrapper {
}
func (ot *tWrapper) ClientCallFunc(ctx context.Context, addr string, req client.Request, rsp interface{}, opts client.CallOptions) error {
endpoint := fmt.Sprintf("%s.%s", req.Service(), req.Endpoint())
for _, ep := range ot.opts.SkipEndpoints {
if ep == endpoint {
return ot.ClientCallFunc(ctx, addr, req, rsp, opts)
}
}
sp := tracer.SpanFromContext(ctx)
defer sp.Finish()