Merge pull request #75 from intelligentfu/start-probe

add be antiaffinity with fe

api/doris/v1/types.go

@@ -142,11 +142,6 @@ type BaseSpec struct {
 	// +patchStrategy=merge
 	ImagePullSecrets []corev1.LocalObjectReference `json:"imagePullSecrets,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,15,rep,name=imagePullSecrets"`
 
-	//+optional
-	//set the fe service for register cn, when not set, will use the fe config to find.
-	//Deprecated,
-	//FeServiceName string `json:"feServiceName,omitempty"`
-
 	//the reference for cn configMap.
 	//+optional
 	ConfigMapInfo ConfigMapInfo `json:"configMapInfo,omitempty"`
@@ -227,6 +222,7 @@ type ConfigMapInfo struct {
 	ConfigMapName string `json:"configMapName,omitempty"`
 
 	//the config response key in configmap.
+	//the config file name for
 	ResolveKey string `json:"resolveKey,omitempty"`
 }
 
@@ -240,6 +236,9 @@ type ExportService struct {
 	//ServicePort config service for NodePort access mode.
 	ServicePorts []DorisServicePort `json:"servicePorts,omitempty"`
 
+	//Annotations for using function on different cloud platform.
+	Annotations map[string]string `json:"annotations,omitempty"`
+
 	// Only applies to Service Type: LoadBalancer.
 	// This feature depends on whether the underlying cloud-provider supports specifying
 	// the loadBalancerIP when a load balancer is created.

cmd/doris-debug/main.go

@@ -24,7 +24,7 @@ func main() {
 
 	fmt.Println("start component " + componentType + "for debugging.....")
 	listenPort := readConfigListenPort()
-	registerMockApiHealth()
+	//registerMockApiHealth()
 	if err := http.ListenAndServe(":"+listenPort, nil); err != nil {
 		fmt.Println("listenAndServe failed," + err.Error())
 		os.Exit(1)
@@ -57,9 +57,9 @@ func readConfigListenPort() string {
 
 	var listenPort string
 	if componentType == "fe" {
-		listenPort = viper.GetString(resource.HTTP_PORT)
+		listenPort = viper.GetString(resource.QUERY_PORT)
 	} else if componentType == "be" {
-		listenPort = viper.GetString(resource.WEBSERVER_PORT)
+		listenPort = viper.GetString(resource.HEARTBEAT_SERVICE_PORT)
 	}
 
 	return listenPort

config/crd/bases/doris.selectdb.com_dorisclusters.yaml

@@ -967,7 +967,8 @@ spec:
                         description: the config info for start progress.
                         type: string
                       resolveKey:
-                        description: the config response key in configmap.
+                        description: the config response key in configmap. the config
+                          file name for
                         type: string
                     type: object
                   envVars:
@@ -1426,6 +1427,12 @@ spec:
                   service:
                     description: expose the be listen ports
                     properties:
+                      annotations:
+                        additionalProperties:
+                          type: string
+                        description: Annotations for using function on different cloud
+                          platform.
+                        type: object
                       loadBalancerIP:
                         description: 'Only applies to Service Type: LoadBalancer.
                           This feature depends on whether the underlying cloud-provider
@@ -2447,7 +2454,8 @@ spec:
                         description: the config info for start progress.
                         type: string
                       resolveKey:
-                        description: the config response key in configmap.
+                        description: the config response key in configmap. the config
+                          file name for
                         type: string
                     type: object
                   envVars:
@@ -2912,6 +2920,12 @@ spec:
                   service:
                     description: expose the be listen ports
                     properties:
+                      annotations:
+                        additionalProperties:
+                          type: string
+                        description: Annotations for using function on different cloud
+                          platform.
+                        type: object
                       loadBalancerIP:
                         description: 'Only applies to Service Type: LoadBalancer.
                           This feature depends on whether the underlying cloud-provider
@@ -4612,7 +4626,8 @@ spec:
                         description: the config info for start progress.
                         type: string
                       resolveKey:
-                        description: the config response key in configmap.
+                        description: the config response key in configmap. the config
+                          file name for
                         type: string
                     type: object
                   envVars:
@@ -5071,6 +5086,12 @@ spec:
                   service:
                     description: expose the be listen ports
                     properties:
+                      annotations:
+                        additionalProperties:
+                          type: string
+                        description: Annotations for using function on different cloud
+                          platform.
+                        type: object
                       loadBalancerIP:
                         description: 'Only applies to Service Type: LoadBalancer.
                           This feature depends on whether the underlying cloud-provider
@@ -6093,7 +6114,8 @@ spec:
                         description: the config info for start progress.
                         type: string
                       resolveKey:
-                        description: the config response key in configmap.
+                        description: the config response key in configmap. the config
+                          file name for
                         type: string
                     type: object
                   electionNumber:
@@ -6557,6 +6579,12 @@ spec:
                   service:
                     description: expose the be listen ports
                     properties:
+                      annotations:
+                        additionalProperties:
+                          type: string
+                        description: Annotations for using function on different cloud
+                          platform.
+                        type: object
                       loadBalancerIP:
                         description: 'Only applies to Service Type: LoadBalancer.
                           This feature depends on whether the underlying cloud-provider

doc/api.md

@@ -1353,6 +1353,17 @@ More info: <a href="https://kubernetes.io/docs/concepts/services-networking/serv
 </tr>
 <tr>
 <td>
+<code>annotations</code><br/>
+<em>
+map[string]string
+</em>
+</td>
+<td>
+<p>Annotations for using function on different cloud platform.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>loadBalancerIP</code><br/>
 <em>
 string
@@ -2450,5 +2461,5 @@ string
 <hr/>
 <p><em>
 Generated with <code>gen-crd-api-reference-docs</code>
-on git commit <code>002533a</code>.
+on git commit <code>490a4cc</code>.
 </em></p>

pkg/common/utils/resource/pod.go

@@ -35,6 +35,18 @@ const (
 	DEFAULT_ROOT_PATH    = "/opt/apache-doris"
 	POD_INFO_PATH        = "/etc/podinfo"
 	POD_INFO_VOLUME_NAME = "podinfo"
+
+	NODE_TOPOLOGYKEY = "kubernetes.io/hostname"
+
+	DEFAULT_INIT_IMAGE = "selectdb/alpine:latest"
+)
+
+type probeType string
+
+var (
+	httpGet   probeType = "httpGet"
+	tcpSocket probeType = "tcpSocket"
+	exec      probeType = "exec"
 )
 
 func NewPodTemplateSpec(dcr *v1.DorisCluster, componentType v1.ComponentType) corev1.PodTemplateSpec {
@@ -52,11 +64,9 @@ func NewPodTemplateSpec(dcr *v1.DorisCluster, componentType v1.ComponentType) co
 		volumes = newVolumesFromBaseSpec(dcr.Spec.BeSpec.BaseSpec)
 		si = dcr.Spec.BeSpec.BaseSpec.SystemInitialization
 		dcrAffinity = dcr.Spec.BeSpec.BaseSpec.Affinity
-		defaultInitContainers = append(defaultInitContainers, constructBeDefaultInitContainer())
 	case v1.Component_CN:
 		si = dcr.Spec.CnSpec.BaseSpec.SystemInitialization
 		dcrAffinity = dcr.Spec.CnSpec.BaseSpec.Affinity
-		defaultInitContainers = append(defaultInitContainers, constructBeDefaultInitContainer())
 	case v1.Component_Broker:
 		si = dcr.Spec.BrokerSpec.BaseSpec.SystemInitialization
 		dcrAffinity = dcr.Spec.BrokerSpec.BaseSpec.Affinity
@@ -92,14 +102,26 @@ func NewPodTemplateSpec(dcr *v1.DorisCluster, componentType v1.ComponentType) co
 		},
 	}
 
+	constructInitContainers(componentType, &pts.Spec, si)
+	pts.Spec.Affinity = constructAffinity(dcrAffinity, componentType)
+
+	return pts
+}
+
+func constructInitContainers(componentType v1.ComponentType, podSpec *corev1.PodSpec, si *v1.SystemInitialization) {
+	defaultImage := ""
+	var defaultInitContains []corev1.Container
 	if si != nil {
 		initContainer := newBaseInitContainer("init", si)
-		pts.Spec.InitContainers = append(pts.Spec.InitContainers, initContainer)
+		defaultImage = si.InitImage
+		defaultInitContains = append(defaultInitContains, initContainer)
 	}
 
-	pts.Spec.Affinity = constructAffinity(dcrAffinity, componentType)
-
-	return pts
+	// the init containers have sequence，should confirm use initial is always in the first priority.
+	if componentType == v1.Component_BE || componentType == v1.Component_CN {
+		podSpec.InitContainers = append(podSpec.InitContainers, constructBeDefaultInitContainer(defaultImage))
+	}
+	podSpec.InitContainers = append(podSpec.InitContainers, defaultInitContains...)
 }
 
 // newVolumesFromBaseSpec return corev1.Volume build from baseSpec.
@@ -168,7 +190,7 @@ func newBaseInitContainer(name string, si *v1.SystemInitialization) corev1.Conta
 	enablePrivileged := true
 	initImage := si.InitImage
 	if initImage == "" {
-		initImage = "selectdb/alpine:latest"
+		initImage = DEFAULT_INIT_IMAGE
 	}
 	c := corev1.Container{
 		Image:           initImage,
@@ -224,32 +246,36 @@ func NewBaseMainContainer(dcr *v1.DorisCluster, config map[string]interface{}, c
 		Resources:       spec.ResourceRequirements,
 	}
 
-	var healthPort int32
+	//livenessPort use heartbeat port for probe service alive.
+	var livenessPort int32
+	//readnessPort use http port for confirm the service can provider service to client.
+	var readnessPort int32
 	var prestopScript string
 	var health_api_path string
 	switch componentType {
 	case v1.Component_FE:
-		healthPort = GetPort(config, HTTP_PORT)
+		readnessPort = GetPort(config, HTTP_PORT)
+		livenessPort = GetPort(config, QUERY_PORT)
 		prestopScript = FE_PRESTOP
 		health_api_path = HEALTH_API_PATH
 	case v1.Component_BE, v1.Component_CN:
-		healthPort = GetPort(config, WEBSERVER_PORT)
+		readnessPort = GetPort(config, WEBSERVER_PORT)
+		livenessPort = GetPort(config, HEARTBEAT_SERVICE_PORT)
 		prestopScript = BE_PRESTOP
 		health_api_path = HEALTH_API_PATH
 	case v1.Component_Broker:
-		healthPort = GetPort(config, BROKER_IPC_PORT)
+		livenessPort = GetPort(config, BROKER_IPC_PORT)
+		readnessPort = GetPort(config, BROKER_IPC_PORT)
 		prestopScript = BROKER_PRESTOP
 		health_api_path = ""
 	default:
 		klog.Infof("the componentType %s is not supported in probe.")
 	}
 
-	if healthPort != 0 {
-		c.LivenessProbe = livenessProbe(healthPort, health_api_path)
-		c.StartupProbe = startupProbe(healthPort, health_api_path)
-		c.ReadinessProbe = readinessProbe(healthPort, health_api_path)
-		c.Lifecycle = lifeCycle(prestopScript)
-	}
+	c.LivenessProbe = livenessProbe(livenessPort, "")
+	c.StartupProbe = startupProbe(readnessPort, health_api_path)
+	c.ReadinessProbe = readinessProbe(readnessPort, health_api_path)
+	c.Lifecycle = lifeCycle(prestopScript)
 
 	return c
 }
@@ -474,7 +500,7 @@ func startupProbe(port int32, path string) *corev1.Probe {
 	return &corev1.Probe{
 		FailureThreshold: 60,
 		PeriodSeconds:    5,
-		ProbeHandler:     getProbe(port, path),
+		ProbeHandler:     getProbe(port, path, httpGet),
 	}
 }
 
@@ -483,7 +509,10 @@ func livenessProbe(port int32, path string) *corev1.Probe {
 	return &corev1.Probe{
 		PeriodSeconds:    5,
 		FailureThreshold: 3,
-		ProbeHandler:     getProbe(port, path),
+		// for pulling image and start doris
+		InitialDelaySeconds: 80,
+		TimeoutSeconds:      180,
+		ProbeHandler:        getProbe(port, path, tcpSocket),
 	}
 }
 
@@ -492,7 +521,7 @@ func readinessProbe(port int32, path string) *corev1.Probe {
 	return &corev1.Probe{
 		PeriodSeconds:    5,
 		FailureThreshold: 3,
-		ProbeHandler:     getProbe(port, path),
+		ProbeHandler:     getProbe(port, path, httpGet),
 	}
 }
 
@@ -507,7 +536,27 @@ func lifeCycle(preStopScriptPath string) *corev1.Lifecycle {
 	}
 }
 
-func getProbe(port int32, path string) corev1.ProbeHandler {
+// getProbe describe a health check.
+func getProbe(port int32, path string, probeType probeType) corev1.ProbeHandler {
+	switch probeType {
+	case tcpSocket:
+		return getTcpSocket(port)
+	case httpGet:
+		return getHttpProbe(port, path)
+	default:
+	}
+	return corev1.ProbeHandler{}
+}
+
+func getTcpSocket(port int32) corev1.ProbeHandler {
+	return corev1.ProbeHandler{
+		TCPSocket: &corev1.TCPSocketAction{
+			Port: intstr.FromInt(int(port)),
+		},
+	}
+}
+
+func getHttpProbe(port int32, path string) corev1.ProbeHandler {
 	var p corev1.ProbeHandler
 	if path != "" {
 		p = corev1.ProbeHandler{
@@ -542,7 +591,7 @@ func getDefaultAffinity(componentType v1.ComponentType) *corev1.Affinity {
 					{Key: v1.ComponentLabelKey, Operator: metav1.LabelSelectorOpIn, Values: []string{string(componentType)}},
 				},
 			},
-			TopologyKey: "kubernetes.io/hostname",
+			TopologyKey: NODE_TOPOLOGYKEY,
 		},
 	}
 	return &corev1.Affinity{
@@ -571,12 +620,13 @@ func constructAffinity(dcrAffinity *corev1.Affinity, componentType v1.ComponentT
 	return affinity
 }
 
-func constructBeDefaultInitContainer() corev1.Container {
+func constructBeDefaultInitContainer(defaultImage string) corev1.Container {
 	return newBaseInitContainer(
 		"default-init",
 		&v1.SystemInitialization{
-			Command: []string{"/bin/sh"},
-			Args:    []string{"-c", "sysctl -w vm.max_map_count=2000000 && swapoff -a"},
+			Command:   []string{"/bin/sh"},
+			InitImage: defaultImage,
+			Args:      []string{"-c", "sysctl -w vm.max_map_count=2000000 && swapoff -a"},
 		},
 	)
 }