【ISSUE #133】RocketMQ-Operator support the rocketmq-5.0 proxy node for cluster pattern

Makefile

@@ -126,13 +126,15 @@ endif
 install: manifests kustomize ## Install CRDs into the K8s cluster specified in ~/.kube/config.
 	kubectl create -f deploy/crds/rocketmq.apache.org_brokers.yaml
 	kubectl create -f deploy/crds/rocketmq.apache.org_nameservices.yaml
+	kubectl create -f deploy/crds/rocketmq.apache.proxys.yaml
 	kubectl create -f deploy/crds/rocketmq.apache.org_consoles.yaml
 	kubectl create -f deploy/crds/rocketmq.apache.org_topictransfers.yaml
 
 .PHONY: uninstall
 uninstall: manifests kustomize ## Uninstall CRDs from the K8s cluster specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource not found errors during deletion.
 	kubectl delete --ignore-not-found=$(ignore-not-found) -f deploy/crds/rocketmq.apache.org_brokers.yaml
 	kubectl delete --ignore-not-found=$(ignore-not-found) -f deploy/crds/rocketmq.apache.org_nameservices.yaml
+    kubectl delete --ignore-not-found=$(ignore-not-found) -f deploy/crds/rocketmq.apache.org_proxys.yaml
 	kubectl delete --ignore-not-found=$(ignore-not-found) -f deploy/crds/rocketmq.apache.org_consoles.yaml
 	kubectl delete --ignore-not-found=$(ignore-not-found) -f deploy/crds/rocketmq.apache.org_topictransfers.yaml
 

example/rocketmq_v1alpha1_proxy_cr.yaml

@@ -0,0 +1,70 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: proxy-config
+data:
+  rmq-proxy.json: |
+    {
+      "rocketMQClsuterName": "DefaultCluster",
+      "namesrvAddr": "127.0.0.1:9876",
+      "proxyMode": "CLUSTER"
+    }
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: broker-config
+data:
+  broker.conf: |
+    deleteWhen = 04
+    fileReservedTime = 48
+    flushDiskType = ASYNC_FLUSH
+    namesrvAddr = 127.0.0.1:9876
+---
+apiVersion: rocketmq.apache.org/v1alpha1
+kind: Proxy
+metadata:
+  name: proxy
+  namespace: default
+spec:
+  brokerConfigPath: "/root/rocketmq/broker/conf/broker.conf"
+  proxyConfigPath: "/root/rocketmq/proxy/conf/rmq-proxy.json"
+  proxyMode: "LOCAL"
+  # proxyStatefulSet define the proxy StatefulSet
+  proxyStatefulSet:
+    apiVersion: apps/v1
+    kind: StatefulSet
+    metadata:
+      labels:
+        app: rocketmq-proxy
+    spec:
+      replicas: 1
+      selector:
+        matchLabels:
+          app: rocketmq-proxy
+      template:
+        metadata:
+          labels:
+            app: rocketmq-proxy
+        spec:
+          containers:
+            - name: proxy
+              image: apacherocketmq/rocketmq-proxy:5.0.0-alpine-operator-0.3.0
+              ports:
+                - containerPort: 8081
+              volumeMounts:
+                # Consistent with proxyConfigPath
+                - mountPath: "/root/rocketmq/proxy/conf/rmq-proxy.json"
+                  subPath: rmq-proxy.json
+                  name: proxy-config
+                # Consistent with brokerConfigPath
+                - mountPath: "/root/rocketmq/proxy/conf/broker.conf"
+                  subPath: broker.conf
+                  name: broker-config
+          volumes:
+            - name: proxy-config
+              configMap:
+                name: proxy-config
+            - name: broker-config
+              configMap:
+                name: broker-config

example/rocketmq_v1alpha1_proxy_service.yaml

@@ -0,0 +1,30 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: proxy-service
+  labels:
+    app: rocketmq-proxy
+spec:
+  type: NodePort
+  selector:
+    app: rocketmq-proxy
+  ports:
+    - port: 8081
+      targetPort: 8081
+      protocol: TCP
+      nodePort: 31000

images/proxy/alpine/Dockerfile

@@ -0,0 +1,72 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+FROM openjdk:8-alpine
+
+RUN apk add --no-cache bash gettext nmap-ncat openssl busybox-extras
+
+ARG version
+
+# Rocketmq version
+ENV ROCKETMQ_VERSION ${version}
+
+# Rocketmq home
+ENV ROCKETMQ_HOME  /root/rocketmq/proxy
+
+WORKDIR  ${ROCKETMQ_HOME}
+
+# Install
+RUN set -eux; \
+    apk add --virtual .build-deps curl gnupg unzip; \
+    curl https://archive.apache.org/dist/rocketmq/${ROCKETMQ_VERSION}/rocketmq-all-${ROCKETMQ_VERSION}-bin-release.zip -o rocketmq.zip; \
+    curl https://archive.apache.org/dist/rocketmq/${ROCKETMQ_VERSION}/rocketmq-all-${ROCKETMQ_VERSION}-bin-release.zip.asc -o rocketmq.zip.asc; \
+	curl -L https://www.apache.org/dist/rocketmq/KEYS -o KEYS; \
+	\
+	gpg --import KEYS; \
+    gpg --batch --verify rocketmq.zip.asc rocketmq.zip; \
+    unzip rocketmq.zip; \
+	mv rocketmq-all*/* . ; \
+	rmdir rocketmq-all* ; \
+	rm rocketmq.zip ; \
+	rm rocketmq.zip.asc KEYS; \
+	apk del .build-deps ; \
+    rm -rf /var/cache/apk/* ; \
+    rm -rf /tmp/*
+
+# Copy customized scripts
+COPY runserver-customize.sh ${ROCKETMQ_HOME}/bin/
+
+# Expose proxy ports
+EXPOSE 8081
+
+# Override customized scripts for proxy
+RUN mv ${ROCKETMQ_HOME}/bin/runserver-customize.sh ${ROCKETMQ_HOME}/bin/runserver.sh \
+ && chmod a+x ${ROCKETMQ_HOME}/bin/runserver.sh \
+ && chmod a+x ${ROCKETMQ_HOME}/bin/mqproxy
+
+# Export Java options
+RUN export JAVA_OPT=" -Duser.home=/opt"
+
+# Add ${JAVA_HOME}/lib/ext as java.ext.dirs
+RUN sed -i 's/${JAVA_HOME}\/jre\/lib\/ext/${JAVA_HOME}\/jre\/lib\/ext:${JAVA_HOME}\/lib\/ext/' ${ROCKETMQ_HOME}/bin/tools.sh
+
+COPY  proxyStart.sh ${ROCKETMQ_HOME}/bin/
+
+RUN chmod a+x ${ROCKETMQ_HOME}/bin/proxyStart.sh
+WORKDIR ${ROCKETMQ_HOME}/bin
+
+CMD ["/bin/bash", "./proxyStart.sh"]

images/proxy/alpine/proxyStart.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+if [ $PROXY_MODE == "LOCAL" ]; then
+./mqproxy -bc $BROKER_CONFIG_PATH -pc $PROXY_CONFIG_PATH -pm $PROXY_MODE
+fi
+./mqproxy -pc $PROXY_CONFIG_PATH -pm $PROXY_MODE
+

images/proxy/alpine/runserver-customize.sh

@@ -0,0 +1,150 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#===========================================================================================
+# Java Environment Setting
+#===========================================================================================
+error_exit ()
+{
+    echo "ERROR: $1 !!"
+    exit 1
+}
+
+find_java_home()
+{
+    case "`uname`" in
+        Darwin)
+            JAVA_HOME=$(/usr/libexec/java_home)
+        ;;
+        *)
+            JAVA_HOME=$(dirname $(dirname $(readlink -f $(which javac))))
+        ;;
+    esac
+}
+
+find_java_home
+
+[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=$HOME/jdk/java
+[ ! -e "$JAVA_HOME/bin/java" ] && JAVA_HOME=/usr/java
+[ ! -e "$JAVA_HOME/bin/java" ] && error_exit "Please set the JAVA_HOME variable in your environment, We need java(x64)!"
+
+export JAVA_HOME
+export JAVA="$JAVA_HOME/bin/java"
+export BASE_DIR=$(dirname $0)/..
+export CLASSPATH=.:${BASE_DIR}/conf:${CLASSPATH}
+
+#===========================================================================================
+# JVM Configuration
+#===========================================================================================
+calculate_heap_sizes()
+{
+    case "`uname`" in
+        Linux)
+            system_memory_in_mb=`free -m| sed -n '2p' | awk '{print $2}'`
+            system_memory_in_mb_in_docker=$(($(cat /sys/fs/cgroup/memory/memory.limit_in_bytes)/1024/1024))
+            if [ $system_memory_in_mb_in_docker -lt $system_memory_in_mb ];then
+              system_memory_in_mb=$system_memory_in_mb_in_docker
+            fi
+            system_cpu_cores=`egrep -c 'processor([[:space:]]+):.*' /proc/cpuinfo`
+            system_cpu_cores_in_docker=$(($(cat /sys/fs/cgroup/cpu/cpu.cfs_quota_us)/$(cat /sys/fs/cgroup/cpu/cpu.cfs_period_us)))
+            if [ $system_cpu_cores_in_docker -lt $system_cpu_cores -a $system_cpu_cores_in_docker -ne 0 ];then
+              system_cpu_cores=$system_cpu_cores_in_docker
+            fi
+        ;;
+        FreeBSD)
+            system_memory_in_bytes=`sysctl hw.physmem | awk '{print $2}'`
+            system_memory_in_mb=`expr $system_memory_in_bytes / 1024 / 1024`
+            system_cpu_cores=`sysctl hw.ncpu | awk '{print $2}'`
+        ;;
+        SunOS)
+            system_memory_in_mb=`prtconf | awk '/Memory size:/ {print $3}'`
+            system_cpu_cores=`psrinfo | wc -l`
+        ;;
+        Darwin)
+            system_memory_in_bytes=`sysctl hw.memsize | awk '{print $2}'`
+            system_memory_in_mb=`expr $system_memory_in_bytes / 1024 / 1024`
+            system_cpu_cores=`sysctl hw.ncpu | awk '{print $2}'`
+        ;;
+        *)
+            # assume reasonable defaults for e.g. a modern desktop or
+            # cheap server
+            system_memory_in_mb="2048"
+            system_cpu_cores="2"
+        ;;
+    esac
+
+    # some systems like the raspberry pi don't report cores, use at least 1
+    if [ "$system_cpu_cores" -lt "1" ]
+    then
+        system_cpu_cores="1"
+    fi
+
+    # set max heap size based on the following
+    # max(min(1/2 ram, 1024MB), min(1/4 ram, 8GB))
+    # calculate 1/2 ram and cap to 1024MB
+    # calculate 1/4 ram and cap to 8192MB
+    # pick the max
+    half_system_memory_in_mb=`expr $system_memory_in_mb / 2`
+    quarter_system_memory_in_mb=`expr $half_system_memory_in_mb / 2`
+    if [ "$half_system_memory_in_mb" -gt "1024" ]
+    then
+        half_system_memory_in_mb="1024"
+    fi
+    if [ "$quarter_system_memory_in_mb" -gt "8192" ]
+    then
+        quarter_system_memory_in_mb="8192"
+    fi
+    if [ "$half_system_memory_in_mb" -gt "$quarter_system_memory_in_mb" ]
+    then
+        max_heap_size_in_mb="$half_system_memory_in_mb"
+    else
+        max_heap_size_in_mb="$quarter_system_memory_in_mb"
+    fi
+    MAX_HEAP_SIZE="${max_heap_size_in_mb}M"
+
+    # Young gen: min(max_sensible_per_modern_cpu_core * num_cores, 1/4 * heap size)
+    max_sensible_yg_per_core_in_mb="100"
+    max_sensible_yg_in_mb=`expr $max_sensible_yg_per_core_in_mb "*" $system_cpu_cores`
+
+    desired_yg_in_mb=`expr $max_heap_size_in_mb / 4`
+
+    if [ "$desired_yg_in_mb" -gt "$max_sensible_yg_in_mb" ]
+    then
+        HEAP_NEWSIZE="${max_sensible_yg_in_mb}M"
+    else
+        HEAP_NEWSIZE="${desired_yg_in_mb}M"
+    fi
+}
+
+calculate_heap_sizes
+
+# Dynamically calculate parameters, for reference.
+Xms=$MAX_HEAP_SIZE
+Xmx=$MAX_HEAP_SIZE
+Xmn=$HEAP_NEWSIZE
+# Set for `JAVA_OPT`.
+JAVA_OPT="${JAVA_OPT} -server -Xms${Xms} -Xmx${Xmx} -Xmn${Xmn}"
+JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFraction=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8  -XX:-UseParNewGC"
+JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:/dev/shm/rmq_srv_gc.log -XX:+PrintGCDetails"
+JAVA_OPT="${JAVA_OPT} -XX:-OmitStackTraceInFastThrow"
+JAVA_OPT="${JAVA_OPT}  -XX:-UseLargePages"
+JAVA_OPT="${JAVA_OPT} -Djava.ext.dirs=${JAVA_HOME}/jre/lib/ext:${BASE_DIR}/lib"
+#JAVA_OPT="${JAVA_OPT} -Xdebug -Xrunjdwp:transport=dt_socket,address=9555,server=y,suspend=n"
+JAVA_OPT="${JAVA_OPT} ${JAVA_OPT_EXT}"
+JAVA_OPT="${JAVA_OPT} -cp ${CLASSPATH}"
+
+$JAVA ${JAVA_OPT} $@

main.go

@@ -23,6 +23,7 @@ import (
 	"github.com/apache/rocketmq-operator/pkg/controller/broker"
 	"github.com/apache/rocketmq-operator/pkg/controller/console"
 	"github.com/apache/rocketmq-operator/pkg/controller/nameservice"
+	"github.com/apache/rocketmq-operator/pkg/controller/proxy"
 	"github.com/apache/rocketmq-operator/pkg/controller/topictransfer"
 	"os"
 
@@ -106,6 +107,12 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err := proxy.SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to add proxy controller to manager")
+		os.Exit(1)
+	}
+
+
 	//+kubebuilder:scaffold:builder
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {