Supports natsmq

apis/milvus.io/v1beta1/dependencies_types.go

@@ -16,7 +16,7 @@ type MilvusDependencies struct {
 	// +kubebuilder:validation:Optional
 	Etcd MilvusEtcd `json:"etcd"`
 
-	// +kubebuilder:validation:Enum:={"pulsar", "kafka", "rocksmq", ""}
+	// +kubebuilder:validation:Enum:={"pulsar", "kafka", "rocksmq", "natsmq", ""}
 	// +kubebuilder:validation:Optional
 	// MsgStreamType default to pulsar for cluster, rocksmq for standalone
 	MsgStreamType MsgStreamType `json:"msgStreamType,omitempty"`
@@ -28,7 +28,10 @@ type MilvusDependencies struct {
 	Kafka MilvusKafka `json:"kafka,omitempty"`
 
 	// +kubebuilder:validation:Optional
-	RocksMQ MilvusRocksMQ `json:"rocksmq,omitempty"`
+	RocksMQ MilvusBuildInMQ `json:"rocksmq,omitempty"`
+
+	// +kubebuilder:validation:Optional
+	NatsMQ MilvusBuildInMQ `json:"natsmq,omitempty"`
 
 	// +kubebuilder:validation:Optional
 	Storage MilvusStorage `json:"storage"`
@@ -40,6 +43,7 @@ const (
 	MsgStreamTypePulsar  MsgStreamType = "pulsar"
 	MsgStreamTypeKafka   MsgStreamType = "kafka"
 	MsgStreamTypeRocksMQ MsgStreamType = "rocksmq"
+	MsgStreamTypeNatsMQ  MsgStreamType = "natsmq"
 )
 
 type MilvusEtcd struct {
@@ -88,8 +92,8 @@ type MilvusStorage struct {
 	External bool `json:"external,omitempty"`
 }
 
-// MilvusRocksMQ configuration
-type MilvusRocksMQ struct {
+// MilvusBuildInMQ (rocksmq or natsmq) configuration
+type MilvusBuildInMQ struct {
 	Persistence Persistence `json:"persistence"`
 }
 

apis/milvus.io/v1beta1/milvus_types.go

@@ -114,6 +114,16 @@ func (ms MilvusSpec) GetMilvusVersionByImage() (*semver.Version, error) {
 	return provisioner.GetSemanticVersion(splited[1])
 }
 
+func (ms *MilvusSpec) GetPersistenceConfig() *Persistence {
+	switch ms.Dep.MsgStreamType {
+	case MsgStreamTypeRocksMQ:
+		return &ms.Dep.RocksMQ.Persistence
+	case MsgStreamTypeNatsMQ:
+		return &ms.Dep.NatsMQ.Persistence
+	}
+	return nil
+}
+
 // MilvusMode defines the mode of Milvus deployment
 type MilvusMode string
 

apis/milvus.io/v1beta1/milvus_types_test.go

@@ -216,3 +216,16 @@ func TestGetMilvusVersionByGlobalImage(t *testing.T) {
 	_, err = m.Spec.GetMilvusVersionByImage()
 	assert.Error(t, err)
 }
+
+func TestGetPersistenceConfig(t *testing.T) {
+	m := Milvus{}
+	m.Spec.Dep.MsgStreamType = MsgStreamTypePulsar
+	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.MsgStreamType = MsgStreamTypeNatsMQ
+	assert.Same(t, &m.Spec.Dep.NatsMQ.Persistence, m.Spec.GetPersistenceConfig())
+}