Etcd数据恢复机制

func bootstrap(cfg config.ServerConfig) (b *bootstrappedServer, err error) {
   if cfg.MaxRequestBytes > recommendedMaxRequestBytes {
      cfg.Logger.Warn(
         "exceeded recommended request limit",
         zap.Uint("max-request-bytes", cfg.MaxRequestBytes),
         zap.String("max-request-size", humanize.Bytes(uint64(cfg.MaxRequestBytes))),
         zap.Int("recommended-request-bytes", recommendedMaxRequestBytes),
         zap.String("recommended-request-size", recommendedMaxRequestBytesString),
      )
   }

   if terr := fileutil.TouchDirAll(cfg.Logger, cfg.DataDir); terr != nil {
      return nil, fmt.Errorf("cannot access data directory: %v", terr)
   }

   if terr := fileutil.TouchDirAll(cfg.Logger, cfg.MemberDir()); terr != nil {
      return nil, fmt.Errorf("cannot access member directory: %v", terr)
   }
   ss := bootstrapSnapshot(cfg)
   prt, err := rafthttp.NewRoundTripper(cfg.PeerTLSInfo, cfg.PeerDialTimeout())
   if err != nil {
      return nil, err
   }

   haveWAL := wal.Exist(cfg.WALDir())
   st := v2store.New(StoreClusterPrefix, StoreKeysPrefix)
   backend, err := bootstrapBackend(cfg, haveWAL, st, ss)
   if err != nil {
      return nil, err
   }
   var bwal *bootstrappedWAL

   if haveWAL {
      if err = fileutil.IsDirWriteable(cfg.WALDir()); err != nil {
         return nil, fmt.Errorf("cannot write to WAL directory: %v", err)
      }
      bwal = bootstrapWALFromSnapshot(cfg, backend.snapshot)
   }

   cluster, err := bootstrapCluster(cfg, bwal, prt)
   if err != nil {
      backend.Close()
      return nil, err
   }

   s, err := bootstrapStorage(cfg, st, backend, bwal, cluster)
   if err != nil {
      backend.Close()
      return nil, err
   }

   if err = cluster.Finalize(cfg, s); err != nil {
      backend.Close()
      return nil, err
   }
   raft := bootstrapRaft(cfg, cluster, s.wal)
   return &bootstrappedServer{
      prt: prt,
      ss: ss,
      storage: s,
      cluster: cluster,
      raft: raft,
   }, nil
}

func recoverSnapshot(cfg config.ServerConfig, st v2store.Store, be backend.Backend, beExist bool, beHooks *serverstorage.BackendHooks, ci cindex.ConsistentIndexer, ss *snap.Snapshotter) (*raftpb.Snapshot, backend.Backend, error) {
   // Find a snapshot to start/restart a raft node
   walSnaps, err := wal.ValidSnapshotEntries(cfg.Logger, cfg.WALDir())
   if err != nil {
      return nil, be, err
   }
   // snapshot files can be orphaned if etcd crashes after writing them but before writing the corresponding
   // bwal log entries
   snapshot, err := ss.LoadNewestAvailable(walSnaps)
   if err != nil && !errors.Is(err, snap.ErrNoSnapshot) {
      return nil, be, err
   }

   if snapshot != nil {
      if err = st.Recovery(snapshot.Data); err != nil {
         cfg.Logger.Panic("failed to recover from snapshot", zap.Error(err))
      }

      if err = serverstorage.AssertNoV2StoreContent(cfg.Logger, st, cfg.V2Deprecation); err != nil {
         cfg.Logger.Error("illegal v2store content", zap.Error(err))
         return nil, be, err
      }

      cfg.Logger.Info(
         "recovered v2 store from snapshot",
         zap.Uint64("snapshot-index", snapshot.Metadata.Index),
         zap.String("snapshot-size", humanize.Bytes(uint64(snapshot.Size()))),
      )

      if be, err = serverstorage.RecoverSnapshotBackend(cfg, be, *snapshot, beExist, beHooks); err != nil {
         cfg.Logger.Panic("failed to recover v3 backend from snapshot", zap.Error(err))
      }
      // A snapshot db may have already been recovered, and the old db should have
      // already been closed in this case, so we should set the backend again.
      ci.SetBackend(be)

      s1, s2 := be.Size(), be.SizeInUse()
      cfg.Logger.Info(
         "recovered v3 backend from snapshot",
         zap.Int64("backend-size-bytes", s1),
         zap.String("backend-size", humanize.Bytes(uint64(s1))),
         zap.Int64("backend-size-in-use-bytes", s2),
         zap.String("backend-size-in-use", humanize.Bytes(uint64(s2))),
      )
      if beExist {
         // TODO: remove kvindex != 0 checking when we do not expect users to upgrade
         // etcd from pre-3.0 release.
         kvindex := ci.ConsistentIndex()
         if kvindex < snapshot.Metadata.Index {
            if kvindex != 0 {
               return nil, be, fmt.Errorf("database file (%v index %d) does not match with snapshot (index %d)", cfg.BackendPath(), kvindex, snapshot.Metadata.Index)
            }
            cfg.Logger.Warn(
               "consistent index was never saved",
               zap.Uint64("snapshot-index", snapshot.Metadata.Index),
            )
         }
      }
   } else {
      cfg.Logger.Info("No snapshot found. Recovering WAL from scratch!")
   }
   return snapshot, be, nil
}

// openWALFromSnapshot reads the WAL at the given snap and returns the wal, its latest HardState and cluster ID, and all entries that appear
// after the position of the given snap in the WAL.
// The snap must have been previously saved to the WAL, or this call will panic.
func openWALFromSnapshot(cfg config.ServerConfig, snapshot *raftpb.Snapshot) (*wal.WAL, *raftpb.HardState, []raftpb.Entry, *raftpb.Snapshot, *snapshotMetadata) {
   var walsnap walpb.Snapshot
   if snapshot != nil {
      walsnap.Index, walsnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
   }
   repaired := false
   for {
      w, err := wal.Open(cfg.Logger, cfg.WALDir(), walsnap)
      if err != nil {
         cfg.Logger.Fatal("failed to open WAL", zap.Error(err))
      }
      if cfg.UnsafeNoFsync {
         w.SetUnsafeNoFsync()
      }
      wmetadata, st, ents, err := w.ReadAll()
      if err != nil {
         w.Close()
         // we can only repair ErrUnexpectedEOF and we never repair twice.
         if repaired || !errors.Is(err, io.ErrUnexpectedEOF) {
            cfg.Logger.Fatal("failed to read WAL, cannot be repaired", zap.Error(err))
         }
         if !wal.Repair(cfg.Logger, cfg.WALDir()) {
            cfg.Logger.Fatal("failed to repair WAL", zap.Error(err))
         } else {
            cfg.Logger.Info("repaired WAL", zap.Error(err))
            repaired = true
         }
         continue
      }
      var metadata etcdserverpb.Metadata
      pbutil.MustUnmarshal(&metadata, wmetadata)
      id := types.ID(metadata.NodeID)
      cid := types.ID(metadata.ClusterID)
      meta := &snapshotMetadata{clusterID: cid, nodeID: id}
      return w, &st, ents, snapshot, meta
   }
}