Compare commits
23 Commits
Author | SHA1 | Date | |
---|---|---|---|
52d8255974 | |||
9830cb48a9 | |||
92d7ab2105 | |||
d2935ef399 | |||
ce4c96ae0a | |||
|
14026d15be | ||
|
2df0c7643e | ||
e13c2c48fd | |||
8db55d2e55 | |||
ed61cad961 | |||
040fc4548f | |||
6189a1b980 | |||
eb2a450a7b | |||
|
d2a30a5da1 | ||
65889c66f6 | |||
dcdf133d5b | |||
8742b55305 | |||
4a64ee72f7 | |||
881d7afeea | |||
8c95448535 | |||
c1dc041d8c | |||
|
25be0ac0f0 | ||
|
86f73cac4e |
@@ -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
|
||||
|
@@ -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
|
||||
|
@@ -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
2
go.mod
@@ -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
|
||||
)
|
||||
|
5
go.sum
5
go.sum
@@ -10,9 +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-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 h1:WUcCxqQqDT0aXO4VnQbfMvp4zh7m1Gb2clVuHUAGGRE=
|
||||
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=
|
||||
|
@@ -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
3
meter/generate.go
Normal 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
34
meter/handler/handler.go
Normal file
@@ -0,0 +1,34 @@
|
||||
package handler
|
||||
|
||||
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
|
||||
}
|
28
meter/handler/handler.proto
Normal file
28
meter/handler/handler.proto
Normal file
@@ -0,0 +1,28 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package micro.meter.handler;
|
||||
option go_package = "github.com/unistack-org/micro/v3/meter/handler;handler";
|
||||
|
||||
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"; };
|
||||
};
|
||||
};
|
24
meter/handler/handler_micro.pb.go
Normal file
24
meter/handler/handler_micro.pb.go
Normal file
@@ -0,0 +1,24 @@
|
||||
// Code generated by protoc-gen-micro
|
||||
// source: handler.proto
|
||||
package handler
|
||||
|
||||
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
|
||||
}
|
33
meter/handler/handler_micro_http.pb.go
Normal file
33
meter/handler/handler_micro_http.pb.go
Normal file
@@ -0,0 +1,33 @@
|
||||
// Code generated by protoc-gen-micro
|
||||
// source: handler.proto
|
||||
package handler
|
||||
|
||||
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...)...))
|
||||
}
|
@@ -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
91
router/dns.go
Normal 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"
|
||||
}
|
@@ -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
3
server/generate.go
Normal 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
87
server/health/health.go
Normal file
@@ -0,0 +1,87 @@
|
||||
package health
|
||||
|
||||
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 NewHandler(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
|
||||
}
|
62
server/health/health.proto
Normal file
62
server/health/health.proto
Normal file
@@ -0,0 +1,62 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package micro.server.health;
|
||||
option go_package = "github.com/unistack-org/micro/v3/server/health;health";
|
||||
|
||||
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"; };
|
||||
};
|
||||
};
|
38
server/health/health_micro.pb.go
Normal file
38
server/health/health_micro.pb.go
Normal file
@@ -0,0 +1,38 @@
|
||||
// Code generated by protoc-gen-micro
|
||||
// source: health.proto
|
||||
package health
|
||||
|
||||
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
|
||||
}
|
43
server/health/health_micro_http.pb.go
Normal file
43
server/health/health_micro_http.pb.go
Normal file
@@ -0,0 +1,43 @@
|
||||
// Code generated by protoc-gen-micro
|
||||
// source: health.proto
|
||||
package health
|
||||
|
||||
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...)...))
|
||||
}
|
@@ -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.
|
||||
|
@@ -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 {
|
||||
|
16
service.go
16
service.go
@@ -103,48 +103,48 @@ func (s *service) Init(opts ...Option) error {
|
||||
// skip config as the struct not passed
|
||||
continue
|
||||
}
|
||||
if err = cfg.Init(config.Context(s.opts.Context)); err != nil {
|
||||
if err = cfg.Init(config.Context(cfg.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = cfg.Load(s.opts.Context); err != nil {
|
||||
if err = cfg.Load(cfg.Options().Context); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
for _, log := range s.opts.Loggers {
|
||||
if err = log.Init(logger.WithContext(s.opts.Context)); err != nil {
|
||||
if err = log.Init(logger.WithContext(log.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for _, reg := range s.opts.Registers {
|
||||
if err = reg.Init(register.Context(s.opts.Context)); err != nil {
|
||||
if err = reg.Init(register.Context(reg.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for _, brk := range s.opts.Brokers {
|
||||
if err = brk.Init(broker.Context(s.opts.Context)); err != nil {
|
||||
if err = brk.Init(broker.Context(brk.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for _, str := range s.opts.Stores {
|
||||
if err = str.Init(store.Context(s.opts.Context)); err != nil {
|
||||
if err = str.Init(store.Context(str.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for _, srv := range s.opts.Servers {
|
||||
if err = srv.Init(server.Context(s.opts.Context)); err != nil {
|
||||
if err = srv.Init(server.Context(srv.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
for _, cli := range s.opts.Clients {
|
||||
if err = cli.Init(client.Context(s.opts.Context)); err != nil {
|
||||
if err = cli.Init(client.Context(cli.Options().Context)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
@@ -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()
|
||||
|
||||
|
@@ -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
|
||||
|
@@ -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)
|
||||
}
|
||||
|
Reference in New Issue
Block a user