RockteMQ-AI commented on code in PR #183:
URL: https://github.com/apache/rocketmq-operator/pull/183#discussion_r3902419948
##########
pkg/apis/rocketmq/v1alpha1/nameservice_types.go:
##########
@@ -22,12 +22,19 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
+const (
Review 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.
##########
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:
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.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -216,17 +229,40 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
}
Review 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.
##########
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
+apiVersion: rbac.authorization.k8s.io/v1
metadata:
- name: {{ include "rocketmq-operator.fullname" . }}
- labels:
- {{- include "rocketmq-operator.labels" . | nindent 4 }}
+ name: rocketmq-operator
+subjects:
+- kind: ServiceAccount
+ name: rocketmq-operator
Review Comment:
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 than `default`, prevents multiple releases in the same
cluster, and makes the chart non-functional for most production deployments.
The original templated approach should be restored.
##########
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") {
Review 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.
##########
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` 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.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -202,19 +204,18 @@ func (r *ReconcileBroker) Reconcile(ctx context.Context,
request reconcile.Reque
// Check for name server scaling
if broker.Spec.AllowRestart {
- // The following code will restart all brokers to update
NAMESRV_ADDR env
- if share.IsNameServersStrUpdated {
- for brokerGroupIndex := 0; brokerGroupIndex <
broker.Spec.Size; brokerGroupIndex++ {
- brokerName := getBrokerName(broker,
brokerGroupIndex)
- // Update master broker
- reqLogger.Info("Update Master Broker
NAMESRV_ADDR of " + brokerName)
- dep := r.getBrokerStatefulSet(broker,
brokerGroupIndex, 0)
- found := &appsv1.StatefulSet{}
- err = r.client.Get(context.TODO(),
types.NamespacedName{Name: dep.Name, Namespace: dep.Namespace}, found)
- if err != nil {
- reqLogger.Error(err, "Failed to get
broker master StatefulSet of "+brokerName)
- } else {
-
found.Spec.Template.Spec.Containers[0].Env[0].Value = share.NameServersStr
+ // The following code will restart all brokers to update
NAMESRV_ADDR env if name server list is updated
+ for brokerGroupIndex := 0; brokerGroupIndex < broker.Spec.Size;
brokerGroupIndex++ {
+ brokerName := getBrokerName(broker, brokerGroupIndex)
+ 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 {
Review 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.
##########
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{
Review 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.
##########
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 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.
##########
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 ""
+ }
- // IsNameServersStrInitialized is whether the name server list is
initialized
- IsNameServersStrInitialized = false
+ nameservice := nameserviceList.Items[0]
+ labelSelector :=
labels.SelectorFromSet(tool.LabelsForNameService(nameservice.Name))
+ listOps := &client.ListOptions{
+ Namespace: nameservice.Namespace,
+ LabelSelector: labelSelector,
+ }
+ podList := &corev1.PodList{}
+ err = r.List(context.Background(), podList, listOps)
+ if err != nil {
+ return ""
+ }
Review 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.
##########
pkg/tool/resource_name.go:
##########
@@ -26,3 +26,7 @@ func BuildHeadlessSvcResourceName(name string) string {
func BuildSvcResourceName(name string) string {
return fmt.Sprintf("%s-svc", name)
}
+
+func LabelsForNameService(name string) map[string]string {
Review 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.
--
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]
