refactor and add docs

Signed-off-by: zhaohehuhu <luoyedeyi@163.com>

Author: zhaohehuhu

config/crd/bases/starrocks.com_starrocksclusters.yaml

@@ -13602,8 +13602,11 @@ spec:
                       of nodeSelectors to match when scheduling pods on nodes"
                     type: object
                   observerSpec:
-                    description: ObserverSpec configures optional FE observer pods.
-                      Observer pods reuse the generic FE resources.
+                    description: |-
+                      ObserverSpec configures optional FE observer pods. Observer pods reuse the generic FE resources.
+                      The FE image must be StarRocks 4.1.0 or later. Removing observers does not automatically
+                      run ALTER SYSTEM DROP OBSERVER; manually drop observer nodes
+                      if SHOW FRONTENDS reports Alive = false.
                     properties:
                       enabled:
                         description: Enabled controls whether FE observer should be

doc/scale_fe_observer_howto.md

@@ -0,0 +1,56 @@
+# Scale FE observer Howto
+
+## Overview
+
+FE observers are mainly used to increase the query concurrency of a StarRocks cluster. Unlike FE followers, observers do
+not participate in leader election, so adding observers does not add leader selection pressure to the cluster.
+
+Use `starRocksFeSpec.observerSpec` when the cluster needs more FE read capacity without changing the FE follower quorum.
+Observers can be scaled with the read workload, while FE followers should still be scaled carefully because they store
+metadata and participate in quorum.
+
+## How to correctly scale out FE read workloads by observer
+
+FE observers can be used to scale FE read workloads without changing the FE follower quorum. To use
+`starRocksFeSpec.observerSpec`, the FE image configured in `starRocksFeSpec.image` must be StarRocks 4.1.0 or later.
+The operator rejects FE observer deployment for older FE image versions because those images do not enable FE observer.
+
+To scale FE observers out, enable `observerSpec` and set the desired observer count:
+
+```yaml
+starRocksFeSpec:
+   image: "starrocks/fe-ubuntu:4.1.0"
+   observerSpec:
+      enabled: true
+      observerNumber: 2
+```
+
+## How to correctly scale in FE observer
+
+Known limitation: neither the operator nor `fe_entrypoint.sh` runs `ALTER SYSTEM DROP OBSERVER` when observers are
+removed. If you decrease `starRocksFeSpec.observerSpec.observerNumber` or set
+`starRocksFeSpec.observerSpec.enabled: false`, the observer Pods are removed from Kubernetes, but the removed observers
+can remain in StarRocks metadata and appear as permanent `Alive=false` rows in `SHOW FRONTENDS`. Because observers are
+intended for dynamic read-workload scaling, stale observer rows can accumulate after repeated scale-in operations.
+
+For planned observer scale-in, manually drop the observers that will be removed before lowering `observerNumber`.
+
+1. Execute `SHOW FRONTENDS` and find the `OBSERVER` rows for the observer Pods that will be removed.
+
+2. Drop each observer from StarRocks metadata by using its FE host and `EditLogPort` from `SHOW FRONTENDS`.
+   ```sql
+   mysql> ALTER SYSTEM DROP OBSERVER "<observer-host>:<edit-log-port>";
+   ```
+
+   For example:
+   ```sql
+   mysql> ALTER SYSTEM DROP OBSERVER "kube-starrocks-fe-observer-1.kube-starrocks-fe-observer-search.default.svc.cluster.local:9010";
+   ```
+
+3. Lower `starRocksFeSpec.observerSpec.observerNumber`, or set `starRocksFeSpec.observerSpec.enabled: false` if all
+   observers should be removed.
+
+4. Execute `SHOW FRONTENDS` again and verify the removed observer rows are gone.
+
+If the observer Pods were already removed from Kubernetes, connect to an FE node, run `SHOW FRONTENDS`, find `OBSERVER`
+rows with `Alive=false`, and run `ALTER SYSTEM DROP OBSERVER "<observer-host>:<edit-log-port>"` for each stale row.

helm-charts/charts/kube-starrocks/charts/starrocks/values.yaml

@@ -515,6 +515,10 @@ starrocksFESpec:
   shareProcessNamespace:
   # Optional FE observer pods. Observer pods reuse the FE image, resources, scheduling,
   # config, env, storage, and services from this starrocksFESpec.
+  # FE observer requires the FE image tag to be StarRocks 4.1.0 or later.
+  # Known limitation: scaling observerNumber down or disabling observerSpec removes
+  # Kubernetes pods, but does not run ALTER SYSTEM DROP OBSERVER. Manually drop
+  # observer nodes if SHOW FRONTENDS reports Alive = false.
   observerSpec:
     # specify the FE observer deployment or not.
     enabled: false

helm-charts/charts/kube-starrocks/values.yaml

@@ -631,6 +631,10 @@ starrocks:
     shareProcessNamespace:
     # Optional FE observer pods. Observer pods reuse the FE image, resources, scheduling,
     # config, env, storage, and services from this starrocksFESpec.
+    # FE observer requires the FE image tag to be StarRocks 4.1.0 or later.
+    # Known limitation: scaling observerNumber down or disabling observerSpec removes
+    # Kubernetes pods, but does not run ALTER SYSTEM DROP OBSERVER. Manually drop
+    # observer nodes if SHOW FRONTENDS reports Alive = false.
     observerSpec:
       # specify the FE observer deployment or not.
       enabled: false

pkg/k8sutils/templates/pod/spec.go

@@ -173,6 +173,10 @@ func Envs(spec v1.SpecInterface, config map[string]interface{},
 				Name:  v1.IS_FE_OBSERVER,
 				Value: "true",
 			},
+			{
+				Name:  v1.FE_SERVICE_NAME,
+				Value: feExternalServiceName,
+			},
 		} {
 			addEnv(envVar)
 		}

pkg/subcontrollers/feobserver/feobserver_controller.go

@@ -18,12 +18,15 @@ package feobserver
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/StarRocks/starrocks-kubernetes-operator/pkg/k8sutils/templates/service"
 	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -39,6 +42,8 @@ import (
 	"github.com/StarRocks/starrocks-kubernetes-operator/pkg/subcontrollers/fe"
 )
 
+const minFeObserverImageVersion = "4.1.0"
+
 type FeObserverController struct {
 	Client   client.Client
 	Recorder record.EventRecorder
@@ -64,13 +69,12 @@ func (fc *FeObserverController) SyncCluster(ctx context.Context, src *srapi.Star
 	feSpec := src.Spec.StarRocksFeSpec
 	observerSpec := feSpec.ToObserverSpec()
 	if observerSpec == nil {
-		logger.Info("fe observer is disabled, clear observer resources")
-		if err := fc.clearObserverResources(ctx, src); err != nil {
-			logger.Error(err, "clear fe observer resources failed")
-			return err
-		}
+		logger.Info("src.Spec.StarRocksFeObserverSpec == nil, skip sync fe observer")
 		return nil
 	}
+	if src.Spec.StarRocksFeSpec == nil {
+		return fmt.Errorf("starRocksFeSpec is required before deploying fe observer")
+	}
 
 	var err error
 	defer func() {
@@ -84,6 +88,10 @@ func (fc *FeObserverController) SyncCluster(ctx context.Context, src *srapi.Star
 		return err
 	}
 
+	if !fe.CheckFEReady(ctx, fc.Client, src.Namespace, src.Name) {
+		logger.Info("FE is not ready, stop sync fe observer")
+		return nil
+	}
 	// get the fe configMap for resolve ports
 	logger.V(log.DebugLevel).Info("get fe configMap to resolve ports", "ConfigMapInfo", feSpec.ConfigMapInfo)
 	feConfig, err := fe.GetFEConfig(ctx, fc.Client, feSpec, src.Namespace)
@@ -95,30 +103,44 @@ func (fc *FeObserverController) SyncCluster(ctx context.Context, src *srapi.Star
 	// generate new fe observer statefulset.
 	logger.V(log.DebugLevel).Info("build fe observer statefulset", "StarRocksCluster", src)
 	object := object.NewFromCluster(src)
-
+	defaultLabels := load.Labels(src.Name, observerSpec)
+	svc := rutils.BuildExternalService(object, observerSpec, feConfig, load.Selector(src.Name, observerSpec), defaultLabels)
+	searchServiceName := service.SearchServiceName(src.Name, observerSpec)
+	internalService := service.MakeSearchService(searchServiceName, &svc, []corev1.ServicePort{
+		{
+			Name:        "query-port",
+			Port:        rutils.GetPort(feConfig, rutils.QUERY_PORT),
+			TargetPort:  intstr.FromInt(int(rutils.GetPort(feConfig, rutils.QUERY_PORT))),
+			AppProtocol: func() *string { mysql := "mysql"; return &mysql }(),
+		},
+	}, defaultLabels)
 	podTemplateSpec, err := fc.buildPodTemplate(src, feConfig)
 	if err != nil {
 		logger.Error(err, "build pod template failed")
 		return err
 	}
 	expectSts := statefulset.MakeStatefulset(object, observerSpec, podTemplateSpec)
-	expectSts.Spec.ServiceName = feSearchServiceName(src.Name)
 	err = k8sutils.ApplyStatefulSet(ctx, fc.Client, &expectSts, false, rutils.StatefulSetDeepEqual)
 	if err != nil {
 		logger.Error(err, "fe observer statefulset failed", "StarRocksCluster", src)
 		return err
 	}
 
-	if err = fc.deleteLegacyObserverServices(ctx, src); err != nil {
-		logger.Error(err, "delete legacy fe observer services failed")
+	if err = k8sutils.ApplyService(ctx, fc.Client, internalService, rutils.ServiceDeepEqual); err != nil {
+		logger.Error(err, "deploy search service failed", "internalService", internalService)
+		fc.Recorder.Event(src, corev1.EventTypeWarning, "DeployFeObserverFailed", err.Error())
 		return err
 	}
 
+	if err = k8sutils.ApplyService(ctx, fc.Client, &svc, rutils.ServiceDeepEqual); err != nil {
+		logger.Error(err, "deploy external service failed", "externalService", svc)
+		return err
+	}
 	return nil
 }
 
 // UpdateClusterStatus update the all resource status about fe observer.
-func (fc *FeObserverController) UpdateClusterStatus(_ context.Context, src *srapi.StarRocksCluster) error {
+func (fc *FeObserverController) UpdateClusterStatus(ctx context.Context, src *srapi.StarRocksCluster) error {
 	feSpec := src.Spec.StarRocksFeSpec
 	observerSpec := feSpec.ToObserverSpec()
 	if observerSpec == nil {
@@ -137,7 +159,7 @@ func (fc *FeObserverController) UpdateClusterStatus(_ context.Context, src *srap
 	}
 
 	src.Status.StarRocksFeObserverStatus = fs
-	fs.ServiceName = feExternalServiceName(src.Name)
+	fs.ServiceName = service.ExternalServiceName(src.Name, feSpec)
 	statefulSetName := load.Name(src.Name, observerSpec)
 	fs.ResourceNames = rutils.MergeSlices(fs.ResourceNames, []string{statefulSetName})
 
@@ -167,7 +189,23 @@ func (fc *FeObserverController) ClearCluster(ctx context.Context, src *srapi.Sta
 		return nil
 	}
 
-	return fc.clearObserverResources(ctx, src)
+	observerSpec := src.Spec.StarRocksFeSpec.ToObserverSpec()
+	statefulSetName := load.Name(src.Name, observerSpec)
+	if err := k8sutils.DeleteStatefulset(ctx, fc.Client, src.Namespace, statefulSetName); err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+	searchServiceName := service.SearchServiceName(src.Name, observerSpec)
+	if err := k8sutils.DeleteService(ctx, fc.Client, src.Namespace, searchServiceName); err != nil && !apierrors.IsNotFound(err) {
+		logger.Error(err, "delete search service failed", "searchServiceName", searchServiceName)
+		return err
+	}
+	externalServiceName := service.ExternalServiceName(src.Name, observerSpec)
+	err := k8sutils.DeleteService(ctx, fc.Client, src.Namespace, externalServiceName)
+	if err != nil && !apierrors.IsNotFound(err) {
+		logger.Error(err, "delete external service failed", "externalServiceName", externalServiceName)
+		return err
+	}
+	return nil
 }
 
 func (fc *FeObserverController) Validating(feSpec *srapi.StarRocksFeSpec) error {
@@ -179,33 +217,20 @@ func (fc *FeObserverController) Validating(feSpec *srapi.StarRocksFeSpec) error
 	if err := srapi.ValidUpdateStrategy(feSpec.UpdateStrategy); err != nil {
 		return err
 	}
-	return nil
-}
-
-func (fc *FeObserverController) clearObserverResources(ctx context.Context, src *srapi.StarRocksCluster) error {
-	statefulSetName := load.Name(src.Name, (*srapi.StarRocksFeObserverSpec)(nil))
-	if err := k8sutils.DeleteStatefulset(ctx, fc.Client, src.Namespace, statefulSetName); err != nil && !apierrors.IsNotFound(err) {
-		return err
+	if feSpec.ToObserverSpec() != nil {
+		if err := validateObserverImageVersion(feSpec.Image); err != nil {
+			return err
+		}
 	}
-	return fc.deleteLegacyObserverServices(ctx, src)
+	return nil
 }
 
-func (fc *FeObserverController) deleteLegacyObserverServices(ctx context.Context, src *srapi.StarRocksCluster) error {
-	searchServiceName := src.Name + "-" + srapi.DEFAULT_FE_OBSERVER + "-search"
-	if err := k8sutils.DeleteService(ctx, fc.Client, src.Namespace, searchServiceName); err != nil && !apierrors.IsNotFound(err) {
-		return err
-	}
-	externalServiceName := src.Name + "-" + srapi.DEFAULT_FE_OBSERVER + "-service"
-	if err := k8sutils.DeleteService(ctx, fc.Client, src.Namespace, externalServiceName); err != nil && !apierrors.IsNotFound(err) {
-		return err
+func validateObserverImageVersion(image string) error {
+	imageVersion := pod.GetImageVersion(image)
+	_, err := pod.IsLowerThanAny(imageVersion, []string{"4.1.0"})
+	if err != nil {
+		return fmt.Errorf("fe observer requires StarRocks FE image version >= %s "+
+			"Current image version %s does not enable FE observer", minFeObserverImageVersion, imageVersion)
 	}
 	return nil
 }
-
-func feExternalServiceName(clusterName string) string {
-	return clusterName + "-" + srapi.DEFAULT_FE + "-service"
-}
-
-func feSearchServiceName(clusterName string) string {
-	return clusterName + "-" + srapi.DEFAULT_FE + "-search"
-}