micro/registry/consul_registry.go

290 lines
5.7 KiB
Go
Raw Normal View History

2015-01-14 02:31:27 +03:00
package registry
import (
2016-01-16 23:25:18 +03:00
"crypto/tls"
"encoding/json"
2015-01-14 02:31:27 +03:00
"errors"
"fmt"
"net"
2016-01-16 23:25:18 +03:00
"net/http"
"runtime"
"time"
2015-01-14 02:31:27 +03:00
consul "github.com/hashicorp/consul/api"
2015-01-14 02:31:27 +03:00
)
type consulRegistry struct {
2015-02-15 02:00:47 +03:00
Address string
Client *consul.Client
2015-12-19 21:28:08 +03:00
Options Options
}
2016-01-17 02:39:47 +03:00
func newTransport(config *tls.Config) *http.Transport {
if config == nil {
config = &tls.Config{
InsecureSkipVerify: true,
}
}
2016-01-16 23:25:18 +03:00
t := &http.Transport{
Proxy: http.ProxyFromEnvironment,
Dial: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
}).Dial,
TLSHandshakeTimeout: 10 * time.Second,
2016-01-17 02:39:47 +03:00
TLSClientConfig: config,
2016-01-16 23:25:18 +03:00
}
runtime.SetFinalizer(&t, func(tr **http.Transport) {
(*tr).CloseIdleConnections()
})
return t
}
2015-10-11 14:05:20 +03:00
func encodeEndpoints(en []*Endpoint) []string {
var tags []string
for _, e := range en {
if b, err := json.Marshal(e); err == nil {
tags = append(tags, "e="+string(b))
}
}
return tags
}
func decodeEndpoints(tags []string) []*Endpoint {
var en []*Endpoint
for _, tag := range tags {
if len(tag) == 0 || tag[0] != 'e' {
continue
}
var e *Endpoint
if err := json.Unmarshal([]byte(tag[2:]), &e); err == nil {
en = append(en, e)
}
}
return en
}
2015-05-27 00:39:48 +03:00
func encodeMetadata(md map[string]string) []string {
var tags []string
for k, v := range md {
if b, err := json.Marshal(map[string]string{
k: v,
}); err == nil {
2015-10-11 14:05:20 +03:00
tags = append(tags, "t="+string(b))
}
}
return tags
}
2015-05-27 00:39:48 +03:00
func decodeMetadata(tags []string) map[string]string {
md := make(map[string]string)
for _, tag := range tags {
2015-10-11 14:05:20 +03:00
if len(tag) == 0 || tag[0] != 't' {
continue
}
var kv map[string]string
2015-10-11 14:05:20 +03:00
if err := json.Unmarshal([]byte(tag[2:]), &kv); err == nil {
for k, v := range kv {
md[k] = v
}
}
}
return md
2015-02-15 02:00:47 +03:00
}
2015-01-14 02:31:27 +03:00
func encodeVersion(v string) string {
return "v=" + v
}
func decodeVersion(tags []string) (string, bool) {
for _, tag := range tags {
if len(tag) == 0 || tag[0] != 'v' {
continue
}
return tag[2:], true
}
return "", false
}
func newConsulRegistry(addrs []string, opts ...Option) Registry {
2015-12-19 21:28:08 +03:00
var opt Options
for _, o := range opts {
o(&opt)
}
// use default config
config := consul.DefaultConfig()
2015-12-19 21:28:08 +03:00
// set timeout
if opt.Timeout > 0 {
config.HttpClient.Timeout = opt.Timeout
}
// check if there are any addrs
if len(addrs) > 0 {
addr, port, err := net.SplitHostPort(addrs[0])
if ae, ok := err.(*net.AddrError); ok && ae.Err == "missing port in address" {
port = "8500"
addr = addrs[0]
config.Address = fmt.Sprintf("%s:%s", addr, port)
} else if err == nil {
config.Address = fmt.Sprintf("%s:%s", addr, port)
}
}
2015-12-19 21:28:08 +03:00
2016-01-16 23:25:18 +03:00
// requires secure connection?
if opt.Secure || opt.TLSConfig != nil {
2016-01-16 23:25:18 +03:00
config.Scheme = "https"
// We're going to support InsecureSkipVerify
2016-01-17 02:39:47 +03:00
config.HttpClient.Transport = newTransport(opt.TLSConfig)
2016-01-16 23:25:18 +03:00
}
2015-12-19 21:28:08 +03:00
// create the client
2015-10-26 23:23:57 +03:00
client, _ := consul.NewClient(config)
cr := &consulRegistry{
2015-12-05 04:12:29 +03:00
Address: config.Address,
Client: client,
2015-12-19 21:28:08 +03:00
Options: opt,
}
return cr
}
func (c *consulRegistry) Deregister(s *Service) error {
if len(s.Nodes) == 0 {
2015-01-14 02:31:27 +03:00
return errors.New("Require at least one node")
}
2016-01-26 23:44:29 +03:00
node := s.Nodes[0]
2016-01-27 02:32:27 +03:00
return c.Client.Agent().ServiceDeregister(node.Id)
2015-01-14 02:31:27 +03:00
}
2016-01-27 02:32:27 +03:00
func (c *consulRegistry) Register(s *Service, opts ...RegisterOption) error {
if len(s.Nodes) == 0 {
2015-01-14 02:31:27 +03:00
return errors.New("Require at least one node")
}
2016-01-27 02:32:27 +03:00
var options RegisterOptions
for _, o := range opts {
o(&options)
}
node := s.Nodes[0]
2015-05-27 00:39:48 +03:00
tags := encodeMetadata(node.Metadata)
2015-10-11 14:05:20 +03:00
tags = append(tags, encodeEndpoints(s.Endpoints)...)
tags = append(tags, encodeVersion(s.Version))
2015-01-14 02:31:27 +03:00
2016-01-27 03:32:16 +03:00
var check *consul.AgentServiceCheck
if options.TTL > time.Duration(0) {
check = &consul.AgentServiceCheck{
TTL: fmt.Sprintf("%v", options.TTL),
}
}
2016-01-27 03:15:46 +03:00
if err := c.Client.Agent().ServiceRegister(&consul.AgentServiceRegistration{
ID: node.Id,
Name: s.Name,
Tags: tags,
Port: node.Port,
Address: node.Address,
2016-01-27 03:32:16 +03:00
Check: check,
2016-01-27 03:15:46 +03:00
}); err != nil {
return err
}
2015-01-14 02:31:27 +03:00
2016-01-27 03:32:16 +03:00
if options.TTL == time.Duration(0) {
return nil
}
2016-01-27 03:15:46 +03:00
return c.Client.Agent().PassTTL("service:"+node.Id, "")
2015-01-14 02:31:27 +03:00
}
2015-11-08 04:48:48 +03:00
func (c *consulRegistry) GetService(name string) ([]*Service, error) {
2016-01-27 02:32:27 +03:00
rsp, _, err := c.Client.Health().Service(name, "", true, nil)
2015-01-14 02:31:27 +03:00
if err != nil {
return nil, err
}
2015-11-08 04:48:48 +03:00
serviceMap := map[string]*Service{}
2015-01-14 02:31:27 +03:00
for _, s := range rsp {
2016-01-27 02:32:27 +03:00
if s.Service.Service != name {
2015-01-14 02:31:27 +03:00
continue
}
// version is now a tag
2016-01-27 02:32:27 +03:00
version, found := decodeVersion(s.Service.Tags)
// service ID is now the node id
2016-01-27 02:32:27 +03:00
id := s.Service.ID
// key is always the version
key := version
// address is service address
2016-01-27 02:32:27 +03:00
address := s.Service.Address
// if we can't get the new type of version
// use old the old ways
if !found {
// id was set as node
2016-01-27 02:32:27 +03:00
id = s.Node.Node
// key was service id
2016-01-27 02:32:27 +03:00
key = s.Service.ID
// version was service id
2016-01-27 02:32:27 +03:00
version = s.Service.ID
// address was address
2016-01-27 02:32:27 +03:00
address = s.Node.Address
2015-11-08 04:48:48 +03:00
}
svc, ok := serviceMap[key]
if !ok {
svc = &Service{
2016-01-27 02:32:27 +03:00
Endpoints: decodeEndpoints(s.Service.Tags),
Name: s.Service.Service,
2015-11-08 04:48:48 +03:00
Version: version,
}
serviceMap[key] = svc
}
svc.Nodes = append(svc.Nodes, &Node{
Id: id,
Address: address,
2016-01-27 02:32:27 +03:00
Port: s.Service.Port,
Metadata: decodeMetadata(s.Service.Tags),
2015-01-14 02:31:27 +03:00
})
}
2015-11-08 04:48:48 +03:00
var services []*Service
for _, service := range serviceMap {
services = append(services, service)
}
return services, nil
2015-01-14 02:31:27 +03:00
}
func (c *consulRegistry) ListServices() ([]*Service, error) {
2015-12-19 21:28:08 +03:00
rsp, _, err := c.Client.Catalog().Services(nil)
if err != nil {
return nil, err
}
2015-12-05 04:12:29 +03:00
var services []*Service
for service, _ := range rsp {
services = append(services, &Service{Name: service})
}
return services, nil
}
func (c *consulRegistry) Watch() (Watcher, error) {
return newConsulWatcher(c)
2015-01-14 02:31:27 +03:00
}
2015-12-20 00:56:14 +03:00
func (c *consulRegistry) String() string {
return "consul"
}