add config

config/source/consul/README.md

@@ -0,0 +1,49 @@
+# Consul Source
+
+The consul source reads config from consul key/values
+
+## Consul Format
+
+The consul source expects keys under the default prefix `/micro/config`
+
+Values are expected to be json
+
+```
+// set database
+consul kv put micro/config/database '{"address": "10.0.0.1", "port": 3306}'
+// set cache
+consul kv put micro/config/cache '{"address": "10.0.0.2", "port": 6379}'
+```
+
+Keys are split on `/` so access becomes
+
+```
+conf.Get("micro", "config", "database")
+```
+
+## New Source
+
+Specify source with data
+
+```go
+consulSource := consul.NewSource(
+	// optionally specify consul address; default to localhost:8500
+	consul.WithAddress("10.0.0.10:8500"),
+	// optionally specify prefix; defaults to /micro/config
+	consul.WithPrefix("/my/prefix"),
+  // optionally strip the provided prefix from the keys, defaults to false
+  consul.StripPrefix(true),
+)
+```
+
+## Load Source
+
+Load the source into config
+
+```go
+// Create new config
+conf := config.NewConfig()
+
+// Load file source
+conf.Load(consulSource)
+```

config/source/consul/consul.go

@@ -0,0 +1,121 @@
+package consul
+
+import (
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/hashicorp/consul/api"
+	"github.com/micro/go-micro/config/source"
+)
+
+// Currently a single consul reader
+type consul struct {
+	prefix      string
+	stripPrefix string
+	addr        string
+	opts        source.Options
+	client      *api.Client
+}
+
+var (
+	// DefaultPrefix is the prefix that consul keys will be assumed to have if you
+	// haven't specified one
+	DefaultPrefix = "/micro/config/"
+)
+
+func (c *consul) Read() (*source.ChangeSet, error) {
+	kv, _, err := c.client.KV().List(c.prefix, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if kv == nil || len(kv) == 0 {
+		return nil, fmt.Errorf("source not found: %s", c.prefix)
+	}
+
+	data, err := makeMap(c.opts.Encoder, kv, c.stripPrefix)
+	if err != nil {
+		return nil, fmt.Errorf("error reading data: %v", err)
+	}
+
+	b, err := c.opts.Encoder.Encode(data)
+	if err != nil {
+		return nil, fmt.Errorf("error reading source: %v", err)
+	}
+
+	cs := &source.ChangeSet{
+		Timestamp: time.Now(),
+		Format:    c.opts.Encoder.String(),
+		Source:    c.String(),
+		Data:      b,
+	}
+	cs.Checksum = cs.Sum()
+
+	return cs, nil
+}
+
+func (c *consul) String() string {
+	return "consul"
+}
+
+func (c *consul) Watch() (source.Watcher, error) {
+	w, err := newWatcher(c.prefix, c.addr, c.String(), c.stripPrefix, c.opts.Encoder)
+	if err != nil {
+		return nil, err
+	}
+	return w, nil
+}
+
+// NewSource creates a new consul source
+func NewSource(opts ...source.Option) source.Source {
+	options := source.NewOptions(opts...)
+
+	// use default config
+	config := api.DefaultConfig()
+
+	// check if there are any addrs
+	a, ok := options.Context.Value(addressKey{}).(string)
+	if ok {
+		addr, port, err := net.SplitHostPort(a)
+		if ae, ok := err.(*net.AddrError); ok && ae.Err == "missing port in address" {
+			port = "8500"
+			addr = a
+			config.Address = fmt.Sprintf("%s:%s", addr, port)
+		} else if err == nil {
+			config.Address = fmt.Sprintf("%s:%s", addr, port)
+		}
+	}
+
+	dc, ok := options.Context.Value(dcKey{}).(string)
+	if ok {
+		config.Datacenter = dc
+	}
+
+	token, ok := options.Context.Value(tokenKey{}).(string)
+	if ok {
+		config.Token = token
+	}
+
+	// create the client
+	client, _ := api.NewClient(config)
+
+	prefix := DefaultPrefix
+	sp := ""
+	f, ok := options.Context.Value(prefixKey{}).(string)
+	if ok {
+		prefix = f
+	}
+
+	if b, ok := options.Context.Value(stripPrefixKey{}).(bool); ok && b {
+		sp = prefix
+	}
+
+	return &consul{
+		prefix:      prefix,
+		stripPrefix: sp,
+		addr:        config.Address,
+		opts:        options,
+		client:      client,
+	}
+}

config/source/consul/options.go

@@ -0,0 +1,63 @@
+package consul
+
+import (
+	"context"
+
+	"github.com/micro/go-micro/config/source"
+)
+
+type addressKey struct{}
+type prefixKey struct{}
+type stripPrefixKey struct{}
+type dcKey struct{}
+type tokenKey struct{}
+
+// WithAddress sets the consul address
+func WithAddress(a string) source.Option {
+	return func(o *source.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, addressKey{}, a)
+	}
+}
+
+// WithPrefix sets the key prefix to use
+func WithPrefix(p string) source.Option {
+	return func(o *source.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, prefixKey{}, p)
+	}
+}
+
+// StripPrefix indicates whether to remove the prefix from config entries, or leave it in place.
+func StripPrefix(strip bool) source.Option {
+	return func(o *source.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+
+		o.Context = context.WithValue(o.Context, stripPrefixKey{}, strip)
+	}
+}
+
+func WithDatacenter(p string) source.Option {
+	return func(o *source.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, dcKey{}, p)
+	}
+}
+
+// WithToken sets the key token to use
+func WithToken(p string) source.Option {
+	return func(o *source.Options) {
+		if o.Context == nil {
+			o.Context = context.Background()
+		}
+		o.Context = context.WithValue(o.Context, tokenKey{}, p)
+	}
+}

config/source/consul/util.go

@@ -0,0 +1,49 @@
+package consul
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/consul/api"
+	"github.com/micro/go-micro/config/encoder"
+)
+
+func makeMap(e encoder.Encoder, kv api.KVPairs, stripPrefix string) (map[string]interface{}, error) {
+	data := make(map[string]interface{})
+
+	// consul guarantees lexicographic order, so no need to sort
+	for _, v := range kv {
+		pathString := strings.TrimPrefix(strings.TrimPrefix(v.Key, stripPrefix), "/")
+		var val map[string]interface{}
+
+		// ensure a valid value is stored at this location
+		if len(v.Value) > 0 {
+			if err := e.Decode(v.Value, &val); err != nil {
+				return nil, fmt.Errorf("faild decode value. path: %s, error: %s", pathString, err)
+			}
+		}
+
+		// set target at the root
+		target := data
+
+		// then descend to the target location, creating as we go, if need be
+		if pathString != "" {
+			path := strings.Split(pathString, "/")
+			// find (or create) the location we want to put this value at
+			for _, dir := range path {
+				if _, ok := target[dir]; !ok {
+					target[dir] = make(map[string]interface{})
+				}
+				target = target[dir].(map[string]interface{})
+			}
+
+		}
+
+		// copy over the keys from the value
+		for k := range val {
+			target[k] = val[k]
+		}
+	}
+
+	return data, nil
+}

config/source/consul/watcher.go

@@ -0,0 +1,96 @@
+package consul
+
+import (
+	"errors"
+	"time"
+
+	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/api/watch"
+	"github.com/micro/go-micro/config/encoder"
+	"github.com/micro/go-micro/config/source"
+)
+
+type watcher struct {
+	e           encoder.Encoder
+	name        string
+	stripPrefix string
+
+	wp   *watch.Plan
+	ch   chan *source.ChangeSet
+	exit chan bool
+}
+
+func newWatcher(key, addr, name, stripPrefix string, e encoder.Encoder) (source.Watcher, error) {
+	w := &watcher{
+		e:           e,
+		name:        name,
+		stripPrefix: stripPrefix,
+		ch:          make(chan *source.ChangeSet),
+		exit:        make(chan bool),
+	}
+
+	wp, err := watch.Parse(map[string]interface{}{"type": "keyprefix", "prefix": key})
+	if err != nil {
+		return nil, err
+	}
+
+	wp.Handler = w.handle
+
+	// wp.Run is a blocking call and will prevent newWatcher from returning
+	go wp.Run(addr)
+
+	w.wp = wp
+
+	return w, nil
+}
+
+func (w *watcher) handle(idx uint64, data interface{}) {
+	if data == nil {
+		return
+	}
+
+	kvs, ok := data.(api.KVPairs)
+	if !ok {
+		return
+	}
+
+	d, err := makeMap(w.e, kvs, w.stripPrefix)
+	if err != nil {
+		return
+	}
+
+	b, err := w.e.Encode(d)
+	if err != nil {
+		return
+	}
+
+	cs := &source.ChangeSet{
+		Timestamp: time.Now(),
+		Format:    w.e.String(),
+		Source:    w.name,
+		Data:      b,
+	}
+	cs.Checksum = cs.Sum()
+
+	w.ch <- cs
+}
+
+func (w *watcher) Next() (*source.ChangeSet, error) {
+	select {
+	case cs := <-w.ch:
+		return cs, nil
+	case <-w.exit:
+		return nil, errors.New("watcher stopped")
+	}
+}
+
+func (w *watcher) Stop() error {
+	select {
+	case <-w.exit:
+		return nil
+	default:
+		w.wp.Stop()
+		close(w.exit)
+	}
+	return nil
+}