Etcd - manager support (#439)

Signed-off-by: David Chung <[email protected]>

Author: David Chung

cmd/manager/etcd.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	"github.com/docker/go-connections/tlsconfig"
+	"github.com/docker/infrakit/pkg/discovery"
+	etcd_leader "github.com/docker/infrakit/pkg/leader/etcd/v3"
+	etcd_store "github.com/docker/infrakit/pkg/store/etcd/v3"
+	etcd "github.com/docker/infrakit/pkg/util/etcd/v3"
+	log "github.com/golang/glog"
+	"github.com/spf13/cobra"
+)
+
+func etcdEnvironment(getConfig func() config) *cobra.Command {
+
+	defaultEndpoint := etcd.LocalIP() + ":2379"
+
+	cmd := &cobra.Command{
+		Use:   "etcd",
+		Short: "etcd v3 for leader detection and storage",
+	}
+	pollInterval := cmd.Flags().Duration("poll-interval", 5*time.Second, "Leader polling interval")
+	requestTimeout := cmd.Flags().Duration("request-timeout", 1*time.Second, "Request timeout")
+	endpoint := cmd.Flags().String("endpoint", defaultEndpoint, "Etcd endpoint (v3 grpc)")
+	caFile := cmd.Flags().String("tlscacert", "", "TLS CA cert file path")
+	certFile := cmd.Flags().String("tlscert", "", "TLS cert file path")
+	tlsKey := cmd.Flags().String("tlskey", "", "TLS key file path")
+	insecureSkipVerify := cmd.Flags().Bool("tlsverify", true, "True to skip TLS")
+	cmd.RunE = func(c *cobra.Command, args []string) error {
+
+		options := etcd.Options{
+			Config: clientv3.Config{
+				Endpoints: []string{*endpoint},
+			},
+			RequestTimeout: *requestTimeout,
+		}
+
+		if *caFile != "" && *certFile != "" && *tlsKey != "" {
+			config, err := tlsconfig.Client(tlsconfig.Options{
+				CAFile:             *caFile,
+				CertFile:           *certFile,
+				KeyFile:            *tlsKey,
+				InsecureSkipVerify: *insecureSkipVerify,
+			})
+
+			if err != nil {
+				return err
+			}
+			options.Config.TLS = config
+		}
+
+		etcdClient, err := etcd.NewClient(options)
+		log.Infoln("Connect to etcd3", *endpoint, "err=", err)
+		if err != nil {
+			return err
+		}
+		defer etcdClient.Close()
+
+		// Start the leader and storage backends
+
+		leader := etcd_leader.NewDetector(*pollInterval, etcdClient)
+		snapshot, err := etcd_store.NewSnapshot(etcdClient)
+		if err != nil {
+			return err
+		}
+
+		plugins, err := discovery.NewPluginDiscovery()
+		if err != nil {
+			return err
+		}
+
+		cfg := getConfig()
+		cfg.plugins = plugins
+		cfg.leader = leader
+		cfg.snapshot = snapshot
+
+		return runMain(cfg)
+	}
+
+	return cmd
+}

cmd/manager/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	goflag "flag"
 	"os"
 	"path/filepath"
 	"time"
@@ -20,6 +21,7 @@ import (
 )
 
 func init() {
+
 	cli.RegisterInfo("manager - swarm option",
 		map[string]interface{}{
 			"DockerClientAPIVersion": docker.ClientVersion,
@@ -58,7 +60,15 @@ func main() {
 		}
 	}
 
-	cmd.AddCommand(cli.VersionCommand(), osEnvironment(buildConfig), swarmEnvironment(buildConfig))
+	cmd.AddCommand(cli.VersionCommand(),
+		osEnvironment(buildConfig),
+		swarmEnvironment(buildConfig),
+		etcdEnvironment(buildConfig),
+	)
+
+	// glog flag compatibility
+	cmd.PersistentFlags().AddGoFlagSet(goflag.CommandLine)
+	goflag.CommandLine.Parse([]string{})
 
 	err := cmd.Execute()
 	if err != nil {

cmd/manager/swarm.go

@@ -26,7 +26,7 @@ func swarmEnvironment(getConfig func() config) *cobra.Command {
 	insecureSkipVerify := cmd.Flags().Bool("tlsverify", true, "True to skip TLS")
 	cmd.RunE = func(c *cobra.Command, args []string) error {
 
-		dockerClient, err := docker.NewDockerClient(*host, &tlsconfig.Options{
+		dockerClient, err := docker.NewClient(*host, &tlsconfig.Options{
 			CAFile:             *caFile,
 			CertFile:           *certFile,
 			KeyFile:            *tlsKey,

examples/flavor/swarm/flavor.go

@@ -57,7 +57,7 @@ func DockerClient(spec Spec) (docker.APIClientCloser, error) {
 		tls = &tlsconfig.Options{}
 	}
 
-	return docker.NewDockerClient(spec.Docker.Host, tls)
+	return docker.NewClient(spec.Docker.Host, tls)
 }
 
 // baseFlavor is the base implementation.  The manager / worker implementations will provide override.

pkg/leader/etcd/v3/etcd.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/coreos/etcd/clientv3"
 	"github.com/docker/infrakit/pkg/leader"
 	"github.com/docker/infrakit/pkg/util/etcd/v3"
 	log "github.com/golang/glog"
@@ -18,25 +19,35 @@ func NewDetector(pollInterval time.Duration, client *etcd.Client) leader.Detecto
 }
 
 // AmILeader checks if this node is a leader
-func AmILeader(ctx context.Context, client *etcd.Client) (bool, error) {
+func AmILeader(ctx context.Context, client *etcd.Client) (isLeader bool, err error) {
 
-	// get status of node
 	endpoint := ""
+	var statusResp *clientv3.StatusResponse
+
+	defer func() {
+		log.V(100).Infoln("checking status at", endpoint, "resp=", statusResp, "err=", err, "leader=", isLeader)
+	}()
+
+	// get status of node
 	if len(client.Options.Config.Endpoints) > 0 {
 		endpoint = client.Options.Config.Endpoints[0]
 	}
 
 	if endpoint == "" {
-		return false, fmt.Errorf("bad config:%v", client.Options)
+		isLeader = false
+		err = fmt.Errorf("bad config:%v", client.Options)
+		return
 	}
 
-	statusResp, err := client.Client.Status(ctx, endpoint)
-	log.V(50).Infoln("checking status at", endpoint, "resp=", statusResp, "err=", err)
+	statusResp, err = client.Client.Status(ctx, endpoint)
 	if err != nil {
-		return false, err
+		isLeader = false
+		return
 	}
 
 	// The header has the self, assuming the endpoint is the self node.
 	// The response has the id of the leader. So just compare self id and the leader id.
-	return statusResp.Leader == statusResp.Header.MemberId, nil
+	isLeader = statusResp.Leader == statusResp.Header.MemberId
+
+	return
 }

pkg/store/etcd/v3/etcd.go

@@ -15,14 +15,10 @@ const (
 	DefaultKey = "infrakit/configs/groups.json"
 )
 
-// NewSnapshot returns a snapshot given the options
-func NewSnapshot(options etcd.Options) (store.Snapshot, error) {
-	cli, err := etcd.NewClient(options)
-	if err != nil {
-		return nil, err
-	}
+// NewSnapshot returns a snapshot given the client
+func NewSnapshot(client *etcd.Client) (store.Snapshot, error) {
 	return &snapshot{
-		client: cli,
+		client: client,
 		key:    DefaultKey,
 	}, nil
 }
@@ -78,11 +74,21 @@ func (s *snapshot) Load(output interface{}) error {
 		}
 	}
 
+	if resp == nil {
+		log.Warningln("response is nil. server down?")
+		return nil
+	}
+
 	if resp.Count > 1 {
 		log.Warningf("more than 1 config %v", resp)
 		return nil
 	}
 
+	if resp.Count == 0 {
+		// no data. therefore no effect on the input
+		return nil
+	}
+
 	pair := resp.Kvs[0]
 	any := types.AnyBytes(pair.Value)
 	return any.Decode(&output)

pkg/store/etcd/v3/etcd_test.go

@@ -54,7 +54,9 @@ func testSaveLoad(t *testing.T) {
 		RequestTimeout: 1 * time.Second,
 	}
 
-	snap, err := NewSnapshot(options)
+	etcdClient, err := etcd.NewClient(options)
+	require.NoError(t, err)
+	snap, err := NewSnapshot(etcdClient)
 	require.NoError(t, err)
 
 	defer snap.Close()

pkg/util/docker/docker.go

@@ -26,8 +26,8 @@ type APIClientCloser interface {
 	client.CommonAPIClient
 }
 
-// NewDockerClient creates a new API client.
-func NewDockerClient(host string, tls *tlsconfig.Options) (APIClientCloser, error) {
+// NewClient creates a new API client.
+func NewClient(host string, tls *tlsconfig.Options) (APIClientCloser, error) {
 	tlsOptions := tls
 	if tls.KeyFile == "" || tls.CAFile == "" || tls.CertFile == "" {
 		// The api doesn't like it when you pass in not nil but with zero field values...