etcdserver: check cluster version compability when joining

etcdserver/cluster_util.go

@@ -118,7 +118,11 @@ func getVersions(cl Cluster, local types.ID, tr *http.Transport) map[string]*ver
 	vers := make(map[string]*version.Versions)
 	for _, m := range members {
 		if m.ID == local {
-			vers[m.ID.String()] = &version.Versions{Server: version.Version, Cluster: cl.Version().String()}
+			cv := "not_decided"
+			if cl.Version() != nil {
+				cv = cl.Version().String()
+			}
+			vers[m.ID.String()] = &version.Versions{Server: version.Version, Cluster: cv}
 			continue
 		}
 		ver, err := getVersion(m, tr)
@@ -161,6 +165,52 @@ func decideClusterVersion(vers map[string]*version.Versions) *semver.Version {
 	return cv
 }
 
+// isCompatibleWithCluster return true if the local member has a compitable version with
+// the current running cluster.
+// The version is considered as compitable when at least one of the other members in the cluster has a
+// cluster version in the range of [MinClusterVersion, Version] and no known members has a cluster version
+// out of the range.
+// We set this rule since when the local member joins, another member might be offline.
+func isCompatibleWithCluster(cl Cluster, local types.ID, tr *http.Transport) bool {
+	vers := getVersions(cl, local, tr)
+	minV := semver.Must(semver.NewVersion(version.MinClusterVersion))
+	maxV := semver.Must(semver.NewVersion(version.Version))
+	maxV = &semver.Version{
+		Major: maxV.Major,
+		Minor: maxV.Minor,
+	}
+
+	return isCompatibleWithVers(vers, local, minV, maxV)
+}
+
+func isCompatibleWithVers(vers map[string]*version.Versions, local types.ID, minV, maxV *semver.Version) bool {
+	var ok bool
+	for id, v := range vers {
+		// ignore comparasion with local version
+		if id == local.String() {
+			continue
+		}
+		if v == nil {
+			continue
+		}
+		clusterv, err := semver.NewVersion(v.Cluster)
+		if err != nil {
+			log.Printf("etcdserver: cannot understand the cluster version of member %s (%v)", id, err)
+			continue
+		}
+		if clusterv.LessThan(*minV) {
+			log.Printf("etcdserver: the running cluster version(%v) is lower than the minimal cluster version(%v) supported", clusterv.String(), minV.String())
+			return false
+		}
+		if maxV.LessThan(*clusterv) {
+			log.Printf("etcdserver: the running cluster version(%v) is higher than the maximum cluster version(%v) supported", clusterv.String(), maxV.String())
+			return false
+		}
+		ok = true
+	}
+	return ok
+}
+
 // getVersion returns the Versions of the given member via its
 // peerURLs. Returns the last error if it fails to get the version.
 func getVersion(m *Member, tr *http.Transport) (*version.Versions, error) {

etcdserver/cluster_util_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/coreos/etcd/Godeps/_workspace/src/github.com/coreos/go-semver/semver"
+	"github.com/coreos/etcd/pkg/types"
 	"github.com/coreos/etcd/version"
 )
 
@@ -57,3 +58,74 @@ func TestDecideClusterVersion(t *testing.T) {
 		}
 	}
 }
+
+func TestIsCompatibleWithVers(t *testing.T) {
+	tests := []struct {
+		vers       map[string]*version.Versions
+		local      types.ID
+		minV, maxV *semver.Version
+		wok        bool
+	}{
+		// too low
+		{
+			map[string]*version.Versions{
+				"a": &version.Versions{Server: "2.0.0", Cluster: "not_decided"},
+				"b": &version.Versions{Server: "2.1.0", Cluster: "2.1.0"},
+				"c": &version.Versions{Server: "2.1.0", Cluster: "2.1.0"},
+			},
+			0xa,
+			semver.Must(semver.NewVersion("2.0.0")), semver.Must(semver.NewVersion("2.0.0")),
+			false,
+		},
+		{
+			map[string]*version.Versions{
+				"a": &version.Versions{Server: "2.1.0", Cluster: "not_decided"},
+				"b": &version.Versions{Server: "2.1.0", Cluster: "2.1.0"},
+				"c": &version.Versions{Server: "2.1.0", Cluster: "2.1.0"},
+			},
+			0xa,
+			semver.Must(semver.NewVersion("2.0.0")), semver.Must(semver.NewVersion("2.1.0")),
+			true,
+		},
+		// too high
+		{
+			map[string]*version.Versions{
+				"a": &version.Versions{Server: "2.2.0", Cluster: "not_decided"},
+				"b": &version.Versions{Server: "2.0.0", Cluster: "2.0.0"},
+				"c": &version.Versions{Server: "2.0.0", Cluster: "2.0.0"},
+			},
+			0xa,
+			semver.Must(semver.NewVersion("2.1.0")), semver.Must(semver.NewVersion("2.2.0")),
+			false,
+		},
+		// cannot get b's version, expect ok
+		{
+			map[string]*version.Versions{
+				"a": &version.Versions{Server: "2.1.0", Cluster: "not_decided"},
+				"b": nil,
+				"c": &version.Versions{Server: "2.1.0", Cluster: "2.1.0"},
+			},
+			0xa,
+			semver.Must(semver.NewVersion("2.0.0")), semver.Must(semver.NewVersion("2.1.0")),
+			true,
+		},
+		// cannot get b and c's version, expect not ok
+		{
+			map[string]*version.Versions{
+				"a": &version.Versions{Server: "2.1.0", Cluster: "not_decided"},
+				"b": nil,
+				"c": nil,
+			},
+			0xa,
+			semver.Must(semver.NewVersion("2.0.0")), semver.Must(semver.NewVersion("2.1.0")),
+			false,
+		},
+	}
+
+	for i, tt := range tests {
+		ok := isCompatibleWithVers(tt.vers, tt.local, tt.minV, tt.maxV)
+		if ok != tt.wok {
+			t.Errorf("#%d: ok = %+v, want %+v", i, ok, tt.wok)
+		}
+	}
+}

etcdserver/server.go

@@ -209,6 +209,10 @@ func NewServer(cfg *ServerConfig) (*EtcdServer, error) {
 		if err := ValidateClusterAndAssignIDs(cl, existingCluster); err != nil {
 			return nil, fmt.Errorf("error validating peerURLs %s: %v", existingCluster, err)
 		}
+		if !isCompatibleWithCluster(cl, cl.MemberByName(cfg.Name).ID, cfg.Transport) {
+			return nil, fmt.Errorf("incomptible with current running cluster")
+		}
+
 		remotes = existingCluster.Members()
 		cl.SetID(existingCluster.id)
 		cl.SetStore(st)