etcdctlv3: add mirrormaker

etcdctlv3/command/make_mirror_command.go

@@ -0,0 +1,124 @@
+// Copyright 2016 CoreOS, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package command
+
+import (
+	"errors"
+
+	"github.com/coreos/etcd/Godeps/_workspace/src/github.com/spf13/cobra"
+	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/clientv3/sync"
+	"github.com/coreos/etcd/etcdserver/api/v3rpc"
+	"github.com/coreos/etcd/storage/storagepb"
+)
+
+var (
+	mmcert   string
+	mmkey    string
+	mmcacert string
+	mmprefix string
+)
+
+// NewMakeMirrorCommand returns the cobra command for "makeMirror".
+func NewMakeMirrorCommand() *cobra.Command {
+	c := &cobra.Command{
+		Use:   "make-mirror [options] [destination]",
+		Short: "make-mirror makes a mirror at the destination etcd cluster",
+		Run:   makeMirrorCommandFunc,
+	}
+
+	c.Flags().StringVar(&mmprefix, "prefix", "", "the key-value prefix to mirror")
+	// TODO: add dest-prefix to mirror a prefix to a different prefix in the destionation cluster?
+	c.Flags().StringVar(&mmcert, "dest-cert", "", "identify secure client using this TLS certificate file for the destination cluster")
+	c.Flags().StringVar(&mmkey, "dest-key", "", "identify secure client using this TLS key file")
+	c.Flags().StringVar(&mmcacert, "dest-cacert", "", "verify certificates of TLS enabled secure servers using this CA bundle")
+
+	return c
+}
+
+func makeMirrorCommandFunc(cmd *cobra.Command, args []string) {
+	if len(args) != 1 {
+		ExitWithError(ExitBadArgs, errors.New("make-mirror takes one destination arguement."))
+	}
+
+	dc := mustClient(args[0], mmcert, mmkey, mmcacert)
+	c := mustClientFromCmd(cmd)
+
+	err := makeMirror(context.TODO(), c, dc)
+	ExitWithError(ExitError, err)
+}
+
+func makeMirror(ctx context.Context, c *clientv3.Client, dc *clientv3.Client) error {
+	// TODO: remove the prefix of the destination cluster?
+	dkv := clientv3.NewKV(dc)
+
+	s := sync.NewSyncer(c, mmprefix, 0)
+
+	rc, errc := s.SyncBase(ctx)
+
+	for r := range rc {
+		for _, kv := range r.Kvs {
+			_, err := dkv.Put(ctx, string(kv.Key), string(kv.Value))
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	err := <-errc
+	if err != nil {
+		return err
+	}
+
+	wc := s.SyncUpdates(ctx)
+
+	for wr := range wc {
+		if wr.CompactRevision != 0 {
+			return v3rpc.ErrCompacted
+		}
+
+		var rev int64
+		ops := []clientv3.Op{}
+
+		for _, ev := range wr.Events {
+			nrev := ev.Kv.ModRevision
+			if rev != 0 && nrev > rev {
+				_, err := dkv.Txn(ctx).Then(ops...).Commit()
+				if err != nil {
+					return err
+				}
+				ops = []clientv3.Op{}
+			}
+			switch ev.Type {
+			case storagepb.PUT:
+				ops = append(ops, clientv3.OpPut(string(ev.Kv.Key), string(ev.Kv.Value)))
+			case storagepb.DELETE, storagepb.EXPIRE:
+				ops = append(ops, clientv3.OpDelete(string(ev.Kv.Key)))
+			default:
+				panic("unexpected event type")
+			}
+		}
+
+		if len(ops) != 0 {
+			_, err := dkv.Txn(ctx).Then(ops...).Commit()
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}

etcdctlv3/main.go

@@ -58,6 +58,7 @@ func init() {
 		command.NewLeaseCommand(),
 		command.NewMemberCommand(),
 		command.NewSnapshotCommand(),
+		command.NewMakeMirrorCommand(),
 	)
 }