Documentation
¶
Overview ¶
Package etcd provides a Subscriber and Registrar implementation for etcd. If you use etcd as your service discovery system, this package will help you implement the registration and client-side load balancing patterns.
Example ¶
package main
import (
"io"
"time"
"golang.org/x/net/context"
"github.com/go-kit/kit/endpoint"
"github.com/go-kit/kit/log"
"github.com/go-kit/kit/sd/lb"
)
func main() {
// Let's say this is a service that means to register itself.
// First, we will set up some context.
var (
etcdServer = "http://10.0.0.1:2379" // don't forget schema and port!
prefix = "/services/foosvc/" // known at compile time
instance = "1.2.3.4:8080" // taken from runtime or platform, somehow
key = prefix + instance // should be globally unique
value = "http://" + instance // based on our transport
ctx = context.Background()
)
// Build the client.
client, err := NewClient(ctx, []string{etcdServer}, ClientOptions{})
if err != nil {
panic(err)
}
// Build the registrar.
registrar := NewRegistrar(client, Service{
Key: key,
Value: value,
}, log.NewNopLogger())
// Register our instance.
registrar.Register()
// At the end of our service lifecycle, for example at the end of func main,
// we should make sure to deregister ourselves. This is important! Don't
// accidentally skip this step by invoking a log.Fatal or os.Exit in the
// interim, which bypasses the defer stack.
defer registrar.Deregister()
// It's likely that we'll also want to connect to other services and call
// their methods. We can build a subscriber to listen for changes from etcd
// and build endpoints, wrap it with a load-balancer to pick a single
// endpoint, and finally wrap it with a retry strategy to get something that
// can be used as an endpoint directly.
barPrefix := "/services/barsvc"
subscriber, err := NewSubscriber(client, barPrefix, barFactory, log.NewNopLogger())
if err != nil {
panic(err)
}
balancer := lb.NewRoundRobin(subscriber)
retry := lb.Retry(3, 3*time.Second, balancer)
// And now retry can be used like any other endpoint.
req := struct{}{}
if _, err = retry(ctx, req); err != nil {
panic(err)
}
}
func barFactory(string) (endpoint.Endpoint, io.Closer, error) { return endpoint.Nop, nil, nil }
Output:
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNoKey indicates a client method needs a key but receives none. ErrNoKey = errors.New("no key provided") // ErrNoValue indicates a client method needs a value but receives none. ErrNoValue = errors.New("no value provided") )
Functions ¶
This section is empty.
Types ¶
type Client ¶
type Client interface {
// GetEntries queries the given prefix in etcd and returns a slice
// containing the values of all keys found, recursively, underneath that
// prefix.
GetEntries(prefix string) ([]string, error)
// WatchPrefix watches the given prefix in etcd for changes. When a change
// is detected, it will signal on the passed channel. Clients are expected
// to call GetEntries to update themselves with the latest set of complete
// values. WatchPrefix will always send an initial sentinel value on the
// channel after establishing the watch, to ensure that clients always
// receive the latest set of values. WatchPrefix will block until the
// context passed to the NewClient constructor is terminated.
WatchPrefix(prefix string, ch chan struct{})
// Register a service with etcd.
Register(s Service) error
// Deregister a service with etcd.
Deregister(s Service) error
}
Client is a wrapper around the etcd client.
func NewClient ¶
NewClient returns Client with a connection to the named machines. It will return an error if a connection to the cluster cannot be made. The parameter machines needs to be a full URL with schemas. e.g. "http://localhost:2379" will work, but "localhost:2379" will not.
type ClientOptions ¶
type ClientOptions struct {
Cert string
Key string
CACert string
DialTimeout time.Duration
DialKeepAlive time.Duration
HeaderTimeoutPerRequest time.Duration
}
ClientOptions defines options for the etcd client. All values are optional. If any duration is not specified, a default of 3 seconds will be used.
type Registrar ¶
type Registrar struct {
// contains filtered or unexported fields
}
Registrar registers service instance liveness information to etcd.
func NewRegistrar ¶
NewRegistrar returns a etcd Registrar acting on the provided catalog registration (service).
func (*Registrar) Deregister ¶
func (r *Registrar) Deregister()
Deregister implements the sd.Registrar interface. Call it when you want your service to be deregistered from etcd, typically just prior to shutdown.
type Service ¶
type Service struct {
Key string // unique key, e.g. "/service/foobar/1.2.3.4:8080"
Value string // returned to subscribers, e.g. "http://1.2.3.4:8080"
DeleteOptions *etcd.DeleteOptions
}
Service holds the instance identifying data you want to publish to etcd. Key must be unique, and value is the string returned to subscribers, typically called the "instance" string in other parts of package sd.
type Subscriber ¶ added in v0.1.0
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber yield endpoints stored in a certain etcd keyspace. Any kind of change in that keyspace is watched and will update the Subscriber endpoints.
func NewSubscriber ¶ added in v0.1.0
func NewSubscriber(c Client, prefix string, factory sd.Factory, logger log.Logger) (*Subscriber, error)
NewSubscriber returns an etcd subscriber. It will start watching the given prefix for changes, and update the endpoints.
func (*Subscriber) Endpoints ¶ added in v0.1.0
func (s *Subscriber) Endpoints() ([]endpoint.Endpoint, error)
Endpoints implements the Subscriber interface.
func (*Subscriber) Stop ¶ added in v0.1.0
func (s *Subscriber) Stop()
Stop terminates the Subscriber.
Source Files