func (etcd *EtcdSource) syncNode(etcdAction string, etcdNode *client.Node) (Node, error) {
	node, err := etcd.parseNode(etcdNode)
	if err != nil {
		return node, err
	}
	switch etcdAction {
	case "create", "set", "update", "compareAndSwap":
	case "delete", "expire", "compareAndDelete":
		node.Remove = true
	default:
		return node, fmt.Errorf("Unknown etcd action: %s", etcdAction)
	}
	return node, nil
}
Handle changed node
package config
import (
	"fmt"
	"github.com/coreos/etcd/client"
	"golang.org/x/net/context"
	"log"
	"math/rand"
	"net/url"
	"strings"
	"time"
)
type EtcdOptions struct {
	Scheme string        `long:"etcd-scheme" value-name:"http|https" default:"http" description:"Set default scheme for etcd:// URLs"`
	Hosts  []string      `long:"etcd-host" value-name:"HOST:PORT" description:"Include hosts"`
	Prefix string        `long:"etcd-prefix" value-name:"/PATH" default:"/clusterf" description:"Namespace all keys under given path"`
	TTL    time.Duration `long:"etcd-ttl" value-name:"DURATION" default:"10s" description:"Write values with given TTL, and refresh at half of that"`
	mockRefreshDelay time.Duration  
}
func (options EtcdOptions) OpenURL(url *url.URL) (*EtcdSource, error) {
	switch url.Scheme {
	case "etcd":
	case "etcd+http":
		options.Scheme = "http"
	case "etcd+https":
		options.Scheme = "https"
	}
	for _, host := range strings.Split(url.Host, ",") {
		options.Hosts = append(options.Hosts, host)
	}
	if url.Path != "" {
		options.Prefix = url.Path
	}
	return options.Open()
}
func (options EtcdOptions) String() string {
	return fmt.Sprintf("etcd+%s://%s%s", options.Scheme, strings.Join(options.Hosts, ","), options.Prefix)
}
func (options EtcdOptions) clientConfig() (clientConfig client.Config, err error) {
	for _, host := range options.Hosts {
		endpointURL := url.URL{Scheme: options.Scheme, Host: host}
		clientConfig.Endpoints = append(clientConfig.Endpoints, endpointURL.String())
	}
	return
}
func (options EtcdOptions) Open() (*EtcdSource, error) {
	etcdSource := EtcdSource{
		options: options,
	}
	if clientConfig, err := options.clientConfig(); err != nil {
		return nil, err
	} else if client, err := client.New(clientConfig); err != nil {
		return nil, err
	} else {
		etcdSource.client = client
	}
	etcdSource.keysAPI = client.NewKeysAPI(etcdSource.client)
	return &etcdSource, nil
}
func fixupClusterError(err error) error {
	if clusterError, ok := err.(*client.ClusterError); ok {
		var errs []string
		for _, clusterErr := range clusterError.Errors {
			errs = append(errs, clusterErr.Error())
		}
		return fmt.Errorf("%s: %s", clusterError.Error(), strings.Join(errs, "; "))
	} else {
		return err
	}
}
type EtcdSource struct {
	options EtcdOptions
	client  client.Client
	keysAPI client.KeysAPI
	syncIndex uint64
	writeChan chan map[string]Node
	flushChan chan error
}
func (etcd *EtcdSource) String() string {
	return etcd.options.String()
}
func (etcd *EtcdSource) path(parts ...string) string {
	return strings.Join(append([]string{etcd.options.Prefix}, parts...), "/")
}
func (etcd *EtcdSource) Init() error {
	if response, err := etcd.keysAPI.Set(context.Background(), etcd.path(), "", &client.SetOptions{Dir: true}); err != nil {
		return fixupClusterError(err)
	} else {
		etcd.syncIndex = response.Node.CreatedIndex
	}
	return nil
}
func (etcd *EtcdSource) Scan() ([]Node, error) {
	response, err := etcd.keysAPI.Get(context.Background(), etcd.path(), &client.GetOptions{Recursive: true})
	if err == nil {
	} else if clientError, ok := err.(client.Error); ok && clientError.Code == client.ErrorCodeKeyNotFound {
		return nil, etcd.Init()
	} else {
		return nil, fixupClusterError(err)
	}
	if response.Node.Dir != true {
		return nil, fmt.Errorf("etcd prefix=%s is not a directory", response.Node.Key)
	}
	etcd.syncIndex = response.Index
	var nodes []Node
	err = etcd.scanNode(response.Node, func(node Node) { nodes = append(nodes, node) })
	return nodes, err
}
func (etcd *EtcdSource) parseNode(etcdNode *client.Node) (node Node, err error) {
	path := etcdNode.Key
	if !strings.HasPrefix(path, etcd.options.Prefix) {
		return node, fmt.Errorf("node outside tree: %s", path)
	}
	path = strings.TrimPrefix(path, etcd.options.Prefix)
	path = strings.Trim(path, "/")
	node.Source = etcd
	node.Path = path
	node.IsDir = etcdNode.Dir
	node.Value = etcdNode.Value
	return
}
func (etcd *EtcdSource) scanNode(etcdNode *client.Node, handler func(node Node)) error {
	if node, err := etcd.parseNode(etcdNode); err != nil {
		return err
	} else {
		handler(node)
	}
	for _, childNode := range etcdNode.Nodes {
		if err := etcd.scanNode(childNode, handler); err != nil {
			return err
		}
	}
	return nil
}
func (etcd *EtcdSource) Sync(syncChan chan Node) error {
	go etcd.watch(syncChan)
	return nil
}
func (etcd *EtcdSource) watch(watchChan chan Node) {
	defer close(watchChan)
	watcher := etcd.keysAPI.Watcher(etcd.path(), &client.WatcherOptions{AfterIndex: etcd.syncIndex, Recursive: true})
	for {
		if response, err := watcher.Next(context.Background()); err != nil {
			err = fixupClusterError(err)
			log.Printf("config:EtcdSource.watch: %v", err)
			return
		} else if node, err := etcd.syncNode(response.Action, response.Node); err != nil {
			log.Printf("config:EtcdSource.watch %#v: syncNode: %s", response, err)
			return
		} else {
			log.Printf("config:EtcdSource.watch: %v %v", response.Action, node)
			watchChan <- node
		}
	}
}