Support ordered rolling for v2.3+ (#272)

Signed-off-by: shaoyue.chen <shaoyue.chen@zilliz.com>

apis/milvus.io/v1beta1/milvus_types.go

@@ -17,8 +17,12 @@ limitations under the License.
 package v1beta1
 
 import (
+	"strings"
 	"time"
 
+	"github.com/coreos/go-semver/semver"
+	"github.com/milvus-io/milvus-operator/pkg/provisioner"
+	"github.com/pkg/errors"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	networkv1 "k8s.io/api/networking/v1"
@@ -100,6 +104,16 @@ func (ms MilvusSpec) GetServiceComponent() *ServiceComponent {
 	return &ms.Com.Standalone.ServiceComponent
 }
 
+// GetMilvusVersionByImage returns the version of Milvus by ms.Com.ComponentSpec.Image
+func (ms MilvusSpec) GetMilvusVersionByImage() (*semver.Version, error) {
+	// parse format: registry/namespace/image:tag
+	splited := strings.Split(ms.Com.ComponentSpec.Image, ":")
+	if len(splited) != 2 {
+		return nil, errors.Errorf("unknown version of image[%s]", splited[0])
+	}
+	return provisioner.GetSemanticVersion(splited[1])
+}
+
 // MilvusMode defines the mode of Milvus deployment
 type MilvusMode string
 

apis/milvus.io/v1beta1/milvus_types_test.go

@@ -194,3 +194,25 @@ func TestMilvus_IsPodServiceLabelAdded(t *testing.T) {
 		assert.False(t, m.IsPodServiceLabelAdded())
 	})
 }
+
+func TestGetMilvusVersionByGlobalImage(t *testing.T) {
+	m := Milvus{}
+	_, err := m.Spec.GetMilvusVersionByImage()
+	assert.Error(t, err)
+
+	m.Default()
+	_, err = m.Spec.GetMilvusVersionByImage()
+	assert.NoError(t, err)
+
+	m.Spec.Com.ComponentSpec.Image = "milvusdb/milvus:v2.3.1-beta1"
+	ver, err := m.Spec.GetMilvusVersionByImage()
+	assert.NoError(t, err)
+	assert.Equal(t, int64(2), ver.Major)
+	assert.Equal(t, int64(3), ver.Minor)
+	assert.Equal(t, int64(1), ver.Patch)
+	assert.Equal(t, "beta1", ver.PreRelease.Slice()[0])
+
+	m.Spec.Com.ComponentSpec.Image = "harbor.milvus.io/milvus/milvus:latest"
+	_, err = m.Spec.GetMilvusVersionByImage()
+	assert.Error(t, err)
+}

pkg/controllers/components.go

@@ -365,10 +365,19 @@ func (c MilvusComponent) GetDependencies(spec v1beta1.MilvusSpec) []MilvusCompon
 		return []MilvusComponent{}
 	}
 	// cluster mode
+	var depGraph = clusterDependencyGraph
+	semeticVersion, err := spec.GetMilvusVersionByImage()
+	if err != nil || semeticVersion.Major < 2 {
+		// unknown version, ordered rolling not supported
+		return []MilvusComponent{}
+	}
+	// new upgrade order starts from v2.3.0
+	if semeticVersion.Major >= 3 || semeticVersion.Minor >= 3 {
+		depGraph = clusterDependencyGraphV2p3
+	}
+
 	isMixCoord := spec.Com.MixCoord != nil
 	isDowngrade := spec.Com.ImageUpdateMode == v1beta1.ImageUpdateModeRollingDowngrade
-
-	var depGraph = clusterDependencyGraph
 	if isMixCoord {
 		depGraph = mixCoordClusterDependencyGraph
 	}

pkg/controllers/components_test.go

@@ -518,12 +518,58 @@ func TestMilvusComponent_SetReplicas(t *testing.T) {
 }
 
 func TestMilvusComponent_GetDependencies(t *testing.T) {
-	m := v1beta1.Milvus{}
-	assert.Len(t, MilvusStandalone.GetDependencies(m.Spec), 0)
+	t.Run("standalone no dep", func(t *testing.T) {
+		m := v1beta1.Milvus{}
+		assert.Len(t, MilvusStandalone.GetDependencies(m.Spec), 0)
+	})
+
+	t.Run("cluster unknown image no dep", func(t *testing.T) {
+		m := v1beta1.Milvus{}
+		m.Spec.Mode = v1beta1.MilvusModeCluster
+		assert.Len(t, Proxy.GetDependencies(m.Spec), 0)
+		assert.Len(t, QueryNode.GetDependencies(m.Spec), 0)
+	})
+
+	t.Run("cluster 2.3+", func(t *testing.T) {
+		m := v1beta1.Milvus{}
+		m.Spec.Mode = v1beta1.MilvusModeCluster
+		m.Spec.Com.Image = "milvusdb/milvus:v2.3.0"
+		assert.Len(t, RootCoord.GetDependencies(m.Spec), 0)
+		assert.Equal(t, RootCoord, IndexCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, IndexCoord, DataCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, DataCoord, QueryCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, QueryCoord, DataNode.GetDependencies(m.Spec)[0])
+		assert.Equal(t, QueryCoord, IndexNode.GetDependencies(m.Spec)[0])
+		assert.Equal(t, QueryCoord, QueryNode.GetDependencies(m.Spec)[0])
+		assert.Len(t, Proxy.GetDependencies(m.Spec), 3)
+
+		// if we have a new major version
+		m.Spec.Com.Image = "milvusdb/milvus:v3.0.0"
+		assert.Equal(t, RootCoord, IndexCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, IndexCoord, DataCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, DataCoord, QueryCoord.GetDependencies(m.Spec)[0])
+	})
+
+	t.Run("clusterModeDowngrade 2.3+", func(t *testing.T) {
+		m := v1beta1.Milvus{}
+		m.Spec.Com.Image = "milvusdb/milvus:2.3.0"
+		m.Spec.Mode = v1beta1.MilvusModeCluster
+		m.Spec.Com.ImageUpdateMode = v1beta1.ImageUpdateModeRollingDowngrade
+		m.Default()
+		assert.Equal(t, IndexCoord, RootCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, DataCoord, IndexCoord.GetDependencies(m.Spec)[0])
+		assert.Equal(t, QueryCoord, DataCoord.GetDependencies(m.Spec)[0])
+		assert.Len(t, QueryCoord.GetDependencies(m.Spec), 3)
+		assert.Equal(t, Proxy, DataNode.GetDependencies(m.Spec)[0])
+		assert.Equal(t, Proxy, IndexNode.GetDependencies(m.Spec)[0])
+		assert.Equal(t, Proxy, QueryNode.GetDependencies(m.Spec)[0])
+		assert.Len(t, Proxy.GetDependencies(m.Spec), 0)
+	})
 
-	t.Run("clusterMode", func(t *testing.T) {
+	t.Run("clusterMode 2.2", func(t *testing.T) {
 		m := v1beta1.Milvus{}
 		m.Spec.Mode = v1beta1.MilvusModeCluster
+		m.Spec.Com.Image = "milvusdb/milvus:v2.2.4"
 		m.Default()
 		assert.Len(t, RootCoord.GetDependencies(m.Spec), 0)
 		assert.Equal(t, RootCoord, DataCoord.GetDependencies(m.Spec)[0])
@@ -535,8 +581,9 @@ func TestMilvusComponent_GetDependencies(t *testing.T) {
 		assert.Len(t, Proxy.GetDependencies(m.Spec), 3)
 	})
 
-	t.Run("clusterModeDowngrade", func(t *testing.T) {
+	t.Run("clusterModeDowngrade 2.2", func(t *testing.T) {
 		m := v1beta1.Milvus{}
+		m.Spec.Com.Image = "milvusdb/milvus:2.2.4"
 		m.Spec.Mode = v1beta1.MilvusModeCluster
 		m.Spec.Com.ImageUpdateMode = v1beta1.ImageUpdateModeRollingDowngrade
 		m.Default()

pkg/controllers/dependency_graph.go

@@ -39,6 +39,7 @@ func (d *dependencyGraphImpl) GetReversedDependencies(component MilvusComponent)
 }
 
 func init() {
+	// v2.2.x
 	clusterDependencyGraph.AddDependency(RootCoord, []MilvusComponent{})
 	clusterDependencyGraph.AddDependency(DataCoord, []MilvusComponent{RootCoord})
 	clusterDependencyGraph.AddDependency(IndexCoord, []MilvusComponent{DataCoord})
@@ -48,11 +49,24 @@ func init() {
 	clusterDependencyGraph.AddDependency(DataNode, []MilvusComponent{QueryCoord})
 	clusterDependencyGraph.AddDependency(Proxy, []MilvusComponent{IndexNode, QueryNode, DataNode})
 
+	// v2.3+
+	clusterDependencyGraphV2p3.AddDependency(RootCoord, []MilvusComponent{})
+	clusterDependencyGraphV2p3.AddDependency(IndexCoord, []MilvusComponent{RootCoord})
+	clusterDependencyGraphV2p3.AddDependency(DataCoord, []MilvusComponent{IndexCoord})
+	clusterDependencyGraphV2p3.AddDependency(QueryCoord, []MilvusComponent{DataCoord})
+	clusterDependencyGraphV2p3.AddDependency(IndexNode, []MilvusComponent{QueryCoord})
+	clusterDependencyGraphV2p3.AddDependency(QueryNode, []MilvusComponent{QueryCoord})
+	clusterDependencyGraphV2p3.AddDependency(DataNode, []MilvusComponent{QueryCoord})
+	clusterDependencyGraphV2p3.AddDependency(Proxy, []MilvusComponent{IndexNode, QueryNode, DataNode})
+
 	mixCoordClusterDependencyGraph.AddDependency(MixCoord, []MilvusComponent{})
 	mixCoordClusterDependencyGraph.AddDependency(IndexNode, []MilvusComponent{MixCoord})
 	mixCoordClusterDependencyGraph.AddDependency(QueryNode, []MilvusComponent{MixCoord})
 	mixCoordClusterDependencyGraph.AddDependency(DataNode, []MilvusComponent{MixCoord})
 	mixCoordClusterDependencyGraph.AddDependency(Proxy, []MilvusComponent{IndexNode, QueryNode, DataNode})
 }
 
-var clusterDependencyGraph, mixCoordClusterDependencyGraph dependencyGraph = newDependencyGraphImpl(), newDependencyGraphImpl()
+var (
+	clusterDependencyGraph, mixCoordClusterDependencyGraph = newDependencyGraphImpl(), newDependencyGraphImpl()
+	clusterDependencyGraphV2p3                             = newDependencyGraphImpl()
+)

pkg/controllers/deployment_updater_test.go

@@ -70,8 +70,8 @@ func TestMilvus_UpdateDeployment(t *testing.T) {
 		assert.Len(t, deployment.Spec.Template.Spec.Containers[0].VolumeMounts, 3)
 	})
 
-	const oldImage = "milvusdb/milvus:old"
-	const newImage = "milvusdb/milvus:new"
+	const oldImage = "milvusdb/milvus:2.2.13"
+	const newImage = "milvusdb/milvus:2.3.0"
 
 	t.Run("rolling update image", func(t *testing.T) {
 		inst := env.Inst.DeepCopy()

pkg/provisioner/cert_manager.go

@@ -177,13 +177,13 @@ func certManagerCRDsExist(crdMap map[string]string) bool {
 	return true
 }
 
-func getSemanticVersion(version string) (*semver.Version, error) {
+func GetSemanticVersion(version string) (*semver.Version, error) {
 	return semver.NewVersion(strings.TrimPrefix(version, "v"))
 }
 
 func certManagerVersionSatisfied(crdVersionMap map[string]string) bool {
 	for _, crdName := range certManagerCrdNames {
-		currentVersion, err := getSemanticVersion(crdVersionMap[crdName])
+		currentVersion, err := GetSemanticVersion(crdVersionMap[crdName])
 		if err != nil {
 			err = errors.Wrapf(err, "failed to parse crd version")
 			logger.Error(err, "crdName", crdName, "version", crdVersionMap[crdName])