vendor/k8s.io/kubernetes/cluster/images/etcd/migrate/rollback_v2.go - cloudstack-kubernetes-provider - Git at Google

 /*
 Copyright 2016 The Kubernetes Authors.

 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 main

 import (
 	"encoding/json"
 	"os"
 	"path"
 	"strconv"
 	"strings"
 	"time"

 	// Uncomment when you want to rollback to 2.2.1 version.
 	oldwal "k8s.io/kubernetes/third_party/forked/etcd221/wal"
 	// Uncomment when you want to rollback to 2.3.7 version.
 	// oldwal "k8s.io/kubernetes/third_party/forked/etcd237/wal"

 	"github.com/coreos/etcd/etcdserver"
 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"github.com/coreos/etcd/etcdserver/membership"
 	"github.com/coreos/etcd/mvcc/backend"
 	"github.com/coreos/etcd/mvcc/mvccpb"
 	"github.com/coreos/etcd/pkg/pbutil"
 	"github.com/coreos/etcd/pkg/types"
 	"github.com/coreos/etcd/raft/raftpb"
 	"github.com/coreos/etcd/snap"
 	"github.com/coreos/etcd/store"
 	"github.com/coreos/etcd/wal"
 	"github.com/coreos/etcd/wal/walpb"
 	"github.com/coreos/go-semver/semver"
 	"k8s.io/klog"
 )

 const rollbackVersion = "2.2.0"

 // RollbackV3ToV2 rolls back an etcd 3.0.x data directory to the 2.x.x version specified by rollbackVersion.
 func RollbackV3ToV2(migrateDatadir string, ttl time.Duration) error {
 	dbpath := path.Join(migrateDatadir, "member", "snap", "db")
 	klog.Infof("Rolling db file %s back to etcd 2.x", dbpath)

 	// etcd3 store backend. We will use it to parse v3 data files and extract information.
 	be := backend.NewDefaultBackend(dbpath)
 	tx := be.BatchTx()

 	// etcd2 store backend. We will use v3 data to update this and then save snapshot to disk.
 	st := store.New(etcdserver.StoreClusterPrefix, etcdserver.StoreKeysPrefix)
 	expireTime := time.Now().Add(ttl)

 	tx.Lock()
 	err := tx.UnsafeForEach([]byte("key"), func(k, v []byte) error {
 		kv := &mvccpb.KeyValue{}
 		kv.Unmarshal(v)

 		// This is compact key.
 		if !strings.HasPrefix(string(kv.Key), "/") {
 			return nil
 		}

 		ttlOpt := store.TTLOptionSet{}
 		if kv.Lease != 0 {
 			ttlOpt = store.TTLOptionSet{ExpireTime: expireTime}
 		}

 		if !isTombstone(k) {
 			sk := path.Join(strings.Trim(etcdserver.StoreKeysPrefix, "/"), string(kv.Key))
 			_, err := st.Set(sk, false, string(kv.Value), ttlOpt)
 			if err != nil {
 				return err
 			}
 		} else {
 			st.Delete(string(kv.Key), false, false)
 		}

 		return nil
 	})
 	if err != nil {
 		return err
 	}
 	tx.Unlock()

 	if err := traverseAndDeleteEmptyDir(st, "/"); err != nil {
 		return err
 	}

 	// rebuild cluster state.
 	metadata, hardstate, oldSt, err := rebuild(migrateDatadir)
 	if err != nil {
 		return err
 	}

 	// In the following, it's low level logic that saves metadata and data into v2 snapshot.
 	backupPath := migrateDatadir + ".rollback.backup"
 	if err := os.Rename(migrateDatadir, backupPath); err != nil {
 		return err
 	}
 	if err := os.MkdirAll(path.Join(migrateDatadir, "member", "snap"), 0777); err != nil {
 		return err
 	}
 	walDir := path.Join(migrateDatadir, "member", "wal")

 	w, err := oldwal.Create(walDir, metadata)
 	if err != nil {
 		return err
 	}
 	err = w.SaveSnapshot(walpb.Snapshot{Index: hardstate.Commit, Term: hardstate.Term})
 	w.Close()
 	if err != nil {
 		return err
 	}

 	event, err := oldSt.Get(etcdserver.StoreClusterPrefix, true, false)
 	if err != nil {
 		return err
 	}
 	// nodes (members info) for ConfState
 	nodes := []uint64{}
 	traverseMetadata(event.Node, func(n *store.NodeExtern) {
 		if n.Key != etcdserver.StoreClusterPrefix {
 			// update store metadata
 			v := ""
 			if !n.Dir {
 				v = *n.Value
 			}
 			if n.Key == path.Join(etcdserver.StoreClusterPrefix, "version") {
 				v = rollbackVersion
 			}
 			if _, err := st.Set(n.Key, n.Dir, v, store.TTLOptionSet{}); err != nil {
 				klog.Error(err)
 			}

 			// update nodes
 			fields := strings.Split(n.Key, "/")
 			if len(fields) == 4 && fields[2] == "members" {
 				nodeID, err := strconv.ParseUint(fields[3], 16, 64)
 				if err != nil {
 					klog.Fatalf("failed to parse member ID (%s): %v", fields[3], err)
 				}
 				nodes = append(nodes, nodeID)
 			}
 		}
 	})

 	data, err := st.Save()
 	if err != nil {
 		return err
 	}
 	raftSnap := raftpb.Snapshot{
 		Data: data,
 		Metadata: raftpb.SnapshotMetadata{
 			Index: hardstate.Commit,
 			Term:  hardstate.Term,
 			ConfState: raftpb.ConfState{
 				Nodes: nodes,
 			},
 		},
 	}
 	snapshotter := snap.New(path.Join(migrateDatadir, "member", "snap"))
 	if err := snapshotter.SaveSnap(raftSnap); err != nil {
 		return err
 	}
 	klog.Infof("Finished successfully")
 	return nil
 }

 func traverseMetadata(head *store.NodeExtern, handleFunc func(*store.NodeExtern)) {
 	q := []*store.NodeExtern{head}

 	for len(q) > 0 {
 		n := q[0]
 		q = q[1:]

 		handleFunc(n)

 		for _, next := range n.Nodes {
 			q = append(q, next)
 		}
 	}
 }

 const (
 	revBytesLen       = 8 + 1 + 8
 	markedRevBytesLen = revBytesLen + 1
 	markBytePosition  = markedRevBytesLen - 1

 	markTombstone byte = 't'
 )

 func isTombstone(b []byte) bool {
 	return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone
 }

 func traverseAndDeleteEmptyDir(st store.Store, dir string) error {
 	e, err := st.Get(dir, true, false)
 	if err != nil {
 		return err
 	}
 	if len(e.Node.Nodes) == 0 {
 		st.Delete(dir, true, true)
 		return nil
 	}
 	for _, node := range e.Node.Nodes {
 		if !node.Dir {
 			klog.V(2).Infof("key: %s", node.Key[len(etcdserver.StoreKeysPrefix):])
 		} else {
 			err := traverseAndDeleteEmptyDir(st, node.Key)
 			if err != nil {
 				return err
 			}
 		}
 	}
 	return nil
 }

 func rebuild(datadir string) ([]byte, *raftpb.HardState, store.Store, error) {
 	waldir := path.Join(datadir, "member", "wal")
 	snapdir := path.Join(datadir, "member", "snap")

 	ss := snap.New(snapdir)
 	snapshot, err := ss.Load()
 	if err != nil && err != snap.ErrNoSnapshot {
 		return nil, nil, nil, err
 	}

 	var walsnap walpb.Snapshot
 	if snapshot != nil {
 		walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
 	}

 	w, err := wal.OpenForRead(waldir, walsnap)
 	if err != nil {
 		return nil, nil, nil, err
 	}
 	defer w.Close()

 	meta, hardstate, ents, err := w.ReadAll()
 	if err != nil {
 		return nil, nil, nil, err
 	}

 	st := store.New(etcdserver.StoreClusterPrefix, etcdserver.StoreKeysPrefix)
 	if snapshot != nil {
 		err := st.Recovery(snapshot.Data)
 		if err != nil {
 			return nil, nil, nil, err
 		}
 	}

 	cluster := membership.NewCluster("")
 	cluster.SetStore(st)
 	cluster.Recover(func(*semver.Version) {})

 	applier := etcdserver.NewApplierV2(st, cluster)
 	for _, ent := range ents {
 		if ent.Type == raftpb.EntryConfChange {
 			var cc raftpb.ConfChange
 			pbutil.MustUnmarshal(&cc, ent.Data)
 			switch cc.Type {
 			case raftpb.ConfChangeAddNode:
 				m := new(membership.Member)
 				if err := json.Unmarshal(cc.Context, m); err != nil {
 					return nil, nil, nil, err
 				}
 				cluster.AddMember(m)
 			case raftpb.ConfChangeRemoveNode:
 				id := types.ID(cc.NodeID)
 				cluster.RemoveMember(id)
 			case raftpb.ConfChangeUpdateNode:
 				m := new(membership.Member)
 				if err := json.Unmarshal(cc.Context, m); err != nil {
 					return nil, nil, nil, err
 				}
 				cluster.UpdateRaftAttributes(m.ID, m.RaftAttributes)
 			}
 			continue
 		}

 		var raftReq pb.InternalRaftRequest
 		if !pbutil.MaybeUnmarshal(&raftReq, ent.Data) { // backward compatible
 			var r pb.Request
 			pbutil.MustUnmarshal(&r, ent.Data)
 			applyRequest(&r, applier)
 		} else {
 			if raftReq.V2 != nil {
 				req := raftReq.V2
 				applyRequest(req, applier)
 			}
 		}
 	}

 	return meta, &hardstate, st, nil
 }

 func toTTLOptions(r *pb.Request) store.TTLOptionSet {
 	refresh, _ := pbutil.GetBool(r.Refresh)
 	ttlOptions := store.TTLOptionSet{Refresh: refresh}
 	if r.Expiration != 0 {
 		ttlOptions.ExpireTime = time.Unix(0, r.Expiration)
 	}
 	return ttlOptions
 }

 func applyRequest(r *pb.Request, applyV2 etcdserver.ApplierV2) {
 	// TODO: find a sane way to perform this cast or avoid it in the first place
 	reqV2 := &etcdserver.RequestV2{
 		ID:               r.ID,
 		Method:           r.Method,
 		Path:             r.Path,
 		Val:              r.Val,
 		Dir:              r.Dir,
 		PrevValue:        r.PrevValue,
 		PrevIndex:        r.PrevIndex,
 		PrevExist:        r.PrevExist,
 		Expiration:       r.Expiration,
 		Wait:             r.Wait,
 		Since:            r.Since,
 		Recursive:        r.Recursive,
 		Sorted:           r.Sorted,
 		Quorum:           r.Quorum,
 		Time:             r.Time,
 		Stream:           r.Stream,
 		Refresh:          r.Refresh,
 		XXX_unrecognized: r.XXX_unrecognized,
 	}
 	toTTLOptions(r)
 	switch r.Method {
 	case "PUT":
 		applyV2.Put(reqV2)
 	case "DELETE":
 		applyV2.Delete(reqV2)
 	case "POST", "QGET", "SYNC":
 		return
 	default:
 		klog.Fatal("unknown command")
 	}
 }
	/*
	Copyright 2016 The Kubernetes Authors.

	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 main

	import (
	"encoding/json"
	"os"
	"path"
	"strconv"
	"strings"
	"time"

	// Uncomment when you want to rollback to 2.2.1 version.
	oldwal "k8s.io/kubernetes/third_party/forked/etcd221/wal"
	// Uncomment when you want to rollback to 2.3.7 version.
	// oldwal "k8s.io/kubernetes/third_party/forked/etcd237/wal"

	"github.com/coreos/etcd/etcdserver"
	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
	"github.com/coreos/etcd/etcdserver/membership"
	"github.com/coreos/etcd/mvcc/backend"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/coreos/etcd/pkg/pbutil"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/coreos/etcd/snap"
	"github.com/coreos/etcd/store"
	"github.com/coreos/etcd/wal"
	"github.com/coreos/etcd/wal/walpb"
	"github.com/coreos/go-semver/semver"
	"k8s.io/klog"
	)

	const rollbackVersion = "2.2.0"

	// RollbackV3ToV2 rolls back an etcd 3.0.x data directory to the 2.x.x version specified by rollbackVersion.
	func RollbackV3ToV2(migrateDatadir string, ttl time.Duration) error {
	dbpath := path.Join(migrateDatadir, "member", "snap", "db")
	klog.Infof("Rolling db file %s back to etcd 2.x", dbpath)

	// etcd3 store backend. We will use it to parse v3 data files and extract information.
	be := backend.NewDefaultBackend(dbpath)
	tx := be.BatchTx()

	// etcd2 store backend. We will use v3 data to update this and then save snapshot to disk.
	st := store.New(etcdserver.StoreClusterPrefix, etcdserver.StoreKeysPrefix)
	expireTime := time.Now().Add(ttl)

	tx.Lock()
	err := tx.UnsafeForEach([]byte("key"), func(k, v []byte) error {
	kv := &mvccpb.KeyValue{}
	kv.Unmarshal(v)

	// This is compact key.
	if !strings.HasPrefix(string(kv.Key), "/") {
	return nil
	}

	ttlOpt := store.TTLOptionSet{}
	if kv.Lease != 0 {
	ttlOpt = store.TTLOptionSet{ExpireTime: expireTime}
	}

	if !isTombstone(k) {
	sk := path.Join(strings.Trim(etcdserver.StoreKeysPrefix, "/"), string(kv.Key))
	_, err := st.Set(sk, false, string(kv.Value), ttlOpt)
	if err != nil {
	return err
	}
	} else {
	st.Delete(string(kv.Key), false, false)
	}

	return nil
	})
	if err != nil {
	return err
	}
	tx.Unlock()

	if err := traverseAndDeleteEmptyDir(st, "/"); err != nil {
	return err
	}

	// rebuild cluster state.
	metadata, hardstate, oldSt, err := rebuild(migrateDatadir)
	if err != nil {
	return err
	}

	// In the following, it's low level logic that saves metadata and data into v2 snapshot.
	backupPath := migrateDatadir + ".rollback.backup"
	if err := os.Rename(migrateDatadir, backupPath); err != nil {
	return err
	}
	if err := os.MkdirAll(path.Join(migrateDatadir, "member", "snap"), 0777); err != nil {
	return err
	}
	walDir := path.Join(migrateDatadir, "member", "wal")

	w, err := oldwal.Create(walDir, metadata)
	if err != nil {
	return err
	}
	err = w.SaveSnapshot(walpb.Snapshot{Index: hardstate.Commit, Term: hardstate.Term})
	w.Close()
	if err != nil {
	return err
	}

	event, err := oldSt.Get(etcdserver.StoreClusterPrefix, true, false)
	if err != nil {
	return err
	}
	// nodes (members info) for ConfState
	nodes := []uint64{}
	traverseMetadata(event.Node, func(n *store.NodeExtern) {
	if n.Key != etcdserver.StoreClusterPrefix {
	// update store metadata
	v := ""
	if !n.Dir {
	v = *n.Value
	}
	if n.Key == path.Join(etcdserver.StoreClusterPrefix, "version") {
	v = rollbackVersion
	}
	if _, err := st.Set(n.Key, n.Dir, v, store.TTLOptionSet{}); err != nil {
	klog.Error(err)
	}

	// update nodes
	fields := strings.Split(n.Key, "/")
	if len(fields) == 4 && fields[2] == "members" {
	nodeID, err := strconv.ParseUint(fields[3], 16, 64)
	if err != nil {
	klog.Fatalf("failed to parse member ID (%s): %v", fields[3], err)
	}
	nodes = append(nodes, nodeID)
	}
	}
	})

	data, err := st.Save()
	if err != nil {
	return err
	}
	raftSnap := raftpb.Snapshot{
	Data: data,
	Metadata: raftpb.SnapshotMetadata{
	Index: hardstate.Commit,
	Term: hardstate.Term,
	ConfState: raftpb.ConfState{
	Nodes: nodes,
	},
	},
	}
	snapshotter := snap.New(path.Join(migrateDatadir, "member", "snap"))
	if err := snapshotter.SaveSnap(raftSnap); err != nil {
	return err
	}
	klog.Infof("Finished successfully")
	return nil
	}

	func traverseMetadata(head store.NodeExtern, handleFunc func(store.NodeExtern)) {
	q := []*store.NodeExtern{head}

	for len(q) > 0 {
	n := q[0]
	q = q[1:]

	handleFunc(n)

	for _, next := range n.Nodes {
	q = append(q, next)
	}
	}
	}

	const (
	revBytesLen = 8 + 1 + 8
	markedRevBytesLen = revBytesLen + 1
	markBytePosition = markedRevBytesLen - 1

	markTombstone byte = 't'
	)

	func isTombstone(b []byte) bool {
	return len(b) == markedRevBytesLen && b[markBytePosition] == markTombstone
	}

	func traverseAndDeleteEmptyDir(st store.Store, dir string) error {
	e, err := st.Get(dir, true, false)
	if err != nil {
	return err
	}
	if len(e.Node.Nodes) == 0 {
	st.Delete(dir, true, true)
	return nil
	}
	for _, node := range e.Node.Nodes {
	if !node.Dir {
	klog.V(2).Infof("key: %s", node.Key[len(etcdserver.StoreKeysPrefix):])
	} else {
	err := traverseAndDeleteEmptyDir(st, node.Key)
	if err != nil {
	return err
	}
	}
	}
	return nil
	}

	func rebuild(datadir string) ([]byte, *raftpb.HardState, store.Store, error) {
	waldir := path.Join(datadir, "member", "wal")
	snapdir := path.Join(datadir, "member", "snap")

	ss := snap.New(snapdir)
	snapshot, err := ss.Load()
	if err != nil && err != snap.ErrNoSnapshot {
	return nil, nil, nil, err
	}

	var walsnap walpb.Snapshot
	if snapshot != nil {
	walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
	}

	w, err := wal.OpenForRead(waldir, walsnap)
	if err != nil {
	return nil, nil, nil, err
	}
	defer w.Close()

	meta, hardstate, ents, err := w.ReadAll()
	if err != nil {
	return nil, nil, nil, err
	}

	st := store.New(etcdserver.StoreClusterPrefix, etcdserver.StoreKeysPrefix)
	if snapshot != nil {
	err := st.Recovery(snapshot.Data)
	if err != nil {
	return nil, nil, nil, err
	}
	}

	cluster := membership.NewCluster("")
	cluster.SetStore(st)
	cluster.Recover(func(*semver.Version) {})

	applier := etcdserver.NewApplierV2(st, cluster)
	for _, ent := range ents {
	if ent.Type == raftpb.EntryConfChange {
	var cc raftpb.ConfChange
	pbutil.MustUnmarshal(&cc, ent.Data)
	switch cc.Type {
	case raftpb.ConfChangeAddNode:
	m := new(membership.Member)
	if err := json.Unmarshal(cc.Context, m); err != nil {
	return nil, nil, nil, err
	}
	cluster.AddMember(m)
	case raftpb.ConfChangeRemoveNode:
	id := types.ID(cc.NodeID)
	cluster.RemoveMember(id)
	case raftpb.ConfChangeUpdateNode:
	m := new(membership.Member)
	if err := json.Unmarshal(cc.Context, m); err != nil {
	return nil, nil, nil, err
	}
	cluster.UpdateRaftAttributes(m.ID, m.RaftAttributes)
	}
	continue
	}

	var raftReq pb.InternalRaftRequest
	if !pbutil.MaybeUnmarshal(&raftReq, ent.Data) { // backward compatible
	var r pb.Request
	pbutil.MustUnmarshal(&r, ent.Data)
	applyRequest(&r, applier)
	} else {
	if raftReq.V2 != nil {
	req := raftReq.V2
	applyRequest(req, applier)
	}
	}
	}

	return meta, &hardstate, st, nil
	}

	func toTTLOptions(r *pb.Request) store.TTLOptionSet {
	refresh, _ := pbutil.GetBool(r.Refresh)
	ttlOptions := store.TTLOptionSet{Refresh: refresh}
	if r.Expiration != 0 {
	ttlOptions.ExpireTime = time.Unix(0, r.Expiration)
	}
	return ttlOptions
	}

	func applyRequest(r *pb.Request, applyV2 etcdserver.ApplierV2) {
	// TODO: find a sane way to perform this cast or avoid it in the first place
	reqV2 := &etcdserver.RequestV2{
	ID: r.ID,
	Method: r.Method,
	Path: r.Path,
	Val: r.Val,
	Dir: r.Dir,
	PrevValue: r.PrevValue,
	PrevIndex: r.PrevIndex,
	PrevExist: r.PrevExist,
	Expiration: r.Expiration,
	Wait: r.Wait,
	Since: r.Since,
	Recursive: r.Recursive,
	Sorted: r.Sorted,
	Quorum: r.Quorum,
	Time: r.Time,
	Stream: r.Stream,
	Refresh: r.Refresh,
	XXX_unrecognized: r.XXX_unrecognized,
	}
	toTTLOptions(r)
	switch r.Method {
	case "PUT":
	applyV2.Put(reqV2)
	case "DELETE":
	applyV2.Delete(reqV2)
	case "POST", "QGET", "SYNC":
	return
	default:
	klog.Fatal("unknown command")
	}
	}