e2e: add corruption checking tests

Signed-off-by: Gyu-Ho Lee <gyuhox@gmail.com>

e2e/cluster_test.go

@@ -20,6 +20,7 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/coreos/etcd/etcdserver"
 )
@@ -114,6 +115,7 @@ type etcdProcessClusterConfig struct {
 
 	forceNewCluster   bool
 	initialToken      string
+	corruptCheckTime  time.Duration
 	quotaBackendBytes int64
 	noStrictReconfig  bool
 }
@@ -221,6 +223,11 @@ func (cfg *etcdProcessClusterConfig) etcdServerProcessConfigs() []*etcdServerPro
 				"--quota-backend-bytes", fmt.Sprintf("%d", cfg.quotaBackendBytes),
 			)
 		}
+		if cfg.corruptCheckTime > 0 {
+			args = append(args,
+				"--experimental-corrupt-check-time", cfg.corruptCheckTime.String(),
+			)
+		}
 		if cfg.noStrictReconfig {
 			args = append(args, "--strict-reconfig-check=false")
 		}

e2e/ctl_v3_test.go

@@ -55,6 +55,8 @@ type ctlCtx struct {
 	t                 *testing.T
 	cfg               etcdProcessClusterConfig
 	quotaBackendBytes int64
+	corruptCheckTime  time.Duration
+	corruptFunc       func(string, int64) error
 	noStrictReconfig  bool
 
 	epc *etcdProcessCluster
@@ -105,6 +107,14 @@ func withCompactPhysical() ctlOption {
 	return func(cx *ctlCtx) { cx.compactPhysical = true }
 }
 
+func withCorruptCheckTime(d time.Duration) ctlOption {
+	return func(cx *ctlCtx) { cx.corruptCheckTime = d }
+}
+
+func withCorruptFunc(f func(string, int64) error) ctlOption {
+	return func(cx *ctlCtx) { cx.corruptFunc = f }
+}
+
 func withNoStrictReconfig() ctlOption {
 	return func(cx *ctlCtx) { cx.noStrictReconfig = true }
 }
@@ -130,6 +140,9 @@ func testCtl(t *testing.T, testFunc func(ctlCtx), opts ...ctlOption) {
 	if ret.quotaBackendBytes > 0 {
 		ret.cfg.quotaBackendBytes = ret.quotaBackendBytes
 	}
+	if ret.corruptCheckTime > 0 {
+		ret.cfg.corruptCheckTime = ret.corruptCheckTime
+	}
 	ret.cfg.noStrictReconfig = ret.noStrictReconfig
 
 	epc, err := newEtcdProcessCluster(&ret.cfg)

e2e/etcd_corrupt_test.go

@@ -0,0 +1,154 @@
+// Copyright 2017 The etcd 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 e2e
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/coreos/etcd/clientv3"
+	"github.com/coreos/etcd/mvcc"
+	"github.com/coreos/etcd/mvcc/backend"
+	"github.com/coreos/etcd/mvcc/mvccpb"
+
+	bolt "github.com/coreos/bbolt"
+)
+
+// TODO: test this in integration tests when integration package uses embedded etcd
+
+func TestEtcdCorruptHash(t *testing.T) {
+	testCtl(t, corruptTest, withCfg(configNoTLS), withQuorum(), withCorruptCheckTime(time.Minute),
+		withCorruptFunc(corruptHash),
+	)
+}
+
+func TestEtcdCorruptFutureRevision(t *testing.T) {
+	testCtl(t, corruptTest, withCfg(configNoTLS), withQuorum(), withCorruptCheckTime(time.Minute),
+		withCorruptFunc(corruptFutureRevision),
+	)
+}
+
+func TestEtcdCorruptCompact(t *testing.T) {
+	testCtl(t, corruptTest, withCfg(configNoTLS), withQuorum(), withCorruptCheckTime(time.Minute),
+		withCorruptFunc(corruptCompact),
+	)
+}
+
+func corruptTest(cx ctlCtx) {
+	for i := 0; i < 10; i++ {
+		if err := ctlV3Put(cx, "foo", fmt.Sprintf("v%05d", i), ""); err != nil {
+			if cx.dialTimeout > 0 && !isGRPCTimedout(err) {
+				cx.t.Fatalf("putTest ctlV3Put error (%v)", err)
+			}
+		}
+	}
+
+	eps := cx.epc.EndpointsV3()
+	cli1, err := clientv3.New(clientv3.Config{Endpoints: []string{eps[1]}, DialTimeout: 3 * time.Second})
+	if err != nil {
+		cx.t.Fatal(err)
+	}
+	defer cli1.Close()
+
+	sresp, err := cli1.Status(context.TODO(), eps[0])
+	if err != nil {
+		cx.t.Fatal(err)
+	}
+	id0 := sresp.Header.GetMemberId()
+
+	cx.epc.procs[0].Stop()
+
+	// Corrupt member 0 by modifying backend offline.
+	fp := filepath.Join(cx.epc.procs[0].Config().dataDirPath, "member", "snap", "db")
+	if err = cx.corruptFunc(fp, sresp.Header.Revision); err != nil {
+		cx.t.Fatal(err)
+	}
+
+	ep := cx.epc.procs[0]
+	proc, err := spawnCmd(append([]string{ep.Config().execPath}, ep.Config().args...))
+	if err != nil {
+		cx.t.Fatal(err)
+	}
+	defer proc.Stop()
+
+	// restarting corrupted member should fail
+	waitReadyExpectProc(proc, []string{fmt.Sprintf("etcdserver: %016x is corrupt", id0)})
+}
+
+func corruptHash(fpath string, rev int64) error {
+	db, derr := bolt.Open(fpath, os.ModePerm, &bolt.Options{})
+	if derr != nil {
+		return derr
+	}
+	defer db.Close()
+
+	return db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket([]byte("key"))
+		if b == nil {
+			return errors.New("got nil bucket for 'key'")
+		}
+		keys, vals := [][]byte{}, [][]byte{}
+		c := b.Cursor()
+		for k, v := c.First(); k != nil; k, v = c.Next() {
+			keys = append(keys, k)
+			var kv mvccpb.KeyValue
+			if uerr := kv.Unmarshal(v); uerr != nil {
+				return uerr
+			}
+			kv.Value[len(kv.Value)/2]++
+			v2, v2err := kv.Marshal()
+			if v2err != nil {
+				return v2err
+			}
+			vals = append(vals, v2)
+		}
+		for i := range keys {
+			if perr := b.Put(keys[i], vals[i]); perr != nil {
+				return perr
+			}
+		}
+		return nil
+	})
+}
+
+func corruptFutureRevision(fpath string, rev int64) error {
+	be := backend.NewDefaultBackend(fpath)
+	s := mvcc.NewStore(be, nil, &fakeConsistentIndex{uint64(rev + 2)})
+	s.Put([]byte("abc"), []byte("def"), 0)
+	s.Put([]byte("xyz"), []byte("123"), 0)
+	s.Commit()
+	s.Close()
+	be.Close()
+	return nil
+}
+
+func corruptCompact(fpath string, rev int64) error {
+	be := backend.NewDefaultBackend(fpath)
+	s := mvcc.NewStore(be, nil, &fakeConsistentIndex{uint64(rev)})
+	s.Compact(rev / 2)
+	s.Commit()
+	s.Close()
+	be.Close()
+	return nil
+}
+
+type fakeConsistentIndex struct{ rev uint64 }
+
+func (f *fakeConsistentIndex) ConsistentIndex() uint64 { return f.rev }