etcd_corrupt_test.go 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. // Copyright 2017 The etcd Authors
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package e2e
  15. import (
  16. "context"
  17. "errors"
  18. "fmt"
  19. "os"
  20. "path/filepath"
  21. "testing"
  22. "time"
  23. "go.etcd.io/etcd/clientv3"
  24. "go.etcd.io/etcd/mvcc/mvccpb"
  25. bolt "go.etcd.io/bbolt"
  26. )
  27. // TODO: test with embedded etcd in integration package
  28. func TestEtcdCorruptHash(t *testing.T) {
  29. // oldenv := os.Getenv("EXPECT_DEBUG")
  30. // defer os.Setenv("EXPECT_DEBUG", oldenv)
  31. // os.Setenv("EXPECT_DEBUG", "1")
  32. cfg := configNoTLS
  33. // trigger snapshot so that restart member can load peers from disk
  34. cfg.snapshotCount = 3
  35. testCtl(t, corruptTest, withQuorum(),
  36. withCfg(cfg),
  37. withInitialCorruptCheck(),
  38. withCorruptFunc(corruptHash),
  39. )
  40. }
  41. func corruptTest(cx ctlCtx) {
  42. for i := 0; i < 10; i++ {
  43. if err := ctlV3Put(cx, fmt.Sprintf("foo%05d", i), fmt.Sprintf("v%05d", i), ""); err != nil {
  44. if cx.dialTimeout > 0 && !isGRPCTimedout(err) {
  45. cx.t.Fatalf("putTest ctlV3Put error (%v)", err)
  46. }
  47. }
  48. }
  49. // enough time for all nodes sync on the same data
  50. time.Sleep(3 * time.Second)
  51. eps := cx.epc.EndpointsV3()
  52. cli1, err := clientv3.New(clientv3.Config{Endpoints: []string{eps[1]}, DialTimeout: 3 * time.Second})
  53. if err != nil {
  54. cx.t.Fatal(err)
  55. }
  56. defer cli1.Close()
  57. sresp, err := cli1.Status(context.TODO(), eps[0])
  58. if err != nil {
  59. cx.t.Fatal(err)
  60. }
  61. id0 := sresp.Header.GetMemberId()
  62. cx.epc.procs[0].Stop()
  63. // corrupt first member by modifying backend offline.
  64. fp := filepath.Join(cx.epc.procs[0].Config().dataDirPath, "member", "snap", "db")
  65. if err = cx.corruptFunc(fp); err != nil {
  66. cx.t.Fatal(err)
  67. }
  68. ep := cx.epc.procs[0]
  69. proc, err := spawnCmd(append([]string{ep.Config().execPath}, ep.Config().args...))
  70. if err != nil {
  71. cx.t.Fatal(err)
  72. }
  73. defer proc.Stop()
  74. // restarting corrupted member should fail
  75. waitReadyExpectProc(proc, []string{fmt.Sprintf("etcdmain: %016x found data inconsistency with peers", id0)})
  76. }
  77. func corruptHash(fpath string) error {
  78. db, derr := bolt.Open(fpath, os.ModePerm, &bolt.Options{})
  79. if derr != nil {
  80. return derr
  81. }
  82. defer db.Close()
  83. return db.Update(func(tx *bolt.Tx) error {
  84. b := tx.Bucket([]byte("key"))
  85. if b == nil {
  86. return errors.New("got nil bucket for 'key'")
  87. }
  88. keys, vals := [][]byte{}, [][]byte{}
  89. c := b.Cursor()
  90. for k, v := c.First(); k != nil; k, v = c.Next() {
  91. keys = append(keys, k)
  92. var kv mvccpb.KeyValue
  93. if uerr := kv.Unmarshal(v); uerr != nil {
  94. return uerr
  95. }
  96. kv.Key[0]++
  97. kv.Value[0]++
  98. v2, v2err := kv.Marshal()
  99. if v2err != nil {
  100. return v2err
  101. }
  102. vals = append(vals, v2)
  103. }
  104. for i := range keys {
  105. if perr := b.Put(keys[i], vals[i]); perr != nil {
  106. return perr
  107. }
  108. }
  109. return nil
  110. })
  111. }