README
¶
sd — Service Discovery
The sd package wires together service discovery, load balancing, and retry
into a single callable endpoint.Endpoint.
Quick start (no Consul needed)
import (
"github.com/dreamsxin/go-kit/sd"
"github.com/dreamsxin/go-kit/sd/instance"
)
// In-memory instancer — perfect for tests and local dev
cache := instance.NewCache()
cache.Update(events.Event{Instances: []string{"host1:8080", "host2:8080"}})
ep := sd.NewEndpoint(cache, factory, logger,
sd.WithMaxRetries(3),
sd.WithTimeout(500*time.Millisecond),
)
resp, err := ep(ctx, request)
With Consul
import "github.com/dreamsxin/go-kit/sd/consul"
instancer := consul.NewInstancer(consulClient, logger, "my-service", true)
defer instancer.Stop()
ep := sd.NewEndpoint(instancer, factory, logger,
sd.WithMaxRetries(3),
sd.WithTimeout(500*time.Millisecond),
sd.WithInvalidateOnError(5*time.Second),
)
Options
| Option | Default | Description |
|---|---|---|
WithMaxRetries(n) |
3 | Max retry attempts; 0 = retry until timeout |
WithTimeout(d) |
500ms | Total budget including all retries |
WithInvalidateOnError(d) |
disabled | Clear cache after SD error grace period |
Architecture
Instancer → Endpointer → RoundRobin → Retry → Endpoint
Each layer is independently usable:
// Manual assembly (full control)
ep := endpointer.NewEndpointer(instancer, factory, logger)
lb := balancer.NewRoundRobin(ep)
call := executor.Retry(3, 500*time.Millisecond, lb)
Retry strategies
// Fixed max attempts
executor.Retry(3, time.Second, lb)
// Unlimited within timeout
executor.RetryAlways(2*time.Second, lb)
// Custom callback — stop on non-retryable errors
executor.RetryWithCallback(time.Second, lb,
func(n int, err error) (keepTrying bool, replacement error) {
if errors.Is(err, ErrInvalidArgument) {
return false, err // stop immediately
}
return n < 5, nil // retry up to 5 times
},
)
Consul registration
registrar := consul.NewRegistrar(client, logger, "my-service", "10.0.0.1", 8080,
consul.IDRegistrarOptions("my-service-1"),
consul.CheckRegistrarOptions(&stdconsul.AgentServiceCheck{
HTTP: "http://10.0.0.1:8080/health",
Interval: "10s",
}),
)
registrar.Register()
defer registrar.Deregister()
See also
examples/sd/— runnable demo of every sd componentexamples/profilesvc/client/— Consul-backed client example
Documentation
¶
Overview ¶
Package sd provides service-discovery helpers that wire together an Instancer, EndpointCache, Balancer, and Retry executor into a single callable endpoint.Endpoint.
Typical usage:
instancer := consul.NewInstancer(consulClient, logger, "my-service", true)
defer instancer.Stop()
ep := sd.NewEndpoint(instancer, factory, logger,
sd.WithMaxRetries(3),
sd.WithTimeout(500*time.Millisecond),
sd.WithInvalidateOnError(5*time.Second),
)
resp, err := ep(ctx, request)
Index ¶
- func NewEndpoint(src interfaces.Instancer, factory endpoint.Factory, logger *kitlog.Logger, ...) endpoint.Endpoint
- func NewEndpointCloser(src interfaces.Instancer, factory endpoint.Factory, logger *kitlog.Logger, ...) (endpoint.Endpoint, io.Closer)
- func NewEndpointWithDefaults(src interfaces.Instancer, factory endpoint.Factory, logger *kitlog.Logger) endpoint.Endpoint
- type Option
- type Options
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewEndpoint ¶
func NewEndpoint( src interfaces.Instancer, factory endpoint.Factory, logger *kitlog.Logger, opts ...Option, ) endpoint.Endpoint
NewEndpoint wires together an Instancer → Endpointer → RoundRobin balancer → Retry executor and returns a single endpoint.Endpoint ready to call.
The returned endpoint automatically distributes requests across healthy instances and retries on transient failures.
func NewEndpointCloser ¶ added in v1.3.0
func NewEndpointCloser( src interfaces.Instancer, factory endpoint.Factory, logger *kitlog.Logger, opts ...Option, ) (endpoint.Endpoint, io.Closer)
NewEndpointCloser is like NewEndpoint but also returns an io.Closer that must be called to stop the background goroutine started by the Endpointer.
ep, closer := sd.NewEndpointCloser(instancer, factory, logger) defer closer.Close()
func NewEndpointWithDefaults ¶
func NewEndpointWithDefaults( src interfaces.Instancer, factory endpoint.Factory, logger *kitlog.Logger, ) endpoint.Endpoint
NewEndpointWithDefaults is identical to NewEndpoint but uses sensible production defaults without requiring any options:
- MaxRetries: 3
- Timeout: 500ms
- InvalidateOnError: 5s
Use NewEndpoint when you need to customise these values.
Types ¶
type Option ¶
type Option func(*Options)
Option is a functional option for NewEndpoint.
func WithInvalidateOnError ¶
WithInvalidateOnError enables cache invalidation after a SD error. The cache is cleared once the given grace period has elapsed.
func WithMaxRetries ¶
WithMaxRetries sets the maximum retry count. 0 means retry until Timeout.
func WithTimeout ¶
WithTimeout sets the total call timeout (including retries).
type Options ¶
type Options struct {
// MaxRetries is the maximum number of retry attempts (default 3).
// Set to 0 for unlimited retries within Timeout.
MaxRetries int
// Timeout is the total time budget for one call including all retries
// (default 500ms).
Timeout time.Duration
// InvalidateOnError, when > 0, causes the endpoint cache to be cleared
// after the given duration following a service-discovery error.
InvalidateOnError time.Duration
}
Options controls the behaviour of NewEndpoint.
Source Files
Directories