RockteMQ-AI commented on code in PR #56:
URL: https://github.com/apache/rocketmq-operator/pull/56#discussion_r3909363540
##########
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
+ actual, _ := share.GetInstance().LoadOrStore(actualKey,
share.ShareItem{})
+ defer func() {
+ reqLogger.Info("Broker actualKey:" + actualKey + "
actual.NameServerStr:" + actual.NameServersStr +
+ " actual.BrokerClusterName:" + actual.BrokerClusterName
+ " IsNameServersStrInitialized:" +
strconv.FormatBool(actual.IsNameServersStrInitialized))
+ share.GetInstance().Store(actualKey, actual)
+ }()
+
+ 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
}
-
if broker.Spec.NameServers == "" {
// wait for name server ready when create broker cluster if
nameServers is omitted
for {
- if share.IsNameServersStrInitialized {
+ actual, _ = share.GetInstance().Load(actualKey)
+ if actual.IsNameServersStrInitialized {
break
} else {
- log.Info("Broker Waiting for name server
ready...")
+ log.Info("Broker Waiting for name server
ready..., actualKey:" +
+ actualKey + " actual.NameServersStr:" +
actual.NameServersStr + " IsNameServersStrInitialized:" +
+
strconv.FormatBool(actual.IsNameServersStrInitialized))
time.Sleep(time.Duration(cons.WaitForNameServerReadyInSecond) * time.Second)
}
}
} else {
- share.NameServersStr = broker.Spec.NameServers
+ actual.NameServersStr = broker.Spec.NameServers
Review Comment:
actual.NameServersStr is only assigned on the local copy fetched at the top
of Reconcile; it is written back to the sync.Map in the deferred Store (line
139), i.e. after getBrokerStatefulSet() has already run. getENV() (line 484)
re-loads the item from the map, so on the first reconcile of a Broker with
spec.nameServers set explicitly, the StatefulSet is created with an empty
NAMESRV_ADDR env var and the brokers cannot start. The old global-variable code
made the mutation immediately visible to getENV; the copy-in/copy-out semantics
break that. Store the item immediately after mutating it, or pass the
name-server string down into getBrokerStatefulSet/getENV as a parameter.
##########
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
+ actual, _ := share.GetInstance().LoadOrStore(actualKey,
share.ShareItem{})
+ defer func() {
+ reqLogger.Info("Broker actualKey:" + actualKey + "
actual.NameServerStr:" + actual.NameServersStr +
+ " actual.BrokerClusterName:" + actual.BrokerClusterName
+ " IsNameServersStrInitialized:" +
strconv.FormatBool(actual.IsNameServersStrInitialized))
+ share.GetInstance().Store(actualKey, actual)
+ }()
+
+ 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
}
-
if broker.Spec.NameServers == "" {
// wait for name server ready when create broker cluster if
nameServers is omitted
for {
Review Comment:
This busy-wait loop blocks the reconcile worker indefinitely. With per-key
state, a mismatch between broker.spec.rocketMQName and the NameService CR's
rocketMQName (a plain string with no cross-CRD validation — a single typo)
means IsNameServersStrInitialized is never set and this goroutine sleeps
forever. With the default MaxConcurrentReconciles=1, one misconfigured cluster
starves every broker CR in every other cluster, defeating the multi-cluster
goal of this PR. Prefer returning reconcile.Result{RequeueAfter: ...} like the
pod-not-ready handling already added further down in this function.
##########
deploy/crds/rocketmq_v1alpha1_broker_crd.yaml:
##########
@@ -93,6 +96,7 @@ spec:
- volumes
- volumeClaimTemplates
- scalePodName
+ - rocketMQName
Review Comment:
rocketMQName is added to `required` (also in the nameservice and
topictransfer CRDs). Existing Broker/NameService/TopicTransfer CRs created by
older operator versions lack this field, so after the CRD update any subsequent
write to those objects (kubectl apply/edit, and potentially the operator's own
status updates) fails OpenAPI validation — there is no migration path
documented. Also note that CRs patched with an empty value silently collide on
the key "<namespace>-". Consider making the field optional with a documented
default, or document a mandatory pre-upgrade migration for all existing CRs.
##########
cmd/manager/main.go:
##########
@@ -100,7 +93,6 @@ func main() {
// Create a new Cmd to provide shared dependencies and start components
Review Comment:
Removing GetWatchNamespace()/the Namespace option silently ignores the
WATCH_NAMESPACE env var that existing operator deployments set. After
upgrading, the operator expands from namespaced to cluster-wide watching with
no notice; deployments that deliberately scoped it for least privilege will
break at startup because the caches now need list/watch across all namespaces
(and old namespaced Role manifests no longer suffice). Keep honoring
WATCH_NAMESPACE, or at minimum log a loud warning when it is set.
##########
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
Review Comment:
The ClusterRoleBinding subject hardcodes namespace: default. If the operator
is installed in any other namespace, its ServiceAccount never receives the
ClusterRole and — since the manager now watches all namespaces — informer cache
sync fails and the operator cannot start. The binding should be parameterized
by the install namespace, and deploy/operator.yaml needs to be updated to
reference this cluster-scoped RBAC (the diff does not touch it, so existing
deployments still ship the namespaced Role and WATCH_NAMESPACE).
##########
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
+ actual, _ := share.GetInstance().LoadOrStore(actualKey,
share.ShareItem{})
+ defer func() {
+ reqLogger.Info("Broker actualKey:" + actualKey + "
actual.NameServerStr:" + actual.NameServersStr +
+ " actual.BrokerClusterName:" + actual.BrokerClusterName
+ " IsNameServersStrInitialized:" +
strconv.FormatBool(actual.IsNameServersStrInitialized))
+ share.GetInstance().Store(actualKey, actual)
Review Comment:
ShareItem is copied out, mutated, and stored back as a whole by different
controllers (broker, nameservice, topictransfer) with no per-key locking.
Concurrent reconciles for the same key perform read-modify-write on independent
struct copies, so the last Store wins and the other controller's changes are
silently lost — e.g. a broker reconcile finishing here can revert a
NameServersStr / IsNameServersStrUpdated value just written by the nameservice
controller. sync.Map only protects the map itself, not lost updates on the
values. Guard each key with a mutex, or have each controller store only the
fields it owns.
##########
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
- if share.IsNameServersStrUpdated {
- for brokerGroupIndex := 0; brokerGroupIndex <
broker.Spec.Size; brokerGroupIndex++ {
+ if actual.IsNameServersStrUpdated {
+ for brokerGroupIndex := 0; brokerGroupIndex <
broker.Spec.Size; brokerGroupIndex++ {
Review Comment:
The for loop under `if actual.IsNameServersStrUpdated` is indented one level
too deep — a gofmt violation introduced by this change; please run
gofmt/goimports on the touched files.
##########
pkg/share/share.go:
##########
@@ -18,19 +18,45 @@
// Package share defines some variables shared by different packages
package share
+import "sync"
+
var (
- // GroupNum is the number of broker group
- GroupNum = 0
+ once sync.Once
+ instance *Singleton
+)
+type ShareItem struct {
Review Comment:
No unit tests accompany the new share package or the reworked reconcile
logic, although both are easily testable (LoadOrStore/Store round-trips,
concurrent updates to one key, and the Reconcile path where spec.nameServers is
set explicitly). A test asserting that a newly created broker StatefulSet
carries the expected NAMESRV_ADDR would have caught the stale-copy bug in
getENV.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -206,25 +222,25 @@ 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 actual.IsNameServersStrUpdated && (len(oldNameServerListStr)
> cons.MinIpListLength) && (len(actual.NameServersStr) > cons.MinIpListLength) {
mqAdmin := cons.AdminToolDir
subCmd := cons.UpdateBrokerConfig
key := cons.ParamNameServiceAddress
- reqLogger.Info("share.GroupNum=broker.Spec.Size=" +
strconv.Itoa(share.GroupNum))
+ reqLogger.Info("share.GroupNum=broker.Spec.Size=" +
strconv.Itoa(actual.GroupNum))
+
- clusterName := share.BrokerClusterName
+ clusterName := actual.BrokerClusterName
Review Comment:
clusterName (and GroupNum) come from in-memory cross-controller state that
is only populated after a Broker reconcile for the same rocketMQName has run.
After an operator restart the map is empty, so a name-server list change
triggers the mqAdmin updateBrokerConfig command with `-c ""`, which fails and
silently leaves brokers configured with stale name-server addresses. Consider
looking the broker cluster name up from the Broker CR (label/owner lookup)
instead of relying on shared memory. This weakness is inherited from the old
globals, but the restart-loss scenario is worth fixing while this state is
being reworked.
##########
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:
The comment is doubled (`// \t// RocketMQ Name...`) — leftover from editing.
Similar "and and" duplications appear in the new README and example YAML
comments; worth a quick copy pass.
--
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]
