[ISSUE #176] Support manage multi cluster - #183
Conversation
|
@caigy PTAL |
caigy
left a comment
caigy
left a comment
There was a problem hiding this comment.
It seems that there're lots of modifications in this pr, you'd better add some doc for your design, especially for the changes in CRs reflecting the status of all modules in the RocketMQ cluster.
| err := mgr.GetFieldIndexer().IndexField(context.TODO(), &rocketmqv1alpha1.NameService{}, rocketmqv1alpha1.NameServiceRocketMqNameIndexKey, | ||
| func(rawObj client.Object) []string { | ||
| n, ok := rawObj.(*rocketmqv1alpha1.NameService) | ||
| if !ok { | ||
| return nil | ||
| } | ||
| return []string{n.Spec.RocketMqName + "-" + n.Namespace} | ||
| }, | ||
| ) | ||
| if err != nil { | ||
| return err | ||
| } |
There was a problem hiding this comment.
What's the usage of this block? You'd better show more about your design.
| // get cluster of output | ||
| clusterName := "" | ||
| for _, line := range strings.Split(string(clusterListOutput), "\n") { | ||
| if strings.HasPrefix(line, "#Cluster Name") { | ||
| continue | ||
| } |
There was a problem hiding this comment.
It's not stable to get cluster name by analyzing the output of a command.
| // #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE | ||
| // broker broker-0 0 192.168.180.40:10911 V4_5_0 0.00(0,0ms) 0.00(0,0ms) 0 471030.34 -1.0000 | ||
| // broker broker-0 1 192.168.137.89:10911 V4_5_0 0.00(0,0ms) 0.00(0,0ms) 0 471030.34 0.2673 | ||
| clusterListCmd := exec.Command("sh", cons.AdminToolDir, cons.ClusterList, "-n", oldNameServerListStr) |
There was a problem hiding this comment.
As isNameServersStrUpdated is true, why using oldNameServerListStr as the addresses of name servers?
| command := mqAdmin + " " + subCmd + " -c " + clusterName + " -k " + key + " -n " + oldNameServerListStr + " -v " + newNameServerListStr | ||
| cmd := exec.Command("sh", mqAdmin, subCmd, "-c", clusterName, "-k", key, "-n", oldNameServerListStr, "-v", newNameServerListStr) |
There was a problem hiding this comment.
Only brokers registering successfully to name servers can receive the command, others will not be updated.
| runningNameServerNum := getRunningNameServersNum(podList.Items) | ||
| if runningNameServerNum == instance.Spec.Size { | ||
| share.IsNameServersStrInitialized = true | ||
| share.NameServersStr = nameServerListStr // reassign if operator restarts | ||
| } |
There was a problem hiding this comment.
Can this block be safely removed?
OK, i would add this. |
|
This PR has conflicts with the base branch and cannot be merged. Please rebase or merge the base branch into your branch and resolve the conflicts: git fetch origin
git checkout multiple-rmqs
git rebase origin/main
# resolve conflicts, then:
git push --force-with-leaseThis is a one-time reminder. Feel free to @mention me for a re-review after conflicts are resolved. Automated notification by github-manager-bot |
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR modifies 25 file(s) with 805 lines of diff. No test changes detected — consider adding test coverage.
Automated review by github-manager-bot
Additional notes (not anchored to a changed line)
- [INFO]
charts/rocketmq-operator/crds/rocketmq.apache.org_brokers.yaml:1— Large diff (805 lines). Consider breaking into smaller, focused PRs for easier review. (line outside diff)
| @@ -28,6 +28,9 @@ import ( | |||
| // BrokerSpec defines the desired state of Broker | |||
There was a problem hiding this comment.
No test changes detected alongside source modifications. Consider adding tests to cover the changes.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
PR received and logged for review. This PR requires detailed code review by a maintainer.
Diff size: 805 lines
Author: drivebyer (CONTRIBUTOR)
Automated review by RockteMQ-AI
| "context" | ||
| "sort" | ||
| "strings" | ||
|
|
There was a problem hiding this comment.
GetNameServersStr requires exactly 1 NameService matching the index key (len != 1 returns empty). If a user has zero or more than one NameService CR with the same rocketMqName in a namespace, the broker controller enters an infinite blocking loop (broker_controller.go busy-wait for {} loop), starving the reconcile goroutine and preventing any progress. The function should either tolerate multiple NameServices or provide a meaningful error/log.
| groupNum = broker.Spec.Size | ||
| } else { | ||
| share.GroupNum = broker.Status.Size | ||
| groupNum = broker.Status.Size |
There was a problem hiding this comment.
The for {} busy-wait loop for name server readiness blocks the reconcile goroutine indefinitely with only a 2-second sleep between iterations. If the NameService never becomes ready, this blocks the controller thread forever. This should use return reconcile.Result{Requeue: true, RequeueAfter: ...}, nil instead, consistent with the pattern already used elsewhere in this same file (e.g., line 163 for controller readiness).
| var nameserverStr string | ||
| if instance.Spec.NameServers == "" { | ||
| // wait for name server ready if nameServers is omitted | ||
| for { |
There was a problem hiding this comment.
Same infinite blocking loop as in the broker controller: for {} with a sleep waiting for name server readiness. This blocks the reconcile goroutine permanently if the NameService never becomes ready. Should requeue instead.
|
|
||
| func (r *ReconcileBroker) getControllerAccessPoint(namespace string, rocketMqName string) string { | ||
| controllerList := &rocketmqv1alpha1.ControllerList{} | ||
| err := r.client.List(context.TODO(), controllerList, &client.MatchingFields{ |
There was a problem hiding this comment.
getControllerAccessPoint requires exactly 1 Controller (len != 1 returns empty). If there are 0 or 2+ Controller CRs with the same rocketMqName, this returns empty, causing the broker to requeue endlessly in CONTROLLER mode without a clear diagnostic message explaining why.
| "github.com/apache/rocketmq-operator/pkg/tool" | ||
| corev1 "k8s.io/api/core/v1" | ||
| "k8s.io/apimachinery/pkg/labels" | ||
| "sigs.k8s.io/controller-runtime/pkg/client" |
There was a problem hiding this comment.
GetNameServersStr uses client.MatchingFields with the NameServiceRocketMqNameIndexKey, but the index is registered only in the nameservice controller's add() function (nameservice_controller.go). If GetNameServersStr is called before that index is registered (e.g., during startup ordering), the List call will fail silently (returning empty string). Similarly, the controller index in broker_controller.go uses mgr.GetCache().IndexField while nameservice uses mgr.GetFieldIndexer().IndexField — these are equivalent but the inconsistency is worth noting.
| kind: ServiceAccount | ||
| metadata: | ||
| name: {{ template "rocketmq-operator.serviceAccountName" . }} | ||
| name: rocketmq-operator |
There was a problem hiding this comment.
The ServiceAccount name is now hardcoded to 'rocketmq-operator' instead of using the Helm template {{ template "rocketmq-operator.serviceAccountName" . }}. This removes the ability to customize the ServiceAccount name and will conflict if multiple operator instances are deployed in the same cluster.
| if len(nameserviceList.Items) != 1 { | ||
| return "" | ||
| } | ||
|
|
There was a problem hiding this comment.
GetNameServersStr does not check whether all NameService pods are running before returning the name server list. It returns as soon as at least one running pod with a non-empty IP is found. Previously, IsNameServersStrInitialized was only set to true when runningNameServerNum == instance.Spec.Size (all name servers running). This change means brokers and consoles may connect to a partially-ready name server cluster, which could cause intermittent failures during initial deployment.
| if isNameServersStrUpdated && (len(oldNameServerListStr) > cons.MinIpListLength) && (len(newNameServerListStr) > cons.MinIpListLength) { | ||
| // bash-4.4$ ./mqadmin clusterList -n 192.168.180.36:9876 | ||
| // #Cluster Name #Broker Name #BID #Addr #Version #InTPS(LOAD) #OutTPS(LOAD) #PCWait(ms) #Hour #SPACE | ||
| // broker broker-0 0 192.168.180.40:10911 V4_5_0 0.00(0,0ms) 0.00(0,0ms) 0 471030.34 -1.0000 |
There was a problem hiding this comment.
The clusterList command is executed on the operator pod itself using exec.Command("sh", ...), which requires the RocketMQ admin tool to be installed in the operator container. Parsing its stdout is fragile — the output format could change between RocketMQ versions. If the admin tool is not available in the operator container (e.g., minimal operator image), this will fail.
| sourceCluster := topicTransfer.Spec.SourceCluster | ||
|
|
||
| nameServer := strings.Split(share.NameServersStr, ";")[0] | ||
| nameServer := strings.Split(share.GetNameServersStr(r.client, topicTransfer.Namespace, topicTransfer.Spec.RocketMqName), ";")[0] |
There was a problem hiding this comment.
GetNameServersStr is called inline and its result is immediately split on ';'. If GetNameServersStr returns an empty string (e.g., no matching NameService), strings.Split("", ";")[0] returns an empty string, which then passes the length check. However, the function will silently proceed with an empty name server address rather than logging the root cause (no matching NameService found for the given rocketMqName).
| if !reflect.DeepEqual(hostIps, instance.Status.NameServers) { | ||
| oldNameServerListStr := "" | ||
| for _, value := range instance.Status.NameServers { | ||
| oldNameServerListStr = oldNameServerListStr + value + ":9876;" |
There was a problem hiding this comment.
The variable newNameServerListStr is declared before the if !reflect.DeepEqual(...) block but is only assigned inside it. If the DeepEqual check passes (no update needed), newNameServerListStr remains empty string. This is not currently used after the block so it's harmless, but the declaration scope is wider than necessary.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
The core idea — replacing global mutable state with per-resource lookups keyed by rocketMqName — is the right direction, but the implementation has critical issues: two infinite busy-wait loops that will block the controller manager, broken Helm chart templates with hardcoded namespace/SA names, and a fragile clusterList parsing path embedded in the status update that can deadlock name server scaling.
Findings
- [WARNING]
pkg/apis/rocketmq/v1alpha1/nameservice_types.go:25— Index key constant has a leading dot (".spec.rocketMqNameNamespaced") whileControllerRocketMqNameIndexKeyincontroller_types.godoes not ("spec.rocketMqNameNamespaced"). While each indexer/lookup pair is internally consistent, this inconsistency across types is confusing and error-prone for future contributors. Pick one convention (the leading-dot form is standard for controller-runtime field indexers) and apply it uniformly. - [CRITICAL]
pkg/controller/nameservice/nameservice_controller.go:229— The newclusterListcommand execution and cluster name parsing are embedded insideupdateNameServiceStatus. If the admin tool call fails or returns an empty cluster name, the function returns early with an error, which prevents the NameService status from being updated at all — even when the pod IP list has legitimately changed. This creates a deadlock: name server scaling triggers a status update, but the update fails because the admin tool can't reach the (just-scaled) name servers. Separate the cluster-list lookup from the status update, or at least make it non-fatal. - [WARNING]
pkg/controller/nameservice/nameservice_controller.go:245— Cluster name parsing takes only the first field of the first non-header line frommqadmin clusterListoutput. When multiple broker clusters share the same NameService (which is the multi-cluster scenario this PR enables), only the first cluster's config will be updated viaupdateBrokerConfig. The remaining broker clusters will have stalenamesrvAddrconfig, leading to message routing failures. Consider iterating over all unique cluster names in the output. - [CRITICAL]
pkg/controller/broker/broker_controller.go:145— Infinite busy-wait loop: whenbroker.Spec.NameServersis empty, this loop pollsGetNameServersStrwith a 2-second sleep and never breaks out until a NameService is found. This blocks the reconcile goroutine indefinitely, preventing other Broker resources from being reconciled and potentially starving the controller manager. Replace with a requeue (return reconcile.Result{RequeueAfter: ...}, nil) when the name server is not yet available. - [WARNING]
pkg/controller/broker/broker_controller.go:213— TheAllowRestartname-server-update loop iterates overbroker.Spec.Sizeinstead of the locally computedgroupNum(which isbroker.Status.Sizeduring scale-down). When scaling down, this will attempt to update StatefulSets for broker groups that have already been deleted, causing spuriousGeterrors. UsegroupNumhere for consistency. - [CRITICAL]
charts/rocketmq-operator/templates/role_binding.yaml:22— The Helm template variables ({{ .Release.Namespace }},{{ include "rocketmq-operator.fullname" . }}) have been replaced with hardcoded values (name: rocketmq-operator,namespace: default). This breaks installation into any namespace other thandefault, prevents multiple releases in the same cluster, and makes the chart non-functional for most production deployments. The original templated approach should be restored. - [WARNING]
pkg/controller/topictransfer/topictransfer_controller.go:131—GetNameServersStrcan return an empty string (e.g., when no matching NameService exists).strings.Split("", ";")[0]yields"", which passes to thelen(nameServer) < cons.MinIpListLengthcheck — so it won't crash, but the error message is misleading ("no available name server" is correct but the root cause — no matching NameService for thisrocketMqName— is lost). Consider logging therocketMqNameand namespace to aid debugging. - [WARNING]
pkg/share/share.go:35—GetNameServersStrrequires exactly one NameService matching therocketMqNameindex (len(nameserviceList.Items) != 1returns empty). In multi-cluster scenarios where a user accidentally creates two NameService CRs with the samerocketMqNamein the same namespace, this silently returns empty with no error or log message. This will be very difficult to debug. Log a warning when zero or multiple matches are found. - [INFO]
pkg/share/share.go:55— This function duplicates the NameService pod-discovery and IP-collection logic fromnameservice_controller.go'supdateNameServiceStatus. If the label selector or the port/pod-filtering logic changes in one place but not the other, the two will diverge silently. Consider extracting the shared logic into a single helper. - [CRITICAL]
pkg/controller/console/console_controller.go:130— Same infinite busy-wait loop as the broker controller: whenNameServersis empty and no NameService is found, this loop blocks the reconcile goroutine forever. Replace with a requeue. - [INFO]
pkg/tool/resource_name.go:30—LabelsForNameServicehardcodes label keys/values ("app": "name_service","name_service_cr": name) that must match the labels set by the NameService controller when creating pods. If those labels are defined as constants elsewhere, they should be referenced here to avoid silent drift.
Automated review by github-manager-bot
| metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" | ||
| ) | ||
|
|
||
| const ( |
There was a problem hiding this comment.
Index key constant has a leading dot (".spec.rocketMqNameNamespaced") while ControllerRocketMqNameIndexKey in controller_types.go does not ("spec.rocketMqNameNamespaced"). While each indexer/lookup pair is internally consistent, this inconsistency across types is confusing and error-prone for future contributors. Pick one convention (the leading-dot form is standard for controller-runtime field indexers) and apply it uniformly.
| @@ -216,17 +229,40 @@ func (r *ReconcileNameService) updateNameServiceStatus(instance *rocketmqv1alpha | |||
| } | |||
There was a problem hiding this comment.
The new clusterList command execution and cluster name parsing are embedded inside updateNameServiceStatus. If the admin tool call fails or returns an empty cluster name, the function returns early with an error, which prevents the NameService status from being updated at all — even when the pod IP list has legitimately changed. This creates a deadlock: name server scaling triggers a status update, but the update fails because the admin tool can't reach the (just-scaled) name servers. Separate the cluster-list lookup from the status update, or at least make it non-fatal.
| } | ||
| // get cluster of output | ||
| clusterName := "" | ||
| for _, line := range strings.Split(string(clusterListOutput), "\n") { |
There was a problem hiding this comment.
Cluster name parsing takes only the first field of the first non-header line from mqadmin clusterList output. When multiple broker clusters share the same NameService (which is the multi-cluster scenario this PR enables), only the first cluster's config will be updated via updateBrokerConfig. The remaining broker clusters will have stale namesrvAddr config, leading to message routing failures. Consider iterating over all unique cluster names in the output.
| share.GroupNum = broker.Status.Size | ||
| groupNum = broker.Status.Size | ||
| } | ||
|
|
There was a problem hiding this comment.
Infinite busy-wait loop: when broker.Spec.NameServers is empty, this loop polls GetNameServersStr with a 2-second sleep and never breaks out until a NameService is found. This blocks the reconcile goroutine indefinitely, preventing other Broker resources from being reconciled and potentially starving the controller manager. Replace with a requeue (return reconcile.Result{RequeueAfter: ...}, nil) when the name server is not yet available.
| dep := r.getBrokerStatefulSet(broker, brokerGroupIndex, 0, controllerAccessPoint, nameServersStr) | ||
| found := &appsv1.StatefulSet{} | ||
| err = r.client.Get(context.TODO(), types.NamespacedName{Name: dep.Name, Namespace: dep.Namespace}, found) | ||
| if err != nil { |
There was a problem hiding this comment.
The AllowRestart name-server-update loop iterates over broker.Spec.Size instead of the locally computed groupNum (which is broker.Status.Size during scale-down). When scaling down, this will attempt to update StatefulSets for broker groups that have already been deleted, causing spurious Get errors. Use groupNum here for consistency.
| sourceCluster := topicTransfer.Spec.SourceCluster | ||
|
|
||
| nameServer := strings.Split(share.NameServersStr, ";")[0] | ||
| nameServer := strings.Split(share.GetNameServersStr(r.client, topicTransfer.Namespace, topicTransfer.Spec.RocketMqName), ";")[0] |
There was a problem hiding this comment.
GetNameServersStr can return an empty string (e.g., when no matching NameService exists). strings.Split("", ";")[0] yields "", which passes to the len(nameServer) < cons.MinIpListLength check — so it won't crash, but the error message is misleading ("no available name server" is correct but the root cause — no matching NameService for this rocketMqName — is lost). Consider logging the rocketMqName and namespace to aid debugging.
| IsNameServersStrUpdated = false | ||
| func GetNameServersStr(r client.Reader, namespace, rocketMqName string) string { | ||
| nameserviceList := &rocketmqv1alpha1.NameServiceList{} | ||
| err := r.List(context.TODO(), nameserviceList, &client.MatchingFields{ |
There was a problem hiding this comment.
GetNameServersStr requires exactly one NameService matching the rocketMqName index (len(nameserviceList.Items) != 1 returns empty). In multi-cluster scenarios where a user accidentally creates two NameService CRs with the same rocketMqName in the same namespace, this silently returns empty with no error or log message. This will be very difficult to debug. Log a warning when zero or multiple matches are found.
| err = r.List(context.Background(), podList, listOps) | ||
| if err != nil { | ||
| return "" | ||
| } |
There was a problem hiding this comment.
This function duplicates the NameService pod-discovery and IP-collection logic from nameservice_controller.go's updateNameServiceStatus. If the label selector or the port/pod-filtering logic changes in one place but not the other, the two will diverge silently. Consider extracting the shared logic into a single helper.
| var nameserverStr string | ||
| if instance.Spec.NameServers == "" { | ||
| // wait for name server ready if nameServers is omitted | ||
| for { |
There was a problem hiding this comment.
Same infinite busy-wait loop as the broker controller: when NameServers is empty and no NameService is found, this loop blocks the reconcile goroutine forever. Replace with a requeue.
| return fmt.Sprintf("%s-svc", name) | ||
| } | ||
|
|
||
| func LabelsForNameService(name string) map[string]string { |
There was a problem hiding this comment.
LabelsForNameService hardcodes label keys/values ("app": "name_service", "name_service_cr": name) that must match the labels set by the NameService controller when creating pods. If those labels are defined as constants elsewhere, they should be referenced here to avoid silent drift.
4 participants
What is the purpose of the change
In order to manage multi cluster by one operator.
task1 && task2 of #176
Brief changelog
XX
Verifying this change
tested against old cluster()
tested against creating two cluster in one namespace
yaml1(set rocketmq cluster name to empty string): https://gist.github.com/drivebyer/8795a96966be4fd6ebf395cc347a159d
yaml2(set rocketmq cluster name to test-rocketmq): https://gist.github.com/drivebyer/338a0cba168249dd3c02ce0f33f775d8
Please go through this checklist to help us incorporate your contribution quickly and easily.
Notice:
It would be helpful if you could finish the following checklist (the last one is not necessary) before request the community to review your PR.[ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.make docker-buildto build docker image for operator, try your changes from Pod inside your Kubernetes cluster, not just locally. Also provide screenshots to show that the RocketMQ cluster is healthy after the changes.make manifeststo make sure the CRD files are updated.