RockteMQ-AI commented on code in PR #56:
URL: https://github.com/apache/rocketmq-operator/pull/56#discussion_r3839467617
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -131,31 +131,43 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
return reconcile.Result{}, err
}
+ actualKey := broker.Namespace + "-" + broker.Spec.RocketMQName
Review Comment:
The broker reconcile loop spins in a tight busy-wait (`for { if
actual.IsNameServersStrInitialized { break } else { time.Sleep(...) } }`) on
the controller goroutine. This blocks the entire reconcile goroutine
indefinitely, preventing any other reconcile requests from being processed for
this controller. This should be replaced with a requeue-based approach (return
reconcile.Result{Requeue: true, RequeueAfter: ...}) so the controller manager
can continue handling other events.
##########
pkg/share/shareitem_sync_map.go:
##########
@@ -0,0 +1,36 @@
+package share
+
+import "sync"
+
+type ItemSyncMap struct {
+ m sync.Map
+}
+
+func (sMap *ItemSyncMap) Delete(key string) {
+ sMap.m.Delete(key)
+}
+
+func (sMap *ItemSyncMap) Load(key string) (value ShareItem, ok bool) {
+ v, ok := sMap.m.Load(key)
+ if v != nil {
+ value = v.(ShareItem)
+ }
+ return
+}
+
+func (sMap *ItemSyncMap) LoadOrStore(key string, value ShareItem) (actual
ShareItem, loaded bool) {
+ a, loaded := sMap.m.LoadOrStore(key, value)
Review Comment:
LoadOrStore will panic if the key exists and the stored value is nil,
because `a.(ShareItem)` is an unconditional type assertion. Although ShareItem
is a struct (not a pointer), any future refactor to a pointer type would cause
a nil panic. More critically, if the underlying sync.Map somehow stores a
non-ShareItem value (e.g., due to a bug), this will panic at runtime. Use a
safe assertion: `actual, _ = a.(ShareItem)` to avoid panic.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -131,31 +131,43 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
return reconcile.Result{}, err
}
+ actualKey := broker.Namespace + "-" + broker.Spec.RocketMQName
Review Comment:
Race condition: `actual` is loaded from the sync map into a local variable
at line ~134, then the busy-wait loop re-loads `actual` from the map inside the
loop body, but the outer `defer` at line ~139 always stores the local `actual`
variable back on function exit. If `actual.IsNameServersStrInitialized` becomes
true during the wait loop (set by the nameservice controller), the defer will
overwrite the map with the snapshot captured at loop-entry time, potentially
clobbering fields like NameServersStr that the nameservice controller wrote.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -454,9 +475,13 @@ func (r *ReconcileBroker) getBrokerStatefulSet(broker
*rocketmqv1alpha1.Broker,
}
func getENV(broker *rocketmqv1alpha1.Broker, replicaIndex int,
brokerGroupIndex int) []corev1.EnvVar {
+
Review Comment:
`getENV` calls `share.GetInstance().LoadOrStore(actualKey,
share.ShareItem{})` independently of the reconcile loop's local `actual`
variable. This means `getENV` may read a stale or empty NameServersStr if
called before the reconcile loop has stored the updated value, since the defer
hasn't fired yet at StatefulSet creation time. The NameServersStr used to
configure the pod environment could be empty on first creation.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -233,7 +249,7 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
Review Comment:
`actual.IsNameServersStrInitialized` is only set to `true` when
`runningNameServerNum == instance.Spec.Size`, but it is never reset to `false`
if name server pods go down. If the cluster shrinks or pods restart,
`IsNameServersStrInitialized` stays `true` indefinitely, and brokers will not
re-wait for name servers to become ready again. Add logic to reset this flag
when running count drops below spec size.
##########
deploy/clusterrole_binding.yaml:
##########
@@ -0,0 +1,27 @@
+# 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.
+
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+ name: rocketmq-operator
+subjects:
+ - kind: ServiceAccount
+ name: rocketmq-operator
Review Comment:
The ClusterRoleBinding hardcodes `namespace: default` for the ServiceAccount
subject. If the operator is deployed in a non-default namespace, this binding
will be incorrect and the operator will lack permissions. This should be
templated or documented clearly, since ClusterRole + hardcoded namespace is a
common misconfiguration in multi-tenant deployments.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -253,10 +265,17 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
podNames := getPodNames(podList.Items)
Review Comment:
After the early return when `len(podNames) == 0`, the reconcile returns
without storing the updated `actual` back to the sync map (the `defer` will
still run, but `actual.GroupNum` and `actual.BrokerClusterName` will have been
set just before this point). However, `actual.NameServersStr` may not be
populated yet if nameServers is empty and the wait loop hasn't run. The defer
stores a potentially incomplete `actual`, which could overwrite a valid
previously-stored value if the nameservice controller already populated it.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -189,8 +201,8 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
// Check for name server scaling
if broker.Spec.AllowRestart {
// The following code will restart all brokers to update
NAMESRV_ADDR env
Review Comment:
Inconsistent indentation in the `if actual.IsNameServersStrUpdated` block:
the inner `for` loop is indented with extra tabs compared to surrounding code.
This is a minor formatting issue but indicates the code may not have been run
through `gofmt`, which can cause CI lint failures.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -174,7 +174,23 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
}
Review Comment:
The guard `len(hostIps) != int(instance.Spec.Size) || len(hostIps) == 0`
returns early before the `actualKey` is looked up and the defer is registered.
This means when this condition triggers, no state is written to the share map.
This is correct behavior, but the comment `// hostIps is empty,
instance.Status.NameServers is empty, also not in` inside the later block is
misleading since that condition can no longer be reached when hostIps is empty.
##########
deploy/cluster_role.yaml:
##########
@@ -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.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+ creationTimestamp: null
+ name: rocketmq-operator
+rules:
+ - apiGroups:
+ - ""
+ resources:
+ - pods
+ - services
+ - endpoints
+ - persistentvolumeclaims
+ - events
+ - configmaps
+ - secrets
+ - pods/exec
+ verbs:
+ - '*'
+ - apiGroups:
+ - ""
+ resources:
+ - namespaces
+ verbs:
+ - get
+ - apiGroups:
+ - apps
+ resources:
+ - deployments
+ - daemonsets
+ - replicasets
+ - statefulsets
+ verbs:
+ - '*'
+ - apiGroups:
+ - monitoring.coreos.com
+ resources:
+ - servicemonitors
+ verbs:
+ - get
+ - create
+ - apiGroups:
+ - apps
+ resourceNames:
+ - rocketmq-operator
+ resources:
+ - deployments/finalizers
+ verbs:
Review Comment:
The `rocketmq.apache.org` ClusterRole rule lists both `'*'` (wildcard) and
explicit resources `brokers`, `pods/exec`, `topictransfers`. The wildcard
already covers everything, making the explicit entries redundant. More
importantly, `pods/exec` is not a valid subresource of the
`rocketmq.apache.org` API group — it belongs to the core `""` group. This is a
copy-paste error that should be removed to avoid confusion and potential future
misinterpretation.
##########
deploy/crds/rocketmq_v1alpha1_broker_crd.yaml:
##########
@@ -93,6 +96,7 @@ spec:
- volumes
Review Comment:
`rocketMQName` is now a required field in the CRD. This is a breaking change
for any existing Broker CR that does not have this field set. Existing clusters
that upgrade to this operator version will have their Broker CRs fail
validation. A defaulting webhook or a non-required field with a documented
migration path is needed for safe upgrades.
##########
pkg/controller/topictransfer/topictransfer_controller.go:
##########
@@ -127,7 +127,9 @@ func (r *ReconcileTopicTransfer) Reconcile(request
reconcile.Request) (reconcile
targetCluster := topicTransfer.Spec.TargetCluster
sourceCluster := topicTransfer.Spec.SourceCluster
- nameServer := strings.Split(share.NameServersStr, ";")[0]
+ actualKey := topicTransfer.Namespace + "-" +
topicTransfer.Spec.RocketMQName
Review Comment:
`strings.Split(actual.NameServersStr, ";")[0]` will return an empty string
(not panic) if `actual.NameServersStr` is empty, and the subsequent
`len(nameServer) < cons.MinIpListLength` check handles that. However, if
`actual` was just default-initialized by `LoadOrStore` (i.e., the nameservice
for this `rocketMQName` has not yet reconciled), the TopicTransfer will
silently terminate rather than requeue with an informative error. Consider
returning a requeue result instead of terminating.
##########
deploy/crds/rocketmq_v1alpha1_nameservice_crd.yaml:
##########
@@ -73,6 +76,7 @@ spec:
- storageMode
Review Comment:
Same breaking change as the Broker CRD: making `rocketMQName` required on
NameService CRD will invalidate all existing NameService CRs on upgrade. Needs
a migration strategy.
##########
pkg/apis/rocketmq/v1alpha1/topictransfer_types.go:
##########
@@ -37,6 +37,8 @@ type TopicTransferSpec struct {
SourceCluster string `json:"sourceCluster,omitempty"`
// The cluster where the topic will be transferred to
TargetCluster string `json:"targetCluster,omitempty"`
+ // // RocketMQ Name, the broker and nameserver in the same cluster
must be filled with the same name
Review Comment:
Double comment marker on line 40: `// // RocketMQ Name, ...` — there
is a tab and an extra `//` before the actual comment text. This is a copy-paste
artifact and should be `// RocketMQ Name, ...`.
##########
cmd/manager/main.go:
##########
@@ -100,7 +93,6 @@ func main() {
Review Comment:
Removing `Namespace` from `manager.Options` changes the operator from
namespace-scoped to cluster-scoped watching. This is intentional (given the new
ClusterRole), but it is a significant behavioral change: the operator will now
watch all namespaces, increasing API server load and requiring the new
ClusterRole/ClusterRoleBinding to be applied. This should be explicitly
documented in the PR and migration notes, and the old namespace-scoped
Role/RoleBinding files should be deprecated or removed to avoid confusion.
--
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]
