2015-01-25 06:19:16 +03:00
|
|
|
// Copyright 2015 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.
|
2014-10-18 02:41:22 +04:00
|
|
|
|
2014-08-26 05:39:02 +04:00
|
|
|
package etcdserver
|
|
|
|
|
|
|
|
import (
|
2014-09-27 01:18:59 +04:00
|
|
|
"encoding/json"
|
2015-04-04 00:26:15 +03:00
|
|
|
"expvar"
|
2014-10-25 02:26:05 +04:00
|
|
|
"fmt"
|
2014-09-16 04:35:02 +04:00
|
|
|
"log"
|
2014-09-18 01:38:10 +04:00
|
|
|
"math/rand"
|
2014-10-24 03:41:58 +04:00
|
|
|
"net/http"
|
2014-10-21 22:47:03 +04:00
|
|
|
"path"
|
2014-10-24 12:13:47 +04:00
|
|
|
"regexp"
|
2014-09-25 05:05:33 +04:00
|
|
|
"sync/atomic"
|
2014-08-27 03:17:02 +04:00
|
|
|
"time"
|
2014-08-27 02:42:10 +04:00
|
|
|
|
2015-01-29 03:42:31 +03:00
|
|
|
"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
|
2014-10-04 16:20:45 +04:00
|
|
|
"github.com/coreos/etcd/discovery"
|
2014-11-17 03:23:47 +03:00
|
|
|
"github.com/coreos/etcd/etcdserver/etcdhttp/httptypes"
|
2014-09-03 08:36:14 +04:00
|
|
|
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
|
2014-10-11 02:53:28 +04:00
|
|
|
"github.com/coreos/etcd/etcdserver/stats"
|
2014-11-29 08:32:28 +03:00
|
|
|
"github.com/coreos/etcd/pkg/fileutil"
|
2015-01-13 22:19:27 +03:00
|
|
|
"github.com/coreos/etcd/pkg/idutil"
|
2014-10-11 16:41:00 +04:00
|
|
|
"github.com/coreos/etcd/pkg/pbutil"
|
2015-04-04 00:26:15 +03:00
|
|
|
"github.com/coreos/etcd/pkg/runtime"
|
2014-12-29 22:15:02 +03:00
|
|
|
"github.com/coreos/etcd/pkg/timeutil"
|
2014-10-31 03:31:14 +03:00
|
|
|
"github.com/coreos/etcd/pkg/types"
|
2014-11-13 20:11:53 +03:00
|
|
|
"github.com/coreos/etcd/pkg/wait"
|
2014-08-26 05:39:02 +04:00
|
|
|
"github.com/coreos/etcd/raft"
|
2014-08-28 05:53:18 +04:00
|
|
|
"github.com/coreos/etcd/raft/raftpb"
|
2014-11-19 00:53:14 +03:00
|
|
|
"github.com/coreos/etcd/rafthttp"
|
2014-10-03 03:20:14 +04:00
|
|
|
"github.com/coreos/etcd/snap"
|
2014-08-27 02:28:21 +04:00
|
|
|
"github.com/coreos/etcd/store"
|
2015-03-25 01:13:34 +03:00
|
|
|
"github.com/coreos/etcd/version"
|
2014-10-03 03:20:14 +04:00
|
|
|
"github.com/coreos/etcd/wal"
|
2014-08-26 05:39:02 +04:00
|
|
|
)
|
|
|
|
|
2014-09-17 05:18:45 +04:00
|
|
|
const (
|
2014-10-03 21:38:35 +04:00
|
|
|
// owner can make/remove files inside the directory
|
|
|
|
privateDirMode = 0700
|
|
|
|
|
2014-09-17 05:18:45 +04:00
|
|
|
defaultSyncTimeout = time.Second
|
|
|
|
DefaultSnapCount = 10000
|
2014-10-03 21:38:35 +04:00
|
|
|
// TODO: calculate based on heartbeat interval
|
2014-09-29 22:52:36 +04:00
|
|
|
defaultPublishRetryInterval = 5 * time.Second
|
2014-10-15 11:19:27 +04:00
|
|
|
|
2014-10-21 22:47:03 +04:00
|
|
|
StoreAdminPrefix = "/0"
|
|
|
|
StoreKeysPrefix = "/1"
|
2014-11-29 08:32:28 +03:00
|
|
|
|
|
|
|
purgeFileInterval = 30 * time.Second
|
2014-09-17 05:18:45 +04:00
|
|
|
)
|
2014-09-16 04:35:02 +04:00
|
|
|
|
2014-08-29 03:41:42 +04:00
|
|
|
var (
|
2014-10-21 22:04:42 +04:00
|
|
|
storeMembersPrefix = path.Join(StoreAdminPrefix, "members")
|
|
|
|
storeRemovedMembersPrefix = path.Join(StoreAdminPrefix, "removed_members")
|
2014-10-24 12:13:47 +04:00
|
|
|
|
|
|
|
storeMemberAttributeRegexp = regexp.MustCompile(path.Join(storeMembersPrefix, "[[:xdigit:]]{1,16}", attributesSuffix))
|
2014-08-29 03:41:42 +04:00
|
|
|
)
|
2014-08-27 02:42:10 +04:00
|
|
|
|
2014-09-18 01:38:10 +04:00
|
|
|
func init() {
|
|
|
|
rand.Seed(time.Now().UnixNano())
|
2015-04-04 00:26:15 +03:00
|
|
|
|
|
|
|
expvar.Publish(
|
|
|
|
"file_descriptor_limit",
|
|
|
|
expvar.Func(
|
|
|
|
func() interface{} {
|
|
|
|
n, _ := runtime.FDLimit()
|
|
|
|
return n
|
|
|
|
},
|
|
|
|
),
|
|
|
|
)
|
2014-09-18 01:38:10 +04:00
|
|
|
}
|
|
|
|
|
2014-08-26 05:39:02 +04:00
|
|
|
type Response struct {
|
2014-08-31 01:19:03 +04:00
|
|
|
Event *store.Event
|
2014-09-09 03:56:10 +04:00
|
|
|
Watcher store.Watcher
|
2014-08-31 02:27:43 +04:00
|
|
|
err error
|
2014-08-26 05:39:02 +04:00
|
|
|
}
|
|
|
|
|
2014-09-12 04:14:41 +04:00
|
|
|
type Server interface {
|
|
|
|
// Start performs any initialization of the Server necessary for it to
|
|
|
|
// begin serving requests. It must be called before Do or Process.
|
|
|
|
// Start must be non-blocking; any long-running server functionality
|
|
|
|
// should be implemented in goroutines.
|
2014-09-30 04:14:22 +04:00
|
|
|
Start()
|
2014-09-12 04:14:41 +04:00
|
|
|
// Stop terminates the Server and performs any necessary finalization.
|
|
|
|
// Do and Process cannot be called after Stop has been invoked.
|
|
|
|
Stop()
|
2014-11-13 22:26:49 +03:00
|
|
|
// ID returns the ID of the Server.
|
|
|
|
ID() types.ID
|
2015-01-06 07:20:11 +03:00
|
|
|
// Leader returns the ID of the leader Server.
|
|
|
|
Leader() types.ID
|
2014-11-11 22:46:49 +03:00
|
|
|
// Do takes a request and attempts to fulfill it, returning a Response.
|
2014-09-12 04:14:41 +04:00
|
|
|
Do(ctx context.Context, r pb.Request) (Response, error)
|
|
|
|
// Process takes a raft message and applies it to the server's raft state
|
|
|
|
// machine, respecting any timeout of the given context.
|
|
|
|
Process(ctx context.Context, m raftpb.Message) error
|
2014-10-16 20:49:26 +04:00
|
|
|
// AddMember attempts to add a member into the cluster. It will return
|
|
|
|
// ErrIDRemoved if member ID is removed from the cluster, or return
|
|
|
|
// ErrIDExists if member ID exists in the cluster.
|
2014-10-14 08:06:03 +04:00
|
|
|
AddMember(ctx context.Context, memb Member) error
|
2014-10-16 20:49:26 +04:00
|
|
|
// RemoveMember attempts to remove a member from the cluster. It will
|
|
|
|
// return ErrIDRemoved if member ID is removed from the cluster, or return
|
|
|
|
// ErrIDNotFound if member ID is not in the cluster.
|
2014-10-14 08:06:03 +04:00
|
|
|
RemoveMember(ctx context.Context, id uint64) error
|
2014-11-11 22:46:49 +03:00
|
|
|
|
|
|
|
// UpdateMember attempts to update a existing member in the cluster. It will
|
|
|
|
// return ErrIDNotFound if the member ID does not exist.
|
|
|
|
UpdateMember(ctx context.Context, updateMemb Member) error
|
2014-09-12 04:14:41 +04:00
|
|
|
}
|
|
|
|
|
2014-10-11 17:03:03 +04:00
|
|
|
// EtcdServer is the production implementation of the Server interface
|
|
|
|
type EtcdServer struct {
|
2015-03-06 00:03:04 +03:00
|
|
|
cfg *ServerConfig
|
|
|
|
snapCount uint64
|
2015-01-16 01:39:30 +03:00
|
|
|
|
|
|
|
r raftNode
|
|
|
|
|
2014-10-11 17:03:03 +04:00
|
|
|
w wait.Wait
|
2014-11-14 00:47:12 +03:00
|
|
|
stop chan struct{}
|
2015-01-03 07:00:29 +03:00
|
|
|
done chan struct{}
|
|
|
|
errorc chan error
|
2014-10-31 03:31:14 +03:00
|
|
|
id types.ID
|
2014-10-11 17:03:03 +04:00
|
|
|
attributes Attributes
|
|
|
|
|
2014-10-23 03:15:37 +04:00
|
|
|
Cluster *Cluster
|
2014-10-11 17:03:03 +04:00
|
|
|
|
2014-12-26 21:56:56 +03:00
|
|
|
store store.Store
|
2014-10-11 17:03:03 +04:00
|
|
|
|
2014-10-11 02:53:28 +04:00
|
|
|
stats *stats.ServerStats
|
|
|
|
lstats *stats.LeaderStats
|
|
|
|
|
2014-10-17 00:35:59 +04:00
|
|
|
SyncTicker <-chan time.Time
|
2014-10-11 17:03:03 +04:00
|
|
|
|
2014-12-16 03:07:47 +03:00
|
|
|
reqIDGen *idutil.Generator
|
2014-10-11 17:03:03 +04:00
|
|
|
}
|
|
|
|
|
2014-10-03 03:20:14 +04:00
|
|
|
// NewServer creates a new EtcdServer from the supplied configuration. The
|
|
|
|
// configuration is considered static for the lifetime of the EtcdServer.
|
2014-11-05 00:09:24 +03:00
|
|
|
func NewServer(cfg *ServerConfig) (*EtcdServer, error) {
|
2015-02-05 03:33:40 +03:00
|
|
|
st := store.New(StoreAdminPrefix, StoreKeysPrefix)
|
2014-10-03 03:20:14 +04:00
|
|
|
var w *wal.WAL
|
|
|
|
var n raft.Node
|
2014-11-03 23:04:24 +03:00
|
|
|
var s *raft.MemoryStorage
|
2014-10-31 03:31:14 +03:00
|
|
|
var id types.ID
|
2015-02-12 02:24:32 +03:00
|
|
|
|
2015-03-25 01:13:34 +03:00
|
|
|
// Run the migrations.
|
|
|
|
dataVer, err := version.DetectDataDir(cfg.DataDir)
|
2014-12-11 21:37:32 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2015-03-25 01:13:34 +03:00
|
|
|
if err := upgradeDataDir(cfg.DataDir, cfg.Name, dataVer); err != nil {
|
|
|
|
return nil, err
|
2014-11-18 21:43:20 +03:00
|
|
|
}
|
2015-03-25 01:13:34 +03:00
|
|
|
|
|
|
|
haveWAL := wal.Exist(cfg.WALDir())
|
2015-02-13 20:51:48 +03:00
|
|
|
ss := snap.New(cfg.SnapDir())
|
2014-11-18 21:43:20 +03:00
|
|
|
|
2014-10-24 03:41:58 +04:00
|
|
|
switch {
|
2014-11-06 22:09:33 +03:00
|
|
|
case !haveWAL && !cfg.NewCluster:
|
2015-03-23 21:56:34 +03:00
|
|
|
if err := cfg.VerifyJoinExisting(); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2015-02-14 06:05:29 +03:00
|
|
|
existingCluster, err := GetClusterFromRemotePeers(getRemotePeerURLs(cfg.Cluster, cfg.Name), cfg.Transport)
|
2014-10-25 02:26:05 +04:00
|
|
|
if err != nil {
|
2014-11-05 23:14:42 +03:00
|
|
|
return nil, fmt.Errorf("cannot fetch cluster info from peer urls: %v", err)
|
2014-10-25 02:26:05 +04:00
|
|
|
}
|
2014-11-18 01:08:45 +03:00
|
|
|
if err := ValidateClusterAndAssignIDs(cfg.Cluster, existingCluster); err != nil {
|
|
|
|
return nil, fmt.Errorf("error validating peerURLs %s: %v", existingCluster, err)
|
2014-10-24 03:41:58 +04:00
|
|
|
}
|
2015-02-21 00:51:38 +03:00
|
|
|
cfg.Cluster.UpdateIndex(existingCluster.index)
|
2014-11-18 01:08:45 +03:00
|
|
|
cfg.Cluster.SetID(existingCluster.id)
|
2014-10-24 03:41:58 +04:00
|
|
|
cfg.Cluster.SetStore(st)
|
2014-11-13 07:11:04 +03:00
|
|
|
cfg.Print()
|
2014-11-03 23:04:24 +03:00
|
|
|
id, n, s, w = startNode(cfg, nil)
|
2014-11-06 22:09:33 +03:00
|
|
|
case !haveWAL && cfg.NewCluster:
|
2015-03-23 21:56:34 +03:00
|
|
|
if err := cfg.VerifyBootstrap(); err != nil {
|
2014-11-05 00:09:24 +03:00
|
|
|
return nil, err
|
2014-10-11 17:03:03 +04:00
|
|
|
}
|
2014-10-23 22:08:42 +04:00
|
|
|
m := cfg.Cluster.MemberByName(cfg.Name)
|
2015-02-12 01:18:10 +03:00
|
|
|
if isMemberBootstrapped(cfg.Cluster, cfg.Name, cfg.Transport) {
|
2014-12-26 08:37:20 +03:00
|
|
|
return nil, fmt.Errorf("member %s has already been bootstrapped", m.ID)
|
|
|
|
}
|
2014-10-11 17:47:44 +04:00
|
|
|
if cfg.ShouldDiscover() {
|
2015-02-18 00:15:48 +03:00
|
|
|
str, err := discovery.JoinCluster(cfg.DiscoveryURL, cfg.DiscoveryProxy, m.ID, cfg.Cluster.String())
|
2014-10-04 16:20:45 +04:00
|
|
|
if err != nil {
|
2014-11-05 00:09:24 +03:00
|
|
|
return nil, err
|
2014-10-04 16:20:45 +04:00
|
|
|
}
|
2015-02-18 00:15:48 +03:00
|
|
|
if cfg.Cluster, err = NewClusterFromString(cfg.Cluster.token, str); err != nil {
|
2014-11-05 00:09:24 +03:00
|
|
|
return nil, err
|
2014-10-04 16:20:45 +04:00
|
|
|
}
|
2015-02-16 20:53:01 +03:00
|
|
|
if err := cfg.Cluster.Validate(); err != nil {
|
2015-02-14 01:32:24 +03:00
|
|
|
return nil, fmt.Errorf("bad discovery cluster: %v", err)
|
|
|
|
}
|
2014-10-03 04:06:35 +04:00
|
|
|
}
|
2014-10-23 03:15:37 +04:00
|
|
|
cfg.Cluster.SetStore(st)
|
2014-11-13 07:11:04 +03:00
|
|
|
cfg.PrintWithInitial()
|
2014-11-03 23:04:24 +03:00
|
|
|
id, n, s, w = startNode(cfg, cfg.Cluster.MemberIDs())
|
2014-10-24 03:41:58 +04:00
|
|
|
case haveWAL:
|
2015-02-12 02:24:32 +03:00
|
|
|
if err := fileutil.IsDirWriteable(cfg.DataDir); err != nil {
|
|
|
|
return nil, fmt.Errorf("cannot write to data directory: %v", err)
|
2014-12-25 09:02:46 +03:00
|
|
|
}
|
|
|
|
|
2015-02-12 20:45:12 +03:00
|
|
|
if err := fileutil.IsDirWriteable(cfg.MemberDir()); err != nil {
|
|
|
|
return nil, fmt.Errorf("cannot write to member directory: %v", err)
|
|
|
|
}
|
|
|
|
|
2014-10-11 17:47:44 +04:00
|
|
|
if cfg.ShouldDiscover() {
|
2014-12-10 02:55:42 +03:00
|
|
|
log.Printf("etcdserver: discovery token ignored since a cluster has already been initialized. Valid log found at %q", cfg.WALDir())
|
2014-10-09 03:09:12 +04:00
|
|
|
}
|
2014-10-03 03:20:14 +04:00
|
|
|
snapshot, err := ss.Load()
|
|
|
|
if err != nil && err != snap.ErrNoSnapshot {
|
2014-11-05 00:09:24 +03:00
|
|
|
return nil, err
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
|
|
|
if snapshot != nil {
|
2014-12-09 01:53:39 +03:00
|
|
|
if err := st.Recovery(snapshot.Data); err != nil {
|
|
|
|
log.Panicf("etcdserver: recovered store from snapshot error: %v", err)
|
|
|
|
}
|
2014-12-09 00:54:45 +03:00
|
|
|
log.Printf("etcdserver: recovered store from snapshot at index %d", snapshot.Metadata.Index)
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
2014-10-30 23:43:38 +03:00
|
|
|
cfg.Cluster = NewClusterFromStore(cfg.Cluster.token, st)
|
2014-11-13 07:11:04 +03:00
|
|
|
cfg.Print()
|
2014-11-08 00:34:43 +03:00
|
|
|
if snapshot != nil {
|
2014-12-09 00:54:45 +03:00
|
|
|
log.Printf("etcdserver: loaded cluster information from store: %s", cfg.Cluster)
|
2014-11-08 00:34:43 +03:00
|
|
|
}
|
2014-11-07 08:50:35 +03:00
|
|
|
if !cfg.ForceNewCluster {
|
2015-01-06 10:27:03 +03:00
|
|
|
id, n, s, w = restartNode(cfg, snapshot)
|
2014-11-07 08:50:35 +03:00
|
|
|
} else {
|
2015-01-06 10:27:03 +03:00
|
|
|
id, n, s, w = restartAsStandaloneNode(cfg, snapshot)
|
2014-11-07 08:50:35 +03:00
|
|
|
}
|
2014-10-24 03:41:58 +04:00
|
|
|
default:
|
2014-11-05 00:09:24 +03:00
|
|
|
return nil, fmt.Errorf("unsupported bootstrap config")
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
|
|
|
|
2014-10-11 02:53:28 +04:00
|
|
|
sstats := &stats.ServerStats{
|
|
|
|
Name: cfg.Name,
|
2014-10-31 03:31:14 +03:00
|
|
|
ID: id.String(),
|
2014-10-11 02:53:28 +04:00
|
|
|
}
|
2014-10-31 03:31:14 +03:00
|
|
|
lstats := stats.NewLeaderStats(id.String())
|
2014-10-11 02:53:28 +04:00
|
|
|
|
2014-11-03 23:04:24 +03:00
|
|
|
srv := &EtcdServer{
|
2015-03-06 00:03:04 +03:00
|
|
|
cfg: cfg,
|
|
|
|
snapCount: cfg.SnapCount,
|
|
|
|
errorc: make(chan error, 1),
|
|
|
|
store: st,
|
2015-01-16 01:39:30 +03:00
|
|
|
r: raftNode{
|
|
|
|
Node: n,
|
|
|
|
ticker: time.Tick(time.Duration(cfg.TickMs) * time.Millisecond),
|
|
|
|
raftStorage: s,
|
|
|
|
storage: NewStorage(w, ss),
|
|
|
|
},
|
|
|
|
id: id,
|
|
|
|
attributes: Attributes{Name: cfg.Name, ClientURLs: cfg.ClientURLs.StringSlice()},
|
|
|
|
Cluster: cfg.Cluster,
|
|
|
|
stats: sstats,
|
|
|
|
lstats: lstats,
|
|
|
|
SyncTicker: time.Tick(500 * time.Millisecond),
|
|
|
|
reqIDGen: idutil.NewGenerator(uint8(id), time.Now()),
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
2014-12-29 23:08:13 +03:00
|
|
|
|
2015-01-03 07:00:29 +03:00
|
|
|
tr := rafthttp.NewTransporter(cfg.Transport, id, cfg.Cluster.ID(), srv, srv.errorc, sstats, lstats)
|
2015-01-16 01:39:30 +03:00
|
|
|
srv.r.transport = tr
|
2015-03-24 09:19:38 +03:00
|
|
|
srv.Cluster.SetTransport(tr)
|
2014-11-03 23:04:24 +03:00
|
|
|
return srv, nil
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
|
|
|
|
2014-09-01 05:48:18 +04:00
|
|
|
// Start prepares and starts server in a new goroutine. It is no longer safe to
|
2014-09-12 04:14:41 +04:00
|
|
|
// modify a server's fields after it has been sent to Start.
|
2014-09-29 22:52:36 +04:00
|
|
|
// It also starts a goroutine to publish its server information.
|
2014-09-30 04:14:22 +04:00
|
|
|
func (s *EtcdServer) Start() {
|
2014-09-29 22:52:36 +04:00
|
|
|
s.start()
|
2014-10-01 20:33:55 +04:00
|
|
|
go s.publish(defaultPublishRetryInterval)
|
2014-11-29 08:32:28 +03:00
|
|
|
go s.purgeFile()
|
2015-04-04 00:26:15 +03:00
|
|
|
go monitorFileDescriptor(s.done)
|
2014-09-29 22:52:36 +04:00
|
|
|
}
|
|
|
|
|
|
|
|
// start prepares and starts server in a new goroutine. It is no longer safe to
|
|
|
|
// modify a server's fields after it has been sent to Start.
|
|
|
|
// This function is just used for testing.
|
|
|
|
func (s *EtcdServer) start() {
|
2015-03-06 00:03:04 +03:00
|
|
|
if s.snapCount == 0 {
|
2014-09-17 05:18:45 +04:00
|
|
|
log.Printf("etcdserver: set snapshot count to default %d", DefaultSnapCount)
|
2015-03-06 00:03:04 +03:00
|
|
|
s.snapCount = DefaultSnapCount
|
2014-09-17 05:18:45 +04:00
|
|
|
}
|
2014-08-29 03:41:42 +04:00
|
|
|
s.w = wait.New()
|
|
|
|
s.done = make(chan struct{})
|
2014-11-14 00:47:12 +03:00
|
|
|
s.stop = make(chan struct{})
|
2014-10-11 02:53:28 +04:00
|
|
|
s.stats.Initialize()
|
2014-09-25 06:51:27 +04:00
|
|
|
// TODO: if this is an empty log, writes all peer infos
|
|
|
|
// into the first entry
|
2014-08-29 03:41:42 +04:00
|
|
|
go s.run()
|
|
|
|
}
|
2014-08-28 00:37:22 +04:00
|
|
|
|
2014-11-29 08:32:28 +03:00
|
|
|
func (s *EtcdServer) purgeFile() {
|
|
|
|
var serrc, werrc <-chan error
|
|
|
|
if s.cfg.MaxSnapFiles > 0 {
|
|
|
|
serrc = fileutil.PurgeFile(s.cfg.SnapDir(), "snap", s.cfg.MaxSnapFiles, purgeFileInterval, s.done)
|
|
|
|
}
|
|
|
|
if s.cfg.MaxWALFiles > 0 {
|
|
|
|
werrc = fileutil.PurgeFile(s.cfg.WALDir(), "wal", s.cfg.MaxWALFiles, purgeFileInterval, s.done)
|
|
|
|
}
|
|
|
|
select {
|
|
|
|
case e := <-werrc:
|
|
|
|
log.Fatalf("etcdserver: failed to purge wal file %v", e)
|
|
|
|
case e := <-serrc:
|
|
|
|
log.Fatalf("etcdserver: failed to purge snap file %v", e)
|
|
|
|
case <-s.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-11-13 22:26:49 +03:00
|
|
|
func (s *EtcdServer) ID() types.ID { return s.id }
|
|
|
|
|
2015-01-16 01:39:30 +03:00
|
|
|
func (s *EtcdServer) RaftHandler() http.Handler { return s.r.transport.Handler() }
|
2014-11-19 00:53:14 +03:00
|
|
|
|
2014-09-12 04:14:41 +04:00
|
|
|
func (s *EtcdServer) Process(ctx context.Context, m raftpb.Message) error {
|
2014-10-31 03:31:14 +03:00
|
|
|
if s.Cluster.IsIDRemoved(types.ID(m.From)) {
|
2014-11-17 03:23:47 +03:00
|
|
|
log.Printf("etcdserver: reject message from removed member %s", types.ID(m.From).String())
|
|
|
|
return httptypes.NewHTTPError(http.StatusForbidden, "cannot process message from removed member")
|
2014-10-21 22:04:42 +04:00
|
|
|
}
|
2014-11-18 01:40:21 +03:00
|
|
|
if m.Type == raftpb.MsgApp {
|
|
|
|
s.stats.RecvAppendReq(types.ID(m.From).String(), m.Size())
|
2014-10-21 22:04:42 +04:00
|
|
|
}
|
2015-01-16 01:39:30 +03:00
|
|
|
return s.r.Step(ctx, m)
|
2014-09-12 04:14:41 +04:00
|
|
|
}
|
|
|
|
|
2015-03-01 18:02:24 +03:00
|
|
|
func (s *EtcdServer) ReportUnreachable(id uint64) { s.r.ReportUnreachable(id) }
|
|
|
|
|
2015-03-01 19:17:14 +03:00
|
|
|
func (s *EtcdServer) ReportSnapshot(id uint64, status raft.SnapshotStatus) {
|
|
|
|
s.r.ReportSnapshot(id, status)
|
|
|
|
}
|
|
|
|
|
2014-09-12 04:14:41 +04:00
|
|
|
func (s *EtcdServer) run() {
|
2015-01-16 01:39:30 +03:00
|
|
|
snap, err := s.r.raftStorage.Snapshot()
|
2014-12-09 03:13:16 +03:00
|
|
|
if err != nil {
|
|
|
|
log.Panicf("etcdserver: get snapshot from raft storage error: %v", err)
|
|
|
|
}
|
2014-12-09 08:17:04 +03:00
|
|
|
confState := snap.Metadata.ConfState
|
2015-03-06 00:03:04 +03:00
|
|
|
snapi := snap.Metadata.Index
|
|
|
|
appliedi := snapi
|
|
|
|
// TODO: get rid of the raft initialization in etcd server
|
|
|
|
s.r.s = s
|
|
|
|
s.r.applyc = make(chan apply)
|
|
|
|
go s.r.run()
|
2015-04-01 21:01:22 +03:00
|
|
|
defer func() {
|
|
|
|
s.r.stopped <- struct{}{}
|
|
|
|
<-s.r.done
|
|
|
|
close(s.done)
|
|
|
|
}()
|
2014-12-09 03:13:16 +03:00
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
var shouldstop bool
|
2014-08-26 05:39:02 +04:00
|
|
|
for {
|
2014-08-27 02:28:21 +04:00
|
|
|
select {
|
2015-03-06 00:03:04 +03:00
|
|
|
case apply := <-s.r.apply():
|
|
|
|
// apply snapshot
|
|
|
|
if !raft.IsEmptySnap(apply.snapshot) {
|
|
|
|
if apply.snapshot.Metadata.Index <= appliedi {
|
|
|
|
log.Panicf("etcdserver: snapshot index [%d] should > appliedi[%d] + 1",
|
|
|
|
apply.snapshot.Metadata.Index, appliedi)
|
2014-10-06 06:12:31 +04:00
|
|
|
}
|
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
if err := s.store.Recovery(apply.snapshot.Data); err != nil {
|
2014-12-09 03:13:16 +03:00
|
|
|
log.Panicf("recovery store error: %v", err)
|
2014-09-17 05:18:45 +04:00
|
|
|
}
|
2015-02-12 21:48:06 +03:00
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
// Avoid snapshot recovery overwriting newer cluster and
|
2015-02-21 00:51:38 +03:00
|
|
|
// transport setting, which may block the communication.
|
2015-03-06 00:03:04 +03:00
|
|
|
if s.Cluster.index < apply.snapshot.Metadata.Index {
|
2015-02-21 00:51:38 +03:00
|
|
|
s.Cluster.Recover()
|
2015-02-12 21:48:06 +03:00
|
|
|
}
|
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
appliedi = apply.snapshot.Metadata.Index
|
|
|
|
snapi = appliedi
|
|
|
|
confState = apply.snapshot.Metadata.ConfState
|
2014-12-09 03:13:16 +03:00
|
|
|
log.Printf("etcdserver: recovered from incoming snapshot at index %d", snapi)
|
2014-09-17 05:18:45 +04:00
|
|
|
}
|
2015-03-06 00:03:04 +03:00
|
|
|
|
|
|
|
// apply entries
|
|
|
|
if len(apply.entries) != 0 {
|
|
|
|
firsti := apply.entries[0].Index
|
2014-11-14 22:14:06 +03:00
|
|
|
if firsti > appliedi+1 {
|
|
|
|
log.Panicf("etcdserver: first index of committed entry[%d] should <= appliedi[%d] + 1", firsti, appliedi)
|
|
|
|
}
|
|
|
|
var ents []raftpb.Entry
|
2015-03-06 00:03:04 +03:00
|
|
|
if appliedi+1-firsti < uint64(len(apply.entries)) {
|
|
|
|
ents = apply.entries[appliedi+1-firsti:]
|
2014-11-14 22:14:06 +03:00
|
|
|
}
|
2015-03-06 00:03:04 +03:00
|
|
|
if appliedi, shouldstop = s.apply(ents, &confState); shouldstop {
|
|
|
|
go s.stopWithDelay(10*100*time.Millisecond, fmt.Errorf("the member has been permanently removed from the cluster"))
|
2014-11-14 10:03:34 +03:00
|
|
|
}
|
2014-11-14 11:10:46 +03:00
|
|
|
}
|
2014-09-17 05:18:45 +04:00
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
// wait for the raft routine to finish the disk writes before triggering a
|
|
|
|
// snapshot. or applied index might be greater than the last index in raft
|
|
|
|
// storage, since the raft routine might be slower than apply routine.
|
|
|
|
apply.done <- struct{}{}
|
2014-11-03 23:26:13 +03:00
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
// trigger snapshot
|
|
|
|
if appliedi-snapi > s.snapCount {
|
2014-11-30 10:17:57 +03:00
|
|
|
log.Printf("etcdserver: start to snapshot (applied: %d, lastsnap: %d)", appliedi, snapi)
|
2015-02-26 23:39:50 +03:00
|
|
|
s.snapshot(appliedi, confState)
|
2014-09-17 05:18:45 +04:00
|
|
|
snapi = appliedi
|
2014-08-28 02:34:33 +04:00
|
|
|
}
|
2015-01-03 07:00:29 +03:00
|
|
|
case err := <-s.errorc:
|
|
|
|
log.Printf("etcdserver: %s", err)
|
|
|
|
log.Printf("etcdserver: the data-dir used by this member must be removed.")
|
2014-11-14 10:03:34 +03:00
|
|
|
return
|
2014-11-14 00:47:12 +03:00
|
|
|
case <-s.stop:
|
2014-08-26 22:37:07 +04:00
|
|
|
return
|
2014-08-26 05:39:02 +04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-10-17 08:37:10 +04:00
|
|
|
// Stop stops the server gracefully, and shuts down the running goroutine.
|
|
|
|
// Stop should be called after a Start(s), otherwise it will block forever.
|
2014-09-12 04:14:41 +04:00
|
|
|
func (s *EtcdServer) Stop() {
|
2014-11-14 00:47:12 +03:00
|
|
|
select {
|
|
|
|
case s.stop <- struct{}{}:
|
|
|
|
case <-s.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
<-s.done
|
2014-08-30 02:32:41 +04:00
|
|
|
}
|
2014-08-29 03:41:42 +04:00
|
|
|
|
2015-01-03 07:00:29 +03:00
|
|
|
func (s *EtcdServer) stopWithDelay(d time.Duration, err error) {
|
2014-12-24 02:47:41 +03:00
|
|
|
time.Sleep(d)
|
2015-01-03 07:00:29 +03:00
|
|
|
select {
|
|
|
|
case s.errorc <- err:
|
|
|
|
default:
|
|
|
|
}
|
2014-12-24 02:47:41 +03:00
|
|
|
}
|
|
|
|
|
2014-11-14 01:16:48 +03:00
|
|
|
// StopNotify returns a channel that receives a empty struct
|
|
|
|
// when the server is stopped.
|
|
|
|
func (s *EtcdServer) StopNotify() <-chan struct{} { return s.done }
|
|
|
|
|
2014-10-03 22:05:56 +04:00
|
|
|
// Do interprets r and performs an operation on s.store according to r.Method
|
2014-09-16 04:35:02 +04:00
|
|
|
// and other fields. If r.Method is "POST", "PUT", "DELETE", or a "GET" with
|
2014-09-01 05:48:18 +04:00
|
|
|
// Quorum == true, r will be sent through consensus before performing its
|
|
|
|
// respective operation. Do will block until an action is performed or there is
|
|
|
|
// an error.
|
2014-09-12 04:14:41 +04:00
|
|
|
func (s *EtcdServer) Do(ctx context.Context, r pb.Request) (Response, error) {
|
2014-12-16 03:07:47 +03:00
|
|
|
r.ID = s.reqIDGen.Next()
|
2014-08-31 01:35:27 +04:00
|
|
|
if r.Method == "GET" && r.Quorum {
|
|
|
|
r.Method = "QGET"
|
|
|
|
}
|
2014-08-27 02:28:21 +04:00
|
|
|
switch r.Method {
|
2014-08-31 01:35:27 +04:00
|
|
|
case "POST", "PUT", "DELETE", "QGET":
|
2014-08-27 02:28:21 +04:00
|
|
|
data, err := r.Marshal()
|
|
|
|
if err != nil {
|
|
|
|
return Response{}, err
|
|
|
|
}
|
2014-09-23 21:17:53 +04:00
|
|
|
ch := s.w.Register(r.ID)
|
2015-02-15 22:36:23 +03:00
|
|
|
|
|
|
|
// TODO: benchmark the cost of time.Now()
|
|
|
|
// might be sampling?
|
|
|
|
start := time.Now()
|
2015-01-16 01:39:30 +03:00
|
|
|
s.r.Propose(ctx, data)
|
2015-02-15 22:36:23 +03:00
|
|
|
|
|
|
|
proposePending.Inc()
|
|
|
|
defer proposePending.Dec()
|
|
|
|
|
2014-08-27 02:28:21 +04:00
|
|
|
select {
|
|
|
|
case x := <-ch:
|
2015-02-15 22:36:23 +03:00
|
|
|
proposeDurations.Observe(float64(time.Since(start).Nanoseconds() / int64(time.Millisecond)))
|
2014-08-27 02:28:21 +04:00
|
|
|
resp := x.(Response)
|
|
|
|
return resp, resp.err
|
|
|
|
case <-ctx.Done():
|
2015-02-15 22:36:23 +03:00
|
|
|
proposeFailed.Inc()
|
2014-09-23 21:17:53 +04:00
|
|
|
s.w.Trigger(r.ID, nil) // GC wait
|
2014-10-26 21:35:59 +03:00
|
|
|
return Response{}, parseCtxErr(ctx.Err())
|
2014-08-29 03:41:42 +04:00
|
|
|
case <-s.done:
|
|
|
|
return Response{}, ErrStopped
|
2014-08-27 02:28:21 +04:00
|
|
|
}
|
|
|
|
case "GET":
|
|
|
|
switch {
|
|
|
|
case r.Wait:
|
2014-10-03 22:05:56 +04:00
|
|
|
wc, err := s.store.Watch(r.Path, r.Recursive, r.Stream, r.Since)
|
2014-08-27 02:28:21 +04:00
|
|
|
if err != nil {
|
|
|
|
return Response{}, err
|
|
|
|
}
|
|
|
|
return Response{Watcher: wc}, nil
|
|
|
|
default:
|
2014-10-03 22:05:56 +04:00
|
|
|
ev, err := s.store.Get(r.Path, r.Recursive, r.Sorted)
|
2014-08-27 02:28:21 +04:00
|
|
|
if err != nil {
|
|
|
|
return Response{}, err
|
|
|
|
}
|
|
|
|
return Response{Event: ev}, nil
|
|
|
|
}
|
2014-11-04 09:01:07 +03:00
|
|
|
case "HEAD":
|
|
|
|
ev, err := s.store.Get(r.Path, r.Recursive, r.Sorted)
|
|
|
|
if err != nil {
|
|
|
|
return Response{}, err
|
|
|
|
}
|
|
|
|
return Response{Event: ev}, nil
|
2014-08-27 02:42:10 +04:00
|
|
|
default:
|
|
|
|
return Response{}, ErrUnknownMethod
|
2014-08-26 05:39:02 +04:00
|
|
|
}
|
|
|
|
}
|
2014-08-26 22:37:07 +04:00
|
|
|
|
2014-11-14 01:01:56 +03:00
|
|
|
func (s *EtcdServer) SelfStats() []byte { return s.stats.JSON() }
|
2014-10-17 03:56:58 +04:00
|
|
|
|
2014-10-17 05:39:55 +04:00
|
|
|
func (s *EtcdServer) LeaderStats() []byte {
|
2015-01-16 01:39:30 +03:00
|
|
|
lead := atomic.LoadUint64(&s.r.lead)
|
2014-12-16 03:41:11 +03:00
|
|
|
if lead != uint64(s.id) {
|
|
|
|
return nil
|
|
|
|
}
|
2014-10-17 05:39:55 +04:00
|
|
|
return s.lstats.JSON()
|
2014-10-11 02:53:28 +04:00
|
|
|
}
|
|
|
|
|
2014-11-14 01:01:56 +03:00
|
|
|
func (s *EtcdServer) StoreStats() []byte { return s.store.JsonStats() }
|
2014-10-11 02:53:28 +04:00
|
|
|
|
2014-10-01 02:14:44 +04:00
|
|
|
func (s *EtcdServer) AddMember(ctx context.Context, memb Member) error {
|
2014-10-08 03:19:51 +04:00
|
|
|
// TODO: move Member to protobuf type
|
2014-10-01 02:14:44 +04:00
|
|
|
b, err := json.Marshal(memb)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2014-09-23 23:02:44 +04:00
|
|
|
cc := raftpb.ConfChange{
|
|
|
|
Type: raftpb.ConfChangeAddNode,
|
2014-10-31 03:31:14 +03:00
|
|
|
NodeID: uint64(memb.ID),
|
2014-10-01 02:14:44 +04:00
|
|
|
Context: b,
|
2014-09-19 23:36:19 +04:00
|
|
|
}
|
2014-09-23 08:59:13 +04:00
|
|
|
return s.configure(ctx, cc)
|
2014-09-19 23:36:19 +04:00
|
|
|
}
|
|
|
|
|
2014-10-08 15:58:53 +04:00
|
|
|
func (s *EtcdServer) RemoveMember(ctx context.Context, id uint64) error {
|
2014-09-23 23:02:44 +04:00
|
|
|
cc := raftpb.ConfChange{
|
|
|
|
Type: raftpb.ConfChangeRemoveNode,
|
2014-09-19 23:36:19 +04:00
|
|
|
NodeID: id,
|
|
|
|
}
|
2014-09-23 08:59:13 +04:00
|
|
|
return s.configure(ctx, cc)
|
2014-09-19 23:36:19 +04:00
|
|
|
}
|
|
|
|
|
2014-11-11 22:46:49 +03:00
|
|
|
func (s *EtcdServer) UpdateMember(ctx context.Context, memb Member) error {
|
|
|
|
b, err := json.Marshal(memb)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
cc := raftpb.ConfChange{
|
|
|
|
Type: raftpb.ConfChangeUpdateNode,
|
|
|
|
NodeID: uint64(memb.ID),
|
|
|
|
Context: b,
|
|
|
|
}
|
|
|
|
return s.configure(ctx, cc)
|
|
|
|
}
|
|
|
|
|
2014-09-25 05:05:33 +04:00
|
|
|
// Implement the RaftTimer interface
|
2015-01-16 01:39:30 +03:00
|
|
|
func (s *EtcdServer) Index() uint64 { return atomic.LoadUint64(&s.r.index) }
|
2014-09-25 05:05:33 +04:00
|
|
|
|
2015-01-16 01:39:30 +03:00
|
|
|
func (s *EtcdServer) Term() uint64 { return atomic.LoadUint64(&s.r.term) }
|
2014-09-25 05:05:33 +04:00
|
|
|
|
2014-11-16 08:28:03 +03:00
|
|
|
// Only for testing purpose
|
|
|
|
// TODO: add Raft server interface to expose raft related info:
|
|
|
|
// Index, Term, Lead, Committed, Applied, LastIndex, etc.
|
2015-01-16 01:39:30 +03:00
|
|
|
func (s *EtcdServer) Lead() uint64 { return atomic.LoadUint64(&s.r.lead) }
|
2014-09-25 05:05:33 +04:00
|
|
|
|
2015-01-06 07:20:11 +03:00
|
|
|
func (s *EtcdServer) Leader() types.ID { return types.ID(s.Lead()) }
|
|
|
|
|
2014-11-06 23:09:21 +03:00
|
|
|
// configure sends a configuration change through consensus and
|
|
|
|
// then waits for it to be applied to the server. It
|
|
|
|
// will block until the change is performed or there is an error.
|
2014-09-23 23:02:44 +04:00
|
|
|
func (s *EtcdServer) configure(ctx context.Context, cc raftpb.ConfChange) error {
|
2014-12-16 03:07:47 +03:00
|
|
|
cc.ID = s.reqIDGen.Next()
|
2014-10-12 04:35:58 +04:00
|
|
|
ch := s.w.Register(cc.ID)
|
2015-01-16 01:39:30 +03:00
|
|
|
if err := s.r.ProposeConfChange(ctx, cc); err != nil {
|
2014-10-12 04:35:58 +04:00
|
|
|
s.w.Trigger(cc.ID, nil)
|
2014-09-19 23:36:19 +04:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
select {
|
2014-10-16 20:49:26 +04:00
|
|
|
case x := <-ch:
|
|
|
|
if err, ok := x.(error); ok {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if x != nil {
|
2014-11-05 23:14:42 +03:00
|
|
|
log.Panicf("return type should always be error")
|
2014-10-16 20:49:26 +04:00
|
|
|
}
|
2014-09-19 23:36:19 +04:00
|
|
|
return nil
|
|
|
|
case <-ctx.Done():
|
2014-10-12 04:35:58 +04:00
|
|
|
s.w.Trigger(cc.ID, nil) // GC wait
|
2014-10-26 21:35:59 +03:00
|
|
|
return parseCtxErr(ctx.Err())
|
2014-09-19 23:36:19 +04:00
|
|
|
case <-s.done:
|
|
|
|
return ErrStopped
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-09-16 04:35:02 +04:00
|
|
|
// sync proposes a SYNC request and is non-blocking.
|
|
|
|
// This makes no guarantee that the request will be proposed or performed.
|
|
|
|
// The request will be cancelled after the given timeout.
|
|
|
|
func (s *EtcdServer) sync(timeout time.Duration) {
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
|
|
req := pb.Request{
|
|
|
|
Method: "SYNC",
|
2014-12-16 03:07:47 +03:00
|
|
|
ID: s.reqIDGen.Next(),
|
2014-09-16 04:35:02 +04:00
|
|
|
Time: time.Now().UnixNano(),
|
|
|
|
}
|
2014-10-11 16:41:00 +04:00
|
|
|
data := pbutil.MustMarshal(&req)
|
2014-09-16 04:35:02 +04:00
|
|
|
// There is no promise that node has leader when do SYNC request,
|
|
|
|
// so it uses goroutine to propose.
|
|
|
|
go func() {
|
2015-01-16 01:39:30 +03:00
|
|
|
s.r.Propose(ctx, data)
|
2014-09-16 04:35:02 +04:00
|
|
|
cancel()
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
2014-09-27 01:18:59 +04:00
|
|
|
// publish registers server information into the cluster. The information
|
2014-10-03 03:20:14 +04:00
|
|
|
// is the JSON representation of this server's member struct, updated with the
|
|
|
|
// static clientURLs of the server.
|
2014-09-27 01:18:59 +04:00
|
|
|
// The function keeps attempting to register until it succeeds,
|
|
|
|
// or its server is stopped.
|
2014-10-01 20:33:55 +04:00
|
|
|
func (s *EtcdServer) publish(retryInterval time.Duration) {
|
2014-10-03 02:48:58 +04:00
|
|
|
b, err := json.Marshal(s.attributes)
|
2014-09-27 01:18:59 +04:00
|
|
|
if err != nil {
|
|
|
|
log.Printf("etcdserver: json marshal error: %v", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
req := pb.Request{
|
|
|
|
Method: "PUT",
|
2014-10-29 23:27:07 +03:00
|
|
|
Path: MemberAttributesStorePath(s.id),
|
2014-09-27 01:18:59 +04:00
|
|
|
Val: string(b),
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), retryInterval)
|
|
|
|
_, err := s.Do(ctx, req)
|
|
|
|
cancel()
|
|
|
|
switch err {
|
|
|
|
case nil:
|
2014-10-31 03:31:14 +03:00
|
|
|
log.Printf("etcdserver: published %+v to cluster %s", s.attributes, s.Cluster.ID())
|
2014-09-27 01:18:59 +04:00
|
|
|
return
|
|
|
|
case ErrStopped:
|
|
|
|
log.Printf("etcdserver: aborting publish because server is stopped")
|
|
|
|
return
|
|
|
|
default:
|
|
|
|
log.Printf("etcdserver: publish error: %v", err)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-11-26 08:38:57 +03:00
|
|
|
func (s *EtcdServer) send(ms []raftpb.Message) {
|
2015-04-13 15:25:16 +03:00
|
|
|
for i, _ := range ms {
|
|
|
|
if s.Cluster.IsIDRemoved(types.ID(ms[i].To)) {
|
|
|
|
ms[i].To = 0
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
|
|
|
}
|
2015-01-16 01:39:30 +03:00
|
|
|
s.r.transport.Send(ms)
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
|
|
|
|
2014-12-09 02:54:52 +03:00
|
|
|
// apply takes entries received from Raft (after it has been committed) and
|
|
|
|
// applies them to the current state of the EtcdServer.
|
|
|
|
// The given entries should not be empty.
|
2014-12-09 08:12:13 +03:00
|
|
|
func (s *EtcdServer) apply(es []raftpb.Entry, confState *raftpb.ConfState) (uint64, bool) {
|
2014-10-11 17:58:39 +04:00
|
|
|
var applied uint64
|
2015-01-03 01:10:07 +03:00
|
|
|
var shouldstop bool
|
|
|
|
var err error
|
2014-10-11 17:58:39 +04:00
|
|
|
for i := range es {
|
|
|
|
e := es[i]
|
|
|
|
switch e.Type {
|
|
|
|
case raftpb.EntryNormal:
|
|
|
|
var r pb.Request
|
2014-10-11 16:41:00 +04:00
|
|
|
pbutil.MustUnmarshal(&r, e.Data)
|
2014-10-11 17:58:39 +04:00
|
|
|
s.w.Trigger(r.ID, s.applyRequest(r))
|
|
|
|
case raftpb.EntryConfChange:
|
|
|
|
var cc raftpb.ConfChange
|
2014-10-11 16:41:00 +04:00
|
|
|
pbutil.MustUnmarshal(&cc, e.Data)
|
2015-03-24 09:19:38 +03:00
|
|
|
shouldstop, err = s.applyConfChange(cc, confState, e.Index)
|
2014-11-14 10:03:34 +03:00
|
|
|
s.w.Trigger(cc.ID, err)
|
2014-10-11 17:58:39 +04:00
|
|
|
default:
|
2014-11-05 23:14:42 +03:00
|
|
|
log.Panicf("entry type should be either EntryNormal or EntryConfChange")
|
2014-10-11 17:58:39 +04:00
|
|
|
}
|
2015-01-16 01:39:30 +03:00
|
|
|
atomic.StoreUint64(&s.r.index, e.Index)
|
|
|
|
atomic.StoreUint64(&s.r.term, e.Term)
|
2014-10-11 17:58:39 +04:00
|
|
|
applied = e.Index
|
|
|
|
}
|
2015-01-03 01:10:07 +03:00
|
|
|
return applied, shouldstop
|
2014-10-11 17:58:39 +04:00
|
|
|
}
|
|
|
|
|
2014-10-01 02:14:44 +04:00
|
|
|
// applyRequest interprets r as a call to store.X and returns a Response interpreted
|
2014-09-24 21:46:22 +04:00
|
|
|
// from store.Event
|
2014-10-01 02:14:44 +04:00
|
|
|
func (s *EtcdServer) applyRequest(r pb.Request) Response {
|
2014-08-31 02:27:43 +04:00
|
|
|
f := func(ev *store.Event, err error) Response {
|
|
|
|
return Response{Event: ev, err: err}
|
|
|
|
}
|
2014-12-29 22:15:02 +03:00
|
|
|
expr := timeutil.UnixNanoToTime(r.Expiration)
|
2014-08-27 03:17:02 +04:00
|
|
|
switch r.Method {
|
|
|
|
case "POST":
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.Create(r.Path, r.Dir, r.Val, true, expr))
|
2014-08-27 03:17:02 +04:00
|
|
|
case "PUT":
|
2014-12-31 01:51:26 +03:00
|
|
|
exists, existsSet := pbutil.GetBool(r.PrevExist)
|
2014-08-27 03:39:26 +04:00
|
|
|
switch {
|
2014-08-27 23:04:00 +04:00
|
|
|
case existsSet:
|
2014-08-27 22:54:29 +04:00
|
|
|
if exists {
|
2015-04-15 08:18:04 +03:00
|
|
|
if r.PrevIndex == 0 && r.PrevValue == "" {
|
|
|
|
return f(s.store.Update(r.Path, r.Val, expr))
|
|
|
|
} else {
|
|
|
|
return f(s.store.CompareAndSwap(r.Path, r.PrevValue, r.PrevIndex, r.Val, expr))
|
|
|
|
}
|
2014-08-27 22:54:29 +04:00
|
|
|
}
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.Create(r.Path, r.Dir, r.Val, false, expr))
|
2014-08-27 22:38:25 +04:00
|
|
|
case r.PrevIndex > 0 || r.PrevValue != "":
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.CompareAndSwap(r.Path, r.PrevValue, r.PrevIndex, r.Val, expr))
|
2014-08-27 22:29:33 +04:00
|
|
|
default:
|
2014-10-24 12:13:47 +04:00
|
|
|
if storeMemberAttributeRegexp.MatchString(r.Path) {
|
2014-10-27 20:35:24 +03:00
|
|
|
id := mustParseMemberIDFromKey(path.Dir(r.Path))
|
2014-11-08 03:04:09 +03:00
|
|
|
var attr Attributes
|
|
|
|
if err := json.Unmarshal([]byte(r.Val), &attr); err != nil {
|
2014-11-05 23:14:42 +03:00
|
|
|
log.Panicf("unmarshal %s should never fail: %v", r.Val, err)
|
2014-10-24 12:13:47 +04:00
|
|
|
}
|
2014-12-26 07:56:30 +03:00
|
|
|
s.Cluster.UpdateAttributes(id, attr)
|
2014-10-24 12:13:47 +04:00
|
|
|
}
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.Set(r.Path, r.Dir, r.Val, expr))
|
2014-08-27 03:39:26 +04:00
|
|
|
}
|
2014-08-27 03:17:02 +04:00
|
|
|
case "DELETE":
|
2014-08-27 22:34:42 +04:00
|
|
|
switch {
|
|
|
|
case r.PrevIndex > 0 || r.PrevValue != "":
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.CompareAndDelete(r.Path, r.PrevValue, r.PrevIndex))
|
2014-08-27 22:34:42 +04:00
|
|
|
default:
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.Delete(r.Path, r.Dir, r.Recursive))
|
2014-08-27 22:34:42 +04:00
|
|
|
}
|
2014-08-31 01:35:27 +04:00
|
|
|
case "QGET":
|
2014-10-03 22:05:56 +04:00
|
|
|
return f(s.store.Get(r.Path, r.Recursive, r.Sorted))
|
2014-09-16 04:35:02 +04:00
|
|
|
case "SYNC":
|
2014-10-03 22:05:56 +04:00
|
|
|
s.store.DeleteExpiredKeys(time.Unix(0, r.Time))
|
2014-09-16 04:35:02 +04:00
|
|
|
return Response{}
|
2014-08-27 03:17:02 +04:00
|
|
|
default:
|
2014-08-28 00:37:22 +04:00
|
|
|
// This should never be reached, but just in case:
|
2014-08-31 02:27:43 +04:00
|
|
|
return Response{err: ErrUnknownMethod}
|
2014-08-27 03:17:02 +04:00
|
|
|
}
|
2014-08-26 22:37:07 +04:00
|
|
|
}
|
2014-08-27 22:54:29 +04:00
|
|
|
|
2015-03-24 09:19:38 +03:00
|
|
|
// applyConfChange applies a ConfChange to the server at the given index. It is only
|
|
|
|
// invoked with a ConfChange that has already passed through Raft.
|
|
|
|
func (s *EtcdServer) applyConfChange(cc raftpb.ConfChange, confState *raftpb.ConfState, index uint64) (bool, error) {
|
2014-11-06 04:16:42 +03:00
|
|
|
if err := s.Cluster.ValidateConfigurationChange(cc); err != nil {
|
2014-10-16 20:49:26 +04:00
|
|
|
cc.NodeID = raft.None
|
2015-01-16 01:39:30 +03:00
|
|
|
s.r.ApplyConfChange(cc)
|
2014-11-14 10:03:34 +03:00
|
|
|
return false, err
|
2014-10-16 20:49:26 +04:00
|
|
|
}
|
2015-01-16 01:39:30 +03:00
|
|
|
*confState = *s.r.ApplyConfChange(cc)
|
2014-10-01 02:14:44 +04:00
|
|
|
switch cc.Type {
|
|
|
|
case raftpb.ConfChangeAddNode:
|
2014-10-23 03:15:37 +04:00
|
|
|
m := new(Member)
|
|
|
|
if err := json.Unmarshal(cc.Context, m); err != nil {
|
2014-11-05 23:14:42 +03:00
|
|
|
log.Panicf("unmarshal member should never fail: %v", err)
|
2014-10-01 02:14:44 +04:00
|
|
|
}
|
2014-10-31 03:31:14 +03:00
|
|
|
if cc.NodeID != uint64(m.ID) {
|
2014-11-05 23:14:42 +03:00
|
|
|
log.Panicf("nodeID should always be equal to member ID")
|
2014-10-04 02:21:10 +04:00
|
|
|
}
|
2015-03-24 09:19:38 +03:00
|
|
|
s.Cluster.AddMember(m, index)
|
2014-11-13 07:37:50 +03:00
|
|
|
if m.ID == s.id {
|
|
|
|
log.Printf("etcdserver: added local member %s %v to cluster %s", m.ID, m.PeerURLs, s.Cluster.ID())
|
|
|
|
} else {
|
|
|
|
log.Printf("etcdserver: added member %s %v to cluster %s", m.ID, m.PeerURLs, s.Cluster.ID())
|
|
|
|
}
|
2014-10-01 02:14:44 +04:00
|
|
|
case raftpb.ConfChangeRemoveNode:
|
2014-11-05 01:39:43 +03:00
|
|
|
id := types.ID(cc.NodeID)
|
2015-03-24 09:19:38 +03:00
|
|
|
s.Cluster.RemoveMember(id, index)
|
2014-11-13 07:37:50 +03:00
|
|
|
if id == s.id {
|
2014-11-14 10:03:34 +03:00
|
|
|
return true, nil
|
2014-11-13 07:37:50 +03:00
|
|
|
} else {
|
|
|
|
log.Printf("etcdserver: removed member %s from cluster %s", id, s.Cluster.ID())
|
|
|
|
}
|
2014-11-11 22:46:49 +03:00
|
|
|
case raftpb.ConfChangeUpdateNode:
|
|
|
|
m := new(Member)
|
|
|
|
if err := json.Unmarshal(cc.Context, m); err != nil {
|
|
|
|
log.Panicf("unmarshal member should never fail: %v", err)
|
|
|
|
}
|
|
|
|
if cc.NodeID != uint64(m.ID) {
|
|
|
|
log.Panicf("nodeID should always be equal to member ID")
|
|
|
|
}
|
2015-03-24 09:19:38 +03:00
|
|
|
s.Cluster.UpdateRaftAttributes(m.ID, m.RaftAttributes, index)
|
2014-11-13 07:37:50 +03:00
|
|
|
if m.ID == s.id {
|
|
|
|
log.Printf("etcdserver: update local member %s %v in cluster %s", m.ID, m.PeerURLs, s.Cluster.ID())
|
|
|
|
} else {
|
|
|
|
log.Printf("etcdserver: update member %s %v in cluster %s", m.ID, m.PeerURLs, s.Cluster.ID())
|
|
|
|
}
|
2014-10-16 20:49:26 +04:00
|
|
|
}
|
2014-11-14 10:03:34 +03:00
|
|
|
return false, nil
|
2014-10-16 20:49:26 +04:00
|
|
|
}
|
|
|
|
|
2014-09-17 05:18:45 +04:00
|
|
|
// TODO: non-blocking snapshot
|
2015-02-26 23:39:50 +03:00
|
|
|
func (s *EtcdServer) snapshot(snapi uint64, confState raftpb.ConfState) {
|
|
|
|
clone := s.store.Clone()
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
d, err := clone.SaveNoCopy()
|
|
|
|
// TODO: current store will never fail to do a snapshot
|
|
|
|
// what should we do if the store might fail?
|
|
|
|
if err != nil {
|
|
|
|
log.Panicf("etcdserver: store save should never fail: %v", err)
|
|
|
|
}
|
|
|
|
snap, err := s.r.raftStorage.CreateSnapshot(snapi, &confState, d)
|
|
|
|
if err != nil {
|
|
|
|
// the snapshot was done asynchronously with the progress of raft.
|
|
|
|
// raft might have already got a newer snapshot.
|
|
|
|
if err == raft.ErrSnapOutOfDate {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
log.Panicf("etcdserver: unexpected create snapshot error %v", err)
|
|
|
|
}
|
|
|
|
if err := s.r.storage.SaveSnap(snap); err != nil {
|
|
|
|
log.Fatalf("etcdserver: save snapshot error: %v", err)
|
2015-01-23 21:50:43 +03:00
|
|
|
}
|
2015-03-01 07:33:28 +03:00
|
|
|
log.Printf("etcdserver: saved snapshot at index %d", snap.Metadata.Index)
|
2015-01-23 21:50:43 +03:00
|
|
|
|
2015-03-01 07:33:28 +03:00
|
|
|
// keep some in memory log entries for slow followers.
|
|
|
|
compacti := uint64(1)
|
|
|
|
if snapi > numberOfCatchUpEntries {
|
|
|
|
compacti = snapi - numberOfCatchUpEntries
|
|
|
|
}
|
|
|
|
err = s.r.raftStorage.Compact(compacti)
|
2015-02-26 23:39:50 +03:00
|
|
|
if err != nil {
|
|
|
|
// the compaction was done asynchronously with the progress of raft.
|
|
|
|
// raft log might already been compact.
|
|
|
|
if err == raft.ErrCompacted {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
log.Panicf("etcdserver: unexpected compaction error %v", err)
|
2014-11-26 02:13:07 +03:00
|
|
|
}
|
2015-03-01 07:33:28 +03:00
|
|
|
log.Printf("etcdserver: compacted raft log at %d", compacti)
|
2015-02-26 23:39:50 +03:00
|
|
|
}()
|
2014-09-17 05:18:45 +04:00
|
|
|
}
|
|
|
|
|
2015-01-16 01:39:30 +03:00
|
|
|
func (s *EtcdServer) PauseSending() { s.r.pauseSending() }
|
2014-12-26 08:37:20 +03:00
|
|
|
|
2015-01-16 01:39:30 +03:00
|
|
|
func (s *EtcdServer) ResumeSending() { s.r.resumeSending() }
|