Compare commits

...

20 Commits

Author SHA1 Message Date
d2935ef399 meter/handler: fix proto and generated code
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-28 19:18:11 +03:00
ce4c96ae0a server/health: add health check handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-28 19:18:11 +03:00
Renovate Bot
14026d15be fix(deps): update golang.org/x/net commit hash to 61e0566 2021-03-27 01:29:30 +00:00
Renovate Bot
2df0c7643e fix(deps): update golang.org/x/net commit hash to 6b15177 2021-03-26 19:17:30 +00:00
e13c2c48fd client: use router.DefaultRouter in NewOptions helper
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-26 17:45:55 +03:00
8db55d2e55 router: use dns as default router
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-26 17:43:39 +03:00
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
65889c66f6 meter/wrapper: add DefaultSkipEndpoints
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-25 00:06:38 +03:00
dcdf133d5b server: mask router
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-24 13:26:36 +03:00
8742b55305 auth: fix Init method
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-24 13:12:03 +03:00
4a64ee72f7 meter/handler: provide default metrics handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-23 17:26:29 +03:00
881d7afeea meter/handler: provide initial meter handler
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-23 17:12:13 +03:00
8c95448535 util/reflect: add IsZero helper
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-21 16:17:50 +03:00
c1dc041d8c client: fix NewOptions with CallOptions filling
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
2021-03-16 19:09:49 +03:00
Renovate Bot
25be0ac0f0 fix(deps): update golang.org/x/net commit hash to d523dce 2021-03-16 13:12:17 +00:00
Renovate Bot
86f73cac4e fix(deps): update golang.org/x/net commit hash to 34ac3e1 2021-03-15 22:09:13 +00:00
24 changed files with 687 additions and 22 deletions

View File

@@ -30,7 +30,7 @@ var (
// Auth provides authentication and authorization
type Auth interface {
// Init the auth
Init(opts ...Option)
Init(opts ...Option) error
// Options set for auth
Options() Options
// Generate a new account

View File

@@ -14,10 +14,11 @@ func (n *noopAuth) String() string {
}
// Init the auth
func (n *noopAuth) Init(opts ...Option) {
func (n *noopAuth) Init(opts ...Option) error {
for _, o := range opts {
o(&n.opts)
}
return nil
}
// Options set for auth

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
@@ -160,20 +163,23 @@ func NewOptions(opts ...Option) Options {
ContentType: DefaultContentType,
Codecs: make(map[string]codec.Codec),
CallOptions: CallOptions{
Context: context.Background(),
Backoff: DefaultBackoff,
Retry: DefaultRetry,
Retries: DefaultRetries,
RequestTimeout: DefaultRequestTimeout,
DialTimeout: transport.DefaultDialTimeout,
},
Lookup: LookupRoute,
PoolSize: DefaultPoolSize,
PoolTTL: DefaultPoolTTL,
Selector: random.NewSelector(),
Logger: logger.DefaultLogger,
Broker: broker.DefaultBroker,
Meter: meter.DefaultMeter,
Tracer: tracer.DefaultTracer,
Lookup: LookupRoute,
PoolSize: DefaultPoolSize,
PoolTTL: DefaultPoolTTL,
Selector: random.NewSelector(),
Logger: logger.DefaultLogger,
Broker: broker.DefaultBroker,
Meter: meter.DefaultMeter,
Tracer: tracer.DefaultTracer,
Router: router.DefaultRouter,
Transport: transport.DefaultTransport,
}
for _, o := range opts {
@@ -311,6 +317,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-20210226172049-e18ecbb05110
golang.org/x/net v0.0.0-20210326220855-61e056675ecf
)

4
go.sum
View File

@@ -10,9 +10,9 @@ 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-20210226172049-e18ecbb05110 h1:qWPm9rbaAMKs8Bq/9LRpbMqxWRVUAQwMI9fVrssnTfw=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20210326220855-61e056675ecf/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-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
}

3
meter/generate.go Normal file
View File

@@ -0,0 +1,3 @@
package meter
//go:generate protoc -I./handler -I../ -I/home/vtolstov/.cache/go-path/pkg/mod/github.com/unistack-org/micro-proto@v0.0.1 --micro_out=components=micro|http|server,standalone=false,debug=true,paths=source_relative:./handler handler/handler.proto

34
meter/handler/handler.go Normal file
View File

@@ -0,0 +1,34 @@
package pb
import (
"bytes"
"context"
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/meter"
)
var (
// guard to fail early
_ MeterServer = &handler{}
)
type handler struct {
meter meter.Meter
opts []meter.Option
}
func NewHandler(meter meter.Meter, opts ...meter.Option) *handler {
return &handler{meter: meter, opts: opts}
}
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
}
rsp.Data = buf.Bytes()
return nil
}

View File

@@ -0,0 +1,28 @@
syntax = "proto3";
package meter;
option go_package = "github.com/unistack-org/micro/v3/meter/handler;pb";
import "api/annotations.proto";
import "openapiv2/annotations.proto";
import "codec/frame.proto";
service Meter {
rpc Metrics(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv2.openapiv2_operation) = {
operation_id: "Metrics";
responses: {
key: "default";
value: {
description: "Error response";
schema: {
json_schema: {
ref: "micro.codec.Frame";
}
}
}
}
};
option (micro.api.http) = { get: "/metrics"; };
};
};

View File

@@ -0,0 +1,24 @@
// Code generated by protoc-gen-micro
// source: handler.proto
package pb
import (
context "context"
api "github.com/unistack-org/micro/v3/api"
codec "github.com/unistack-org/micro/v3/codec"
)
func NewMeterEndpoints() []*api.Endpoint {
return []*api.Endpoint{
&api.Endpoint{
Name: "Meter.Metrics",
Path: []string{"/metrics"},
Method: []string{"GET"},
Handler: "rpc",
},
}
}
type MeterServer interface {
Metrics(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}

View File

@@ -0,0 +1,33 @@
// Code generated by protoc-gen-micro
// source: handler.proto
package pb
import (
context "context"
api "github.com/unistack-org/micro/v3/api"
codec "github.com/unistack-org/micro/v3/codec"
server "github.com/unistack-org/micro/v3/server"
)
type meterServer struct {
MeterServer
}
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 *codec.Frame, rsp *codec.Frame) error
}
type Meter struct {
meter
}
h := &meterServer{sh}
var nopts []server.HandlerOption
for _, endpoint := range NewMeterEndpoints() {
nopts = append(nopts, api.WithEndpoint(endpoint))
}
return s.Handle(s.NewHandler(&Meter{h}, append(nopts, opts...)...))
}

View File

@@ -28,19 +28,24 @@ var (
labelFailure = "failure"
labelStatus = "status"
labelEndpoint = "endpoint"
// DefaultSkipEndpoints contains list of endpoints that not evaluted by wrapper
DefaultSkipEndpoints = []string{"Meter.Metrics"}
)
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)
@@ -72,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
@@ -100,7 +111,11 @@ 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 w.opts.SkipEndpoints {
if ep == endpoint {
return w.callFunc(ctx, addr, req, rsp, opts)
}
}
ts := time.Now()
err := w.callFunc(ctx, addr, req, rsp, opts)
te := time.Since(ts)
@@ -123,6 +138,11 @@ 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 w.opts.SkipEndpoints {
if ep == endpoint {
return w.Client.Call(ctx, req, rsp, opts...)
}
}
ts := time.Now()
err := w.Client.Call(ctx, req, rsp, opts...)
@@ -146,6 +166,11 @@ 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 w.opts.SkipEndpoints {
if ep == endpoint {
return w.Client.Stream(ctx, req, opts...)
}
}
ts := time.Now()
stream, err := w.Client.Stream(ctx, req, opts...)
@@ -200,6 +225,11 @@ 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 w.opts.SkipEndpoints {
if ep == endpoint {
return fn(ctx, req, rsp)
}
}
ts := time.Now()
err := fn(ctx, req, rsp)

91
router/dns.go Normal file
View File

@@ -0,0 +1,91 @@
package router
import (
"fmt"
"net"
"strconv"
)
// NewRouter returns an initialized dns router
func NewRouter(opts ...Option) Router {
options := NewOptions(opts...)
return &dns{options}
}
type dns struct {
options Options
}
func (d *dns) Init(opts ...Option) error {
for _, o := range opts {
o(&d.options)
}
return nil
}
func (d *dns) Options() Options {
return d.options
}
func (d *dns) Table() Table {
return nil
}
func (d *dns) Close() error {
return nil
}
func (d *dns) Lookup(opts ...QueryOption) ([]Route, error) {
options := NewQuery(opts...)
// check to see if we have the port provided in the service, e.g. go-micro-srv-foo:8000
host, port, err := net.SplitHostPort(options.Service)
if err == nil {
// lookup the service using A records
ips, err := net.LookupHost(host)
if err != nil {
return nil, err
}
p, _ := strconv.Atoi(port)
// convert the ip addresses to routes
result := make([]Route, len(ips))
for i, ip := range ips {
result[i] = Route{
Service: options.Service,
Address: fmt.Sprintf("%s:%d", ip, uint16(p)),
}
}
return result, nil
}
// we didn't get the port so we'll lookup the service using SRV records. If we can't lookup the
// service using the SRV record, we return the error.
_, nodes, err := net.LookupSRV(options.Service, "tcp", d.options.Network)
if err != nil {
return nil, err
}
// convert the nodes (net services) to routes
result := make([]Route, len(nodes))
for i, n := range nodes {
result[i] = Route{
Service: options.Service,
Address: fmt.Sprintf("%s:%d", n.Target, n.Port),
Network: d.options.Network,
}
}
return result, nil
}
func (d *dns) Watch(opts ...WatchOption) (Watcher, error) {
return nil, nil
}
func (d *dns) Name() string {
return d.options.Name
}
func (d *dns) String() string {
return "dns"
}

View File

@@ -7,7 +7,7 @@ import (
var (
// DefaultRouter is the global default router
DefaultRouter Router
DefaultRouter Router = NewRouter()
// DefaultNetwork is default micro network
DefaultNetwork = "micro"
// ErrRouteNotFound is returned when no route was found in the routing table

3
server/generate.go Normal file
View File

@@ -0,0 +1,3 @@
package server
//go:generate protoc -I./health -I../ -I/home/vtolstov/.cache/go-path/pkg/mod/github.com/unistack-org/micro-proto@v0.0.1 --micro_out=components=micro|http|server,standalone=false,debug=true,paths=source_relative:./health health/health.proto

87
server/health/health.go Normal file
View File

@@ -0,0 +1,87 @@
package pb
import (
"context"
"github.com/unistack-org/micro/v3/codec"
"github.com/unistack-org/micro/v3/errors"
"github.com/unistack-org/micro/v3/server"
)
var (
// guard to fail early
_ HealthServer = &handler{}
)
type handler struct {
server server.Server
opts Options
}
type CheckFunc func(context.Context) error
type Option func(*Options)
type Options struct {
LiveChecks []CheckFunc
ReadyChecks []CheckFunc
Version string
Name string
}
func LiveChecks(fns ...CheckFunc) Option {
return func(o *Options) {
o.LiveChecks = append(o.LiveChecks, fns...)
}
}
func ReadyChecks(fns ...CheckFunc) Option {
return func(o *Options) {
o.ReadyChecks = append(o.ReadyChecks, fns...)
}
}
func Name(name string) Option {
return func(o *Options) {
o.Name = name
}
}
func Version(version string) Option {
return func(o *Options) {
o.Version = version
}
}
func NewHealth(opts ...Option) *handler {
options := Options{}
for _, o := range opts {
o(&options)
}
return &handler{opts: options}
}
func (h *handler) Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
var err error
for _, fn := range h.opts.LiveChecks {
if err = fn(ctx); err != nil {
return errors.ServiceUnavailable(h.opts.Name, "%v", err)
}
}
return nil
}
func (h *handler) Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
var err error
for _, fn := range h.opts.ReadyChecks {
if err = fn(ctx); err != nil {
return errors.ServiceUnavailable(h.opts.Name, "%v", err)
}
}
return nil
}
func (h *handler) Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
rsp.Data = []byte(h.opts.Version)
return nil
}

View File

@@ -0,0 +1,62 @@
syntax = "proto3";
package health;
option go_package = "github.com/unistack-org/micro/v3/server/health;pb";
import "api/annotations.proto";
import "openapiv2/annotations.proto";
import "codec/frame.proto";
service Health {
rpc Live(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv2.openapiv2_operation) = {
operation_id: "Live";
responses: {
key: "default";
value: {
description: "Error response";
schema: {
json_schema: {
ref: "micro.codec.Frame";
}
}
}
}
};
option (micro.api.http) = { get: "/live"; };
};
rpc Ready(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv2.openapiv2_operation) = {
operation_id: "Ready";
responses: {
key: "default";
value: {
description: "Error response";
schema: {
json_schema: {
ref: "micro.codec.Frame";
}
}
}
}
};
option (micro.api.http) = { get: "/ready"; };
};
rpc Version(micro.codec.Frame) returns (micro.codec.Frame) {
option (micro.openapiv2.openapiv2_operation) = {
operation_id: "Version";
responses: {
key: "default";
value: {
description: "Error response";
schema: {
json_schema: {
ref: "micro.codec.Frame";
}
}
}
}
};
option (micro.api.http) = { get: "/version"; };
};
};

View File

@@ -0,0 +1,38 @@
// Code generated by protoc-gen-micro
// source: health.proto
package pb
import (
context "context"
api "github.com/unistack-org/micro/v3/api"
codec "github.com/unistack-org/micro/v3/codec"
)
func NewHealthEndpoints() []*api.Endpoint {
return []*api.Endpoint{
&api.Endpoint{
Name: "Health.Live",
Path: []string{"/live"},
Method: []string{"GET"},
Handler: "rpc",
},
&api.Endpoint{
Name: "Health.Ready",
Path: []string{"/ready"},
Method: []string{"GET"},
Handler: "rpc",
},
&api.Endpoint{
Name: "Health.Version",
Path: []string{"/version"},
Method: []string{"GET"},
Handler: "rpc",
},
}
}
type HealthServer interface {
Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}

View File

@@ -0,0 +1,43 @@
// Code generated by protoc-gen-micro
// source: health.proto
package pb
import (
context "context"
api "github.com/unistack-org/micro/v3/api"
codec "github.com/unistack-org/micro/v3/codec"
server "github.com/unistack-org/micro/v3/server"
)
type healthServer struct {
HealthServer
}
func (h *healthServer) Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.HealthServer.Live(ctx, req, rsp)
}
func (h *healthServer) Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.HealthServer.Ready(ctx, req, rsp)
}
func (h *healthServer) Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error {
return h.HealthServer.Version(ctx, req, rsp)
}
func RegisterHealthServer(s server.Server, sh HealthServer, opts ...server.HandlerOption) error {
type health interface {
Live(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Ready(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
Version(ctx context.Context, req *codec.Frame, rsp *codec.Frame) error
}
type Health struct {
health
}
h := &healthServer{sh}
var nopts []server.HandlerOption
for _, endpoint := range NewHealthEndpoints() {
nopts = append(nopts, api.WithEndpoint(endpoint))
}
return s.Handle(s.NewHandler(&Health{h}, append(nopts, opts...)...))
}

View File

@@ -39,8 +39,12 @@ type Options struct {
Meter meter.Meter
// Transport holds the transport
Transport transport.Transport
// Router for requests
Router Router
/*
// Router for requests
Router Router
*/
// Listener may be passed if already created
Listener net.Listener
// Wait group
@@ -262,12 +266,14 @@ func TLSConfig(t *tls.Config) Option {
}
}
/*
// WithRouter sets the request router
func WithRouter(r Router) Option {
return func(o *Options) {
o.Router = r
}
}
*/
// Wait tells the server to wait for requests to finish before exiting
// If `wg` is nil, server only wait for completion of rpc handler.

View File

@@ -65,6 +65,7 @@ type Server interface {
String() string
}
/*
// Router handle serving messages
type Router interface {
// ProcessMessage processes a message
@@ -72,6 +73,7 @@ type Router interface {
// ServeRequest processes a request to completion
ServeRequest(ctx context.Context, req Request, rsp Response) error
}
*/
// Message is an async message interface
type Message interface {

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()

View File

@@ -65,8 +65,24 @@ func IsEmpty(v reflect.Value) bool {
return v.IsNil()
case reflect.Invalid:
return true
case reflect.Struct:
var ok bool
for i := 0; i < v.NumField(); i++ {
ok = IsEmpty(v.FieldByIndex([]int{i}))
if !ok {
return false
}
}
default:
return false
}
return false
return true
}
// IsZero returns true if struct is zero (not have any defined values)
func IsZero(src interface{}) bool {
v := reflect.ValueOf(src)
return IsEmpty(v)
}
// Zero creates new zero interface

View File

@@ -43,3 +43,50 @@ func TestURLVars(t *testing.T) {
}
_ = mp
}
func TestIsZero(t *testing.T) {
testStr1 := struct {
Name string
Value string
Nested struct {
NestedName string
}
}{
Name: "test_name",
Value: "test_value",
}
testStr1.Nested.NestedName = "nested_name"
if ok := IsZero(testStr1); ok {
t.Fatalf("zero ret on non zero struct: %#+v", testStr1)
}
testStr1.Name = ""
testStr1.Value = ""
testStr1.Nested.NestedName = ""
if ok := IsZero(testStr1); !ok {
t.Fatalf("non zero ret on zero struct: %#+v", testStr1)
}
type testStr3 struct {
Nested string
}
type testStr2 struct {
Name string
Nested *testStr3
}
vtest := &testStr2{
Name: "test_name",
Nested: &testStr3{Nested: "nested_name"},
}
if ok := IsZero(vtest); ok {
t.Fatalf("zero ret on non zero struct: %#+v", vtest)
}
vtest.Nested = nil
vtest.Name = ""
if ok := IsZero(vtest); !ok {
t.Fatalf("non zero ret on zero struct: %#+v", vtest)
}
//t.Logf("XX %#+v\n", ok)
}