support streaming-node (#212)

Signed-off-by: haorenfsa <[email protected]>

apis/milvus.io/v1alpha1/milvus_types.go

@@ -111,9 +111,9 @@ type MilvusReplicas struct {
 
 // +genclient
 // +genclient:noStatus
-//+kubebuilder:object:root=true
-//+kubebuilder:subresource:status
-//+kubebuilder:resource:path=milvuses,singular=milvus,shortName=mi
+// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
+// +kubebuilder:resource:path=milvuses,singular=milvus,shortName=mi
 // Milvus is the Schema for the milvus API
 type Milvus struct {
 	metav1.TypeMeta   `json:",inline"`
@@ -123,7 +123,7 @@ type Milvus struct {
 	Status v1beta1.MilvusStatus `json:"status,omitempty"`
 }
 
-//+kubebuilder:object:root=true
+// +kubebuilder:object:root=true
 // MilvusList contains a list of Milvus
 type MilvusList struct {
 	metav1.TypeMeta `json:",inline"`

apis/milvus.io/v1beta1/components_types.go

@@ -7,25 +7,27 @@ import (
 type ComponentType string
 
 const (
-	RootCoord  ComponentType = "rootCoord"
-	DataCoord  ComponentType = "dataCoord"
-	QueryCoord ComponentType = "queryCoord"
-	IndexCoord ComponentType = "indexCoord"
-	DataNode   ComponentType = "dataNode"
-	QueryNode  ComponentType = "queryNode"
-	IndexNode  ComponentType = "indexNode"
-	Proxy      ComponentType = "proxy"
-
-	MixCoordName   = "mixcoord"
-	RootCoordName  = "rootcoord"
-	DataCoordName  = "datacoord"
-	QueryCoordName = "querycoord"
-	IndexCoordName = "indexcoord"
-	DataNodeName   = "datanode"
-	QueryNodeName  = "querynode"
-	IndexNodeName  = "indexnode"
-	ProxyName      = "proxy"
-	StandaloneName = "standalone"
+	RootCoord     ComponentType = "rootCoord"
+	DataCoord     ComponentType = "dataCoord"
+	QueryCoord    ComponentType = "queryCoord"
+	IndexCoord    ComponentType = "indexCoord"
+	DataNode      ComponentType = "dataNode"
+	QueryNode     ComponentType = "queryNode"
+	IndexNode     ComponentType = "indexNode"
+	StreamingNode ComponentType = "streamingNode"
+	Proxy         ComponentType = "proxy"
+
+	MixCoordName      = "mixcoord"
+	RootCoordName     = "rootcoord"
+	DataCoordName     = "datacoord"
+	QueryCoordName    = "querycoord"
+	IndexCoordName    = "indexcoord"
+	DataNodeName      = "datanode"
+	QueryNodeName     = "querynode"
+	IndexNodeName     = "indexnode"
+	ProxyName         = "proxy"
+	StandaloneName    = "standalone"
+	StreamingNodeName = "streamingnode"
 )
 
 var (
@@ -141,7 +143,6 @@ type MilvusComponents struct {
 
 	// Note: it's still in beta, do not use for production. EnableRollingUpdate whether to enable rolling update for milvus component
 	// there is nearly zero downtime for rolling update
-	// TODO: enable rolling update by default for next major version
 	// +kubebuilder:validation:Optional
 	EnableRollingUpdate *bool `json:"enableRollingUpdate,omitempty"`
 
@@ -220,6 +221,9 @@ type MilvusComponents struct {
 	// +kubebuilder:validation:Optional
 	QueryNode *MilvusQueryNode `json:"queryNode,omitempty"`
 
+	// +kubebuilder:validation:Optional
+	StreamingNode *MilvusStreamingNode `json:"streamingNode,omitempty"`
+
 	// +kubebuilder:validation:Optional
 	Standalone *MilvusStandalone `json:"standalone,omitempty"`
 }
@@ -280,6 +284,10 @@ type MilvusIndexCoord struct {
 	Component `json:",inline"`
 }
 
+type MilvusStreamingNode struct {
+	Component `json:",inline"`
+}
+
 type MilvusStandalone struct {
 	ServiceComponent `json:",inline"`
 }

apis/milvus.io/v1beta1/milvus_types.go

@@ -132,6 +132,14 @@ func (ms *MilvusSpec) GetPersistenceConfig() *Persistence {
 	return nil
 }
 
+func (ms *MilvusSpec) UseMixCoord() bool {
+	return ms.Com.MixCoord != nil
+}
+
+func (ms *MilvusSpec) UseStreamingNode() bool {
+	return ms.Com.StreamingNode != nil
+}
+
 // MilvusMode defines the mode of Milvus deployment
 type MilvusMode string
 
@@ -159,11 +167,6 @@ type MilvusStatus struct {
 	// IngressStatus of the ingress created by milvus
 	IngressStatus networkv1.IngressStatus `json:"ingress,omitempty"`
 
-	// DeprecatedReplicas is deprecated, will be removed in next major version, use ComponentsDeployStatus instead
-	// DeprecatedReplicas is the number of updated replicas in ready status
-	// +kubebuilder:validation:Optional
-	DeprecatedReplicas MilvusReplicas `json:"replicas,omitempty"`
-
 	// ComponentsDeployStatus contains the map of component's name to the status of each component deployment
 	// it is used to check the status of rolling update of each component
 	// +optional
@@ -286,6 +289,8 @@ type MilvusReplicas struct {
 	//+kubebuilder:validation:Optional
 	QueryNode int `json:"queryNode,omitempty"`
 	//+kubebuilder:validation:Optional
+	StreamingNode int `json:"streamingNode,omitempty"`
+	//+kubebuilder:validation:Optional
 	Standalone int `json:"standalone,omitempty"`
 }
 

apis/milvus.io/v1beta1/milvus_webhook.go

@@ -261,7 +261,7 @@ func (r *Milvus) DefaultComponents() {
 		if spec.Com.Proxy == nil {
 			spec.Com.Proxy = &MilvusProxy{}
 		}
-		if spec.Com.MixCoord == nil {
+		if !r.Spec.UseMixCoord() {
 			if r.noCoordSpecifiedByUser() {
 				// default to use mixcoord
 				spec.Com.MixCoord = &MilvusMixCoord{}
@@ -301,6 +301,11 @@ func (r *Milvus) defaultComponentsReplicas() {
 		if spec.Com.Standalone.Replicas == nil {
 			spec.Com.Standalone.Replicas = &defaultNoReplicas
 		}
+		if r.Spec.UseStreamingNode() {
+			if spec.Com.StreamingNode.Replicas == nil {
+				spec.Com.StreamingNode.Replicas = &defaultReplicas
+			}
+		}
 		if spec.Com.MixCoord != nil {
 			if spec.Com.MixCoord.Replicas == nil {
 				spec.Com.MixCoord.Replicas = &defaultReplicas
@@ -505,7 +510,9 @@ func (r *Milvus) DefaultConf() {
 	if !r.isRollingUpdateSupportedByConfig() {
 		r.Spec.Com.EnableRollingUpdate = util.BoolPtr(false)
 	}
-	setEnableActiveStandby(&r.Spec, true)
+	if *r.Spec.Com.EnableRollingUpdate {
+		setEnableActiveStandby(&r.Spec, true)
+	}
 }
 
 var rollingUpdateConfigFields = []string{

apis/milvus.io/v1beta1/milvus_webhook_test.go

@@ -67,7 +67,6 @@ func TestMilvus_Default_NotExternal(t *testing.T) {
 			Data: map[string]interface{}{},
 		},
 	}
-	setEnableActiveStandby(&standaloneDefault, true)
 
 	t.Run("standalone not external ok", func(t *testing.T) {
 		mc := Milvus{ObjectMeta: metav1.ObjectMeta{Name: crName}}
@@ -131,6 +130,7 @@ func TestMilvus_Default_NotExternal(t *testing.T) {
 			},
 		},
 	}
+	setEnableActiveStandby(&clusterDefault, true)
 	t.Run("cluster not external dep ok", func(t *testing.T) {
 		mc := Milvus{ObjectMeta: metav1.ObjectMeta{Name: crName}}
 		mc.Spec.Mode = MilvusModeCluster