RockteMQ-AI commented on code in PR #87:
URL: https://github.com/apache/rocketmq-operator/pull/87#discussion_r3902449693
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -104,6 +105,43 @@ type ReconcileBroker struct {
scheme *runtime.Scheme
}
+func (r *ReconcileBroker) CreateService(request reconcile.Request, broker
*rocketmqv1alpha1.Broker, brokerGroupIndex int, replicaIndex int) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Create a Broker Service...")
+
+ svc := r.getBrokerService(broker, brokerGroupIndex, replicaIndex)
+ svcObj := &corev1.Service{}
+ err := r.client.Get(context.TODO(), types.NamespacedName{Name:
svc.Name, Namespace: svc.Namespace}, svcObj)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating a Broker Service.",
"Service.Namespace", svc.Namespace, "Service.Name", svc.Name)
+ err = r.client.Create(context.TODO(), svc)
+ if err != nil {
+ reqLogger.Error(err, "Failed to create new Service",
"Service.Namespace", svc.Namespace, "Service.Name", svc.Name)
+ }
+ } else if err != nil {
+ reqLogger.Error(err, "Failed to get Broker Service.")
+ }
+}
+
+func (r *ReconcileBroker) DLegerHostIp(request reconcile.Request, broker
*rocketmqv1alpha1.Broker, brokerGroupIndex int, replicaIndex int) string {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Get DLeger Host IP lists...")
+
+ statefulSetName := getBrokerStatefulSetName(broker, brokerGroupIndex,
replicaIndex)
+
Review Comment:
`DLegerHostIp` contains an unbounded `for {}` loop with `time.Sleep` that
polls forever until the Service exists. This blocks the reconcile goroutine
indefinitely if the Service creation failed (e.g., RBAC denial, quota
exceeded). The controller will deadlock — no other reconcile events can be
processed on this worker. Replace with a bounded retry or, better, return the
error from Reconcile and let the controller framework requeue with backoff.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -104,6 +105,43 @@ type ReconcileBroker struct {
scheme *runtime.Scheme
}
+func (r *ReconcileBroker) CreateService(request reconcile.Request, broker
*rocketmqv1alpha1.Broker, brokerGroupIndex int, replicaIndex int) {
Review Comment:
`CreateService` logs the error from `r.client.Create` but does not return
it. The caller proceeds to build a StatefulSet and call `DLegerHostIp` which
will then block forever waiting for the Service that failed to create. This
should return an `error` to the caller so Reconcile can requeue.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -154,9 +192,18 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
share.BrokerClusterName = broker.Name
replicaPerGroup := broker.Spec.ReplicaPerGroup
reqLogger.Info("brokerGroupNum=" + strconv.Itoa(share.GroupNum) + ",
replicaPerGroup=" + strconv.Itoa(replicaPerGroup))
+
for brokerGroupIndex := 0; brokerGroupIndex < share.GroupNum;
brokerGroupIndex++ {
Review Comment:
DLeger peer construction and `CreateService`/`DLegerHostIp` calls are
unconditional — they run even when `broker.Spec.EnableDLeger` is `false`. This
creates unnecessary Services and blocks on them for every non-DLedger
deployment. Wrap the DLeger-specific logic in an `if broker.Spec.EnableDLeger`
guard.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -157,9 +158,18 @@ func (r *ReconcileNameService) Reconcile(request
reconcile.Request) (reconcile.R
return r.updateNameServiceStatus(instance, request, true)
}
+func getNameServiceName(nameService *rocketmqv1alpha1.NameService) string {
Review Comment:
`getNameServiceName` panics when `nameService.Spec.Size` is 0: the loop body
never executes, `nameserviceName` remains `""`, and
`nameserviceName[:len(nameserviceName)-1]` evaluates to `""[:-1]` which is an
out-of-bounds slice. Add a guard for `Size == 0`.
##########
pkg/share/share.go:
##########
@@ -25,6 +25,9 @@ var (
// NameServersStr is the name server list
NameServersStr = ""
+ // NameServersStr is the name server list
Review Comment:
`NameServersServiceStr` is a package-level mutable global written by the
NameService controller (`nameservice_controller.go:168`) and read by the Broker
controller (`broker_controller.go:575`). This is a data race in a concurrent
controller-runtime environment where both reconcilers run in parallel
goroutines. Use a thread-safe mechanism (mutex, or pass via CR status) instead
of a bare global string.
##########
pkg/apis/rocketmq/v1alpha1/broker_types.go:
##########
@@ -34,14 +34,18 @@ type BrokerSpec struct {
Size int `json:"size"`
// NameServers defines the name service list e.g.
192.168.1.1:9876;192.168.1.2:9876
NameServers string `json:"nameServers,omitempty"`
+ // Whether enable rocketmq-on-dleger group deploy, false in default
+ EnableDLeger bool `json:"enableDLeger"`
// ReplicaPerGroup each broker cluster's replica number
ReplicaPerGroup int `json:"replicaPerGroup"`
// BaseImage is the broker image to use for the Pods
BrokerImage string `json:"brokerImage"`
// ImagePullPolicy defines how the image is pulled
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy"`
// AllowRestart defines whether allow pod restart
- AllowRestart bool `json:"allowRestart"`
+ AllowRestart bool `json:"allowRestart"`
+ Affinity corev1.Affinity `json:"affinity"`
Review Comment:
`Tolerations` and other new fields (`Env` in NameService) also lack
`omitempty` in their JSON tags. This forces users to provide empty arrays in
their CRs and makes the CRD schema unnecessarily strict.
##########
pkg/apis/rocketmq/v1alpha1/broker_types.go:
##########
@@ -34,14 +34,18 @@ type BrokerSpec struct {
Size int `json:"size"`
// NameServers defines the name service list e.g.
192.168.1.1:9876;192.168.1.2:9876
NameServers string `json:"nameServers,omitempty"`
+ // Whether enable rocketmq-on-dleger group deploy, false in default
+ EnableDLeger bool `json:"enableDLeger"`
// ReplicaPerGroup each broker cluster's replica number
ReplicaPerGroup int `json:"replicaPerGroup"`
// BaseImage is the broker image to use for the Pods
BrokerImage string `json:"brokerImage"`
// ImagePullPolicy defines how the image is pulled
ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy"`
// AllowRestart defines whether allow pod restart
- AllowRestart bool `json:"allowRestart"`
+ AllowRestart bool `json:"allowRestart"`
Review Comment:
`Affinity` is declared as `corev1.Affinity` (value type) without
`omitempty`. An empty Affinity struct will always be serialized into the
StatefulSet pod spec even when the user sets none, and the CRD makes it
required. Use `*corev1.Affinity` (pointer) with `json:"affinity,omitempty"` so
it is nil when unset.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -104,6 +105,43 @@ type ReconcileBroker struct {
scheme *runtime.Scheme
}
+func (r *ReconcileBroker) CreateService(request reconcile.Request, broker
*rocketmqv1alpha1.Broker, brokerGroupIndex int, replicaIndex int) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Create a Broker Service...")
+
+ svc := r.getBrokerService(broker, brokerGroupIndex, replicaIndex)
+ svcObj := &corev1.Service{}
+ err := r.client.Get(context.TODO(), types.NamespacedName{Name:
svc.Name, Namespace: svc.Namespace}, svcObj)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating a Broker Service.",
"Service.Namespace", svc.Namespace, "Service.Name", svc.Name)
+ err = r.client.Create(context.TODO(), svc)
+ if err != nil {
+ reqLogger.Error(err, "Failed to create new Service",
"Service.Namespace", svc.Namespace, "Service.Name", svc.Name)
+ }
+ } else if err != nil {
+ reqLogger.Error(err, "Failed to get Broker Service.")
+ }
+}
+
+func (r *ReconcileBroker) DLegerHostIp(request reconcile.Request, broker
*rocketmqv1alpha1.Broker, brokerGroupIndex int, replicaIndex int) string {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Get DLeger Host IP lists...")
+
+ statefulSetName := getBrokerStatefulSetName(broker, brokerGroupIndex,
replicaIndex)
+
+ svcObj := &corev1.Service{}
+ for {
+ err := r.client.Get(context.TODO(), types.NamespacedName{Name:
statefulSetName, Namespace: broker.Namespace}, svcObj)
+ if err != nil && errors.IsNotFound(err) {
+ log.Info("Waiting for broker service created...")
+
time.Sleep(time.Duration(cons.WaitForNameServerReadyInSecond) * time.Second)
Review Comment:
`DLegerHostIp` returns `svcObj.Spec.ClusterIP` from a Service that may have
just been created. For non-headless Services (the broker services here have no
`ClusterIP: "None"`), the ClusterIP is assigned asynchronously by the API
server. A tight poll loop may read an empty ClusterIP. Consider using a
headless Service + pod DNS (e.g., `<pod>.<svc>.<ns>.svc.cluster.local`) for
stable addressing.
##########
pkg/controller/console/console_controller.go:
##########
@@ -200,12 +195,12 @@ func newDeploymentForCR(cr *rocketmqv1alpha1.Console)
*appsv1.Deployment {
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{{
- Resources:
cr.Spec.ConsoleDeployment.Spec.Template.Spec.Containers[0].Resources,
- Image:
cr.Spec.ConsoleDeployment.Spec.Template.Spec.Containers[0].Image,
- Name:
cr.Spec.ConsoleDeployment.Spec.Template.Spec.Containers[0].Name,
+ Resources:
cr.Spec.ConsoleDeployment.Spec.Template.Spec.Containers[0].Resources,
Review Comment:
The hardcoded `JAVA_OPTS` env var (containing `-Drocketmq.namesrv.addr=...`)
was removed without a replacement mechanism. Users who relied on the operator
injecting the name server address into the console will now get a console that
cannot connect to RocketMQ. The example CRs add `JAVA_OPTS` manually, but
existing deployments that upgrade will silently break. Consider keeping a
default or documenting this as a breaking change.
##########
deploy/crds/rocketmq_v1alpha1_broker_crd.yaml:
##########
@@ -95,24 +841,26 @@ spec:
type: object
type: array
required:
Review Comment:
`enableDLeger` is added to the `required` fields list. This is a breaking
change — any existing Broker CR that does not specify `enableDLeger` will be
rejected by the API server on update. New optional fields should not be
required; rely on the Go zero-value (`false`) or add a default via admission
webhook.
##########
images/broker/alpine/brokerGenConfig.sh:
##########
@@ -30,6 +30,17 @@ function create_config() {
if [ $BROKER_ID != 0 ]; then
sed -i 's/brokerRole=.*/brokerRole=SLAVE/g' $BROKER_CONFIG_FILE
fi
+
+ # Enable RocketMQ-on-DLedger Group
Review Comment:
The shell test `[ ! -z $ENABLE_DLEGER ]` is unquoted. If `ENABLE_DLEGER` is
unset (not just empty), this becomes `[ ! -z ]` which evaluates to true (tests
the `-z` flag itself as a non-empty string), enabling DLedger unintentionally.
Quote it: `[ ! -z "$ENABLE_DLEGER" ]` or use `[ -n "$ENABLE_DLEGER" ]`.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -301,7 +311,53 @@ func labelsForNameService(name string) map[string]string {
return map[string]string{"app": "name_service", "name_service_cr": name}
}
+func (r *ReconcileNameService) getNameServiceService(nameService
*rocketmqv1alpha1.NameService) *corev1.Service {
+ statefulSetName := nameService.Name
+ ls := labelsForNameService(nameService.Name)
+
+ svc := &corev1.Service{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: statefulSetName,
+ Labels: ls,
+ Namespace: nameService.Namespace,
+ },
+ Spec: corev1.ServiceSpec{
+ Selector: ls,
+ ClusterIP: "None",
+ Ports: []corev1.ServicePort{{
+ Name:
cons.NameServiceMainContainerPortName,
+ Port: cons.NameServiceMainContainerPort,
+ TargetPort:
intstr.FromInt(cons.NameServiceMainContainerPort),
+ }},
+ },
+ }
+
+ // Set bind for broker crd and svc
+ controllerutil.SetControllerReference(nameService, svc, r.scheme)
+
+ return svc
+}
+
+func (r *ReconcileNameService) CreateService(request reconcile.Request,
nameService *rocketmqv1alpha1.NameService) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Create a Name Service...")
+
Review Comment:
The comment says `// Set bind for broker crd and svc` but this is in the
NameService controller and the owner reference is set to `nameService`, not a
broker. Copy-paste artifact — update the comment to reflect the actual owner.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -453,10 +569,10 @@ func (r *ReconcileBroker) getBrokerStatefulSet(broker
*rocketmqv1alpha1.Broker,
}
Review Comment:
`share.NameServersStr` was replaced with `share.NameServersServiceStr` in
the broker env. `NameServersServiceStr` is built from the CR name and assumes a
specific headless service naming convention. If the NameService controller
hasn't reconciled yet, this value may be stale or empty, causing brokers to
start with no valid name server address.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -193,8 +241,14 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
for brokerGroupIndex := 0; brokerGroupIndex <
broker.Spec.Size; brokerGroupIndex++ {
Review Comment:
In the scale-down/update path, `CreateService` is called for replicas inside
the nameServers-update loop but never for the master (replicaIndex 0). If the
master's Service was deleted externally, it will not be recreated during
updates, while replica services will be. This is inconsistent with the initial
creation path at line 196.
##########
pkg/share/share.go:
##########
@@ -25,6 +25,9 @@ var (
// NameServersStr is the name server list
NameServersStr = ""
+ // NameServersStr is the name server list
Review Comment:
The comment on `NameServersServiceStr` says `// NameServersStr is the name
server list` — it's a copy of the comment above it. Update to describe what
this variable actually holds (service-based name server addresses).
--
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]
