RockteMQ-AI commented on code in PR #63: URL: https://github.com/apache/rocketmq-operator/pull/63#discussion_r3909311477
########## pkg/apis/rocketmq/v1alpha1/rocketmq_types.go: ########## @@ -0,0 +1,70 @@ +/* + * 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 v1alpha1 + +import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + +// EDIT THIS FILE! THIS IS SCAFFOLDING FOR YOU TO OWN! +// NOTE: json tags are required. Any new fields you add must have json tags for the fields to be serialized. + +// RocketmqSpec defines the desired state of Rocketmq +// +k8s:openapi-gen=true +// +kubebuilder:subresource:status +type RocketmqSpec struct { + // Broker defines broker info + Broker BrokerSpec `json:"broker"` + // NameService defines name Service info + NameService NameServiceSpec `json:"nameService"` + // Console defines console info + Console ConsoleSpec `json:"console,omitempty"` +} + +// RocketmqStatus defines the observed state of Rocketmq +// +k8s:openapi-gen=true +type RocketmqStatus struct { + // INSERT ADDITIONAL STATUS FIELD - define observed state of cluster + // Important: Run "operator-sdk generate k8s" to regenerate code after modifying this file + // Add custom validation using kubebuilder tags: https://book-v1.book.kubebuilder.io/beyond_basics/generating_crd.html + Broker string `json:"broker"` + NameService string `json:"nameService"` +} + +// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object + +// Rocketmq is the Schema for the Rocketmq API +// +k8s:openapi-gen=true +type Rocketmq struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + Spec RocketmqSpec `json:"spec,omitempty"` + Status RocketmqStatus `json:"status,omitempty"` +} + +// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object + +// RocketmqList contains a list of Rocketmq +type RocketmqList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []Rocketmq `json:"items"` +} + +func init() { + SchemeBuilder.Register(&Rocketmq{}, &RocketmqList{}) Review Comment: Rocketmq and RocketmqList have no generated DeepCopyObject methods. SchemeBuilder.Register requires runtime.Object, so this package will not compile until deepcopy generation output is added. ########## 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 { Review Comment: consoleDep is a newly constructed object with no ResourceVersion, so Update will be rejected for an existing Console CR. Update consoleFound after copying the desired fields instead. ########## install-operator.sh: ########## @@ -19,6 +19,7 @@ kubectl create -f deploy/crds/rocketmq_v1alpha1_broker_crd.yaml kubectl create -f deploy/crds/rocketmq_v1alpha1_nameservice_crd.yaml kubectl create -f deploy/crds/rocketmq_v1alpha1_consoles_crd.yaml kubectl create -f deploy/crds/rocketmq_v1alpha1_topictransfer_crd.yaml +kubectl create -f deploy/crds/rocketmq_v1alpha1_rocketmq_crd.yaml kubectl create -f deploy/service_account.yaml Review Comment: The new CRD is installed without extending deploy/role.yaml for rocketmqs and rocketmqs/status. The operator will be forbidden from listing/watching Rocketmq resources and updating their 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) { + 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.") Review Comment: Broker reconciliation only propagates size, replica count, image, and pull policy. Changes to nameServers, resources, env, volumes, allowRestart, or scalePodName are silently ignored, preventing configuration updates. ########## 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 Service now requires name_service_cr=${rocketmq-name}-name-service, but generated Console pods only receive app and console-cr labels. The Service will have no endpoints; kubectl also does not expand this placeholder. ########## 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 does not require broker or nameService and provides no nested schema. Invalid or misspelled cluster configuration is accepted and reconciled into empty child CRs instead of being rejected at admission. ########## 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) Review Comment: The Broker CR is created immediately after the NameService CR is created; no readiness/status check is made. This violates the required NameServer-before-Broker ordering and starts brokers before discovery is available. -- 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]
