feat(main): add broker config

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

api/v1beta1/automq_types.go

@@ -157,6 +157,8 @@ type AutoMQSpec struct {
 	ClusterID string `json:"clusterID,omitempty"`
 	// Image is the image of the AutoMQ
 	Image string `json:"image,omitempty"`
+	// NodePort is the node port of the AutoMQ
+	NodePort int32 `json:"nodePort,omitempty"`
 	// Metrics is the metrics configuration for the AutoMQ
 	Metrics MetricsSpec `json:"metrics,omitempty"`
 	// Controller is the controller configuration for the AutoMQ

cmd/main.go

@@ -18,14 +18,17 @@ package main
 
 import (
 	"flag"
+	v1 "k8s.io/api/core/v1"
 	"os"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 
 	infrav1beta1 "github.com/cuisongliu/automq-operator/api/v1beta1"
 	"github.com/cuisongliu/automq-operator/internal/controller"
+	"github.com/gin-gonic/gin"
 	promv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -111,8 +114,49 @@ func main() {
 			os.Exit(1)
 		}
 	}
+
+	if os.Getenv("OPERATOR_APIS_SVC_NAME") == "" {
+		setupLog.Error(err, "OPERATOR_APIS_SVC_NAME is empty")
+		os.Exit(1)
+	}
+
+	if os.Getenv("NAMESPACE_NAME") == "" {
+		_ = os.Setenv("NAMESPACE_NAME", "default")
+	}
+
 	//+kubebuilder:scaffold:builder
 
+	ctx := ctrl.SetupSignalHandler()
+
+	go func() {
+		if mgr.GetCache().WaitForCacheSync(ctx) {
+			setupLog.Info("cache sync success")
+			router := gin.Default()
+			router.GET("/api/v1/nodes/:name", func(c *gin.Context) {
+				name := c.Param("name")
+				node := &v1.Node{}
+				node.Name = name
+				if noe := mgr.GetClient().Get(ctx, client.ObjectKeyFromObject(node), node); noe != nil {
+					c.JSON(500, gin.H{"message": noe.Error()})
+					return
+				}
+				nodeIP := ""
+				for _, addr := range node.Status.Addresses {
+					if addr.Type == v1.NodeInternalIP {
+						nodeIP = addr.Address
+						break
+					}
+				}
+				if nodeIP == "" {
+					c.JSON(500, gin.H{"message": "node ip not found"})
+					return
+				}
+				c.String(200, nodeIP)
+			})
+			router.Run(":9090")
+		}
+	}()
+
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
 		setupLog.Error(err, "unable to set up health check")
 		os.Exit(1)
@@ -123,7 +167,7 @@ func main() {
 	}
 
 	setupLog.Info("starting manager")
-	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
+	if err := mgr.Start(ctx); err != nil {
 		setupLog.Error(err, "problem running manager")
 		os.Exit(1)
 	}

config/crd/bases/infra.cuisongliu.github.com_automqs.yaml

@@ -597,6 +597,10 @@ spec:
                 required:
                 - enable
                 type: object
+              nodePort:
+                description: NodePort is the node port of the AutoMQ
+                format: int32
+                type: integer
               s3:
                 description: S3 is the S3 configuration for the AutoMQ
                 properties:

defaults/up.sh

@@ -186,15 +186,26 @@ kafka_monitor_ip() {
 
     # get private ip first
     local_private_ip="0.0.0.0"
-    advertised_ip="${local_private_ip}"
+    advertised_ip_port="${local_private_ip}:9092"
+    if [[ -n "${OPERATOR_APIS_ADDR}" ]]; then
+        node_ip=$(curl -f -s "${OPERATOR_APIS_ADDR}:/api/v1/nodes/${NODE_NAME}")
+        if [[ $? -eq 0 && -n "$node_ip" ]]; then
+            echo "kafka_monitor_ip: node_ip=${node_ip}"
+            advertised_ip_port="${node_ip}:${NODEPORT_DEFAULT_PORT}"
+        else
+            echo "Failed to retrieve node_ip from ${OPERATOR_APIS_ADDR}"
+            exit 1
+        fi
+    fi
+
 
     # change ip settings for this node
     if [[ "${process_role}" == "server" ]]; then
         setup_value "listeners" "PLAINTEXT://${local_private_ip}:9092,CONTROLLER://${local_private_ip}:9093" "${kafka_dir}/config/kraft/${process_role}.properties"
-        setup_value "advertised.listeners" "PLAINTEXT://${advertised_ip}:9092" "${kafka_dir}/config/kraft/${process_role}.properties"
+        setup_value "advertised.listeners" "PLAINTEXT://${advertised_ip_port}" "${kafka_dir}/config/kraft/${process_role}.properties"
     elif [[ "${process_role}" == "broker" ]]; then
         setup_value "listeners" "PLAINTEXT://${local_private_ip}:9092" "${kafka_dir}/config/kraft/${process_role}.properties"
-        setup_value "advertised.listeners" "PLAINTEXT://${advertised_ip}:9092" "${kafka_dir}/config/kraft/${process_role}.properties"
+        setup_value "advertised.listeners" "PLAINTEXT://${advertised_ip_port}" "${kafka_dir}/config/kraft/${process_role}.properties"
     elif [[ "${process_role}" == "controller" ]]; then
         setup_value "listeners" "CONTROLLER://${local_private_ip}:9093" "${kafka_dir}/config/kraft/${process_role}.properties"
     else

deploy/charts/automq-operator/crds/infra.cuisongliu.github.com_automqs.yaml

@@ -597,6 +597,10 @@ spec:
                 required:
                 - enable
                 type: object
+              nodePort:
+                description: NodePort is the node port of the AutoMQ
+                format: int32
+                type: integer
               s3:
                 description: S3 is the S3 configuration for the AutoMQ
                 properties:

deploy/charts/automq-operator/templates/deployment.yaml

@@ -53,13 +53,18 @@ spec:
                   apiVersion: v1
             - name: ENABLE_WEBHOOKS
               value: "{{.Values.webhook.enabled}}"
+            - name: OPERATOR_APIS_SVC_NAME
+              value: "{{ include "automq-operator.fullname" . }}-apis"
           ports:
             - containerPort: 9443
               name: webhook-server
               protocol: TCP
             - name: http
               containerPort: 8081
               protocol: TCP
+            - name: apis
+              containerPort: 9090
+              protocol: TCP
           livenessProbe:
             httpGet:
               path: /healthz

deploy/charts/automq-operator/templates/service.yaml

@@ -16,4 +16,18 @@ spec:
     {{- include "automq-operator.selectorLabels" . | nindent 4 }}
 {{- end }}
 ---
-
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ include "automq-operator.fullname" . }}-apis
+  labels:
+    {{- include "automq-operator.labels" . | nindent 4 }}
+    app: k8s-api
+spec:
+  ports:
+    - name: apis
+      port: 9090
+      protocol: TCP
+      targetPort: apis
+  selector:
+    {{- include "automq-operator.selectorLabels" . | nindent 4 }}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.38
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.64.1
 	github.com/cuisongliu/logger v0.0.0-20230412024334-6d0345c427ba
+	github.com/gin-gonic/gin v1.10.0
 	github.com/go-bindata/go-bindata v3.1.2+incompatible
 	github.com/labring/operator-sdk v1.0.5
 	github.com/onsi/ginkgo/v2 v2.14.0
@@ -36,17 +37,27 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sts v1.31.4 // indirect
 	github.com/aws/smithy-go v1.21.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/bytedance/sonic v1.11.6 // indirect
+	github.com/bytedance/sonic/loader v0.1.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cloudwego/base64x v0.1.4 // indirect
+	github.com/cloudwego/iasm v0.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.8.0 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
+	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/zapr v1.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.3 // indirect
+	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/go-playground/validator/v10 v10.20.0 // indirect
 	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect
+	github.com/goccy/go-json v0.10.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
@@ -58,30 +69,38 @@ require (
 	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
+	github.com/leodido/go-urn v1.4.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pelletier/go-toml/v2 v2.2.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.18.0 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/ugorji/go/codec v1.2.12 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
+	golang.org/x/arch v0.8.0 // indirect
+	golang.org/x/crypto v0.23.0 // indirect
 	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
-	golang.org/x/net v0.21.0 // indirect
+	golang.org/x/net v0.25.0 // indirect
 	golang.org/x/oauth2 v0.12.0 // indirect
-	golang.org/x/sys v0.17.0 // indirect
-	golang.org/x/term v0.17.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/sys v0.20.0 // indirect
+	golang.org/x/term v0.20.0 // indirect
+	golang.org/x/text v0.15.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	golang.org/x/tools v0.16.1 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/protobuf v1.33.0 // indirect
+	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect