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"
|
*: 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
|
|
|
"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
|
|
|
|
*: 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
|
|
|
"github.com/coreos/etcd/Godeps/_workspace/src/github.com/coreos/go-semver/semver"
|
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
|
|
|
|
*: 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
|
|
|
purgeFileInterval = 30 * time.Second
|
|
|
|
monitorVersionInterval = 10 * 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
|
*: 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.
|
|
|
|
// Cluster version is set to the min version that a etcd member is
|
|
|
|
// compatible with when first bootstrap.
|
|
|
|
//
|
|
|
|
// During a rolling upgrades, the ClusterVersion will be updated
|
|
|
|
// automatically after a sync. (10 second by default)
|
|
|
|
//
|
|
|
|
// The API/raft component can utilize ClusterVersion to determine if
|
|
|
|
// it can accept a client request or a raft RPC.
|
|
|
|
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 {
|
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
|
*: 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
|
|
|
|
|
|
|
verMu sync.Mutex
|
|
|
|
clusterVersion *semver.Version
|
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
|
|
|
|
2015-04-21 09:48:36 +03:00
|
|
|
var remotes []*Member
|
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-04-21 09:48:36 +03:00
|
|
|
remotes = existingCluster.Members()
|
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
|
|
|
}
|
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
|
|
|
|
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),
|
|
|
|
},
|
*: 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
|
|
|
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()),
|
|
|
|
clusterVersion: semver.Must(semver.NewVersion(version.MinClusterVersion)),
|
2014-10-03 03:20:14 +04:00
|
|
|
}
|
2014-12-29 23:08:13 +03:00
|
|
|
|
2015-04-21 09:48:36 +03:00
|
|
|
// TODO: move transport initialization near the definition of remote
|
2015-01-03 07:00:29 +03:00
|
|
|
tr := rafthttp.NewTransporter(cfg.Transport, id, cfg.Cluster.ID(), srv, srv.errorc, sstats, lstats)
|
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-19 21:08:03 +03:00
|
|
|
for _, m := range cfg.Cluster.Members() {
|
|
|
|
if m.ID != id {
|
|
|
|
tr.AddPeer(m.ID, m.PeerURLs)
|
|
|
|
}
|
|
|
|
}
|
2015-01-16 01:39:30 +03:00
|
|
|
srv.r.transport = 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)
|
*: 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 {
|
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-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-04-26 09:24:59 +03:00
|
|
|
func (s *EtcdServer) IsIDRemoved(id uint64) bool { return s.Cluster.IsIDRemoved(types.ID(id)) }
|
|
|
|
|
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-04-19 21:34:33 +03:00
|
|
|
s.Cluster.Recover()
|
2015-02-12 21:48:06 +03:00
|
|
|
|
2015-04-19 21:34:33 +03:00
|
|
|
// recover raft transport
|
|
|
|
s.r.transport.RemoveAllPeers()
|
|
|
|
for _, m := range s.Cluster.Members() {
|
|
|
|
if m.ID == s.ID() {
|
|
|
|
continue
|
2015-04-19 21:08:03 +03:00
|
|
|
}
|
2015-04-19 21:34:33 +03:00
|
|
|
s.r.transport.AddPeer(m.ID, m.PeerURLs)
|
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:
|
2015-04-20 12:49:54 +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 {
|
|
|
|
break
|
|
|
|
}
|
2014-10-11 17:58:39 +04:00
|
|
|
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-04-19 21:08:03 +03:00
|
|
|
shouldstop, err = s.applyConfChange(cc, confState)
|
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-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) {
|
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-04-19 21:08:03 +03:00
|
|
|
s.Cluster.AddMember(m)
|
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 {
|
2015-04-19 21:08:03 +03:00
|
|
|
s.r.transport.AddPeer(m.ID, m.PeerURLs)
|
2014-11-13 07:37:50 +03:00
|
|
|
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-04-19 21:08:03 +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
|
|
|
} else {
|
2015-04-19 21:08:03 +03:00
|
|
|
s.r.transport.RemovePeer(id)
|
2014-11-13 07:37:50 +03:00
|
|
|
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-04-19 21:08:03 +03:00
|
|
|
s.Cluster.UpdateRaftAttributes(m.ID, m.RaftAttributes)
|
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 {
|
2015-04-19 21:08:03 +03:00
|
|
|
s.r.transport.UpdatePeer(m.ID, m.PeerURLs)
|
2014-11-13 07:37:50 +03:00
|
|
|
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() }
|
*: 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 {
|
|
|
|
s.verMu.Lock()
|
|
|
|
defer s.verMu.Unlock()
|
|
|
|
// deep copy
|
|
|
|
return semver.Must(semver.NewVersion(s.clusterVersion.String()))
|
|
|
|
}
|
|
|
|
|
|
|
|
// monitorVersions checks the member's version every monitorVersion interval.
|
|
|
|
// 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 {
|
|
|
|
case <-time.After(monitorVersionInterval):
|
|
|
|
v := decideClusterVersion(getVersions(s.Cluster, s.cfg.Transport))
|
|
|
|
if v == nil {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
|
|
|
s.verMu.Lock()
|
|
|
|
// clear patch version
|
|
|
|
v.Patch = 0
|
|
|
|
if s.clusterVersion.LessThan(*v) {
|
|
|
|
log.Printf("etcdsever: updated the cluster version from %v to %v", s.clusterVersion, v.String())
|
|
|
|
// TODO: persist the version upgrade via raft. Then etcdserver will be able to use the
|
|
|
|
// upgraded version without syncing with others after a restart.
|
|
|
|
s.clusterVersion = v
|
|
|
|
}
|
|
|
|
s.verMu.Unlock()
|
|
|
|
case <-s.done:
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|