controllers: implements woodpecker as mq

haorenfsa · haorenfsa · commit 48b972fac90d · 2025-03-28T10:48:57.000+08:00
Signed-off-by: haorenfsa &lt;shaoyue.chen@zilliz.com&gt;
diff --git a/apis/milvus.io/v1beta1/dependencies_types.go b/apis/milvus.io/v1beta1/dependencies_types.go
@@ -31,13 +31,13 @@ type MilvusDependencies struct {
 	Kafka MilvusKafka `json:"kafka,omitempty"`
 
 	// +kubebuilder:validation:Optional
-	WoodPecker MilvusBuildInMQ `json:"woodpecker,omitempty"`
+	WoodPecker MilvusBuiltInMQ `json:"woodpecker,omitempty"`
 
 	// +kubebuilder:validation:Optional
-	RocksMQ MilvusBuildInMQ `json:"rocksmq,omitempty"`
+	RocksMQ MilvusBuiltInMQ `json:"rocksmq,omitempty"`
 
 	// +kubebuilder:validation:Optional
-	NatsMQ MilvusBuildInMQ `json:"natsmq,omitempty"`
+	NatsMQ MilvusBuiltInMQ `json:"natsmq,omitempty"`
 
 	// +kubebuilder:validation:Optional
 	Storage MilvusStorage `json:"storage"`
@@ -50,7 +50,7 @@ type MilvusDependencies struct {
 	CustomMsgStream Values `json:"customMsgStream,omitempty"`
 }
 
-func (m *MilvusDependencies) GetMilvusBuildInMQ() *MilvusBuildInMQ {
+func (m *MilvusDependencies) GetMilvusBuiltInMQ() *MilvusBuiltInMQ {
 	switch m.MsgStreamType {
 	case MsgStreamTypePulsar, MsgStreamTypeKafka, MsgStreamTypeCustom:
 		return nil
@@ -139,8 +139,8 @@ type MilvusStorage struct {
 	External bool `json:"external,omitempty"`
 }
 
-// MilvusBuildInMQ (rocksmq or natsmq) configuration
-type MilvusBuildInMQ struct {
+// MilvusBuiltInMQ (rocksmq or natsmq) configuration
+type MilvusBuiltInMQ struct {
 	Persistence Persistence `json:"persistence,omitempty"`
 }
 
diff --git a/apis/milvus.io/v1beta1/dependencies_types_test.go b/apis/milvus.io/v1beta1/dependencies_types_test.go
@@ -6,29 +6,29 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestMilvusDependencies_GetMilvusBuildInMQ(t *testing.T) {
+func TestMilvusDependencies_GetMilvusBuiltInMQ(t *testing.T) {
 	m := &MilvusDependencies{
 		MsgStreamType: MsgStreamTypePulsar,
 	}
-	assert.Nil(t, m.GetMilvusBuildInMQ())
+	assert.Nil(t, m.GetMilvusBuiltInMQ())
 	m.MsgStreamType = MsgStreamTypeKafka
-	assert.Nil(t, m.GetMilvusBuildInMQ())
+	assert.Nil(t, m.GetMilvusBuiltInMQ())
 
 	m.MsgStreamType = MsgStreamTypeCustom
-	assert.Nil(t, m.GetMilvusBuildInMQ())
+	assert.Nil(t, m.GetMilvusBuiltInMQ())
 
 	m.MsgStreamType = MsgStreamType("unknown")
-	assert.Nil(t, m.GetMilvusBuildInMQ())
+	assert.Nil(t, m.GetMilvusBuiltInMQ())
 
 	m.MsgStreamType = MsgStreamTypeWoodPecker
-	assert.NotNil(t, m.GetMilvusBuildInMQ())
-	assert.Equal(t, &m.WoodPecker, m.GetMilvusBuildInMQ())
+	assert.NotNil(t, m.GetMilvusBuiltInMQ())
+	assert.Equal(t, &m.WoodPecker, m.GetMilvusBuiltInMQ())
 
 	m.MsgStreamType = MsgStreamTypeRocksMQ
-	assert.NotNil(t, m.GetMilvusBuildInMQ())
-	assert.Equal(t, &m.RocksMQ, m.GetMilvusBuildInMQ())
+	assert.NotNil(t, m.GetMilvusBuiltInMQ())
+	assert.Equal(t, &m.RocksMQ, m.GetMilvusBuiltInMQ())
 
 	m.MsgStreamType = MsgStreamTypeNatsMQ
-	assert.NotNil(t, m.GetMilvusBuildInMQ())
-	assert.Equal(t, &m.NatsMQ, m.GetMilvusBuildInMQ())
+	assert.NotNil(t, m.GetMilvusBuiltInMQ())
+	assert.Equal(t, &m.NatsMQ, m.GetMilvusBuiltInMQ())
 }
diff --git a/apis/milvus.io/v1beta1/milvus_types.go b/apis/milvus.io/v1beta1/milvus_types.go
@@ -128,6 +128,8 @@ func (ms *MilvusSpec) GetPersistenceConfig() *Persistence {
 		return &ms.Dep.RocksMQ.Persistence
 	case MsgStreamTypeNatsMQ:
 		return &ms.Dep.NatsMQ.Persistence
+	case MsgStreamTypeWoodPecker:
+		return &ms.Dep.WoodPecker.Persistence
 	}
 	return nil
 }
diff --git a/apis/milvus.io/v1beta1/milvus_types_test.go b/apis/milvus.io/v1beta1/milvus_types_test.go
@@ -224,11 +224,15 @@ func TestGetPersistenceConfig(t *testing.T) {
 	assert.Nil(t, m.Spec.GetPersistenceConfig())
 
 	m.Spec.Dep.MsgStreamType = MsgStreamTypeRocksMQ
-	m.Spec.Dep.NatsMQ.Persistence.Enabled = true
 	assert.Same(t, &m.Spec.Dep.RocksMQ.Persistence, m.Spec.GetPersistenceConfig())
 
+	m.Spec.Dep.NatsMQ.Persistence.Enabled = true
 	m.Spec.Dep.MsgStreamType = MsgStreamTypeNatsMQ
 	assert.Same(t, &m.Spec.Dep.NatsMQ.Persistence, m.Spec.GetPersistenceConfig())
+
+	m.Spec.Dep.WoodPecker.Persistence.Enabled = true
+	m.Spec.Dep.MsgStreamType = MsgStreamTypeWoodPecker
+	assert.Same(t, &m.Spec.Dep.WoodPecker.Persistence, m.Spec.GetPersistenceConfig())
 }
 
 func TestGetActiveConfigMap_SetActiveConfigMap(t *testing.T) {
diff --git a/apis/milvus.io/v1beta1/milvus_webhook.go b/apis/milvus.io/v1beta1/milvus_webhook.go
@@ -434,15 +434,18 @@ func (r *Milvus) defaultValuesMerged() bool {
 	return r.Annotations[DependencyValuesMergedAnnotation] == TrueStr
 }
 
-func (r *Milvus) defaultMsgStream() {
+func (r *Milvus) setDefaultMsgStreamType() {
 	if r.Spec.Dep.MsgStreamType == "" {
 		switch r.Spec.Mode {
 		case MilvusModeStandalone:
 			r.Spec.Dep.MsgStreamType = MsgStreamTypeRocksMQ
-		case MilvusModeCluster:
+		default:
 			r.Spec.Dep.MsgStreamType = MsgStreamTypePulsar
 		}
 	}
+}
+
+func (r *Milvus) setDefaultMsgStreamConfigs() {
 	switch r.Spec.Dep.MsgStreamType {
 	case MsgStreamTypeKafka:
 		if !r.Spec.Dep.Kafka.External {
@@ -483,6 +486,11 @@ func (r *Milvus) defaultMsgStream() {
 	}
 }
 
+func (r *Milvus) defaultMsgStream() {
+	r.setDefaultMsgStreamType()
+	r.setDefaultMsgStreamConfigs()
+}
+
 func (r *Milvus) defaultStorage() {
 	setDefaultStr(&r.Spec.Dep.Storage.Type, "MinIO")
 	if !r.Spec.Dep.Storage.External {
diff --git a/apis/milvus.io/v1beta1/zz_generated.deepcopy.go b/apis/milvus.io/v1beta1/zz_generated.deepcopy.go
diff --git a/config/samples/beta/milvus_woodpecker.yaml b/config/samples/beta/milvus_woodpecker.yaml
@@ -0,0 +1,20 @@
+# the manifest to enable streaming node for milvus cluster mode
+# ref https://milvus.io/docs/release_notes.md#Streaming-Node-Beta
+apiVersion: milvus.io/v1beta1
+kind: Milvus
+metadata:
+  name: milvus
+  namespace: mc-sit
+  labels:
+    app: milvus
+spec:
+  mode: 'cluster'
+  components:
+    # image not formally released yet
+    image: harbor.milvus.io/milvus/milvus:tinswzy-support_woodpecker_wal_type-36319c0-20250326
+    streamingNode:
+      replicas: 1
+    indexNode:
+      replicas: 0
+  dependencies:
+    msgStreamType: woodpecker
diff --git a/pkg/controllers/dependencies.go b/pkg/controllers/dependencies.go
@@ -224,11 +224,11 @@ func (r *MilvusReconciler) ReconcileMsgStream(ctx context.Context, mc v1beta1.Mi
 	switch mc.Spec.Dep.MsgStreamType {
 	case v1beta1.MsgStreamTypeKafka:
 		return r.ReconcileKafka(ctx, mc)
-	case v1beta1.MsgStreamTypeRocksMQ, v1beta1.MsgStreamTypeNatsMQ, v1beta1.MsgStreamTypeCustom:
-		// built in, do nothing
-		return nil
-	default:
+	case v1beta1.MsgStreamTypePulsar:
 		return r.ReconcilePulsar(ctx, mc)
+	default:
+		// built in mq or custom mq, do nothing
+		return nil
 	}
 }
 
diff --git a/pkg/controllers/deployment_updater.go b/pkg/controllers/deployment_updater.go
@@ -28,7 +28,6 @@ type deploymentUpdater interface {
 	GetMergedComponentSpec() ComponentSpec
 	GetArgs() []string
 	GetSecretRef() string
-	GetPersistenceConfig() *v1beta1.Persistence
 	GetMilvus() *v1beta1.Milvus
 	RollingUpdateImageDependencyReady() bool
 	HasHookConfig() bool
@@ -207,13 +206,20 @@ func updateUserDefinedVolumes(template *corev1.PodTemplateSpec, updater deployme
 		fillConfigMapVolumeDefaultValues(&volume)
 		userDefinedVolumes = append(userDefinedVolumes, volume)
 	}
-	if persistence := updater.GetPersistenceConfig(); persistence != nil && persistence.Enabled {
-		rocketMqPvcName := getPVCNameByInstName(updater.GetIntanceName())
-		if len(persistence.PersistentVolumeClaim.ExistingClaim) > 0 {
-			rocketMqPvcName = persistence.PersistentVolumeClaim.ExistingClaim
+	builtInMq := updater.GetMilvus().Spec.Dep.GetMilvusBuiltInMQ()
+
+	if builtInMq != nil {
+		if builtInMq.Persistence.Enabled {
+			rocketMqPvcName := getPVCNameByInstName(updater.GetIntanceName())
+			if len(builtInMq.Persistence.PersistentVolumeClaim.ExistingClaim) > 0 {
+				rocketMqPvcName = builtInMq.Persistence.PersistentVolumeClaim.ExistingClaim
+			}
+			userDefinedVolumes = append(userDefinedVolumes, persisentDataVolumeByName(rocketMqPvcName))
+		} else {
+			userDefinedVolumes = append(userDefinedVolumes, emptyDirDataVolume())
 		}
-		userDefinedVolumes = append(userDefinedVolumes, persisentVolumeByName(rocketMqPvcName))
 	}
+
 	for _, volume := range userDefinedVolumes {
 		addVolume(&template.Spec.Volumes, volume)
 	}
@@ -315,8 +321,9 @@ func updateBuiltInVolumeMounts(template *corev1.PodTemplateSpec, updater deploym
 
 func getUserDefinedVolumeMounts(updater deploymentUpdater) []corev1.VolumeMount {
 	ret := updater.GetMergedComponentSpec().VolumeMounts
-	if persistence := updater.GetPersistenceConfig(); persistence != nil && persistence.Enabled {
-		ret = append(ret, persistentVolumeMount())
+	builtInMq := updater.GetMilvus().Spec.Dep.GetMilvusBuiltInMQ()
+	if builtInMq != nil {
+		ret = append(ret, dataVolumeMount())
 	}
 	return ret
 }
diff --git a/pkg/controllers/deployments.go b/pkg/controllers/deployments.go
@@ -320,7 +320,16 @@ func configVolumeByName(name string) corev1.Volume {
 	}
 }
 
-func persisentVolumeByName(name string) corev1.Volume {
+func emptyDirDataVolume() corev1.Volume {
+	return corev1.Volume{
+		Name: MilvusDataVolumeName,
+		VolumeSource: corev1.VolumeSource{
+			EmptyDir: &corev1.EmptyDirVolumeSource{},
+		},
+	}
+}
+
+func persisentDataVolumeByName(name string) corev1.Volume {
 	return corev1.Volume{
 		Name: MilvusDataVolumeName,
 		VolumeSource: corev1.VolumeSource{
@@ -332,7 +341,7 @@ func persisentVolumeByName(name string) corev1.Volume {
 	}
 }
 
-func persistentVolumeMount() corev1.VolumeMount {
+func dataVolumeMount() corev1.VolumeMount {
 	return corev1.VolumeMount{
 		Name:      MilvusDataVolumeName,
 		ReadOnly:  false,
diff --git a/pkg/controllers/milvus.go b/pkg/controllers/milvus.go
@@ -101,9 +101,12 @@ var Finalize = func(ctx context.Context, r *MilvusReconciler, mc v1beta1.Milvus)
 			deletingReleases[mc.Name+"-pulsar"] = mc.Spec.Dep.Pulsar.InCluster.PVCDeletion
 		}
 	default:
-		// rocksmq
+		builtInMQ := mc.Spec.Dep.GetMilvusBuiltInMQ()
+		if builtInMQ == nil {
+			break
+		}
 		// delete data pvc if need
-		persist := mc.Spec.Dep.RocksMQ.Persistence
+		persist := builtInMQ.Persistence
 		if persist.Enabled && persist.PVCDeletion {
 			pvcName := getPVCNameByInstName(mc.Name)
 			pvc := &corev1.PersistentVolumeClaim{}
diff --git a/pkg/controllers/status_cluster.go b/pkg/controllers/status_cluster.go
@@ -393,9 +393,9 @@ func (r *MilvusStatusSyncer) GetMsgStreamCondition(
 	var eps = []string{}
 	var getter func() v1beta1.MilvusCondition
 	switch mc.Spec.Dep.MsgStreamType {
-	case v1beta1.MsgStreamTypeRocksMQ, v1beta1.MsgStreamTypeNatsMQ, v1beta1.MsgStreamTypeCustom:
-		// rocksmq / natsmq is built in, assume ok
-		return msgStreamReadyCondition, nil
+	case v1beta1.MsgStreamTypePulsar:
+		getter = external.NewPulsarConditionGetter(&mc).GetCondition
+		eps = []string{mc.Spec.Dep.Pulsar.Endpoint}
 	case v1beta1.MsgStreamTypeKafka:
 		kafkaConf, err := GetKafkaConfFromCR(mc)
 		if err != nil {
@@ -409,9 +409,8 @@ func (r *MilvusStatusSyncer) GetMsgStreamCondition(
 		getter = wrapKafkaConditonGetter(ctx, r.logger, mc.Spec.Dep.Kafka, *kafkaConf)
 		eps = mc.Spec.Dep.Kafka.BrokerList
 	default:
-		// default pulsar
-		getter = external.NewPulsarConditionGetter(&mc).GetCondition
-		eps = []string{mc.Spec.Dep.Pulsar.Endpoint}
+		// default built-in mqs, assume ok
+		return msgStreamReadyCondition, nil
 	}
 	return GetCondition(getter, eps), nil
 }
diff --git a/test/min-mc-feature.yaml b/test/min-mc-feature.yaml
@@ -68,7 +68,7 @@ spec:
         pvcDeletion: true
         values:
           replicaCount: 1
-    msgStreamType: pulsar
+    msgStreamType: woodpecker
     pulsar:
       inCluster:
         chartVersion: pulsar-v3
diff --git a/test/min-mc-mixture.yaml b/test/min-mc-mixture.yaml
@@ -35,7 +35,7 @@ spec:
         pvcDeletion: true
         values:
           replicaCount: 1
-    msgStreamType: kafka
+    msgStreamType: woodpecker
     kafka:
       inCluster:
         deletionPolicy: Delete

Original file line number	Diff line number	Diff line change
`@@ -128,6 +128,8 @@ func (ms MilvusSpec) GetPersistenceConfig() Persistence {`
`128`	`128`	`return &ms.Dep.RocksMQ.Persistence`
`129`	`129`	`case MsgStreamTypeNatsMQ:`
`130`	`130`	`return &ms.Dep.NatsMQ.Persistence`
	`131`	`+ case MsgStreamTypeWoodPecker:`
	`132`	`+ return &ms.Dep.WoodPecker.Persistence`
`131`	`133`	`}`
`132`	`134`	`return nil`
`133`	`135`	`}`
Original file line number	Diff line number	Diff line change
`@@ -320,7 +320,16 @@ func configVolumeByName(name string) corev1.Volume {`
`320`	`320`	`}`
`321`	`321`	`}`
`322`	`322`
`323`		`-func persisentVolumeByName(name string) corev1.Volume {`
	`323`	`+func emptyDirDataVolume() corev1.Volume {`
	`324`	`+ return corev1.Volume{`
	`325`	`+ Name: MilvusDataVolumeName,`
	`326`	`+ VolumeSource: corev1.VolumeSource{`
	`327`	`+ EmptyDir: &corev1.EmptyDirVolumeSource{},`
	`328`	`+ },`
	`329`	`+ }`
	`330`	`+}`
	`331`	`+`
	`332`	`+func persisentDataVolumeByName(name string) corev1.Volume {`
`324`	`333`	`return corev1.Volume{`
`325`	`334`	`Name: MilvusDataVolumeName,`
`326`	`335`	`VolumeSource: corev1.VolumeSource{`
`@@ -332,7 +341,7 @@ func persisentVolumeByName(name string) corev1.Volume {`
`332`	`341`	`}`
`333`	`342`	`}`
`334`	`343`
`335`		`-func persistentVolumeMount() corev1.VolumeMount {`
	`344`	`+func dataVolumeMount() corev1.VolumeMount {`
`336`	`345`	`return corev1.VolumeMount{`
`337`	`346`	`Name: MilvusDataVolumeName,`
`338`	`347`	`ReadOnly: false,`