@@ -72,7 +72,7 @@ clone git github.com/imdario/mergo 0.2.1

 #get libnetwork packages

 clone git github.com/docker/libnetwork 82fb373e3eaa4e9bbb5b5ac148b0a3a71f80fca6

-clone git github.com/docker/go-events afb2b9f2c23f33ada1a22b03651775fdc65a5089

+clone git github.com/docker/go-events 18b43f1bc85d9cdd42c05a6cd2d444c7a200a894

 clone git github.com/armon/go-radix e39d623f12e8e41c7b5529e9a9dd67a1e2261f80

 clone git github.com/armon/go-metrics eb0af217e5e9747e41dd5303755356b62d28e3ec

 clone git github.com/hashicorp/go-msgpack 71c2886f5a673a35f909803f38ece5810165097b

@@ -85,7 +85,7 @@ clone git github.com/vishvananda/netlink e73bad418fd727ed3a02830b1af1ad0283a1de6

 clone git github.com/BurntSushi/toml f706d00e3de6abe700c994cdd545a1a4915af060

 clone git github.com/samuel/go-zookeeper d0e0d8e11f318e000a8cc434616d69e329edc374

 clone git github.com/deckarep/golang-set ef32fa3046d9f249d399f98ebaf9be944430fd1d

-clone git github.com/coreos/etcd 06e2338108fdc694349aed923f4a7e45cf0cec1f

+clone git github.com/coreos/etcd 3a49cbb769ebd8d1dd25abb1e83386e9883a5707

 clone git github.com/ugorji/go f1f1a805ed361a0e078bb537e4ea78cd37dcf065

 clone git github.com/hashicorp/consul v0.5.2

 clone git github.com/boltdb/bolt v1.2.1

@@ -144,11 +144,11 @@ clone git github.com/docker/docker-credential-helpers v0.3.0

 clone git github.com/docker/containerd 8508d2bec90b96403143a1104cdcbd56f6aeb361

 # cluster

-clone git github.com/docker/swarmkit 8a761950fb4d9251c335dc6149a8a02756cb3b10

+clone git github.com/docker/swarmkit 27fbaef4ceed648bb575969ccc9083a6e104a719

 clone git github.com/golang/mock bd3c8e81be01eef76d4b503f5e687d2d1354d2d9

 clone git github.com/gogo/protobuf 43a2e0b1c32252bfbbdf81f7faa7a88fb3fa4028

-clone git github.com/cloudflare/cfssl b895b0549c0ff676f92cf09ba971ae02bb41367b

-clone git github.com/google/certificate-transparency 025a5cab06f6a819c455d9fdc9e2a1b6d0982284

+clone git github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a

+clone git github.com/google/certificate-transparency 0f6e3d1d1ba4d03fdaab7cd716f36255c2e48341

 clone git golang.org/x/crypto 3fbbcd23f1cb824e69491a5930cfeff09b12f4d2 https://github.com/golang/crypto.git

 clone git golang.org/x/time a4bde12657593d5e90d0533a3e4fd95e635124cb https://github.com/golang/time.git

 clone git github.com/mreiferson/go-httpclient 63fe23f7434723dc904c901043af07931f293c47

@@ -69,3 +69,7 @@ JSON dictionary:

 or

     {"driver":"postgres","data_source":"postgres://user:password@host/db"}

+ 

+or

+

+    {"driver":"mysql","data_source":"user:password@tcp(hostname:3306)/db?parseTime=true"}

@@ -2,6 +2,7 @@

 package config

 import (

+	"crypto/tls"

 	"crypto/x509"

 	"encoding/asn1"

 	"encoding/json"

@@ -59,26 +60,35 @@ type AuthRemote struct {

 	AuthKeyName string `json:"auth_key"`

 }

+// CAConstraint specifies various CA constraints on the signed certificate.

+// CAConstraint would verify against (and override) the CA

+// extensions in the given CSR.

+type CAConstraint struct {

+	IsCA           bool `json:"is_ca"`

+	MaxPathLen     int  `json:"max_path_len"`

+	MaxPathLenZero bool `json:"max_path_len_zero"`

+}

+

 // A SigningProfile stores information that the CA needs to store

 // signature policy.

 type SigningProfile struct {

-	Usage               []string   `json:"usages"`

-	IssuerURL           []string   `json:"issuer_urls"`

-	OCSP                string     `json:"ocsp_url"`

-	CRL                 string     `json:"crl_url"`

-	CA                  bool       `json:"is_ca"`

-	OCSPNoCheck         bool       `json:"ocsp_no_check"`

-	ExpiryString        string     `json:"expiry"`

-	BackdateString      string     `json:"backdate"`

-	AuthKeyName         string     `json:"auth_key"`

-	RemoteName          string     `json:"remote"`

-	NotBefore           time.Time  `json:"not_before"`

-	NotAfter            time.Time  `json:"not_after"`

-	NameWhitelistString string     `json:"name_whitelist"`

-	AuthRemote          AuthRemote `json:"auth_remote"`

-	CTLogServers        []string   `json:"ct_log_servers"`

-	AllowedExtensions   []OID      `json:"allowed_extensions"`

-	CertStore           string     `json:"cert_store"`

+	Usage               []string     `json:"usages"`

+	IssuerURL           []string     `json:"issuer_urls"`

+	OCSP                string       `json:"ocsp_url"`

+	CRL                 string       `json:"crl_url"`

+	CAConstraint        CAConstraint `json:"ca_constraint"`

+	OCSPNoCheck         bool         `json:"ocsp_no_check"`

+	ExpiryString        string       `json:"expiry"`

+	BackdateString      string       `json:"backdate"`

+	AuthKeyName         string       `json:"auth_key"`

+	RemoteName          string       `json:"remote"`

+	NotBefore           time.Time    `json:"not_before"`

+	NotAfter            time.Time    `json:"not_after"`

+	NameWhitelistString string       `json:"name_whitelist"`

+	AuthRemote          AuthRemote   `json:"auth_remote"`

+	CTLogServers        []string     `json:"ct_log_servers"`

+	AllowedExtensions   []OID        `json:"allowed_extensions"`

+	CertStore           string       `json:"cert_store"`

 	Policies                    []CertificatePolicy

 	Expiry                      time.Duration

@@ -86,6 +96,8 @@ type SigningProfile struct {

 	Provider                    auth.Provider

 	RemoteProvider              auth.Provider

 	RemoteServer                string

+	RemoteCAs                   *x509.CertPool

+	ClientCert                  *tls.Certificate

 	CSRWhitelist                *CSRWhitelist

 	NameWhitelist               *regexp.Regexp

 	ExtensionWhitelist          map[string]bool

@@ -303,6 +315,44 @@ func (p *Signing) OverrideRemotes(remote string) error {

 	return nil

 }

+// SetClientCertKeyPairFromFile updates the properties to set client certificates for mutual

+// authenticated TLS remote requests

+func (p *Signing) SetClientCertKeyPairFromFile(certFile string, keyFile string) error {

+	if certFile != "" && keyFile != "" {

+		cert, err := helpers.LoadClientCertificate(certFile, keyFile)

+		if err != nil {

+			return err

+		}

+		for _, profile := range p.Profiles {

+			profile.ClientCert = cert

+		}

+		p.Default.ClientCert = cert

+	}

+	return nil

+}

+

+// SetRemoteCAsFromFile reads root CAs from file and updates the properties to set remote CAs for TLS

+// remote requests

+func (p *Signing) SetRemoteCAsFromFile(caFile string) error {

+	if caFile != "" {

+		remoteCAs, err := helpers.LoadPEMCertPool(caFile)

+		if err != nil {

+			return err

+		}

+		p.SetRemoteCAs(remoteCAs)

+	}

+	return nil

+}

+

+// SetRemoteCAs updates the properties to set remote CAs for TLS

+// remote requests

+func (p *Signing) SetRemoteCAs(remoteCAs *x509.CertPool) {

+	for _, profile := range p.Profiles {

+		profile.RemoteCAs = remoteCAs

+	}

+	p.Default.RemoteCAs = remoteCAs

+}

+

 // NeedsRemoteSigner returns true if one of the profiles has a remote set

 func (p *Signing) NeedsRemoteSigner() bool {

 	for _, profile := range p.Profiles {

@@ -360,6 +410,11 @@ func (p *SigningProfile) validProfile(isDefault bool) bool {

 		return false

 	}

+	if p.AuthRemote.RemoteName == "" && p.AuthRemote.AuthKeyName != "" {

+		log.Debugf("invalid auth remote profile: no remote signer specified")

+		return false

+	}

+

 	if p.RemoteName != "" {

 		log.Debugf("validate remote profile")

@@ -375,6 +430,7 @@ func (p *SigningProfile) validProfile(isDefault bool) bool {

 		if p.AuthRemote.RemoteName != "" {

 			log.Debugf("invalid remote profile: auth remote is also specified")

+			return false

 		}

 	} else if p.AuthRemote.RemoteName != "" {

 		log.Debugf("validate auth remote profile")

@@ -409,6 +465,43 @@ func (p *SigningProfile) validProfile(isDefault bool) bool {

 	return true

 }

+// This checks if the SigningProfile object contains configurations that are only effective with a local signer

+// which has access to CA private key.

+func (p *SigningProfile) hasLocalConfig() bool {

+	if p.Usage != nil ||

+		p.IssuerURL != nil ||

+		p.OCSP != "" ||

+		p.ExpiryString != "" ||

+		p.BackdateString != "" ||

+		p.CAConstraint.IsCA != false ||

+		!p.NotBefore.IsZero() ||

+		!p.NotAfter.IsZero() ||

+		p.NameWhitelistString != "" ||

+		len(p.CTLogServers) != 0 {

+		return true

+	}

+	return false

+}

+

+// warnSkippedSettings prints a log warning message about skipped settings

+// in a SigningProfile, usually due to remote signer.

+func (p *Signing) warnSkippedSettings() {

+	const warningMessage = `The configuration value by "usages", "issuer_urls", "ocsp_url", "crl_url", "ca_constraint", "expiry", "backdate", "not_before", "not_after", "cert_store" and "ct_log_servers" are skipped`

+	if p == nil {

+		return

+	}

+

+	if (p.Default.RemoteName != "" || p.Default.AuthRemote.RemoteName != "") && p.Default.hasLocalConfig() {

+		log.Warning("default profile points to a remote signer: ", warningMessage)

+	}

+

+	for name, profile := range p.Profiles {

+		if (profile.RemoteName != "" || profile.AuthRemote.RemoteName != "") && profile.hasLocalConfig() {

+			log.Warningf("Profiles[%s] points to a remote signer: %s", name, warningMessage)

+		}

+	}

+}

+

 // Signing codifies the signature configuration policy for a CA.

 type Signing struct {

 	Profiles map[string]*SigningProfile `json:"profiles"`

@@ -450,6 +543,9 @@ func (p *Signing) Valid() bool {

 			return false

 		}

 	}

+

+	p.warnSkippedSettings()

+

 	return true

 }

@@ -149,6 +149,8 @@ const (

 	// UnknownProfile indicates that the profile does not exist.

 	UnknownProfile // 54XX

+

+	UnmatchedWhitelist // 55xx

 )

 // The following are API client related errors, and should be

@@ -313,6 +315,8 @@ func New(category Category, reason Reason) *Error {

 			msg = "Policy violation request"

 		case UnknownProfile:

 			msg = "Unknown policy profile"

+		case UnmatchedWhitelist:

+			msg = "Request does not match policy whitelist"

 		default:

 			panic(fmt.Sprintf("Unsupported CFSSL error reason %d under category PolicyError.",

 				reason))

@@ -8,6 +8,7 @@ import (

 	"crypto/ecdsa"

 	"crypto/elliptic"

 	"crypto/rsa"

+	"crypto/tls"

 	"crypto/x509"

 	"encoding/asn1"

 	"encoding/pem"

@@ -311,11 +312,23 @@ func ParseOneCertificateFromPEM(certsPEM []byte) ([]*x509.Certificate, []byte, e

 // LoadPEMCertPool loads a pool of PEM certificates from file.

 func LoadPEMCertPool(certsFile string) (*x509.CertPool, error) {

+	if certsFile == "" {

+		return nil, nil

+	}

 	pemCerts, err := ioutil.ReadFile(certsFile)

 	if err != nil {

 		return nil, err

 	}

+	return PEMToCertPool(pemCerts)

+}

+

+// PEMToCertPool concerts PEM certificates to a CertPool.

+func PEMToCertPool(pemCerts []byte) (*x509.CertPool, error) {

+	if len(pemCerts) == 0 {

+		return nil, nil

+	}

+

 	certPool := x509.NewCertPool()

 	if !certPool.AppendCertsFromPEM(pemCerts) {

 		return nil, errors.New("failed to load cert pool")

@@ -477,3 +490,29 @@ func SignerAlgo(priv crypto.Signer) x509.SignatureAlgorithm {

 		return x509.UnknownSignatureAlgorithm

 	}

 }

+

+// LoadClientCertificate load key/certificate from pem files

+func LoadClientCertificate(certFile string, keyFile string) (*tls.Certificate, error) {

+	if certFile != "" && keyFile != "" {

+		cert, err := tls.LoadX509KeyPair(certFile, keyFile)

+		if err != nil {

+			log.Critical("Unable to read client certificate from file: %s or key from file: %s", certFile, keyFile)

+			return nil, err

+		}

+		log.Debug("Client certificate loaded ")

+		return &cert, nil

+	}

+	return nil, nil

+}

+

+// CreateTLSConfig creates a tls.Config object from certs and roots

+func CreateTLSConfig(remoteCAs *x509.CertPool, cert *tls.Certificate) *tls.Config {

+	var certs []tls.Certificate

+	if cert != nil {

+		certs = []tls.Certificate{*cert}

+	}

+	return &tls.Config{

+		Certificates: certs,

+		RootCAs:      remoteCAs,

+	}

+}

@@ -48,13 +48,16 @@ func New(req *csr.CertificateRequest) (cert, csrPEM, key []byte, err error) {

 		if req.CA.Expiry != "" {

 			policy.Default.ExpiryString = req.CA.Expiry

 			policy.Default.Expiry, err = time.ParseDuration(req.CA.Expiry)

+			if err != nil {

+				return

+			}

 		}

-		signer.MaxPathLen = req.CA.PathLength

+		policy.Default.CAConstraint.MaxPathLen = req.CA.PathLength

 		if req.CA.PathLength != 0 && req.CA.PathLenZero == true {

 			log.Infof("ignore invalid 'pathlenzero' value")

 		} else {

-			signer.MaxPathLenZero = req.CA.PathLenZero

+			policy.Default.CAConstraint.MaxPathLenZero = req.CA.PathLenZero

 		}

 	}

@@ -72,12 +75,11 @@ func New(req *csr.CertificateRequest) (cert, csrPEM, key []byte, err error) {

 		return

 	}

-	s, err := local.NewSigner(priv, nil, signer.DefaultSigAlgo(priv), nil)

+	s, err := local.NewSigner(priv, nil, signer.DefaultSigAlgo(priv), policy)

 	if err != nil {

 		log.Errorf("failed to create signer: %v", err)

 		return

 	}

-	s.SetPolicy(policy)

 	signReq := signer.SignRequest{Hosts: req.Hosts, Request: string(csrPEM)}

 	cert, err = s.Sign(signReq)

@@ -143,11 +145,11 @@ func NewFromSigner(req *csr.CertificateRequest, priv crypto.Signer) (cert, csrPE

 			}

 		}

-		signer.MaxPathLen = req.CA.PathLength

+		policy.Default.CAConstraint.MaxPathLen = req.CA.PathLength

 		if req.CA.PathLength != 0 && req.CA.PathLenZero == true {

 			log.Infof("ignore invalid 'pathlenzero' value")

 		} else {

-			signer.MaxPathLenZero = req.CA.PathLenZero

+			policy.Default.CAConstraint.MaxPathLenZero = req.CA.PathLenZero

 		}

 	}

@@ -156,12 +158,11 @@ func NewFromSigner(req *csr.CertificateRequest, priv crypto.Signer) (cert, csrPE

 		return nil, nil, err

 	}

-	s, err := local.NewSigner(priv, nil, signer.DefaultSigAlgo(priv), nil)

+	s, err := local.NewSigner(priv, nil, signer.DefaultSigAlgo(priv), policy)

 	if err != nil {

 		log.Errorf("failed to create signer: %v", err)

 		return

 	}

-	s.SetPolicy(policy)

 	signReq := signer.SignRequest{Request: string(csrPEM)}

 	cert, err = s.Sign(signReq)

@@ -217,7 +218,7 @@ var CAPolicy = func() *config.Signing {

 			Usage:        []string{"cert sign", "crl sign"},

 			ExpiryString: "43800h",

 			Expiry:       5 * helpers.OneYear,

-			CA:           true,

+			CAConstraint: config.CAConstraint{IsCA: true},

 		},

 	}

 }

@@ -6,7 +6,6 @@

 package log

 import (

-	"flag"

 	"fmt"

 	"log"

 	"os"

@@ -63,13 +62,6 @@ func SetLogger(logger SyslogWriter) {

 	syslogWriter = logger

 }

-func init() {

-	// Only define loglevel flag once.

-	if flag.Lookup("loglevel") == nil {

-		flag.IntVar(&Level, "loglevel", LevelInfo, "Log level (0 = DEBUG, 5 = FATAL)")

-	}

-}

-

 func print(l int, msg string) {

 	if l >= Level {

 		if syslogWriter != nil {

@@ -115,9 +115,6 @@ func (s *Signer) sign(template *x509.Certificate, profile *config.SigningProfile

 		template.EmailAddresses = nil

 		s.ca = template

 		initRoot = true

-	} else if template.IsCA {

-		template.DNSNames = nil

-		template.EmailAddresses = nil

 	}

 	derBytes, err := x509.CreateCertificate(rand.Reader, template, s.ca, template.PublicKey, s.priv)

@@ -250,18 +247,21 @@ func (s *Signer) Sign(req signer.SignRequest) (cert []byte, err error) {

 	}

 	if safeTemplate.IsCA {

-		if !profile.CA {

-			return nil, cferr.New(cferr.CertificateError, cferr.InvalidRequest)

+		if !profile.CAConstraint.IsCA {

+			log.Error("local signer policy disallows issuing CA certificate")

+			return nil, cferr.New(cferr.PolicyError, cferr.InvalidRequest)

 		}

 		if s.ca != nil && s.ca.MaxPathLen > 0 {

 			if safeTemplate.MaxPathLen >= s.ca.MaxPathLen {

+				log.Error("local signer certificate disallows CA MaxPathLen extending")

 				// do not sign a cert with pathlen > current

-				return nil, cferr.New(cferr.CertificateError, cferr.InvalidRequest)

+				return nil, cferr.New(cferr.PolicyError, cferr.InvalidRequest)

 			}

 		} else if s.ca != nil && s.ca.MaxPathLen == 0 && s.ca.MaxPathLenZero {

+			log.Error("local signer certificate disallows issuing CA certificate")

 			// signer has pathlen of 0, do not sign more intermediate CAs

-			return nil, cferr.New(cferr.CertificateError, cferr.InvalidRequest)

+			return nil, cferr.New(cferr.PolicyError, cferr.InvalidRequest)

 		}

 	}

@@ -272,17 +272,17 @@ func (s *Signer) Sign(req signer.SignRequest) (cert []byte, err error) {

 	if profile.NameWhitelist != nil {

 		if safeTemplate.Subject.CommonName != "" {

 			if profile.NameWhitelist.Find([]byte(safeTemplate.Subject.CommonName)) == nil {

-				return nil, cferr.New(cferr.PolicyError, cferr.InvalidPolicy)

+				return nil, cferr.New(cferr.PolicyError, cferr.UnmatchedWhitelist)

 			}

 		}

 		for _, name := range safeTemplate.DNSNames {

 			if profile.NameWhitelist.Find([]byte(name)) == nil {

-				return nil, cferr.New(cferr.PolicyError, cferr.InvalidPolicy)

+				return nil, cferr.New(cferr.PolicyError, cferr.UnmatchedWhitelist)

 			}

 		}

 		for _, name := range safeTemplate.EmailAddresses {

 			if profile.NameWhitelist.Find([]byte(name)) == nil {

-				return nil, cferr.New(cferr.PolicyError, cferr.InvalidPolicy)

+				return nil, cferr.New(cferr.PolicyError, cferr.UnmatchedWhitelist)

 			}

 		}

 	}

@@ -352,7 +352,7 @@ func (s *Signer) Sign(req signer.SignRequest) (cert []byte, err error) {

 		for _, server := range profile.CTLogServers {

 			log.Infof("submitting poisoned precertificate to %s", server)

-			var ctclient = client.New(server)

+			var ctclient = client.New(server, nil)

 			var resp *ct.SignedCertificateTimestamp

 			resp, err = ctclient.AddPreChain(prechain)

 			if err != nil {

@@ -23,12 +23,6 @@ import (

 	"github.com/cloudflare/cfssl/info"

 )

-// MaxPathLen is the default path length for a new CA certificate.

-var MaxPathLen = 2

-

-// MaxPathLenZero indicates whether a new CA certificate has pathlen=0

-var MaxPathLenZero = false

-

 // Subject contains the information that should be used to override the

 // subject information when signing a certificate.

 type Subject struct {

@@ -294,7 +288,15 @@ func FillTemplate(template *x509.Certificate, defaultProfile, profile *config.Si

 	template.KeyUsage = ku

 	template.ExtKeyUsage = eku

 	template.BasicConstraintsValid = true

-	template.IsCA = profile.CA

+	template.IsCA = profile.CAConstraint.IsCA

+	if template.IsCA {

+		template.MaxPathLen = profile.CAConstraint.MaxPathLen

+		if template.MaxPathLen == 0 {

+			template.MaxPathLenZero = profile.CAConstraint.MaxPathLenZero

+		}

+		template.DNSNames = nil

+		template.EmailAddresses = nil

+	}

 	template.SubjectKeyId = ski

 	if ocspURL != "" {

@@ -96,3 +96,26 @@ func Exist(name string) bool {

 	_, err := os.Stat(name)

 	return err == nil

 }

+

+// ZeroToEnd zeros a file starting from SEEK_CUR to its SEEK_END. May temporarily

+// shorten the length of the file.

+func ZeroToEnd(f *os.File) error {

+	// TODO: support FALLOC_FL_ZERO_RANGE

+	off, err := f.Seek(0, os.SEEK_CUR)

+	if err != nil {

+		return err

+	}

+	lenf, lerr := f.Seek(0, os.SEEK_END)

+	if lerr != nil {

+		return lerr

+	}

+	if err = f.Truncate(off); err != nil {

+		return err

+	}

+	// make sure blocks remain allocated

+	if err = Preallocate(f, lenf, true); err != nil {

+		return err

+	}

+	_, err = f.Seek(off, os.SEEK_SET)

+	return err

+}

@@ -232,7 +232,7 @@ func (l *raftLog) term(i uint64) (uint64, error) {

 	if err == nil {

 		return t, nil

 	}

-	if err == ErrCompacted {

+	if err == ErrCompacted || err == ErrUnavailable {

 		return 0, err

 	}

 	panic(err) // TODO(bdarnell)

@@ -339,7 +339,7 @@ func (l *raftLog) mustCheckOutOfBounds(lo, hi uint64) error {

 		return ErrCompacted

 	}

-	length := l.lastIndex() - fi + 1

+	length := l.lastIndex() + 1 - fi

 	if lo < fi || hi > fi+length {

 		l.logger.Panicf("slice[%d,%d) out of bound [%d,%d]", lo, hi, fi, l.lastIndex())

 	}

@@ -144,6 +144,9 @@ type Node interface {

 	// to match MemoryStorage.Compact.

 	ApplyConfChange(cc pb.ConfChange) *pb.ConfState

+	// TransferLeadership attempts to transfer leadership to the given transferee.

+	TransferLeadership(ctx context.Context, lead, transferee uint64)

+

 	// ReadIndex request a read state. The read state will be set in the ready.

 	// Read state has a read index. Once the application advances further than the read

 	// index, any linearizable read requests issued before the read request can be

@@ -485,6 +488,15 @@ func (n *node) ReportSnapshot(id uint64, status SnapshotStatus) {

 	}

 }

+func (n *node) TransferLeadership(ctx context.Context, lead, transferee uint64) {

+	select {

+	// manually set 'from' and 'to', so that leader can voluntarily transfers its leadership

+	case n.recvc <- pb.Message{Type: pb.MsgTransferLeader, From: transferee, To: lead}:

+	case <-n.done:

+	case <-ctx.Done():

+	}

+}

+

 func (n *node) ReadIndex(ctx context.Context, rctx []byte) error {

 	return n.step(ctx, pb.Message{Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: rctx}}})

 }

@@ -590,11 +590,6 @@ func (r *raft) Step(m pb.Message) error {

 		}

 		return nil

 	}

-	if m.Type == pb.MsgTransferLeader {

-		if r.state != StateLeader {

-			r.logger.Debugf("%x [term %d state %v] ignoring MsgTransferLeader to %x", r.id, r.Term, r.state, m.From)

-		}

-	}

 	switch {

 	case m.Term == 0:

@@ -874,6 +869,13 @@ func stepFollower(r *raft, m pb.Message) {

 				r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)

 			r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})

 		}

+	case pb.MsgTransferLeader:

+		if r.lead == None {

+			r.logger.Infof("%x no leader at term %d; dropping leader transfer msg", r.id, r.Term)

+			return

+		}

+		m.To = r.lead

+		r.send(m)

 	case pb.MsgTimeoutNow:

 		r.logger.Infof("%x [term %d] received MsgTimeoutNow from %x and starts an election to get leadership.", r.id, r.Term, m.From)

 		r.campaign(campaignTransfer)

@@ -130,6 +130,9 @@ func (ms *MemoryStorage) Term(i uint64) (uint64, error) {

 	if i < offset {

 		return 0, ErrCompacted

 	}

+	if int(i-offset) >= len(ms.ents) {

+		return 0, ErrUnavailable

+	}

 	return ms.ents[i-offset].Term, nil

 }

@@ -48,7 +48,7 @@ func max(a, b uint64) uint64 {

 func IsLocalMsg(msgt pb.MessageType) bool {

 	return msgt == pb.MsgHup || msgt == pb.MsgBeat || msgt == pb.MsgUnreachable ||

-		msgt == pb.MsgSnapStatus || msgt == pb.MsgCheckQuorum || msgt == pb.MsgTransferLeader

+		msgt == pb.MsgSnapStatus || msgt == pb.MsgCheckQuorum

 }

 func IsResponseMsg(msgt pb.MessageType) bool {

@@ -131,22 +131,7 @@ func Create(dirpath string, metadata []byte) (*WAL, error) {

 		return nil, err

 	}

-	// rename of directory with locked files doesn't work on windows; close

-	// the WAL to release the locks so the directory can be renamed

-	w.Close()

-	if err := os.Rename(tmpdirpath, dirpath); err != nil {

-		return nil, err

-	}

-	// reopen and relock

-	newWAL, oerr := Open(dirpath, walpb.Snapshot{})

-	if oerr != nil {

-		return nil, oerr

-	}

-	if _, _, _, err := newWAL.ReadAll(); err != nil {

-		newWAL.Close()

-		return nil, err

-	}

-	return newWAL, nil

+	return w.renameWal(tmpdirpath)

 }

 // Open opens the WAL at the given snap.

@@ -301,6 +286,18 @@ func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.

 			state.Reset()

 			return nil, state, nil, err

 		}

+		// decodeRecord() will return io.EOF if it detects a zero record,

+		// but this zero record may be followed by non-zero records from

+		// a torn write. Overwriting some of these non-zero records, but

+		// not all, will cause CRC errors on WAL open. Since the records

+		// were never fully synced to disk in the first place, it's safe

+		// to zero them out to avoid any CRC errors from new writes.

+		if _, err = w.tail().Seek(w.decoder.lastOffset(), os.SEEK_SET); err != nil {

+			return nil, state, nil, err

+		}

+		if err = fileutil.ZeroToEnd(w.tail().File); err != nil {

+			return nil, state, nil, err

+		}

 	}

 	err = nil

@@ -319,7 +316,6 @@ func (w *WAL) ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.

 	if w.tail() != nil {

 		// create encoder (chain crc with the decoder), enable appending

-		_, err = w.tail().Seek(w.decoder.lastOffset(), os.SEEK_SET)

 		w.encoder = newEncoder(w.tail(), w.decoder.lastCRC())

 	}

 	w.decoder = nil

new file mode 100644

@@ -0,0 +1,38 @@

+// Copyright 2016 The etcd Authors

+//

+// 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.

+

+// +build !windows

+

+package wal

+

+import "os"

+

+func (w *WAL) renameWal(tmpdirpath string) (*WAL, error) {

+	// On non-Windows platforms, hold the lock while renaming. Releasing

+	// the lock and trying to reacquire it quickly can be flaky because

+	// it's possible the process will fork to spawn a process while this is

+	// happening. The fds are set up as close-on-exec by the Go runtime,

+	// but there is a window between the fork and the exec where another

+	// process holds the lock.

+

+	if err := os.RemoveAll(w.dir); err != nil {

+		return nil, err

+	}

+	if err := os.Rename(tmpdirpath, w.dir); err != nil {

+		return nil, err

+	}

+

+	w.fp = newFilePipeline(w.dir, SegmentSizeBytes)

+	return w, nil

+}

new file mode 100644

@@ -0,0 +1,41 @@

+// Copyright 2016 The etcd Authors

+//

+// 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.

+

+package wal

+

+import (

+	"os"

+

+	"github.com/coreos/etcd/wal/walpb"

+)

+

+func (w *WAL) renameWal(tmpdirpath string) (*WAL, error) {

+	// rename of directory with locked files doesn't work on

+	// windows; close the WAL to release the locks so the directory

+	// can be renamed

+	w.Close()

+	if err := os.Rename(tmpdirpath, w.dir); err != nil {

+		return nil, err

+	}

+	// reopen and relock

+	newWAL, oerr := Open(w.dir, walpb.Snapshot{})

+	if oerr != nil {

+		return nil, oerr

+	}

+	if _, _, _, err := newWAL.ReadAll(); err != nil {

+		newWAL.Close()

+		return nil, err

+	}

+	return newWAL, nil

+}

@@ -7,7 +7,7 @@ import (

 )

 // NewTCPSocket creates a TCP socket listener with the specified address and

-// the specified tls configuration. If TLSConfig is set, will encapsulate the

+// and the specified tls configuration. If TLSConfig is set, will encapsulate the

 // TCP listener inside a TLS one.

 func NewTCPSocket(addr string, tlsConfig *tls.Config) (net.Listener, error) {

 	l, err := net.Listen("tcp", addr)

@@ -1,6 +1,11 @@

 package events

-import "github.com/Sirupsen/logrus"

+import (

+	"fmt"

+	"sync"

+

+	"github.com/Sirupsen/logrus"

+)

 // Broadcaster sends events to multiple, reliable Sinks. The goal of this

 // component is to dispatch events to configured endpoints. Reliability can be

@@ -10,7 +15,10 @@ type Broadcaster struct {

 	events  chan Event

 	adds    chan configureRequest

 	removes chan configureRequest

-	closed  chan chan struct{}

+

+	shutdown chan struct{}

+	closed   chan struct{}

+	once     sync.Once

 }

 // NewBroadcaster appends one or more sinks to the list of sinks. The

@@ -19,11 +27,12 @@ type Broadcaster struct {

 // its own. Use of EventQueue and RetryingSink should be used here.

 func NewBroadcaster(sinks ...Sink) *Broadcaster {

 	b := Broadcaster{

-		sinks:   sinks,

-		events:  make(chan Event),

-		adds:    make(chan configureRequest),

-		removes: make(chan configureRequest),

-		closed:  make(chan chan struct{}),

+		sinks:    sinks,

+		events:   make(chan Event),

+		adds:     make(chan configureRequest),

+		removes:  make(chan configureRequest),

+		shutdown: make(chan struct{}),

+		closed:   make(chan struct{}),

 	}

 	// Start the broadcaster

@@ -82,24 +91,19 @@ func (b *Broadcaster) configure(ch chan configureRequest, sink Sink) error {

 // Close the broadcaster, ensuring that all messages are flushed to the

 // underlying sink before returning.

 func (b *Broadcaster) Close() error {

-	select {

-	case <-b.closed:

-		// already closed

-		return ErrSinkClosed

-	default:

-		// do a little chan handoff dance to synchronize closing

-		closed := make(chan struct{})

-		b.closed <- closed

-		close(b.closed)

-		<-closed

-		return nil

-	}

+	b.once.Do(func() {

+		close(b.shutdown)

+	})

+

+	<-b.closed

+	return nil

 }

 // run is the main broadcast loop, started when the broadcaster is created.

 // Under normal conditions, it waits for events on the event channel. After

 // Close is called, this goroutine will exit.

 func (b *Broadcaster) run() {

+	defer close(b.closed)

 	remove := func(target Sink) {

 		for i, sink := range b.sinks {

 			if sink == target {

@@ -143,7 +147,7 @@ func (b *Broadcaster) run() {

 		case request := <-b.removes:

 			remove(request.sink)

 			request.response <- nil

-		case closing := <-b.closed:

+		case <-b.shutdown:

 			// close all the underlying sinks

 			for _, sink := range b.sinks {

 				if err := sink.Close(); err != nil && err != ErrSinkClosed {

@@ -151,8 +155,24 @@ func (b *Broadcaster) run() {

 						Errorf("broadcaster: closing sink failed")

 				}

 			}

-			closing <- struct{}{}

 			return

 		}

 	}