RockteMQ-AI commented on code in PR #183:
URL: https://github.com/apache/rocketmq-operator/pull/183#discussion_r3839447768
##########
pkg/share/share.go:
##########
@@ -18,22 +18,61 @@
// Package share defines some variables shared by different packages
package share
-var (
- // GroupNum is the number of broker group
- GroupNum = 0
+import (
+ "context"
+ "sort"
+ "strings"
Review 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.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -136,41 +136,43 @@ func (r *ReconcileBroker) Reconcile(ctx context.Context,
request reconcile.Reque
return reconcile.Result{}, err
}
+ var groupNum int
if broker.Status.Size == 0 {
- share.GroupNum = broker.Spec.Size
+ groupNum = broker.Spec.Size
} else {
- share.GroupNum = broker.Status.Size
+ groupNum = broker.Status.Size
Review 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).
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -343,6 +344,27 @@ func (r *ReconcileBroker) Reconcile(ctx context.Context,
request reconcile.Reque
return reconcile.Result{Requeue: true, RequeueAfter:
time.Duration(cons.RequeueIntervalInSecond) * time.Second}, nil
}
+func (r *ReconcileBroker) getControllerAccessPoint(namespace string,
rocketMqName string) string {
+ controllerList := &rocketmqv1alpha1.ControllerList{}
+ err := r.client.List(context.TODO(), controllerList,
&client.MatchingFields{
Review 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.
##########
pkg/apis/rocketmq/v1alpha1/controller_types.go:
##########
@@ -22,12 +22,19 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
+const (
Review Comment:
ControllerRocketMqNameIndexKey is defined as `"spec.rocketMqNameNamespaced"`
(without leading dot), while NameServiceRocketMqNameIndexKey is defined as
`".spec.rocketMqNameNamespaced"` (with leading dot). While both work as index
keys (they are opaque strings), the inconsistency suggests a typo. The
leading-dot convention is typically used for field indexer keys in
controller-runtime.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -216,17 +229,40 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
}
// use admin tool to update broker config
- if share.IsNameServersStrUpdated && (len(oldNameServerListStr)
> cons.MinIpListLength) && (len(share.NameServersStr) > cons.MinIpListLength) {
+ 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
+ // 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)
+ clusterListOutput, err := clusterListCmd.Output()
+ if err != nil {
+ reqLogger.Error(err, "Get cluster list failed,
command: "+cons.AdminToolDir+" "+cons.ClusterList+" -n "+oldNameServerListStr)
+ return reconcile.Result{Requeue: true}, err
+ }
+ // get cluster of output
+ clusterName := ""
+ for _, line := range
strings.Split(string(clusterListOutput), "\n") {
+ if strings.HasPrefix(line, "#Cluster Name") {
+ continue
+ }
+ for _, f := range strings.Fields(line) {
+ clusterName = f
+ break
+ }
+ }
Review Comment:
The `err` variable used in the error log at the 'Get empty cluster name'
line is nil — it was last set by the successful `clusterListCmd.Output()` call
above (which succeeded since we didn't return early). This means the error log
and the returned `err` will be nil, which may cause the caller to not properly
handle the failure. A new error should be constructed here, e.g.,
`fmt.Errorf("empty cluster name from clusterList output")`.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -187,22 +199,23 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
for _, value := range hostIps {
nameServerListStr = nameServerListStr + value + ":9876;"
}
-
+ newNameServerListStr := ""
// Update status.NameServers if needed
if !reflect.DeepEqual(hostIps, instance.Status.NameServers) {
oldNameServerListStr := ""
for _, value := range instance.Status.NameServers {
oldNameServerListStr = oldNameServerListStr + value +
":9876;"
Review 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.
##########
pkg/controller/topictransfer/topictransfer_controller.go:
##########
@@ -128,7 +128,7 @@ func (r *ReconcileTopicTransfer) Reconcile(ctx
context.Context, request reconcil
targetCluster := topicTransfer.Spec.TargetCluster
sourceCluster := topicTransfer.Spec.SourceCluster
- nameServer := strings.Split(share.NameServersStr, ";")[0]
+ nameServer := strings.Split(share.GetNameServersStr(r.client,
topicTransfer.Namespace, topicTransfer.Spec.RocketMqName), ";")[0]
Review 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).
##########
pkg/share/share.go:
##########
@@ -18,22 +18,61 @@
// Package share defines some variables shared by different packages
package share
-var (
- // GroupNum is the number of broker group
- GroupNum = 0
+import (
+ "context"
+ "sort"
+ "strings"
- // NameServersStr is the name server list
- NameServersStr = ""
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ "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"
+)
- // IsNameServersStrUpdated is whether the name server list is updated
- IsNameServersStrUpdated = false
+func GetNameServersStr(r client.Reader, namespace, rocketMqName string) string
{
+ nameserviceList := &rocketmqv1alpha1.NameServiceList{}
+ err := r.List(context.TODO(), nameserviceList, &client.MatchingFields{
+ rocketmqv1alpha1.NameServiceRocketMqNameIndexKey: rocketMqName
+ "-" + namespace,
+ })
+ if err != nil {
+ return ""
+ }
+ if len(nameserviceList.Items) != 1 {
+ return ""
+ }
Review 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.
##########
pkg/controller/console/console_controller.go:
##########
@@ -124,21 +124,22 @@ func (r *ReconcileConsole) Reconcile(ctx context.Context,
request reconcile.Requ
return reconcile.Result{}, err
}
+ var nameserverStr string
if instance.Spec.NameServers == "" {
// wait for name server ready if nameServers is omitted
for {
Review 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.
##########
pkg/share/share.go:
##########
@@ -18,22 +18,61 @@
// Package share defines some variables shared by different packages
package share
-var (
- // GroupNum is the number of broker group
- GroupNum = 0
+import (
+ "context"
+ "sort"
+ "strings"
- // NameServersStr is the name server list
- NameServersStr = ""
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ "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"
Review 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.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -216,17 +229,40 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
}
Review Comment:
The cluster name is extracted from `mqadmin clusterList` output by parsing
the first non-header field of the first data line. This only picks up the first
broker cluster name in the output. If multiple broker clusters share the same
NameService, only the first one gets its config updated. Additionally, if the
clusterList output has no data lines (all brokers are down), clusterName will
be empty and the function returns an error, but the error variable `err` at
that point is nil (from the successful clusterListCmd.Output()), so `err`
passed to the error log is nil, producing a misleading log message.
##########
charts/rocketmq-operator/templates/role_binding.yaml:
##########
@@ -13,17 +13,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
Review Comment:
The ClusterRoleBinding is now hardcoded to namespace 'default' and
ServiceAccount name 'rocketmq-operator'. This breaks any deployment that
installs the operator into a non-default namespace (which is very common). The
original templated version used `{{ .Release.Namespace }}` and `{{ template
"rocketmq-operator.serviceAccountName" . }}` which correctly adapted to the
Helm release configuration. This is a backward-incompatible regression for all
users deploying to namespaces other than 'default'.
##########
charts/rocketmq-operator/templates/service_account.yaml:
##########
@@ -16,4 +16,4 @@
apiVersion: v1
kind: ServiceAccount
metadata:
- name: {{ template "rocketmq-operator.serviceAccountName" . }}
+ name: rocketmq-operator
Review 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.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -216,17 +229,40 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
}
// use admin tool to update broker config
- if share.IsNameServersStrUpdated && (len(oldNameServerListStr)
> cons.MinIpListLength) && (len(share.NameServersStr) > cons.MinIpListLength) {
+ 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
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
