2014-08-26 05:39:02 +04:00
package etcdserver
import (
2014-09-27 01:18:59 +04:00
"encoding/json"
2014-08-27 02:42:10 +04:00
"errors"
2014-09-16 04:35:02 +04:00
"log"
2014-09-18 01:38:10 +04:00
"math/rand"
2014-10-03 21:38:35 +04:00
"os"
"path"
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
2014-10-04 16:20:45 +04:00
"github.com/coreos/etcd/discovery"
2014-09-03 08:36:14 +04:00
pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
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-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"
2014-09-04 07:06:16 +04:00
"github.com/coreos/etcd/third_party/code.google.com/p/go.net/context"
2014-08-26 05:39:02 +04:00
"github.com/coreos/etcd/wait"
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-09-17 05:18:45 +04:00
)
2014-09-16 04:35:02 +04:00
2014-08-29 03:41:42 +04:00
var (
ErrUnknownMethod = errors . New ( "etcdserver: unknown method" )
ErrStopped = errors . New ( "etcdserver: server stopped" )
)
2014-08-27 02:42:10 +04:00
2014-09-18 01:38:10 +04:00
func init ( ) {
rand . Seed ( time . Now ( ) . UnixNano ( ) )
}
2014-10-03 22:05:56 +04:00
type sendFunc func ( m [ ] raftpb . Message )
2014-08-28 00:37:22 +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-17 05:18:45 +04:00
type Storage interface {
// Save function saves ents and state to the underlying stable storage.
// Save MUST block until st and ents are on stable storage.
Save ( st raftpb . HardState , ents [ ] raftpb . Entry )
// SaveSnap function saves snapshot to the underlying stable storage.
SaveSnap ( snap raftpb . Snapshot )
// TODO: WAL should be able to control cut itself. After implement self-controled cut,
// remove it in this interface.
// Cut cuts out a new wal file for saving new state and entries.
Cut ( ) error
}
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 ( )
// Do takes a request and attempts to fulfil it, returning a Response.
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-09-25 05:05:33 +04:00
type RaftTimer interface {
2014-10-08 15:58:53 +04:00
Index ( ) uint64
Term ( ) uint64
2014-09-25 05:05:33 +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.
func NewServer ( cfg * ServerConfig ) * EtcdServer {
2014-10-07 03:42:13 +04:00
err := cfg . Verify ( )
if err != nil {
log . Fatalln ( err )
2014-10-03 03:20:14 +04:00
}
2014-10-03 21:38:35 +04:00
snapdir := path . Join ( cfg . DataDir , "snap" )
if err := os . MkdirAll ( snapdir , privateDirMode ) ; err != nil {
log . Fatalf ( "etcdserver: cannot create snapshot directory: %v" , err )
}
ss := snap . New ( snapdir )
2014-10-03 03:20:14 +04:00
st := store . New ( )
var w * wal . WAL
var n raft . Node
2014-10-07 03:42:13 +04:00
m := cfg . Cluster . FindName ( cfg . Name )
2014-10-03 21:38:35 +04:00
waldir := path . Join ( cfg . DataDir , "wal" )
if ! wal . Exist ( waldir ) {
2014-10-04 16:20:45 +04:00
if cfg . DiscoveryURL != "" {
d , err := discovery . New ( cfg . DiscoveryURL , m . ID , cfg . Cluster . String ( ) )
if err != nil {
log . Fatalf ( "etcd: cannot init discovery %v" , err )
}
s , err := d . Discover ( )
if err != nil {
log . Fatalf ( "etcd: %v" , err )
}
if err = cfg . Cluster . Set ( s ) ; err != nil {
log . Fatalf ( "etcd: %v" , err )
}
2014-10-04 00:10:34 +04:00
} else if ( cfg . ClusterState ) != ClusterStateValueNew {
log . Fatalf ( "etcd: initial cluster state unset and no wal or discovery URL found" )
2014-10-04 16:20:45 +04:00
}
2014-10-03 04:06:35 +04:00
i := pb . Info { ID : m . ID }
b , err := i . Marshal ( )
if err != nil {
log . Fatal ( err )
}
if w , err = wal . Create ( waldir , b ) ; err != nil {
2014-10-03 03:20:14 +04:00
log . Fatal ( err )
}
2014-10-08 04:38:58 +04:00
ids := cfg . Cluster . IDs ( )
2014-10-11 04:00:14 +04:00
peers := make ( [ ] raft . Peer , len ( ids ) )
for i , id := range ids {
2014-10-08 04:38:58 +04:00
ctx , err := json . Marshal ( ( * cfg . Cluster ) [ id ] )
if err != nil {
log . Fatal ( err )
}
2014-10-11 04:00:14 +04:00
peers [ i ] = raft . Peer { ID : id , Context : ctx }
2014-10-08 04:38:58 +04:00
}
n = raft . StartNode ( m . ID , peers , 10 , 1 )
2014-10-03 03:20:14 +04:00
} else {
2014-10-09 03:09:12 +04:00
if cfg . DiscoveryURL != "" {
log . Printf ( "etcd: warn: ignoring discovery URL: etcd has already been initialized and has a valid log in %q" , waldir )
}
2014-10-08 15:58:53 +04:00
var index uint64
2014-10-03 03:20:14 +04:00
snapshot , err := ss . Load ( )
if err != nil && err != snap . ErrNoSnapshot {
log . Fatal ( err )
}
if snapshot != nil {
log . Printf ( "etcdserver: restart from snapshot at index %d" , snapshot . Index )
st . Recovery ( snapshot . Data )
index = snapshot . Index
}
// restart a node from previous wal
2014-10-03 21:38:35 +04:00
if w , err = wal . OpenAtIndex ( waldir , index ) ; err != nil {
2014-10-03 03:20:14 +04:00
log . Fatal ( err )
}
2014-10-03 04:06:35 +04:00
md , st , ents , err := w . ReadAll ( )
2014-10-03 03:20:14 +04:00
if err != nil {
log . Fatal ( err )
}
2014-10-03 04:06:35 +04:00
var info pb . Info
if err := info . Unmarshal ( md ) ; err != nil {
log . Fatal ( err )
}
2014-10-03 03:20:14 +04:00
// TODO(xiangli): save/recovery nodeID?
2014-10-03 04:06:35 +04:00
if info . ID != m . ID {
log . Fatalf ( "unexpected nodeid %x, want %x: nodeid should always be the same until we support name/peerURLs update or dynamic configuration" , info . ID , m . ID )
2014-10-03 03:20:14 +04:00
}
2014-10-11 04:00:14 +04:00
n = raft . RestartNode ( m . ID , 10 , 1 , snapshot , st , ents )
2014-10-03 03:20:14 +04:00
}
2014-10-11 04:00:14 +04:00
cls := & clusterStore { Store : st }
2014-10-03 03:20:14 +04:00
s := & EtcdServer {
2014-10-03 02:48:58 +04:00
store : st ,
node : n ,
id : m . ID ,
attributes : Attributes { Name : cfg . Name , ClientURLs : cfg . ClientURLs . StringSlice ( ) } ,
2014-10-03 22:05:56 +04:00
storage : struct {
2014-10-03 03:20:14 +04:00
* wal . WAL
* snap . Snapshotter
} { w , ss } ,
2014-10-03 22:05:56 +04:00
send : Sender ( cfg . Transport , cls ) ,
ticker : time . Tick ( 100 * time . Millisecond ) ,
syncTicker : time . Tick ( 500 * time . Millisecond ) ,
snapCount : cfg . SnapCount ,
2014-10-03 03:20:14 +04:00
ClusterStore : cls ,
}
return s
}
2014-09-12 04:14:41 +04:00
// EtcdServer is the production implementation of the Server interface
type EtcdServer struct {
2014-10-03 03:20:14 +04:00
w wait . Wait
done chan struct { }
2014-10-03 02:48:58 +04:00
id uint64
attributes Attributes
2014-08-27 00:39:33 +04:00
2014-10-03 03:20:14 +04:00
ClusterStore ClusterStore
2014-09-30 04:14:22 +04:00
2014-10-03 22:05:56 +04:00
node raft . Node
store store . Store
2014-08-27 02:28:21 +04:00
2014-10-03 22:05:56 +04:00
// send specifies the send function for sending msgs to members. send
2014-08-27 00:39:33 +04:00
// MUST NOT block. It is okay to drop messages, since clients should
2014-10-03 22:05:56 +04:00
// timeout and reissue their messages. If send is nil, server will
2014-08-27 00:39:33 +04:00
// panic.
2014-10-03 22:05:56 +04:00
send sendFunc
2014-08-27 00:39:33 +04:00
2014-10-03 22:05:56 +04:00
storage Storage
2014-09-03 03:59:29 +04:00
2014-10-03 22:05:56 +04:00
ticker <- chan time . Time
syncTicker <- chan time . Time
2014-09-17 05:18:45 +04:00
2014-10-08 15:58:53 +04:00
snapCount uint64 // number of entries to trigger a snapshot
2014-09-25 05:05:33 +04:00
// Cache of the latest raft index and raft term the server has seen
2014-10-08 15:58:53 +04:00
raftIndex uint64
raftTerm uint64
2014-08-26 05:39:02 +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-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 ( ) {
2014-10-03 22:05:56 +04:00
if s . snapCount == 0 {
2014-09-17 05:18:45 +04:00
log . Printf ( "etcdserver: set snapshot count to default %d" , DefaultSnapCount )
2014-10-03 22:05:56 +04: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-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-09-12 04:14:41 +04:00
func ( s * EtcdServer ) Process ( ctx context . Context , m raftpb . Message ) error {
2014-10-03 22:05:56 +04:00
return s . node . Step ( ctx , m )
2014-09-12 04:14:41 +04:00
}
func ( s * EtcdServer ) run ( ) {
2014-09-16 04:35:02 +04:00
var syncC <- chan time . Time
2014-09-17 05:18:45 +04:00
// snapi indicates the index of the last submitted snapshot request
2014-10-08 15:58:53 +04:00
var snapi , appliedi uint64
var nodes [ ] uint64
2014-08-26 05:39:02 +04:00
for {
2014-08-27 02:28:21 +04:00
select {
2014-10-03 22:05:56 +04:00
case <- s . ticker :
s . node . Tick ( )
case rd := <- s . node . Ready ( ) :
s . storage . Save ( rd . HardState , rd . Entries )
s . storage . SaveSnap ( rd . Snapshot )
s . send ( rd . Messages )
2014-08-28 02:34:33 +04:00
// TODO(bmizerany): do this in the background, but take
// care to apply entries in a single goroutine, and not
// race them.
2014-09-25 06:51:27 +04:00
// TODO: apply configuration change into ClusterStore.
2014-08-28 02:34:33 +04:00
for _ , e := range rd . CommittedEntries {
2014-09-19 23:36:19 +04:00
switch e . Type {
2014-09-21 09:48:17 +04:00
case raftpb . EntryNormal :
2014-09-19 23:36:19 +04:00
var r pb . Request
if err := r . Unmarshal ( e . Data ) ; err != nil {
panic ( "TODO: this is bad, what do we do about it?" )
}
2014-10-01 02:14:44 +04:00
s . w . Trigger ( r . ID , s . applyRequest ( r ) )
2014-09-23 23:02:44 +04:00
case raftpb . EntryConfChange :
var cc raftpb . ConfChange
2014-09-23 08:59:13 +04:00
if err := cc . Unmarshal ( e . Data ) ; err != nil {
2014-09-19 23:36:19 +04:00
panic ( "TODO: this is bad, what do we do about it?" )
}
2014-10-01 02:14:44 +04:00
s . applyConfChange ( cc )
2014-10-12 04:35:58 +04:00
s . w . Trigger ( cc . ID , nil )
2014-09-19 23:36:19 +04:00
default :
2014-09-23 08:59:13 +04:00
panic ( "unexpected entry type" )
2014-08-29 03:41:42 +04:00
}
2014-10-08 15:58:53 +04:00
atomic . StoreUint64 ( & s . raftIndex , e . Index )
atomic . StoreUint64 ( & s . raftTerm , e . Term )
2014-09-17 05:18:45 +04:00
appliedi = e . Index
}
2014-10-06 06:12:31 +04:00
if rd . SoftState != nil {
nodes = rd . SoftState . Nodes
if rd . RaftState == raft . StateLeader {
syncC = s . syncTicker
} else {
syncC = nil
}
if rd . SoftState . ShouldStop {
s . Stop ( )
return
}
}
2014-09-17 05:18:45 +04:00
if rd . Snapshot . Index > snapi {
snapi = rd . Snapshot . Index
}
// recover from snapshot if it is more updated than current applied
if rd . Snapshot . Index > appliedi {
2014-10-03 22:05:56 +04:00
if err := s . store . Recovery ( rd . Snapshot . Data ) ; err != nil {
2014-09-17 05:18:45 +04:00
panic ( "TODO: this is bad, what do we do about it?" )
}
appliedi = rd . Snapshot . Index
}
2014-10-03 22:05:56 +04:00
if appliedi - snapi > s . snapCount {
2014-10-06 06:12:31 +04:00
s . snapshot ( appliedi , nodes )
2014-09-17 05:18:45 +04:00
snapi = appliedi
2014-08-28 02:34:33 +04:00
}
2014-09-16 04:35:02 +04:00
case <- syncC :
s . sync ( defaultSyncTimeout )
2014-08-29 03:41:42 +04:00
case <- s . done :
2014-08-26 22:37:07 +04:00
return
2014-08-26 05:39:02 +04:00
}
}
}
2014-09-12 04:14:41 +04:00
// Stop stops the server, and shuts down the running goroutine. Stop should be
// called after a Start(s), otherwise it will block forever.
func ( s * EtcdServer ) Stop ( ) {
2014-10-03 22:05:56 +04:00
s . node . Stop ( )
2014-09-11 04:12:58 +04:00
close ( s . done )
2014-08-30 02:32:41 +04:00
}
2014-08-29 03:41:42 +04:00
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-09-23 21:17:53 +04:00
if r . ID == 0 {
2014-08-26 05:39:02 +04:00
panic ( "r.Id cannot be 0" )
}
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 )
2014-10-03 22:05:56 +04:00
s . node . Propose ( ctx , data )
2014-08-27 02:28:21 +04:00
select {
case x := <- ch :
resp := x . ( Response )
return resp , resp . err
case <- ctx . Done ( ) :
2014-09-23 21:17:53 +04:00
s . w . Trigger ( r . ID , nil ) // GC wait
2014-08-27 02:28:21 +04:00
return Response { } , 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-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-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 {
2014-09-21 05:56:40 +04:00
ID : GenID ( ) ,
2014-09-23 23:02:44 +04:00
Type : raftpb . ConfChangeAddNode ,
2014-10-01 02:14:44 +04:00
NodeID : memb . ID ,
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 {
2014-09-21 05:56:40 +04:00
ID : GenID ( ) ,
2014-09-23 23:02:44 +04:00
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-09-25 05:05:33 +04:00
// Implement the RaftTimer interface
2014-10-08 15:58:53 +04:00
func ( s * EtcdServer ) Index ( ) uint64 {
return atomic . LoadUint64 ( & s . raftIndex )
2014-09-25 05:05:33 +04:00
}
2014-10-08 15:58:53 +04:00
func ( s * EtcdServer ) Term ( ) uint64 {
return atomic . LoadUint64 ( & s . raftTerm )
2014-09-25 05:05:33 +04:00
}
2014-09-19 23:36:19 +04:00
// configure sends configuration change through consensus then performs it.
// 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-10-12 04:35:58 +04:00
ch := s . w . Register ( cc . ID )
2014-10-03 22:05:56 +04:00
if err := s . node . ProposeConfChange ( ctx , cc ) ; err != nil {
2014-09-23 01:38:21 +04:00
log . Printf ( "configure error: %v" , err )
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 {
case <- ch :
return nil
case <- ctx . Done ( ) :
2014-10-12 04:35:58 +04:00
s . w . Trigger ( cc . ID , nil ) // GC wait
2014-09-19 23:36:19 +04:00
return ctx . Err ( )
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-10-12 04:35:58 +04:00
ID : GenID ( ) ,
2014-09-16 04:35:02 +04:00
Time : time . Now ( ) . UnixNano ( ) ,
}
data , err := req . Marshal ( )
if err != nil {
log . Printf ( "marshal request %#v error: %v" , req , err )
return
}
// There is no promise that node has leader when do SYNC request,
// so it uses goroutine to propose.
go func ( ) {
2014-10-03 22:05:56 +04:00
s . node . 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 {
2014-10-12 04:35:58 +04:00
ID : GenID ( ) ,
2014-09-27 01:18:59 +04:00
Method : "PUT" ,
2014-10-03 02:48:58 +04:00
Path : Member { ID : s . id } . storeKey ( ) + attributesSuffix ,
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-03 02:48:58 +04:00
log . Printf ( "etcdserver: published %+v to the cluster" , s . attributes )
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-09-23 01:16:12 +04:00
func getExpirationTime ( r * pb . Request ) time . Time {
var t time . Time
if r . Expiration != 0 {
t = time . Unix ( 0 , r . Expiration )
}
return t
}
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-09-23 01:16:12 +04:00
expr := getExpirationTime ( & r )
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-09-19 08:22:10 +04:00
exists , existsSet := 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 {
2014-10-03 22:05:56 +04:00
return f ( s . store . Update ( r . Path , 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-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
2014-10-01 02:14:44 +04:00
func ( s * EtcdServer ) applyConfChange ( cc raftpb . ConfChange ) {
s . node . ApplyConfChange ( cc )
switch cc . Type {
case raftpb . ConfChangeAddNode :
var m Member
if err := json . Unmarshal ( cc . Context , & m ) ; err != nil {
panic ( "unexpected unmarshal error" )
}
2014-10-04 02:21:10 +04:00
if cc . NodeID != m . ID {
2014-10-07 21:59:02 +04:00
panic ( "unexpected nodeID mismatch" )
2014-10-04 02:21:10 +04:00
}
2014-10-08 03:19:51 +04:00
s . ClusterStore . Add ( m )
2014-10-01 02:14:44 +04:00
case raftpb . ConfChangeRemoveNode :
2014-10-08 03:19:51 +04:00
s . ClusterStore . Remove ( cc . NodeID )
2014-10-01 02:14:44 +04:00
default :
panic ( "unexpected ConfChange type" )
}
}
2014-09-17 05:18:45 +04:00
// TODO: non-blocking snapshot
2014-10-08 15:58:53 +04:00
func ( s * EtcdServer ) snapshot ( snapi uint64 , snapnodes [ ] uint64 ) {
2014-10-03 22:05:56 +04:00
d , err := s . store . Save ( )
2014-09-17 05:18:45 +04:00
// TODO: current store will never fail to do a snapshot
// what should we do if the store might fail?
if err != nil {
panic ( "TODO: this is bad, what do we do about it?" )
}
2014-10-06 06:12:31 +04:00
s . node . Compact ( snapi , snapnodes , d )
2014-10-03 22:05:56 +04:00
s . storage . Cut ( )
2014-09-17 05:18:45 +04:00
}
2014-09-16 04:35:02 +04:00
// TODO: move the function to /id pkg maybe?
// GenID generates a random id that is not equal to 0.
2014-10-08 15:58:53 +04:00
func GenID ( ) ( n uint64 ) {
2014-09-18 01:38:10 +04:00
for n == 0 {
2014-10-08 15:58:53 +04:00
n = uint64 ( rand . Int63 ( ) )
2014-09-16 04:35:02 +04:00
}
2014-09-18 01:38:10 +04:00
return
2014-09-16 04:35:02 +04:00
}
2014-08-27 22:54:29 +04:00
func getBool ( v * bool ) ( vv bool , set bool ) {
if v == nil {
return false , false
}
return * v , true
}