2016-05-13 06:49:40 +03:00
|
|
|
// Copyright 2015 The etcd Authors
|
2015-01-25 06:19:16 +03:00
|
|
|
//
|
|
|
|
// 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"
|
2016-08-02 20:29:23 +03:00
|
|
|
"math"
|
2014-09-18 01:38:10 +04:00
|
|
|
"math/rand"
|
2014-10-24 03:41:58 +04:00
|
|
|
"net/http"
|
2015-08-08 15:58:29 +03:00
|
|
|
"os"
|
2014-10-21 22:47:03 +04:00
|
|
|
"path"
|
2014-10-24 12:13:47 +04:00
|
|
|
"regexp"
|
2016-03-10 08:37:10 +03:00
|
|
|
"sync"
|
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
|
|
|
|
2016-03-24 03:20:52 +03:00
|
|
|
"github.com/coreos/etcd/alarm"
|
2016-03-08 11:31:48 +03:00
|
|
|
"github.com/coreos/etcd/auth"
|
2016-02-25 06:35:15 +03:00
|
|
|
"github.com/coreos/etcd/compactor"
|
2014-10-04 16:20:45 +04:00
|
|
|
"github.com/coreos/etcd/discovery"
|
2016-07-26 21:33:11 +03:00
|
|
|
"github.com/coreos/etcd/etcdserver/api"
|
2016-04-05 02:31:35 +03:00
|
|
|
"github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
|
2014-09-03 08:36:14 +04:00
|
|
|
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
|
2016-04-07 22:02:37 +03:00
|
|
|
"github.com/coreos/etcd/etcdserver/membership"
|
2014-10-11 02:53:28 +04:00
|
|
|
"github.com/coreos/etcd/etcdserver/stats"
|
2016-01-06 00:49:25 +03:00
|
|
|
"github.com/coreos/etcd/lease"
|
2016-04-25 22:32:58 +03:00
|
|
|
"github.com/coreos/etcd/mvcc"
|
|
|
|
"github.com/coreos/etcd/mvcc/backend"
|
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"
|
2016-02-12 06:21:30 +03:00
|
|
|
"github.com/coreos/etcd/pkg/schedule"
|
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"
|
2016-03-23 03:10:28 +03:00
|
|
|
"github.com/coreos/go-semver/semver"
|
|
|
|
"github.com/coreos/pkg/capnslog"
|
|
|
|
"golang.org/x/net/context"
|
2014-08-26 05:39:02 +04:00
|
|
|
)
|
|
|
|
|
2014-09-17 05:18:45 +04:00
|
|
|
const (
|
2015-08-12 07:09:03 +03:00
|
|
|
DefaultSnapCount = 10000
|
2014-10-15 11:19:27 +04:00
|
|
|
|
2015-04-29 22:05:40 +03:00
|
|
|
StoreClusterPrefix = "/0"
|
|
|
|
StoreKeysPrefix = "/1"
|
2014-11-29 08:32:28 +03:00
|
|
|
|
2016-01-25 03:16:05 +03:00
|
|
|
purgeFileInterval = 30 * time.Second
|
|
|
|
// monitorVersionInterval should be smaller than the timeout
|
2016-02-01 08:42:39 +03:00
|
|
|
// on the connection. Or we will not be able to reuse the connection
|
2016-01-25 03:16:05 +03:00
|
|
|
// (since it will timeout).
|
|
|
|
monitorVersionInterval = rafthttp.ConnWriteTimeout - time.Second
|
2015-09-15 08:15:23 +03:00
|
|
|
|
|
|
|
databaseFilename = "db"
|
2015-12-08 18:52:54 +03:00
|
|
|
// max number of in-flight snapshot messages etcdserver allows to have
|
|
|
|
// This number is more than enough for most clusters with 5 machines.
|
|
|
|
maxInFlightMsgSnap = 16
|
2015-12-24 00:55:39 +03:00
|
|
|
|
2015-12-24 03:03:23 +03:00
|
|
|
releaseDelayAfterSnapshot = 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 (
|
2015-06-08 23:28:34 +03:00
|
|
|
plog = capnslog.NewPackageLogger("github.com/coreos/etcd", "etcdserver")
|
|
|
|
|
2016-04-07 22:02:37 +03:00
|
|
|
storeMemberAttributeRegexp = regexp.MustCompile(path.Join(membership.StoreMembersPrefix, "[[:xdigit:]]{1,16}", "attributes"))
|
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.
|
2016-04-07 22:02:37 +03:00
|
|
|
AddMember(ctx context.Context, memb membership.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
|
|
|
|
2015-09-14 23:41:13 +03:00
|
|
|
// UpdateMember attempts to update an existing member in the cluster. It will
|
2014-11-11 22:46:49 +03:00
|
|
|
// return ErrIDNotFound if the member ID does not exist.
|
2016-04-07 22:02:37 +03:00
|
|
|
UpdateMember(ctx context.Context, updateMemb membership.Member) error
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
|
|
|
|
// ClusterVersion is the cluster-wide minimum major.minor version.
|
2016-01-28 13:16:29 +03:00
|
|
|
// Cluster version is set to the min version that an etcd member is
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
// compatible with when first bootstrap.
|
|
|
|
//
|
2015-05-08 21:10:12 +03:00
|
|
|
// ClusterVersion is nil until the cluster is bootstrapped (has a quorum).
|
|
|
|
//
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
// During a rolling upgrades, the ClusterVersion will be updated
|
2015-05-08 21:10:12 +03:00
|
|
|
// automatically after a sync. (5 second by default)
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
//
|
|
|
|
// The API/raft component can utilize ClusterVersion to determine if
|
|
|
|
// it can accept a client request or a raft RPC.
|
2015-05-08 21:10:12 +03:00
|
|
|
// NOTE: ClusterVersion might be nil when etcd 2.1 works with etcd 2.0 and
|
|
|
|
// the leader is etcd 2.0. etcd 2.0 leader will not update clusterVersion since
|
|
|
|
// this feature is introduced post 2.0.
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
ClusterVersion() *semver.Version
|
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 {
|
2016-07-08 21:20:47 +03:00
|
|
|
// inflightSnapshots holds count the number of snapshots currently inflight.
|
|
|
|
inflightSnapshots int64 // must use atomic operations to access; keep 64-bit aligned.
|
|
|
|
appliedIndex uint64 // must use atomic operations to access; keep 64-bit aligned.
|
2016-07-12 23:16:35 +03:00
|
|
|
committedIndex uint64 // must use atomic operations to access; keep 64-bit aligned.
|
2016-07-08 21:20:47 +03:00
|
|
|
// consistIndex used to hold the offset of current executing entry
|
|
|
|
// It is initialized to 0 before executing any entry.
|
|
|
|
consistIndex consistentIndex // must use atomic operations to access; keep 64-bit aligned.
|
|
|
|
Cfg *ServerConfig
|
2016-04-23 05:57:33 +03:00
|
|
|
|
|
|
|
readych chan struct{}
|
|
|
|
r raftNode
|
2015-08-09 03:40:12 +03:00
|
|
|
|
2015-03-06 00:03:04 +03:00
|
|
|
snapCount uint64
|
2015-01-16 01:39:30 +03:00
|
|
|
|
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
|
2016-04-07 22:02:37 +03:00
|
|
|
attributes membership.Attributes
|
2014-10-11 17:03:03 +04:00
|
|
|
|
2016-04-07 22:02:37 +03:00
|
|
|
cluster *membership.RaftCluster
|
2014-10-11 17:03:03 +04:00
|
|
|
|
2014-12-26 21:56:56 +03:00
|
|
|
store store.Store
|
2015-12-11 04:29:36 +03:00
|
|
|
|
2016-05-17 00:40:52 +03:00
|
|
|
applyV2 ApplierV2
|
2016-04-20 10:31:58 +03:00
|
|
|
|
2016-07-27 21:49:58 +03:00
|
|
|
// applyV3 is the applier with auth and quotas
|
|
|
|
applyV3 applierV3
|
|
|
|
// applyV3Base is the core applier without auth or quotas
|
|
|
|
applyV3Base applierV3
|
|
|
|
kv mvcc.ConsistentWatchableKV
|
|
|
|
lessor lease.Lessor
|
|
|
|
bemu sync.Mutex
|
|
|
|
be backend.Backend
|
|
|
|
authStore auth.AuthStore
|
|
|
|
alarmStore *alarm.AlarmStore
|
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
|
2016-02-25 06:35:15 +03:00
|
|
|
// compactor is used to auto-compact the KV.
|
|
|
|
compactor *compactor.Periodic
|
2014-10-11 17:03:03 +04:00
|
|
|
|
2016-01-25 01:56:07 +03:00
|
|
|
// peerRt used to send requests (version, lease) to peers.
|
|
|
|
peerRt http.RoundTripper
|
|
|
|
reqIDGen *idutil.Generator
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
|
2015-05-08 21:10:12 +03:00
|
|
|
// forceVersionC is used to force the version monitor loop
|
|
|
|
// to detect the cluster version immediately.
|
|
|
|
forceVersionC chan struct{}
|
2015-12-08 18:52:54 +03:00
|
|
|
|
|
|
|
msgSnapC chan raftpb.Message
|
2016-04-21 18:48:11 +03:00
|
|
|
|
|
|
|
// wg is used to wait for the go routines that depends on the server state
|
|
|
|
// to exit when stopping the server.
|
|
|
|
wg sync.WaitGroup
|
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.
|
2016-04-08 23:52:19 +03:00
|
|
|
func NewServer(cfg *ServerConfig) (srv *EtcdServer, err error) {
|
2015-04-29 22:05:40 +03:00
|
|
|
st := store.New(StoreClusterPrefix, StoreKeysPrefix)
|
2016-04-08 23:52:19 +03:00
|
|
|
|
2015-12-08 18:52:54 +03:00
|
|
|
var (
|
|
|
|
w *wal.WAL
|
|
|
|
n raft.Node
|
|
|
|
s *raft.MemoryStorage
|
|
|
|
id types.ID
|
2016-04-07 22:02:37 +03:00
|
|
|
cl *membership.RaftCluster
|
2015-12-08 18:52:54 +03:00
|
|
|
)
|
2015-02-12 02:24:32 +03:00
|
|
|
|
2015-12-29 21:40:58 +03:00
|
|
|
if terr := fileutil.TouchDirAll(cfg.DataDir); terr != nil {
|
|
|
|
return nil, fmt.Errorf("cannot access data directory: %v", terr)
|
|
|
|
}
|
|
|
|
|
2015-03-25 01:13:34 +03:00
|
|
|
haveWAL := wal.Exist(cfg.WALDir())
|
2016-04-08 23:52:19 +03:00
|
|
|
|
2016-06-19 21:23:11 +03:00
|
|
|
if err = fileutil.TouchDirAll(cfg.SnapDir()); err != nil {
|
2016-04-08 23:52:19 +03:00
|
|
|
plog.Fatalf("create snapshot directory error: %v", err)
|
|
|
|
}
|
2015-02-13 20:51:48 +03:00
|
|
|
ss := snap.New(cfg.SnapDir())
|
2016-06-27 06:14:43 +03:00
|
|
|
|
|
|
|
bepath := path.Join(cfg.SnapDir(), databaseFilename)
|
|
|
|
beExist := fileutil.Exist(bepath)
|
|
|
|
be := backend.NewDefaultBackend(bepath)
|
2016-04-08 23:52:19 +03:00
|
|
|
defer func() {
|
|
|
|
if err != nil {
|
|
|
|
be.Close()
|
|
|
|
}
|
|
|
|
}()
|
2014-11-18 21:43:20 +03:00
|
|
|
|
2015-12-29 10:16:08 +03:00
|
|
|
prt, err := rafthttp.NewRoundTripper(cfg.PeerTLSInfo, cfg.peerDialTimeout())
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2015-10-11 09:04:10 +03:00
|
|
|
}
|
2016-06-27 20:27:27 +03:00
|
|
|
var (
|
|
|
|
remotes []*membership.Member
|
|
|
|
snapshot *raftpb.Snapshot
|
|
|
|
)
|
|
|
|
|
2014-10-24 03:41:58 +04:00
|
|
|
switch {
|
2014-11-06 22:09:33 +03:00
|
|
|
case !haveWAL && !cfg.NewCluster:
|
2016-04-08 23:52:19 +03:00
|
|
|
if err = cfg.VerifyJoinExisting(); err != nil {
|
2015-03-23 21:56:34 +03:00
|
|
|
return nil, err
|
|
|
|
}
|
2016-04-07 22:02:37 +03:00
|
|
|
cl, err = membership.NewClusterFromURLsMap(cfg.InitialClusterToken, cfg.InitialPeerURLsMap)
|
2015-04-27 05:49:01 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2016-04-08 23:52:19 +03:00
|
|
|
existingCluster, gerr := GetClusterFromRemotePeers(getRemotePeerURLs(cl, cfg.Name), prt)
|
|
|
|
if gerr != nil {
|
|
|
|
return nil, fmt.Errorf("cannot fetch cluster info from peer urls: %v", gerr)
|
2014-10-25 02:26:05 +04:00
|
|
|
}
|
2016-04-08 23:52:19 +03:00
|
|
|
if err = membership.ValidateClusterAndAssignIDs(cl, existingCluster); err != nil {
|
2014-11-18 01:08:45 +03:00
|
|
|
return nil, fmt.Errorf("error validating peerURLs %s: %v", existingCluster, err)
|
2014-10-24 03:41:58 +04:00
|
|
|
}
|
2015-11-04 21:49:42 +03:00
|
|
|
if !isCompatibleWithCluster(cl, cl.MemberByName(cfg.Name).ID, prt) {
|
2016-05-31 00:54:30 +03:00
|
|
|
return nil, fmt.Errorf("incompatible with current running cluster")
|
2015-05-14 17:57:25 +03:00
|
|
|
}
|
|
|
|
|
2015-04-21 09:48:36 +03:00
|
|
|
remotes = existingCluster.Members()
|
2016-04-07 22:02:37 +03:00
|
|
|
cl.SetID(existingCluster.ID())
|
2015-04-27 05:49:01 +03:00
|
|
|
cl.SetStore(st)
|
2016-04-08 23:52:19 +03:00
|
|
|
cl.SetBackend(be)
|
2014-11-13 07:11:04 +03:00
|
|
|
cfg.Print()
|
2015-04-27 05:49:01 +03:00
|
|
|
id, n, s, w = startNode(cfg, cl, nil)
|
2014-11-06 22:09:33 +03:00
|
|
|
case !haveWAL && cfg.NewCluster:
|
2016-04-08 23:52:19 +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
|
|
|
}
|
2016-04-07 22:02:37 +03:00
|
|
|
cl, err = membership.NewClusterFromURLsMap(cfg.InitialClusterToken, cfg.InitialPeerURLsMap)
|
2015-04-27 05:49:01 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
m := cl.MemberByName(cfg.Name)
|
2016-02-03 01:31:50 +03:00
|
|
|
if isMemberBootstrapped(cl, cfg.Name, prt, cfg.bootstrapTimeout()) {
|
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-12-29 10:16:08 +03:00
|
|
|
var str string
|
|
|
|
str, err = discovery.JoinCluster(cfg.DiscoveryURL, cfg.DiscoveryProxy, m.ID, cfg.InitialPeerURLsMap.String())
|
|
|
|
if err != nil {
|
|
|
|
return nil, &DiscoveryError{Op: "join", Err: err}
|
2014-10-04 16:20:45 +04:00
|
|
|
}
|
2016-04-15 09:56:26 +03:00
|
|
|
var urlsmap types.URLsMap
|
|
|
|
urlsmap, err = types.NewURLsMap(str)
|
2015-12-29 10:16:08 +03:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
2014-10-04 16:20:45 +04:00
|
|
|
}
|
2015-04-27 05:49:01 +03:00
|
|
|
if checkDuplicateURL(urlsmap) {
|
|
|
|
return nil, fmt.Errorf("discovery cluster %s has duplicate url", urlsmap)
|
|
|
|
}
|
2016-04-07 22:02:37 +03:00
|
|
|
if cl, err = membership.NewClusterFromURLsMap(cfg.InitialClusterToken, urlsmap); err != nil {
|
2015-04-27 05:49:01 +03:00
|
|
|
return nil, err
|
2015-02-14 01:32:24 +03:00
|
|
|
}
|
2014-10-03 04:06:35 +04:00
|
|
|
}
|
2015-04-27 05:49:01 +03:00
|
|
|
cl.SetStore(st)
|
2016-04-08 23:52:19 +03:00
|
|
|
cl.SetBackend(be)
|
2014-11-13 07:11:04 +03:00
|
|
|
cfg.PrintWithInitial()
|
2015-04-27 05:49:01 +03:00
|
|
|
id, n, s, w = startNode(cfg, cl, cl.MemberIDs())
|
2014-10-24 03:41:58 +04:00
|
|
|
case haveWAL:
|
2016-04-08 23:52:19 +03:00
|
|
|
if err = fileutil.IsDirWriteable(cfg.MemberDir()); err != nil {
|
2015-02-12 20:45:12 +03:00
|
|
|
return nil, fmt.Errorf("cannot write to member directory: %v", err)
|
|
|
|
}
|
|
|
|
|
2016-04-08 23:52:19 +03:00
|
|
|
if err = fileutil.IsDirWriteable(cfg.WALDir()); err != nil {
|
2015-09-01 01:01:18 +03:00
|
|
|
return nil, fmt.Errorf("cannot write to WAL directory: %v", err)
|
|
|
|
}
|
|
|
|
|
2014-10-11 17:47:44 +04:00
|
|
|
if cfg.ShouldDiscover() {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Warningf("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
|
|
|
}
|
2015-12-29 10:16:08 +03:00
|
|
|
snapshot, err = ss.Load()
|
|
|
|
if err != nil && err != snap.ErrNoSnapshot {
|
|
|
|
return nil, err
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
|
|
|
if snapshot != nil {
|
2016-04-15 09:56:26 +03:00
|
|
|
if err = st.Recovery(snapshot.Data); err != nil {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("recovered store from snapshot error: %v", err)
|
2014-12-09 01:53:39 +03:00
|
|
|
}
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("recovered store from snapshot at index %d", snapshot.Metadata.Index)
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
2014-11-13 07:11:04 +03:00
|
|
|
cfg.Print()
|
2014-11-07 08:50:35 +03:00
|
|
|
if !cfg.ForceNewCluster {
|
2015-04-27 05:49:01 +03:00
|
|
|
id, cl, n, s, w = restartNode(cfg, snapshot)
|
2014-11-07 08:50:35 +03:00
|
|
|
} else {
|
2015-04-27 05:49:01 +03:00
|
|
|
id, cl, n, s, w = restartAsStandaloneNode(cfg, snapshot)
|
2014-11-07 08:50:35 +03:00
|
|
|
}
|
2015-04-27 05:49:01 +03:00
|
|
|
cl.SetStore(st)
|
2016-04-08 23:52:19 +03:00
|
|
|
cl.SetBackend(be)
|
2016-07-26 21:33:11 +03:00
|
|
|
cl.Recover(api.UpdateCapability)
|
2016-07-01 19:10:01 +03:00
|
|
|
if cl.Version() != nil && !cl.Version().LessThan(semver.Version{Major: 3}) && !beExist {
|
2016-06-27 06:14:43 +03:00
|
|
|
os.RemoveAll(bepath)
|
|
|
|
return nil, fmt.Errorf("database file (%v) of the backend is missing", bepath)
|
|
|
|
}
|
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
|
|
|
}
|
|
|
|
|
2015-12-29 21:40:58 +03:00
|
|
|
if terr := fileutil.TouchDirAll(cfg.MemberDir()); terr != nil {
|
|
|
|
return nil, fmt.Errorf("cannot access member directory: %v", terr)
|
2015-12-29 10:17:54 +03: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
|
|
|
}
|
2015-04-22 18:05:50 +03:00
|
|
|
sstats.Initialize()
|
2014-10-31 03:31:14 +03:00
|
|
|
lstats := stats.NewLeaderStats(id.String())
|
2014-10-11 02:53:28 +04:00
|
|
|
|
2016-04-08 23:52:19 +03:00
|
|
|
srv = &EtcdServer{
|
2016-04-23 05:57:33 +03:00
|
|
|
readych: make(chan struct{}),
|
2016-05-12 22:34:03 +03:00
|
|
|
Cfg: cfg,
|
2015-03-06 00:03:04 +03:00
|
|
|
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),
|
|
|
|
},
|
2015-05-08 21:10:12 +03:00
|
|
|
id: id,
|
2016-04-07 22:02:37 +03:00
|
|
|
attributes: membership.Attributes{Name: cfg.Name, ClientURLs: cfg.ClientURLs.StringSlice()},
|
2015-05-13 03:22:06 +03:00
|
|
|
cluster: cl,
|
2015-05-08 21:10:12 +03:00
|
|
|
stats: sstats,
|
|
|
|
lstats: lstats,
|
|
|
|
SyncTicker: time.Tick(500 * time.Millisecond),
|
2016-01-25 01:56:07 +03:00
|
|
|
peerRt: prt,
|
2016-02-05 00:06:27 +03:00
|
|
|
reqIDGen: idutil.NewGenerator(uint16(id), time.Now()),
|
2015-05-08 21:10:12 +03:00
|
|
|
forceVersionC: make(chan struct{}),
|
2015-12-08 18:52:54 +03:00
|
|
|
msgSnapC: make(chan raftpb.Message, maxInFlightMsgSnap),
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
2014-12-29 23:08:13 +03:00
|
|
|
|
2016-05-17 00:40:52 +03:00
|
|
|
srv.applyV2 = &applierV2store{store: srv.store, cluster: srv.cluster}
|
2016-04-20 10:31:58 +03:00
|
|
|
|
2016-04-08 23:52:19 +03:00
|
|
|
srv.be = be
|
2016-08-02 20:29:23 +03:00
|
|
|
minTTL := time.Duration((3*cfg.ElectionTicks)/2) * time.Duration(cfg.TickMs) * time.Millisecond
|
|
|
|
srv.lessor = lease.NewLessor(srv.be, int64(math.Ceil(minTTL.Seconds())))
|
2016-04-25 22:32:58 +03:00
|
|
|
srv.kv = mvcc.New(srv.be, srv.lessor, &srv.consistIndex)
|
2016-06-27 20:27:27 +03:00
|
|
|
if beExist {
|
|
|
|
kvindex := srv.kv.ConsistentIndex()
|
|
|
|
if snapshot != nil && kvindex < snapshot.Metadata.Index {
|
|
|
|
return nil, fmt.Errorf("database file (%v index %d) does not match with snapshot (index %d).", bepath, kvindex, snapshot.Metadata.Index)
|
|
|
|
}
|
|
|
|
}
|
2016-03-30 21:37:55 +03:00
|
|
|
srv.consistIndex.setConsistentIndex(srv.kv.ConsistentIndex())
|
2016-03-24 02:39:15 +03:00
|
|
|
srv.authStore = auth.NewAuthStore(srv.be)
|
|
|
|
if h := cfg.AutoCompactionRetention; h != 0 {
|
|
|
|
srv.compactor = compactor.NewPeriodic(h, srv.kv, srv)
|
|
|
|
srv.compactor.Run()
|
2015-08-08 15:58:29 +03:00
|
|
|
}
|
2016-03-24 03:20:52 +03:00
|
|
|
|
2016-07-27 21:49:58 +03:00
|
|
|
srv.applyV3Base = &applierV3backend{srv}
|
2016-04-08 23:52:19 +03:00
|
|
|
if err = srv.restoreAlarms(); err != nil {
|
2016-03-26 01:14:32 +03:00
|
|
|
return nil, err
|
2016-03-24 03:20:52 +03:00
|
|
|
}
|
2015-08-08 15:58:29 +03:00
|
|
|
|
2015-04-21 09:48:36 +03:00
|
|
|
// TODO: move transport initialization near the definition of remote
|
2015-10-11 01:58:02 +03:00
|
|
|
tr := &rafthttp.Transport{
|
2015-10-11 09:04:10 +03:00
|
|
|
TLSInfo: cfg.PeerTLSInfo,
|
|
|
|
DialTimeout: cfg.peerDialTimeout(),
|
|
|
|
ID: id,
|
2016-02-05 09:05:59 +03:00
|
|
|
URLs: cfg.PeerURLs,
|
2015-10-11 09:04:10 +03:00
|
|
|
ClusterID: cl.ID(),
|
|
|
|
Raft: srv,
|
2015-12-08 18:52:54 +03:00
|
|
|
Snapshotter: ss,
|
2015-10-11 09:04:10 +03:00
|
|
|
ServerStats: sstats,
|
|
|
|
LeaderStats: lstats,
|
|
|
|
ErrorC: srv.errorc,
|
|
|
|
}
|
2016-04-08 23:52:19 +03:00
|
|
|
if err = tr.Start(); err != nil {
|
2015-10-11 09:04:10 +03:00
|
|
|
return nil, err
|
2015-10-11 01:58:02 +03:00
|
|
|
}
|
2015-04-21 09:48:36 +03:00
|
|
|
// add all remotes into transport
|
|
|
|
for _, m := range remotes {
|
|
|
|
if m.ID != id {
|
|
|
|
tr.AddRemote(m.ID, m.PeerURLs)
|
|
|
|
}
|
|
|
|
}
|
2015-04-27 05:49:01 +03:00
|
|
|
for _, m := range cl.Members() {
|
2015-04-19 21:08:03 +03:00
|
|
|
if m.ID != id {
|
|
|
|
tr.AddPeer(m.ID, m.PeerURLs)
|
|
|
|
}
|
|
|
|
}
|
2015-01-16 01:39:30 +03:00
|
|
|
srv.r.transport = tr
|
2015-10-11 21:49:52 +03:00
|
|
|
|
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()
|
2016-05-12 22:34:03 +03:00
|
|
|
go s.publish(s.Cfg.ReqTimeout())
|
2014-11-29 08:32:28 +03:00
|
|
|
go s.purgeFile()
|
2015-04-04 00:26:15 +03:00
|
|
|
go monitorFileDescriptor(s.done)
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
go s.monitorVersions()
|
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 {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("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{})
|
2015-05-08 21:10:12 +03:00
|
|
|
if s.ClusterVersion() != nil {
|
2015-08-15 18:30:06 +03:00
|
|
|
plog.Infof("starting server... [version: %v, cluster version: %v]", version.Version, version.Cluster(s.ClusterVersion().String()))
|
2015-05-08 21:10:12 +03:00
|
|
|
} else {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("starting server... [version: %v, cluster version: to_be_decided]", version.Version)
|
2015-05-08 21:10:12 +03:00
|
|
|
}
|
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
|
2016-05-12 22:34:03 +03:00
|
|
|
if s.Cfg.MaxSnapFiles > 0 {
|
|
|
|
serrc = fileutil.PurgeFile(s.Cfg.SnapDir(), "snap", s.Cfg.MaxSnapFiles, purgeFileInterval, s.done)
|
2014-11-29 08:32:28 +03:00
|
|
|
}
|
2016-05-12 22:34:03 +03:00
|
|
|
if s.Cfg.MaxWALFiles > 0 {
|
|
|
|
werrc = fileutil.PurgeFile(s.Cfg.WALDir(), "wal", s.Cfg.MaxWALFiles, purgeFileInterval, s.done)
|
2014-11-29 08:32:28 +03:00
|
|
|
}
|
|
|
|
select {
|
|
|
|
case e := <-werrc:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Fatalf("failed to purge wal file %v", e)
|
2014-11-29 08:32:28 +03:00
|
|
|
case e := <-serrc:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Fatalf("failed to purge snap file %v", e)
|
2014-11-29 08:32:28 +03:00
|
|
|
case <-s.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-11-13 22:26:49 +03:00
|
|
|
func (s *EtcdServer) ID() types.ID { return s.id }
|
|
|
|
|
2016-04-07 22:02:37 +03:00
|
|
|
func (s *EtcdServer) Cluster() *membership.RaftCluster { return s.cluster }
|
2015-05-13 03:22:06 +03:00
|
|
|
|
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
|
|
|
|
2016-01-20 08:09:09 +03:00
|
|
|
func (s *EtcdServer) Lessor() lease.Lessor { return s.lessor }
|
|
|
|
|
2014-09-12 04:14:41 +04:00
|
|
|
func (s *EtcdServer) Process(ctx context.Context, m raftpb.Message) error {
|
2015-05-13 03:22:06 +03:00
|
|
|
if s.cluster.IsIDRemoved(types.ID(m.From)) {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Warningf("reject message from removed member %s", types.ID(m.From).String())
|
2014-11-17 03:23:47 +03:00
|
|
|
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-05-13 03:22:06 +03:00
|
|
|
func (s *EtcdServer) IsIDRemoved(id uint64) bool { return s.cluster.IsIDRemoved(types.ID(id)) }
|
2015-04-26 09:24:59 +03:00
|
|
|
|
2015-03-01 18:02:24 +03:00
|
|
|
func (s *EtcdServer) ReportUnreachable(id uint64) { s.r.ReportUnreachable(id) }
|
|
|
|
|
2015-10-14 03:29:01 +03:00
|
|
|
// ReportSnapshot reports snapshot sent status to the raft state machine,
|
|
|
|
// and clears the used snapshot from the snapshot store.
|
2015-03-01 19:17:14 +03:00
|
|
|
func (s *EtcdServer) ReportSnapshot(id uint64, status raft.SnapshotStatus) {
|
|
|
|
s.r.ReportSnapshot(id, status)
|
|
|
|
}
|
|
|
|
|
2015-12-22 14:41:22 +03:00
|
|
|
type etcdProgress struct {
|
|
|
|
confState raftpb.ConfState
|
|
|
|
snapi uint64
|
|
|
|
appliedi uint64
|
|
|
|
}
|
|
|
|
|
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 {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("get snapshot from raft storage error: %v", err)
|
2014-12-09 03:13:16 +03:00
|
|
|
}
|
2015-07-11 02:31:22 +03:00
|
|
|
s.r.start(s)
|
2014-12-09 03:13:16 +03:00
|
|
|
|
2015-12-22 14:42:56 +03:00
|
|
|
// asynchronously accept apply packets, dispatch progress in-order
|
2016-02-12 06:21:30 +03:00
|
|
|
sched := schedule.NewFIFOScheduler()
|
|
|
|
ep := etcdProgress{
|
2015-12-22 14:41:22 +03:00
|
|
|
confState: snap.Metadata.ConfState,
|
|
|
|
snapi: snap.Metadata.Index,
|
|
|
|
appliedi: snap.Metadata.Index,
|
2016-02-12 06:21:30 +03:00
|
|
|
}
|
2015-12-22 14:42:56 +03:00
|
|
|
|
|
|
|
defer func() {
|
2016-02-12 06:21:30 +03:00
|
|
|
sched.Stop()
|
|
|
|
|
2016-05-18 00:16:29 +03:00
|
|
|
// wait for snapshots before closing raft so wal stays open
|
|
|
|
s.wg.Wait()
|
|
|
|
|
2016-04-23 03:13:57 +03:00
|
|
|
// must stop raft after scheduler-- etcdserver can leak rafthttp pipelines
|
|
|
|
// by adding a peer after raft stops the transport
|
|
|
|
s.r.stop()
|
|
|
|
|
2016-01-12 20:00:57 +03:00
|
|
|
// kv, lessor and backend can be nil if running without v3 enabled
|
2016-01-07 03:25:25 +03:00
|
|
|
// or running unit tests.
|
2016-01-12 20:00:57 +03:00
|
|
|
if s.lessor != nil {
|
|
|
|
s.lessor.Stop()
|
|
|
|
}
|
2016-01-07 03:25:25 +03:00
|
|
|
if s.kv != nil {
|
|
|
|
s.kv.Close()
|
|
|
|
}
|
|
|
|
if s.be != nil {
|
|
|
|
s.be.Close()
|
|
|
|
}
|
2016-02-25 06:35:15 +03:00
|
|
|
if s.compactor != nil {
|
|
|
|
s.compactor.Stop()
|
|
|
|
}
|
2015-12-22 14:42:56 +03:00
|
|
|
close(s.done)
|
|
|
|
}()
|
2015-12-22 14:41:22 +03:00
|
|
|
|
2016-01-09 00:17:54 +03:00
|
|
|
var expiredLeaseC <-chan []*lease.Lease
|
|
|
|
if s.lessor != nil {
|
|
|
|
expiredLeaseC = s.lessor.ExpiredLeasesC()
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
2016-02-12 06:21:30 +03:00
|
|
|
case ap := <-s.r.apply():
|
2016-07-12 23:16:35 +03:00
|
|
|
var ci uint64
|
|
|
|
if len(ap.entries) != 0 {
|
|
|
|
ci = ap.entries[len(ap.entries)-1].Index
|
|
|
|
}
|
|
|
|
if ap.snapshot.Metadata.Index > ci {
|
|
|
|
ci = ap.snapshot.Metadata.Index
|
|
|
|
}
|
|
|
|
if ci != 0 {
|
|
|
|
s.setCommittedIndex(ci)
|
|
|
|
}
|
2016-02-12 06:21:30 +03:00
|
|
|
f := func(context.Context) { s.applyAll(&ep, &ap) }
|
|
|
|
sched.Schedule(f)
|
2016-01-09 00:17:54 +03:00
|
|
|
case leases := <-expiredLeaseC:
|
|
|
|
go func() {
|
|
|
|
for _, l := range leases {
|
|
|
|
s.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: int64(l.ID)})
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
case err := <-s.errorc:
|
|
|
|
plog.Errorf("%s", err)
|
|
|
|
plog.Infof("the data-dir used by this member must be removed.")
|
|
|
|
return
|
|
|
|
case <-s.stop:
|
|
|
|
return
|
|
|
|
}
|
2014-08-26 05:39:02 +04:00
|
|
|
}
|
2015-12-22 14:42:56 +03:00
|
|
|
}
|
|
|
|
|
|
|
|
func (s *EtcdServer) applyAll(ep *etcdProgress, apply *apply) {
|
|
|
|
s.applySnapshot(ep, apply)
|
2016-06-10 21:17:47 +03:00
|
|
|
st := time.Now()
|
2015-12-22 14:42:56 +03:00
|
|
|
s.applyEntries(ep, apply)
|
2016-06-10 21:17:47 +03:00
|
|
|
d := time.Since(st)
|
|
|
|
entriesNum := len(apply.entries)
|
|
|
|
if entriesNum != 0 && d > time.Duration(entriesNum)*warnApplyDuration {
|
|
|
|
plog.Warningf("apply entries took too long [%v for %d entries]", d, len(apply.entries))
|
|
|
|
plog.Warningf("avoid queries with large range/delete range!")
|
|
|
|
}
|
2016-06-17 21:35:51 +03:00
|
|
|
proposalsApplied.Set(float64(ep.appliedi))
|
2015-12-22 14:42:56 +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.
|
2015-12-22 14:40:39 +03:00
|
|
|
<-apply.raftDone
|
2015-12-22 14:42:56 +03:00
|
|
|
s.triggerSnapshot(ep)
|
2016-02-12 06:21:30 +03:00
|
|
|
select {
|
|
|
|
// snapshot requested via send()
|
|
|
|
case m := <-s.msgSnapC:
|
|
|
|
merged := s.createMergedSnapshotMessage(m, ep.appliedi, ep.confState)
|
|
|
|
s.sendMergedSnap(merged)
|
|
|
|
default:
|
|
|
|
}
|
2014-08-26 05:39:02 +04:00
|
|
|
}
|
|
|
|
|
2015-12-22 14:41:22 +03:00
|
|
|
func (s *EtcdServer) applySnapshot(ep *etcdProgress, apply *apply) {
|
|
|
|
if raft.IsEmptySnap(apply.snapshot) {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Infof("applying snapshot at index %d...", ep.snapi)
|
|
|
|
defer plog.Infof("finished applying incoming snapshot at index %d", ep.snapi)
|
|
|
|
|
2015-12-22 14:41:22 +03:00
|
|
|
if apply.snapshot.Metadata.Index <= ep.appliedi {
|
|
|
|
plog.Panicf("snapshot index [%d] should > appliedi[%d] + 1",
|
|
|
|
apply.snapshot.Metadata.Index, ep.appliedi)
|
|
|
|
}
|
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
snapfn, err := s.r.storage.DBFilePath(apply.snapshot.Metadata.Index)
|
|
|
|
if err != nil {
|
|
|
|
plog.Panicf("get database snapshot file path error: %v", err)
|
|
|
|
}
|
2015-12-22 14:41:22 +03:00
|
|
|
|
2016-05-12 22:34:03 +03:00
|
|
|
fn := path.Join(s.Cfg.SnapDir(), databaseFilename)
|
2016-03-24 02:39:15 +03:00
|
|
|
if err := os.Rename(snapfn, fn); err != nil {
|
|
|
|
plog.Panicf("rename snapshot file error: %v", err)
|
|
|
|
}
|
2015-12-22 14:41:22 +03:00
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
newbe := backend.NewDefaultBackend(fn)
|
2016-06-01 07:12:41 +03:00
|
|
|
|
|
|
|
plog.Info("restoring mvcc store...")
|
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
if err := s.kv.Restore(newbe); err != nil {
|
|
|
|
plog.Panicf("restore KV error: %v", err)
|
|
|
|
}
|
2016-03-30 21:37:55 +03:00
|
|
|
s.consistIndex.setConsistentIndex(s.kv.ConsistentIndex())
|
2015-12-22 14:41:22 +03:00
|
|
|
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished restoring mvcc store")
|
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
// Closing old backend might block until all the txns
|
|
|
|
// on the backend are finished.
|
|
|
|
// We do not want to wait on closing the old backend.
|
|
|
|
s.bemu.Lock()
|
|
|
|
oldbe := s.be
|
|
|
|
go func() {
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("closing old backend...")
|
|
|
|
defer plog.Info("finished closing old backend")
|
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
if err := oldbe.Close(); err != nil {
|
|
|
|
plog.Panicf("close backend error: %v", err)
|
|
|
|
}
|
|
|
|
}()
|
2016-01-07 03:25:25 +03:00
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
s.be = newbe
|
|
|
|
s.bemu.Unlock()
|
2016-03-11 04:06:10 +03:00
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
if s.lessor != nil {
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("recovering lessor...")
|
2016-03-24 02:39:15 +03:00
|
|
|
s.lessor.Recover(newbe, s.kv)
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished recovering lessor")
|
2016-03-24 02:39:15 +03:00
|
|
|
}
|
2016-03-21 16:43:41 +03:00
|
|
|
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("recovering alarms...")
|
2016-03-26 01:14:32 +03:00
|
|
|
if err := s.restoreAlarms(); err != nil {
|
|
|
|
plog.Panicf("restore alarms error: %v", err)
|
|
|
|
}
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished recovering alarms")
|
2016-03-26 01:14:32 +03:00
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
if s.authStore != nil {
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("recovering auth store...")
|
2016-03-24 02:39:15 +03:00
|
|
|
s.authStore.Recover(newbe)
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished recovering auth store")
|
2015-12-22 14:41:22 +03:00
|
|
|
}
|
2016-03-24 02:39:15 +03:00
|
|
|
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("recovering store v2...")
|
2015-12-22 14:41:22 +03:00
|
|
|
if err := s.store.Recovery(apply.snapshot.Data); err != nil {
|
|
|
|
plog.Panicf("recovery store error: %v", err)
|
|
|
|
}
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished recovering store v2")
|
|
|
|
|
2016-04-08 23:52:19 +03:00
|
|
|
s.cluster.SetBackend(s.be)
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("recovering cluster configuration...")
|
2016-07-26 21:33:11 +03:00
|
|
|
s.cluster.Recover(api.UpdateCapability)
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished recovering cluster configuration")
|
2015-12-22 14:41:22 +03:00
|
|
|
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("removing old peers from network...")
|
2015-12-22 14:41:22 +03:00
|
|
|
// recover raft transport
|
|
|
|
s.r.transport.RemoveAllPeers()
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished removing old peers from network")
|
|
|
|
|
|
|
|
plog.Info("adding peers from new cluster configuration into network...")
|
2015-12-22 14:41:22 +03:00
|
|
|
for _, m := range s.cluster.Members() {
|
|
|
|
if m.ID == s.ID() {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
s.r.transport.AddPeer(m.ID, m.PeerURLs)
|
|
|
|
}
|
2016-06-01 07:12:41 +03:00
|
|
|
plog.Info("finished adding peers from new cluster configuration into network...")
|
2015-12-22 14:41:22 +03:00
|
|
|
|
|
|
|
ep.appliedi = apply.snapshot.Metadata.Index
|
|
|
|
ep.snapi = ep.appliedi
|
|
|
|
ep.confState = apply.snapshot.Metadata.ConfState
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *EtcdServer) applyEntries(ep *etcdProgress, apply *apply) {
|
|
|
|
if len(apply.entries) == 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
firsti := apply.entries[0].Index
|
|
|
|
if firsti > ep.appliedi+1 {
|
|
|
|
plog.Panicf("first index of committed entry[%d] should <= appliedi[%d] + 1", firsti, ep.appliedi)
|
|
|
|
}
|
|
|
|
var ents []raftpb.Entry
|
|
|
|
if ep.appliedi+1-firsti < uint64(len(apply.entries)) {
|
|
|
|
ents = apply.entries[ep.appliedi+1-firsti:]
|
|
|
|
}
|
2016-01-26 21:19:39 +03:00
|
|
|
if len(ents) == 0 {
|
|
|
|
return
|
|
|
|
}
|
2015-12-22 14:41:22 +03:00
|
|
|
var shouldstop bool
|
|
|
|
if ep.appliedi, shouldstop = s.apply(ents, &ep.confState); shouldstop {
|
|
|
|
go s.stopWithDelay(10*100*time.Millisecond, fmt.Errorf("the member has been permanently removed from the cluster"))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *EtcdServer) triggerSnapshot(ep *etcdProgress) {
|
|
|
|
if ep.appliedi-ep.snapi <= s.snapCount {
|
|
|
|
return
|
|
|
|
}
|
2015-12-24 03:03:23 +03:00
|
|
|
|
2015-12-22 14:41:22 +03:00
|
|
|
plog.Infof("start to snapshot (applied: %d, lastsnap: %d)", ep.appliedi, ep.snapi)
|
|
|
|
s.snapshot(ep.appliedi, ep.confState)
|
|
|
|
ep.snapi = ep.appliedi
|
|
|
|
}
|
|
|
|
|
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
|
|
|
|
2016-04-23 05:57:33 +03:00
|
|
|
// ReadyNotify returns a channel that will be closed when the server
|
|
|
|
// is ready to serve client requests
|
|
|
|
func (s *EtcdServer) ReadyNotify() <-chan struct{} { return s.readych }
|
|
|
|
|
2015-01-03 07:00:29 +03:00
|
|
|
func (s *EtcdServer) stopWithDelay(d time.Duration, err error) {
|
2016-01-26 06:44:18 +03:00
|
|
|
select {
|
|
|
|
case <-time.After(d):
|
|
|
|
case <-s.done:
|
|
|
|
}
|
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-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
|
|
|
|
2016-04-07 22:02:37 +03:00
|
|
|
func (s *EtcdServer) AddMember(ctx context.Context, memb membership.Member) error {
|
2016-05-12 22:34:03 +03:00
|
|
|
if s.Cfg.StrictReconfigCheck && !s.cluster.IsReadyToAddNewMember() {
|
2016-03-21 20:54:27 +03:00
|
|
|
// If s.cfg.StrictReconfigCheck is false, it means the option --strict-reconfig-check isn't passed to etcd.
|
2015-09-10 09:45:34 +03:00
|
|
|
// In such a case adding a new member is allowed unconditionally
|
|
|
|
return ErrNotEnoughStartedMembers
|
|
|
|
}
|
|
|
|
|
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 {
|
2016-05-12 22:34:03 +03:00
|
|
|
if s.Cfg.StrictReconfigCheck && !s.cluster.IsReadyToRemoveMember(id) {
|
2016-03-21 20:54:27 +03:00
|
|
|
// If s.cfg.StrictReconfigCheck is false, it means the option --strict-reconfig-check isn't passed to etcd.
|
2015-09-16 07:18:05 +03:00
|
|
|
// In such a case removing a member is allowed unconditionally
|
|
|
|
return ErrNotEnoughStartedMembers
|
|
|
|
}
|
|
|
|
|
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
|
|
|
}
|
|
|
|
|
2016-04-07 22:02:37 +03:00
|
|
|
func (s *EtcdServer) UpdateMember(ctx context.Context, memb membership.Member) error {
|
2014-11-11 22:46:49 +03:00
|
|
|
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
|
2016-02-21 16:05:03 +03:00
|
|
|
|
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
|
|
|
|
2016-02-21 16:05:03 +03:00
|
|
|
// Lead is only for testing purposes.
|
2014-11-16 08:28:03 +03:00
|
|
|
// 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()) }
|
|
|
|
|
2016-05-12 22:34:03 +03:00
|
|
|
func (s *EtcdServer) IsPprofEnabled() bool { return s.Cfg.EnablePprof }
|
2016-01-07 10:15:27 +03:00
|
|
|
|
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-08-12 23:38:43 +03:00
|
|
|
start := time.Now()
|
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 {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.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
|
2015-08-12 23:38:43 +03:00
|
|
|
return s.parseProposeCtxErr(ctx.Err(), start)
|
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.
|
2016-01-08 11:21:19 +03:00
|
|
|
// The request will be canceled after the given timeout.
|
2014-09-16 04:35:02 +04:00
|
|
|
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.
|
2015-07-23 20:09:28 +03:00
|
|
|
func (s *EtcdServer) publish(timeout 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 {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("json marshal error: %v", err)
|
2014-09-27 01:18:59 +04:00
|
|
|
return
|
|
|
|
}
|
|
|
|
req := pb.Request{
|
|
|
|
Method: "PUT",
|
2016-04-07 22:02:37 +03:00
|
|
|
Path: membership.MemberAttributesStorePath(s.id),
|
2014-09-27 01:18:59 +04:00
|
|
|
Val: string(b),
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
2015-07-23 20:09:28 +03:00
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
2014-09-27 01:18:59 +04:00
|
|
|
_, err := s.Do(ctx, req)
|
|
|
|
cancel()
|
|
|
|
switch err {
|
|
|
|
case nil:
|
2016-04-23 05:57:33 +03:00
|
|
|
close(s.readych)
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("published %+v to cluster %s", s.attributes, s.cluster.ID())
|
2014-09-27 01:18:59 +04:00
|
|
|
return
|
|
|
|
case ErrStopped:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("aborting publish because server is stopped")
|
2014-09-27 01:18:59 +04:00
|
|
|
return
|
|
|
|
default:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Errorf("publish error: %v", err)
|
2014-09-27 01:18:59 +04:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-03-01 03:04:14 +03:00
|
|
|
// TODO: move this function into raft.go
|
2014-11-26 08:38:57 +03:00
|
|
|
func (s *EtcdServer) send(ms []raftpb.Message) {
|
2016-04-10 09:30:00 +03:00
|
|
|
sentAppResp := false
|
|
|
|
for i := len(ms) - 1; i >= 0; i-- {
|
2015-05-13 03:22:06 +03:00
|
|
|
if s.cluster.IsIDRemoved(types.ID(ms[i].To)) {
|
2015-04-13 15:25:16 +03:00
|
|
|
ms[i].To = 0
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
2015-12-08 18:52:54 +03:00
|
|
|
|
2016-04-10 09:30:00 +03:00
|
|
|
if ms[i].Type == raftpb.MsgAppResp {
|
|
|
|
if sentAppResp {
|
|
|
|
ms[i].To = 0
|
|
|
|
} else {
|
|
|
|
sentAppResp = true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-03-24 02:39:15 +03:00
|
|
|
if ms[i].Type == raftpb.MsgSnap {
|
|
|
|
// There are two separate data store: the store for v2, and the KV for v3.
|
|
|
|
// The msgSnap only contains the most recent snapshot of store without KV.
|
|
|
|
// So we need to redirect the msgSnap to etcd server main loop for merging in the
|
|
|
|
// current store snapshot and KV snapshot.
|
|
|
|
select {
|
|
|
|
case s.msgSnapC <- ms[i]:
|
|
|
|
default:
|
|
|
|
// drop msgSnap if the inflight chan if full.
|
2015-12-08 18:52:54 +03:00
|
|
|
}
|
2016-03-24 02:39:15 +03:00
|
|
|
ms[i].To = 0
|
2015-12-08 18:52:54 +03:00
|
|
|
}
|
2016-03-01 03:04:14 +03:00
|
|
|
if ms[i].Type == raftpb.MsgHeartbeat {
|
|
|
|
ok, exceed := s.r.td.Observe(ms[i].To)
|
|
|
|
if !ok {
|
|
|
|
// TODO: limit request rate.
|
2016-05-12 22:34:03 +03:00
|
|
|
plog.Warningf("failed to send out heartbeat on time (exceeded the %dms timeout for %v)", s.Cfg.TickMs, exceed)
|
2016-03-01 03:04:14 +03:00
|
|
|
plog.Warningf("server is likely overloaded")
|
|
|
|
}
|
|
|
|
}
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
2015-12-08 18:52:54 +03:00
|
|
|
|
2015-01-16 01:39:30 +03:00
|
|
|
s.r.transport.Send(ms)
|
2014-11-26 08:38:57 +03:00
|
|
|
}
|
|
|
|
|
2015-12-24 03:03:23 +03:00
|
|
|
func (s *EtcdServer) sendMergedSnap(merged snap.Message) {
|
|
|
|
atomic.AddInt64(&s.inflightSnapshots, 1)
|
|
|
|
|
|
|
|
s.r.transport.SendSnapshot(merged)
|
|
|
|
go func() {
|
|
|
|
select {
|
|
|
|
case ok := <-merged.CloseNotify():
|
|
|
|
// delay releasing inflight snapshot for another 30 seconds to
|
|
|
|
// block log compaction.
|
|
|
|
// If the follower still fails to catch up, it is probably just too slow
|
|
|
|
// to catch up. We cannot avoid the snapshot cycle anyway.
|
|
|
|
if ok {
|
2015-12-30 00:03:12 +03:00
|
|
|
select {
|
|
|
|
case <-time.After(releaseDelayAfterSnapshot):
|
|
|
|
case <-s.done:
|
|
|
|
}
|
2015-12-24 03:03:23 +03:00
|
|
|
}
|
|
|
|
atomic.AddInt64(&s.inflightSnapshots, -1)
|
|
|
|
case <-s.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
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
|
2014-10-11 17:58:39 +04:00
|
|
|
for i := range es {
|
|
|
|
e := es[i]
|
|
|
|
switch e.Type {
|
|
|
|
case raftpb.EntryNormal:
|
2016-04-20 10:31:58 +03:00
|
|
|
s.applyEntryNormal(&e)
|
2014-10-11 17:58:39 +04:00
|
|
|
case raftpb.EntryConfChange:
|
|
|
|
var cc raftpb.ConfChange
|
2014-10-11 16:41:00 +04:00
|
|
|
pbutil.MustUnmarshal(&cc, e.Data)
|
2015-12-24 02:51:00 +03:00
|
|
|
removedSelf, err := s.applyConfChange(cc, confState)
|
|
|
|
shouldstop = shouldstop || removedSelf
|
2014-11-14 10:03:34 +03:00
|
|
|
s.w.Trigger(cc.ID, err)
|
2014-10-11 17:58:39 +04:00
|
|
|
default:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.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
|
|
|
}
|
|
|
|
|
2016-04-20 10:31:58 +03:00
|
|
|
// applyEntryNormal apples an EntryNormal type raftpb request to the EtcdServer
|
|
|
|
func (s *EtcdServer) applyEntryNormal(e *raftpb.Entry) {
|
2016-06-27 20:27:27 +03:00
|
|
|
shouldApplyV3 := false
|
|
|
|
if e.Index > s.consistIndex.ConsistentIndex() {
|
|
|
|
// set the consistent index of current executing entry
|
|
|
|
s.consistIndex.setConsistentIndex(e.Index)
|
|
|
|
shouldApplyV3 = true
|
|
|
|
}
|
2016-07-16 20:24:56 +03:00
|
|
|
defer s.setAppliedIndex(e.Index)
|
2016-06-27 20:27:27 +03:00
|
|
|
|
2016-04-20 10:31:58 +03:00
|
|
|
// raft state machine may generate noop entry when leader confirmation.
|
|
|
|
// skip it in advance to avoid some potential bug in the future
|
|
|
|
if len(e.Data) == 0 {
|
|
|
|
select {
|
|
|
|
case s.forceVersionC <- struct{}{}:
|
|
|
|
default:
|
|
|
|
}
|
|
|
|
return
|
2014-08-31 02:27:43 +04:00
|
|
|
}
|
2016-03-01 04:05:28 +03:00
|
|
|
|
2016-04-20 10:31:58 +03:00
|
|
|
var raftReq pb.InternalRaftRequest
|
|
|
|
if !pbutil.MaybeUnmarshal(&raftReq, e.Data) { // backward compatible
|
|
|
|
var r pb.Request
|
|
|
|
pbutil.MustUnmarshal(&r, e.Data)
|
|
|
|
s.w.Trigger(r.ID, s.applyV2Request(&r))
|
|
|
|
return
|
|
|
|
}
|
|
|
|
if raftReq.V2 != nil {
|
|
|
|
req := raftReq.V2
|
|
|
|
s.w.Trigger(req.ID, s.applyV2Request(req))
|
|
|
|
return
|
2016-03-01 04:05:28 +03:00
|
|
|
}
|
|
|
|
|
2016-04-20 10:31:58 +03:00
|
|
|
// do not re-apply applied entries.
|
2016-06-27 20:27:27 +03:00
|
|
|
if !shouldApplyV3 {
|
2016-04-20 10:31:58 +03:00
|
|
|
return
|
|
|
|
}
|
2016-05-17 09:20:50 +03:00
|
|
|
|
|
|
|
id := raftReq.ID
|
|
|
|
if id == 0 {
|
|
|
|
id = raftReq.Header.ID
|
|
|
|
}
|
|
|
|
|
2016-06-16 07:31:07 +03:00
|
|
|
var ar *applyResult
|
2016-07-08 09:29:37 +03:00
|
|
|
needResult := s.w.IsRegistered(id)
|
|
|
|
if needResult || !noSideEffect(&raftReq) {
|
|
|
|
if !needResult && raftReq.Txn != nil {
|
|
|
|
removeNeedlessRangeReqs(raftReq.Txn)
|
|
|
|
}
|
2016-06-16 07:31:07 +03:00
|
|
|
ar = s.applyV3.Apply(&raftReq)
|
|
|
|
}
|
|
|
|
|
|
|
|
if ar == nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-04-20 10:31:58 +03:00
|
|
|
if ar.err != ErrNoSpace || len(s.alarmStore.Get(pb.AlarmType_NOSPACE)) > 0 {
|
2016-05-17 09:20:50 +03:00
|
|
|
s.w.Trigger(id, ar)
|
2016-04-20 10:31:58 +03:00
|
|
|
return
|
2014-08-27 03:17:02 +04:00
|
|
|
}
|
2016-04-20 10:31:58 +03:00
|
|
|
|
|
|
|
plog.Errorf("applying raft message exceeded backend quota")
|
|
|
|
go func() {
|
|
|
|
a := &pb.AlarmRequest{
|
|
|
|
MemberID: uint64(s.ID()),
|
|
|
|
Action: pb.AlarmRequest_ACTIVATE,
|
|
|
|
Alarm: pb.AlarmType_NOSPACE,
|
|
|
|
}
|
|
|
|
r := pb.InternalRaftRequest{Alarm: a}
|
|
|
|
s.processInternalRaftRequest(context.TODO(), r)
|
2016-05-17 09:20:50 +03:00
|
|
|
s.w.Trigger(id, ar)
|
2016-04-20 10:31:58 +03:00
|
|
|
}()
|
2014-08-26 22:37:07 +04:00
|
|
|
}
|
2014-08-27 22:54:29 +04:00
|
|
|
|
2015-04-19 21:08:03 +03:00
|
|
|
// applyConfChange applies a ConfChange to the server. It is only
|
|
|
|
// invoked with a ConfChange that has already passed through Raft
|
|
|
|
func (s *EtcdServer) applyConfChange(cc raftpb.ConfChange, confState *raftpb.ConfState) (bool, error) {
|
2015-05-13 03:22:06 +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:
|
2016-04-07 22:02:37 +03:00
|
|
|
m := new(membership.Member)
|
2014-10-23 03:15:37 +04:00
|
|
|
if err := json.Unmarshal(cc.Context, m); err != nil {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.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) {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("nodeID should always be equal to member ID")
|
2014-10-04 02:21:10 +04:00
|
|
|
}
|
2015-05-13 03:22:06 +03:00
|
|
|
s.cluster.AddMember(m)
|
2016-06-01 07:12:41 +03:00
|
|
|
if m.ID != s.id {
|
2015-04-19 21:08:03 +03:00
|
|
|
s.r.transport.AddPeer(m.ID, m.PeerURLs)
|
2014-11-13 07:37:50 +03:00
|
|
|
}
|
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-05-13 03:22:06 +03:00
|
|
|
s.cluster.RemoveMember(id)
|
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
|
|
|
}
|
2016-06-01 07:12:41 +03:00
|
|
|
s.r.transport.RemovePeer(id)
|
2014-11-11 22:46:49 +03:00
|
|
|
case raftpb.ConfChangeUpdateNode:
|
2016-04-07 22:02:37 +03:00
|
|
|
m := new(membership.Member)
|
2014-11-11 22:46:49 +03:00
|
|
|
if err := json.Unmarshal(cc.Context, m); err != nil {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("unmarshal member should never fail: %v", err)
|
2014-11-11 22:46:49 +03:00
|
|
|
}
|
|
|
|
if cc.NodeID != uint64(m.ID) {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("nodeID should always be equal to member ID")
|
2014-11-11 22:46:49 +03:00
|
|
|
}
|
2015-05-13 03:22:06 +03:00
|
|
|
s.cluster.UpdateRaftAttributes(m.ID, m.RaftAttributes)
|
2016-06-01 07:12:41 +03:00
|
|
|
if m.ID != s.id {
|
2015-04-19 21:08:03 +03:00
|
|
|
s.r.transport.UpdatePeer(m.ID, m.PeerURLs)
|
2014-11-13 07:37:50 +03:00
|
|
|
}
|
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()
|
|
|
|
|
2016-04-21 18:48:11 +03:00
|
|
|
s.wg.Add(1)
|
2015-02-26 23:39:50 +03:00
|
|
|
go func() {
|
2016-04-21 18:48:11 +03:00
|
|
|
defer s.wg.Done()
|
|
|
|
|
2015-02-26 23:39:50 +03:00
|
|
|
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 {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("store save should never fail: %v", err)
|
2015-02-26 23:39:50 +03:00
|
|
|
}
|
|
|
|
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
|
|
|
|
}
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("unexpected create snapshot error %v", err)
|
2015-02-26 23:39:50 +03:00
|
|
|
}
|
2016-07-03 20:16:44 +03:00
|
|
|
// commit kv to write metadata (for example: consistent index) to disk.
|
2016-03-29 03:15:58 +03:00
|
|
|
s.KV().Commit()
|
2015-10-27 07:06:08 +03:00
|
|
|
// SaveSnap saves the snapshot and releases the locked wal files
|
|
|
|
// to the snapshot index.
|
2015-12-12 15:25:27 +03:00
|
|
|
if err = s.r.storage.SaveSnap(snap); err != nil {
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Fatalf("save snapshot error: %v", err)
|
2015-01-23 21:50:43 +03:00
|
|
|
}
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("saved snapshot at index %d", snap.Metadata.Index)
|
2015-01-23 21:50:43 +03:00
|
|
|
|
2016-06-16 18:57:02 +03:00
|
|
|
// When sending a snapshot, etcd will pause compaction.
|
|
|
|
// After receives a snapshot, the slow follower needs to get all the entries right after
|
|
|
|
// the snapshot sent to catch up. If we do not pause compaction, the log entries right after
|
|
|
|
// the snapshot sent might already be compacted. It happens when the snapshot takes long time
|
|
|
|
// to send and save. Pausing compaction avoids triggering a snapshot sending cycle.
|
|
|
|
if atomic.LoadInt64(&s.inflightSnapshots) != 0 {
|
|
|
|
plog.Infof("skip compaction since there is an inflight snapshot")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
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
|
|
|
|
}
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Panicf("unexpected compaction error %v", err)
|
2014-11-26 02:13:07 +03:00
|
|
|
}
|
2015-06-14 16:19:19 +03:00
|
|
|
plog.Infof("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() }
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) ClusterVersion() *semver.Version {
|
2015-05-13 03:22:06 +03:00
|
|
|
if s.cluster == nil {
|
2015-05-08 21:10:12 +03:00
|
|
|
return nil
|
|
|
|
}
|
2015-05-13 03:22:06 +03:00
|
|
|
return s.cluster.Version()
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
}
|
|
|
|
|
2016-02-01 08:42:39 +03:00
|
|
|
// monitorVersions checks the member's version every monitorVersionInterval.
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
// It updates the cluster version if all members agrees on a higher one.
|
|
|
|
// It prints out log if there is a member with a higher version than the
|
|
|
|
// local version.
|
|
|
|
func (s *EtcdServer) monitorVersions() {
|
|
|
|
for {
|
|
|
|
select {
|
2015-05-08 21:10:12 +03:00
|
|
|
case <-s.forceVersionC:
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
case <-time.After(monitorVersionInterval):
|
2015-05-08 21:10:12 +03:00
|
|
|
case <-s.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
if s.Leader() != s.ID() {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2016-01-25 01:56:07 +03:00
|
|
|
v := decideClusterVersion(getVersions(s.cluster, s.id, s.peerRt))
|
2015-05-08 21:10:12 +03:00
|
|
|
if v != nil {
|
2016-01-08 11:21:19 +03:00
|
|
|
// only keep major.minor version for comparison
|
2015-05-08 21:10:12 +03:00
|
|
|
v = &semver.Version{
|
|
|
|
Major: v.Major,
|
|
|
|
Minor: v.Minor,
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
}
|
2015-05-08 21:10:12 +03:00
|
|
|
}
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
|
2015-05-08 21:10:12 +03:00
|
|
|
// if the current version is nil:
|
|
|
|
// 1. use the decided version if possible
|
|
|
|
// 2. or use the min cluster version
|
2015-05-13 03:22:06 +03:00
|
|
|
if s.cluster.Version() == nil {
|
2015-05-08 21:10:12 +03:00
|
|
|
if v != nil {
|
|
|
|
go s.updateClusterVersion(v.String())
|
|
|
|
} else {
|
|
|
|
go s.updateClusterVersion(version.MinClusterVersion)
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
}
|
2015-05-08 21:10:12 +03:00
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
// update cluster version only if the decided version is greater than
|
|
|
|
// the current cluster version
|
2015-05-13 03:22:06 +03:00
|
|
|
if v != nil && s.cluster.Version().LessThan(*v) {
|
2015-05-08 21:10:12 +03:00
|
|
|
go s.updateClusterVersion(v.String())
|
*: add cluster version and cluster version detection.
Cluster version is the min major.minor of all members in
the etcd cluster. Cluster version is set to the min version
that a etcd member is compatible with when first bootstrapp.
During a rolling upgrades, the cluster version will be updated
automatically.
For example:
```
Cluster [a:1, b:1 ,c:1] -> clusterVersion 1
update a -> 2, b -> 2
after a detection
Cluster [a:2, b:2 ,c:1] -> clusterVersion 1, since c is still 1
update c -> 2
after a detection
Cluster [a:2, b:2 ,c:2] -> clusterVersion 2
```
The API/raft component can utilize clusterVersion to determine if
it can accept a client request or a raft RPC.
We choose polling rather than pushing since we want to use the same
logic for cluster version detection and (TODO) cluster version checking.
Before a member actually joins a etcd cluster, it should check the version
of the cluster. Push does not work since the other members cannot push
version info to it before it actually joins. Moreover, we do not want our
raft RPC system (which is doing the heartbeat pushing) to coordinate cluster version.
2015-04-29 20:56:34 +03:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2015-05-08 21:10:12 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) updateClusterVersion(ver string) {
|
2015-05-13 03:22:06 +03:00
|
|
|
if s.cluster.Version() == nil {
|
2015-08-15 18:30:06 +03:00
|
|
|
plog.Infof("setting up the initial cluster version to %s", version.Cluster(ver))
|
2015-05-08 21:10:12 +03:00
|
|
|
} else {
|
2015-08-15 18:30:06 +03:00
|
|
|
plog.Infof("updating the cluster version from %s to %s", version.Cluster(s.cluster.Version().String()), version.Cluster(ver))
|
2015-05-08 21:10:12 +03:00
|
|
|
}
|
|
|
|
req := pb.Request{
|
|
|
|
Method: "PUT",
|
2016-04-12 21:37:22 +03:00
|
|
|
Path: membership.StoreClusterVersionKey(),
|
2015-05-08 21:10:12 +03:00
|
|
|
Val: ver,
|
|
|
|
}
|
2016-05-12 22:34:03 +03:00
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), s.Cfg.ReqTimeout())
|
2015-05-08 21:10:12 +03:00
|
|
|
_, err := s.Do(ctx, req)
|
|
|
|
cancel()
|
|
|
|
switch err {
|
|
|
|
case nil:
|
|
|
|
return
|
|
|
|
case ErrStopped:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Infof("aborting update cluster version because server is stopped")
|
2015-05-08 21:10:12 +03:00
|
|
|
return
|
|
|
|
default:
|
2015-06-08 23:28:34 +03:00
|
|
|
plog.Errorf("error updating cluster version (%v)", err)
|
2015-05-08 21:10:12 +03:00
|
|
|
}
|
|
|
|
}
|
2015-08-12 23:38:43 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) parseProposeCtxErr(err error, start time.Time) error {
|
|
|
|
switch err {
|
|
|
|
case context.Canceled:
|
|
|
|
return ErrCanceled
|
|
|
|
case context.DeadlineExceeded:
|
|
|
|
curLeadElected := s.r.leadElectedTime()
|
2016-05-12 22:34:03 +03:00
|
|
|
prevLeadLost := curLeadElected.Add(-2 * time.Duration(s.Cfg.ElectionTicks) * time.Duration(s.Cfg.TickMs) * time.Millisecond)
|
2015-08-12 23:38:43 +03:00
|
|
|
if start.After(prevLeadLost) && start.Before(curLeadElected) {
|
2015-08-14 01:46:21 +03:00
|
|
|
return ErrTimeoutDueToLeaderFail
|
2015-08-12 23:38:43 +03:00
|
|
|
}
|
2015-08-26 01:14:41 +03:00
|
|
|
|
|
|
|
lead := types.ID(atomic.LoadUint64(&s.r.lead))
|
|
|
|
switch lead {
|
|
|
|
case types.ID(raft.None):
|
|
|
|
// TODO: return error to specify it happens because the cluster does not have leader now
|
|
|
|
case s.ID():
|
|
|
|
if !isConnectedToQuorumSince(s.r.transport, start, s.ID(), s.cluster.Members()) {
|
|
|
|
return ErrTimeoutDueToConnectionLost
|
|
|
|
}
|
|
|
|
default:
|
|
|
|
if !isConnectedSince(s.r.transport, start, lead) {
|
|
|
|
return ErrTimeoutDueToConnectionLost
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-08-12 23:38:43 +03:00
|
|
|
return ErrTimeout
|
|
|
|
default:
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
2015-08-26 01:14:41 +03:00
|
|
|
|
2016-04-25 22:32:58 +03:00
|
|
|
func (s *EtcdServer) KV() mvcc.ConsistentWatchableKV { return s.kv }
|
2016-03-10 08:37:10 +03:00
|
|
|
func (s *EtcdServer) Backend() backend.Backend {
|
|
|
|
s.bemu.Lock()
|
|
|
|
defer s.bemu.Unlock()
|
|
|
|
return s.be
|
|
|
|
}
|
2016-03-08 11:31:48 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) AuthStore() auth.AuthStore { return s.authStore }
|
2016-03-26 01:14:32 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) restoreAlarms() error {
|
2016-06-15 08:45:09 +03:00
|
|
|
s.applyV3 = s.newApplierV3()
|
2016-03-26 01:14:32 +03:00
|
|
|
as, err := alarm.NewAlarmStore(s)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
s.alarmStore = as
|
|
|
|
if len(as.Get(pb.AlarmType_NOSPACE)) > 0 {
|
|
|
|
s.applyV3 = newApplierV3Capped(s.applyV3)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
}
|
2016-06-09 03:11:26 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) getAppliedIndex() uint64 {
|
|
|
|
return atomic.LoadUint64(&s.appliedIndex)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *EtcdServer) setAppliedIndex(v uint64) {
|
|
|
|
atomic.StoreUint64(&s.appliedIndex, v)
|
|
|
|
}
|
2016-07-12 23:16:35 +03:00
|
|
|
|
|
|
|
func (s *EtcdServer) getCommittedIndex() uint64 {
|
|
|
|
return atomic.LoadUint64(&s.committedIndex)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *EtcdServer) setCommittedIndex(v uint64) {
|
|
|
|
atomic.StoreUint64(&s.committedIndex, v)
|
|
|
|
}
|