RockteMQ-AI commented on code in PR #63: URL: https://github.com/apache/rocketmq-operator/pull/63#discussion_r3839458654
########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { + nameServiceFound.Spec.ImagePullPolicy = nameServiceSts.Spec.ImagePullPolicy + nameServiceFound.Spec.NameServiceImage = nameServiceSts.Spec.NameServiceImage + nameServiceFound.Spec.Size = nameServiceSts.Spec.Size + err = r.client.Update(context.TODO(), nameServiceFound) + if err != nil { + reqLogger.Error(err, "should update nameservice resource", "spec.nameService", + nameServiceSts.Spec, "nameServiceFound.spec", nameServiceFound.Spec) + return reconcile.Result{}, err + } + } + } + + // Check if broker already exists, if not create a new one + brokerFound := &rocketmqv1alpha1.Broker{} + brokerSts := r.brokerForRocketmq(instance) + + err = r.client.Get(context.TODO(), types.NamespacedName{Name: brokerSts.Name, Namespace: brokerSts.Namespace}, brokerFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), brokerSts) + if err != nil { + reqLogger.Error(err, "Failed to create new broker of rocketmq", "broker.namespace", + brokerSts.Namespace, "broker.Name", brokerSts.Name) + } + return reconcile.Result{Requeue: true}, nil + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq broker.") + } else { + // Resource broker will change; Only ReplicaPerGroup Size ImagePullPolicy BrokerImage can update + if !reflect.DeepEqual(brokerSts.Spec.ReplicaPerGroup, brokerFound.Spec.ReplicaPerGroup) || + !reflect.DeepEqual(brokerSts.Spec.Size, brokerFound.Spec.Size) || Review Comment: When broker Get fails with a non-NotFound error, the error is logged but not returned. The code falls through to the status update section where `brokerFound` is still a zero-value `&Broker{}` (never populated), causing the status to be set to empty strings and potentially overwriting valid status. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { Review Comment: No readiness gate between NameService and Broker creation: the controller creates the Broker immediately after the NameService resource exists, without waiting for NameService pods to be Running/Ready. The README states 'the name server cluster will be created first, after all name server cluster is in running state, the operator will create the broker cluster' but the code does not enforce this ordering, which can cause brokers to fail on startup. ########## pkg/controller/console/console_controller.go: ########## @@ -178,11 +178,20 @@ func (r *ReconcileConsole) Reconcile(request reconcile.Request) (reconcile.Resul } // newDeploymentForCR returns a deployment pod with modifying the ENV -func newDeploymentForCR(cr *rocketmqv1alpha1.Console) *appsv1.Deployment { +func (r *ReconcileConsole) newDeploymentForCR(cr *rocketmqv1alpha1.Console) *appsv1.Deployment { env := corev1.EnvVar{ Name: "JAVA_OPTS", Value: fmt.Sprintf("-Drocketmq.namesrv.addr=%s -Dcom.rocketmq.sendMessageWithVIPChannel=false", share.NameServersStr), } Review Comment: The label mutation `selectorLabels["console-cr"] = cr.Name` and `labels["console-cr"] = cr.Name` modifies the maps from `cr.Spec` in place. Since maps are reference types in Go, this permanently mutates the CR's spec in memory, which can cause spurious diffs on subsequent reconcile passes and incorrect `Update` calls to the API server. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { + nameServiceFound.Spec.ImagePullPolicy = nameServiceSts.Spec.ImagePullPolicy + nameServiceFound.Spec.NameServiceImage = nameServiceSts.Spec.NameServiceImage + nameServiceFound.Spec.Size = nameServiceSts.Spec.Size + err = r.client.Update(context.TODO(), nameServiceFound) + if err != nil { + reqLogger.Error(err, "should update nameservice resource", "spec.nameService", + nameServiceSts.Spec, "nameServiceFound.spec", nameServiceFound.Spec) + return reconcile.Result{}, err Review Comment: After NameService creation, the code does not return early or requeue to wait for the NameService to become ready before proceeding to Broker creation. The controller should requeue after creating NameService and only proceed to Broker creation once NameService pods are confirmed running. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { + nameServiceFound.Spec.ImagePullPolicy = nameServiceSts.Spec.ImagePullPolicy + nameServiceFound.Spec.NameServiceImage = nameServiceSts.Spec.NameServiceImage + nameServiceFound.Spec.Size = nameServiceSts.Spec.Size + err = r.client.Update(context.TODO(), nameServiceFound) + if err != nil { + reqLogger.Error(err, "should update nameservice resource", "spec.nameService", + nameServiceSts.Spec, "nameServiceFound.spec", nameServiceFound.Spec) + return reconcile.Result{}, err + } + } + } + + // Check if broker already exists, if not create a new one + brokerFound := &rocketmqv1alpha1.Broker{} + brokerSts := r.brokerForRocketmq(instance) + + err = r.client.Get(context.TODO(), types.NamespacedName{Name: brokerSts.Name, Namespace: brokerSts.Namespace}, brokerFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), brokerSts) + if err != nil { + reqLogger.Error(err, "Failed to create new broker of rocketmq", "broker.namespace", + brokerSts.Namespace, "broker.Name", brokerSts.Name) + } + return reconcile.Result{Requeue: true}, nil + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq broker.") + } else { + // Resource broker will change; Only ReplicaPerGroup Size ImagePullPolicy BrokerImage can update + if !reflect.DeepEqual(brokerSts.Spec.ReplicaPerGroup, brokerFound.Spec.ReplicaPerGroup) || + !reflect.DeepEqual(brokerSts.Spec.Size, brokerFound.Spec.Size) || + !reflect.DeepEqual(brokerSts.Spec.ImagePullPolicy, brokerFound.Spec.ImagePullPolicy) || + !reflect.DeepEqual(brokerSts.Spec.BrokerImage, brokerFound.Spec.BrokerImage) { + brokerFound.Spec.ReplicaPerGroup = brokerSts.Spec.ReplicaPerGroup + brokerFound.Spec.Size = brokerSts.Spec.Size + brokerFound.Spec.ImagePullPolicy = brokerSts.Spec.ImagePullPolicy + brokerFound.Spec.BrokerImage = brokerSts.Spec.BrokerImage + err = r.client.Update(context.TODO(), brokerFound) + if err != nil { + reqLogger.Error(err, "should update broker resource", "spec.Broker", + brokerSts.Spec, "brokerFound.spec", brokerFound.Spec) + return reconcile.Result{}, err + } + } + } + if instance.Spec.Console.ConsoleDeployment.Spec.Replicas != nil { + consoleFound := &rocketmqv1alpha1.Console{} Review Comment: Console update calls `r.client.Update(context.TODO(), consoleDep)` on the desired-state object (`consoleDep`) instead of the existing cluster object (`consoleFound`). This will fail because `consoleDep` lacks the server-assigned ResourceVersion, and even if it succeeded it would overwrite the existing resource with a stale version. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { + nameServiceFound.Spec.ImagePullPolicy = nameServiceSts.Spec.ImagePullPolicy + nameServiceFound.Spec.NameServiceImage = nameServiceSts.Spec.NameServiceImage + nameServiceFound.Spec.Size = nameServiceSts.Spec.Size + err = r.client.Update(context.TODO(), nameServiceFound) + if err != nil { + reqLogger.Error(err, "should update nameservice resource", "spec.nameService", + nameServiceSts.Spec, "nameServiceFound.spec", nameServiceFound.Spec) + return reconcile.Result{}, err + } + } + } + + // Check if broker already exists, if not create a new one + brokerFound := &rocketmqv1alpha1.Broker{} + brokerSts := r.brokerForRocketmq(instance) + + err = r.client.Get(context.TODO(), types.NamespacedName{Name: brokerSts.Name, Namespace: brokerSts.Namespace}, brokerFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), brokerSts) + if err != nil { + reqLogger.Error(err, "Failed to create new broker of rocketmq", "broker.namespace", + brokerSts.Namespace, "broker.Name", brokerSts.Name) + } + return reconcile.Result{Requeue: true}, nil + } else if err != nil { Review Comment: Broker creation error is logged but not returned. The function returns `reconcile.Result{Requeue: true}, nil` which discards the error, defeating the controller-runtime's exponential backoff retry mechanism. The error should be returned so the framework can properly retry with backoff. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { + nameServiceFound.Spec.ImagePullPolicy = nameServiceSts.Spec.ImagePullPolicy + nameServiceFound.Spec.NameServiceImage = nameServiceSts.Spec.NameServiceImage + nameServiceFound.Spec.Size = nameServiceSts.Spec.Size + err = r.client.Update(context.TODO(), nameServiceFound) + if err != nil { + reqLogger.Error(err, "should update nameservice resource", "spec.nameService", + nameServiceSts.Spec, "nameServiceFound.spec", nameServiceFound.Spec) + return reconcile.Result{}, err + } + } + } + + // Check if broker already exists, if not create a new one + brokerFound := &rocketmqv1alpha1.Broker{} + brokerSts := r.brokerForRocketmq(instance) + + err = r.client.Get(context.TODO(), types.NamespacedName{Name: brokerSts.Name, Namespace: brokerSts.Namespace}, brokerFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), brokerSts) + if err != nil { + reqLogger.Error(err, "Failed to create new broker of rocketmq", "broker.namespace", + brokerSts.Namespace, "broker.Name", brokerSts.Name) + } + return reconcile.Result{Requeue: true}, nil + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq broker.") + } else { + // Resource broker will change; Only ReplicaPerGroup Size ImagePullPolicy BrokerImage can update + if !reflect.DeepEqual(brokerSts.Spec.ReplicaPerGroup, brokerFound.Spec.ReplicaPerGroup) || + !reflect.DeepEqual(brokerSts.Spec.Size, brokerFound.Spec.Size) || + !reflect.DeepEqual(brokerSts.Spec.ImagePullPolicy, brokerFound.Spec.ImagePullPolicy) || + !reflect.DeepEqual(brokerSts.Spec.BrokerImage, brokerFound.Spec.BrokerImage) { + brokerFound.Spec.ReplicaPerGroup = brokerSts.Spec.ReplicaPerGroup + brokerFound.Spec.Size = brokerSts.Spec.Size + brokerFound.Spec.ImagePullPolicy = brokerSts.Spec.ImagePullPolicy + brokerFound.Spec.BrokerImage = brokerSts.Spec.BrokerImage + err = r.client.Update(context.TODO(), brokerFound) + if err != nil { + reqLogger.Error(err, "should update broker resource", "spec.Broker", + brokerSts.Spec, "brokerFound.spec", brokerFound.Spec) + return reconcile.Result{}, err + } + } + } + if instance.Spec.Console.ConsoleDeployment.Spec.Replicas != nil { + consoleFound := &rocketmqv1alpha1.Console{} + consoleDep := r.consoleForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: consoleDep.Name, Namespace: consoleDep.Namespace}, consoleFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), consoleDep) + if err != nil { + reqLogger.Error(err, "Failed to create new console of rocketmq") + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq console.") + } else { + if !reflect.DeepEqual(consoleDep.Spec.ConsoleDeployment.Spec.Replicas, consoleFound.Spec.ConsoleDeployment.Spec.Replicas) { + err = r.client.Update(context.TODO(), consoleDep) + if err != nil { + reqLogger.Error(err, "should update console resource") + } + } Review Comment: Status update at the end of Reconcile uses `brokerFound.ObjectMeta.Name` and `nameServiceFound.ObjectMeta.Name`, but if either resource creation failed (non-NotFound error path), these objects are still zero-valued structs. This will set status fields to empty strings and may mask real failures. ########## pkg/controller/console/console_controller.go: ########## @@ -211,6 +220,6 @@ func newDeploymentForCR(cr *rocketmqv1alpha1.Console) *appsv1.Deployment { }, }, } - + controllerutil.SetControllerReference(cr, dep, r.scheme) Review Comment: `controllerutil.SetControllerReference` is called here inside `newDeploymentForCR`, and is also called by the caller at line ~145 of the same file (`SetControllerReference(instance, consoleDeployment, r.scheme)`). Setting the controller reference twice is redundant and may produce warnings or errors if the owner reference already exists. ########## pkg/apis/rocketmq/v1alpha1/broker_types.go: ########## @@ -33,7 +33,7 @@ type BrokerSpec struct { // Add custom validation using kubebuilder tags: https://book-v1.book.kubebuilder.io/beyond_basics/generating_crd.html 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"` + NameServers string `json:"nameServers"` Review Comment: Removing `omitempty` from the `nameServers` JSON tag (`json:"nameServers,omitempty"` -> `json:"nameServers"`) is a breaking API change. Existing Broker resources serialized without `nameServers` will now fail deserialization or validation. This breaks backward compatibility for users of the standalone Broker CRD. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { Review Comment: No unit or integration tests are included for the new Rocketmq controller, which orchestrates three sub-resources (NameService, Broker, Console) with non-trivial reconciliation logic including conditional creation, selective field updates, and status management. This is a significant gap in test coverage for a new controller. ########## deploy/crds/rocketmq_v1alpha1_rocketmq_crd.yaml: ########## @@ -0,0 +1,65 @@ +apiVersion: apiextensions.k8s.io/v1beta1 +kind: CustomResourceDefinition +metadata: + name: rocketmqs.rocketmq.apache.org +spec: + group: rocketmq.apache.org + names: + kind: Rocketmq + listKind: RocketmqList + plural: rocketmqs + singular: rocketmq + scope: Namespaced + subresources: + status: {} + validation: + openAPIV3Schema: + properties: + apiVersion: + description: "APIVersion defines the versioned schema of this representation + of an object. Servers should convert recognized schemas to the latest + internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/api-conventions.md#resources"; + type: string + kind: + description: 'Kind is a string value representing the REST resource this + object represents. Servers may infer this from the endpoint the client + submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/api-conventions.md#types-kinds' + type: string + metadata: + type: object + spec: + properties: + broker: + description: Broker defines rocketmq broker spec info Review Comment: The CRD validation schema defines `broker`, `nameService`, and `console` as `type: object` with no sub-properties. This provides zero validation for nested fields, meaning any malformed spec will be accepted by the API server. At minimum, required sub-fields should be enumerated, or the full nested schemas from the Broker/NameService/Console CRDs should be inlined. ########## example/rocketmq_v1alpha1_cluster_service.yaml: ########## @@ -22,6 +22,7 @@ metadata: spec: type: NodePort selector: + name_service_cr: ${rocketmq-name}-name-service Review Comment: The selector `name_service_cr: ${rocketmq-name}-name-service` uses a shell variable placeholder `${rocketmq-name}` that is not substituted by kubectl. Users must manually replace this value, but this is not documented and will silently result in a Service with no matching pods if applied as-is. ########## pkg/controller/rocketmq/rocketmq_controller.go: ########## @@ -0,0 +1,258 @@ +/* + * 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. + */ + +// Package rocketmq contains the implementation of the Rocketmq CRD reconcile function +package rocketmq + +import ( + "context" + rocketmqv1alpha1 "github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1" + cons "github.com/apache/rocketmq-operator/pkg/constants" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "reflect" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" + "sigs.k8s.io/controller-runtime/pkg/source" + "time" +) + +var log = logf.Log.WithName("controller_rocketmq") + +// Add creates a new Rocketmq Controller and adds it to the Manager. The Manager will set fields on the Controller +// and Start it when the Manager is Started. +func Add(mgr manager.Manager) error { + return add(mgr, newReconciler(mgr)) +} + +// newReconciler returns a new reconcile.Reconciler +func newReconciler(mgr manager.Manager) reconcile.Reconciler { + return &ReconcileRocketmq{client: mgr.GetClient(), scheme: mgr.GetScheme()} +} + +// add adds a new Controller to mgr with r as the reconcile.Reconciler +func add(mgr manager.Manager, r reconcile.Reconciler) error { + c, err := controller.New("rocketmq-controller", mgr, controller.Options{Reconciler: r}) + if err != nil { + return err + } + // Watch for changes to primary resource Rocketmq + err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Rocketmq{}}, &handler.EnqueueRequestForObject{}) + if err != nil { + return err + } + // Watch for changes to secondary resource Pods and requeue the owner Rocketmq + err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{ + IsController: true, + OwnerType: &rocketmqv1alpha1.Rocketmq{}, + }) + if err != nil { + return err + } + return nil +} + +// blank assignment to verify that ReconcileRocketmq implements reconcile.Reconciler +var _ reconcile.Reconciler = &ReconcileRocketmq{} + +// ReconcileRocketmq reconciles a Rocketmq object +type ReconcileRocketmq struct { + // This client, initialized using mgr.Client() above, is a split client + // that reads objects from the cache and writes to the apiserver + client client.Client + scheme *runtime.Scheme +} + +// Reconcile reads that state of the cluster for a NameService object and makes changes based on the state read +// and what is in the NameService.Spec +// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates +// a Pod as an example +// Note: +// The Controller will requeue the Request to be processed again if the returned error is non-nil or +// Result.Requeue is true, otherwise upon completion it will remove the work from the queue. +func (r *ReconcileRocketmq) Reconcile(request reconcile.Request) (reconcile.Result, error) { + reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name) + reqLogger.Info("Reconciling Rocketmq") + // Fetch the Rocketmq instance + instance := &rocketmqv1alpha1.Rocketmq{} + err := r.client.Get(context.TODO(), request.NamespacedName, instance) + if err != nil { + if errors.IsNotFound(err) { + // Request object not found, could have been deleted after reconcile request. + // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers. + // Return and don't requeue + return reconcile.Result{}, nil + } + // Error reading the object, requeue the request + return reconcile.Result{}, err + } + + // Check if nameserver already exist, if not create a new one + nameServiceFound := &rocketmqv1alpha1.NameService{} + nameServiceSts := r.nameServiceForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: nameServiceSts.Name, Namespace: nameServiceSts.Namespace}, nameServiceFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), nameServiceSts) + if err != nil { + reqLogger.Error(err, "Failed to create new nameService of rocketmq", "nameservice.namespace", + nameServiceSts.Namespace, "nameservice.Name", nameServiceSts.Name) + return reconcile.Result{}, err + } + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq nameservice.") + return reconcile.Result{}, err + } else { + // Resource NameService will change; Only size nameServiceImage imagePullPolicy can update + if !reflect.DeepEqual(nameServiceSts.Spec.Size, nameServiceFound.Spec.Size) || + !reflect.DeepEqual(nameServiceSts.Spec.NameServiceImage, nameServiceFound.Spec.NameServiceImage) || + !reflect.DeepEqual(nameServiceSts.Spec.ImagePullPolicy, nameServiceFound.Spec.ImagePullPolicy) { + nameServiceFound.Spec.ImagePullPolicy = nameServiceSts.Spec.ImagePullPolicy + nameServiceFound.Spec.NameServiceImage = nameServiceSts.Spec.NameServiceImage + nameServiceFound.Spec.Size = nameServiceSts.Spec.Size + err = r.client.Update(context.TODO(), nameServiceFound) + if err != nil { + reqLogger.Error(err, "should update nameservice resource", "spec.nameService", + nameServiceSts.Spec, "nameServiceFound.spec", nameServiceFound.Spec) + return reconcile.Result{}, err + } + } + } + + // Check if broker already exists, if not create a new one + brokerFound := &rocketmqv1alpha1.Broker{} + brokerSts := r.brokerForRocketmq(instance) + + err = r.client.Get(context.TODO(), types.NamespacedName{Name: brokerSts.Name, Namespace: brokerSts.Namespace}, brokerFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), brokerSts) + if err != nil { + reqLogger.Error(err, "Failed to create new broker of rocketmq", "broker.namespace", + brokerSts.Namespace, "broker.Name", brokerSts.Name) + } + return reconcile.Result{Requeue: true}, nil + } else if err != nil { + reqLogger.Error(err, "Failed to get rocketmq broker.") + } else { + // Resource broker will change; Only ReplicaPerGroup Size ImagePullPolicy BrokerImage can update + if !reflect.DeepEqual(brokerSts.Spec.ReplicaPerGroup, brokerFound.Spec.ReplicaPerGroup) || + !reflect.DeepEqual(brokerSts.Spec.Size, brokerFound.Spec.Size) || + !reflect.DeepEqual(brokerSts.Spec.ImagePullPolicy, brokerFound.Spec.ImagePullPolicy) || + !reflect.DeepEqual(brokerSts.Spec.BrokerImage, brokerFound.Spec.BrokerImage) { + brokerFound.Spec.ReplicaPerGroup = brokerSts.Spec.ReplicaPerGroup + brokerFound.Spec.Size = brokerSts.Spec.Size + brokerFound.Spec.ImagePullPolicy = brokerSts.Spec.ImagePullPolicy + brokerFound.Spec.BrokerImage = brokerSts.Spec.BrokerImage + err = r.client.Update(context.TODO(), brokerFound) + if err != nil { + reqLogger.Error(err, "should update broker resource", "spec.Broker", + brokerSts.Spec, "brokerFound.spec", brokerFound.Spec) + return reconcile.Result{}, err + } + } + } + if instance.Spec.Console.ConsoleDeployment.Spec.Replicas != nil { + consoleFound := &rocketmqv1alpha1.Console{} + consoleDep := r.consoleForRocketmq(instance) + err = r.client.Get(context.TODO(), types.NamespacedName{Name: consoleDep.Name, Namespace: consoleDep.Namespace}, consoleFound) + if err != nil && errors.IsNotFound(err) { + err = r.client.Create(context.TODO(), consoleDep) + if err != nil { Review Comment: Console creation error is logged but not returned (`reqLogger.Error(err, ...)` without `return reconcile.Result{}, err`). This means console creation failures are silently swallowed and won't trigger proper retry with backoff. -- 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]
