RockteMQ-AI commented on code in PR #134:
URL: https://github.com/apache/rocketmq-operator/pull/134#discussion_r3839452578
##########
Makefile:
##########
@@ -126,13 +126,15 @@ endif
install: manifests kustomize ## Install CRDs into the K8s cluster specified in
~/.kube/config.
kubectl create -f deploy/crds/rocketmq.apache.org_brokers.yaml
kubectl create -f deploy/crds/rocketmq.apache.org_nameservices.yaml
+ kubectl create -f deploy/crds/rocketmq.apache.proxys.yaml
kubectl create -f deploy/crds/rocketmq.apache.org_consoles.yaml
kubectl create -f deploy/crds/rocketmq.apache.org_topictransfers.yaml
.PHONY: uninstall
uninstall: manifests kustomize ## Uninstall CRDs from the K8s cluster
specified in ~/.kube/config. Call with ignore-not-found=true to ignore resource
not found errors during deletion.
kubectl delete --ignore-not-found=$(ignore-not-found) -f
deploy/crds/rocketmq.apache.org_brokers.yaml
kubectl delete --ignore-not-found=$(ignore-not-found) -f
deploy/crds/rocketmq.apache.org_nameservices.yaml
+ kubectl delete --ignore-not-found=$(ignore-not-found) -f
deploy/crds/rocketmq.apache.org_proxys.yaml
Review Comment:
The added uninstall command is indented with spaces instead of a tab. Make
requires recipe lines to begin with a tab, so `make uninstall` will fail with a
`missing separator` error.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
Review Comment:
The RBAC marker uses `resources=Deployments`. Kubernetes RBAC resource names
should be lowercase plural (`deployments`); `Deployments` is non-idiomatic and
may not be accepted by controller-gen or the API server.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
Review Comment:
The controller watches `corev1.Pod` as a secondary resource, but there is no
corresponding RBAC marker granting `get/list/watch` on pods. The controller
will fail to start the watch with a Forbidden error unless the role is manually
fixed.
##########
Makefile:
##########
@@ -126,13 +126,15 @@ endif
install: manifests kustomize ## Install CRDs into the K8s cluster specified in
~/.kube/config.
kubectl create -f deploy/crds/rocketmq.apache.org_brokers.yaml
kubectl create -f deploy/crds/rocketmq.apache.org_nameservices.yaml
+ kubectl create -f deploy/crds/rocketmq.apache.proxys.yaml
Review Comment:
The install target references `deploy/crds/rocketmq.apache.proxys.yaml`,
which is missing `.org_`. The actual generated file is
`rocketmq.apache.org_proxys.yaml`, so the CRD will not be installed and the
operator cannot reconcile Proxy resources.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
+ reqLogger.Error(err, "The value of proxyConfigPath and
proxyMode must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ if instance.Spec.BrokerConfigPath == "" && instance.Spec.ProxyMode ==
"LOCAL" {
Review Comment:
Same as above: `reqLogger.Error(err, ...)` is called when `err` is `nil`
during the LOCAL-mode brokerConfigPath validation.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
+ reqLogger.Error(err, "The value of proxyConfigPath and
proxyMode must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ if instance.Spec.BrokerConfigPath == "" && instance.Spec.ProxyMode ==
"LOCAL" {
+ reqLogger.Error(err, "ProxyMode is LOCAL, the value of
brokerConfigPath must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ proxyDeployment := newDeploymentForCR(instance)
+
+ // Set Proxy instance as the owner and controller
+ if err := controllerutil.SetControllerReference(instance,
proxyDeployment, r.scheme); err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Check if this Pod already exists
+ found := &appsv1.Deployment{}
+ err = r.client.Get(context.TODO(), types.NamespacedName{Name:
proxyDeployment.Name, Namespace: proxyDeployment.Namespace}, found)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating RocketMQ Proxy Deployment",
"Namespace", proxyDeployment, "Name", proxyDeployment.Name)
+ err = r.client.Create(context.TODO(), proxyDeployment)
+ if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // created successfully - don't requeue
+ return reconcile.Result{}, nil
+ } else if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Support proxy Deployment scaling
+ if !reflect.DeepEqual(instance.Spec.ProxyDeployment.Spec.Replicas,
found.Spec.Replicas) {
+ found.Spec.Replicas =
instance.Spec.ProxyDeployment.Spec.Replicas
+ err = r.client.Update(context.TODO(), found)
+ if err != nil {
+ reqLogger.Error(err, "Failed to update proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ } else {
+ reqLogger.Info("Successfully updated proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ }
+ }
+ // CR already exists - don't requeue
+ reqLogger.Info("Skip reconcile: RocketMQ Proxy Deployment already
exists", "Namespace", found.Namespace, "Name", found.Name)
+ return reconcile.Result{}, nil
+}
+
+// newDeploymentForCR returns a Deployment pod with modifying the ENV
+func newDeploymentForCR(cr *rocketmqv1alpha1.Proxy) *appsv1.Deployment {
+ dep := &appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: cr.Name,
+ Namespace: cr.Namespace,
+ },
+ Spec: appsv1.DeploymentSpec{
+ Replicas: cr.Spec.ProxyDeployment.Spec.Replicas,
+ Selector: &metav1.LabelSelector{
+ MatchLabels:
cr.Spec.ProxyDeployment.Spec.Selector.MatchLabels,
Review Comment:
`newDeploymentForCR` dereferences
`cr.Spec.ProxyDeployment.Spec.Selector.MatchLabels` without checking that
`Selector` is non-nil. A Proxy CR that omits the selector will cause a panic.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
+ reqLogger.Error(err, "The value of proxyConfigPath and
proxyMode must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ if instance.Spec.BrokerConfigPath == "" && instance.Spec.ProxyMode ==
"LOCAL" {
+ reqLogger.Error(err, "ProxyMode is LOCAL, the value of
brokerConfigPath must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ proxyDeployment := newDeploymentForCR(instance)
+
+ // Set Proxy instance as the owner and controller
+ if err := controllerutil.SetControllerReference(instance,
proxyDeployment, r.scheme); err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Check if this Pod already exists
+ found := &appsv1.Deployment{}
+ err = r.client.Get(context.TODO(), types.NamespacedName{Name:
proxyDeployment.Name, Namespace: proxyDeployment.Namespace}, found)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating RocketMQ Proxy Deployment",
"Namespace", proxyDeployment, "Name", proxyDeployment.Name)
+ err = r.client.Create(context.TODO(), proxyDeployment)
+ if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // created successfully - don't requeue
+ return reconcile.Result{}, nil
+ } else if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Support proxy Deployment scaling
+ if !reflect.DeepEqual(instance.Spec.ProxyDeployment.Spec.Replicas,
found.Spec.Replicas) {
+ found.Spec.Replicas =
instance.Spec.ProxyDeployment.Spec.Replicas
+ err = r.client.Update(context.TODO(), found)
+ if err != nil {
+ reqLogger.Error(err, "Failed to update proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ } else {
+ reqLogger.Info("Successfully updated proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ }
+ }
+ // CR already exists - don't requeue
+ reqLogger.Info("Skip reconcile: RocketMQ Proxy Deployment already
exists", "Namespace", found.Namespace, "Name", found.Name)
+ return reconcile.Result{}, nil
+}
+
+// newDeploymentForCR returns a Deployment pod with modifying the ENV
+func newDeploymentForCR(cr *rocketmqv1alpha1.Proxy) *appsv1.Deployment {
+ dep := &appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: cr.Name,
+ Namespace: cr.Namespace,
+ },
+ Spec: appsv1.DeploymentSpec{
+ Replicas: cr.Spec.ProxyDeployment.Spec.Replicas,
+ Selector: &metav1.LabelSelector{
+ MatchLabels:
cr.Spec.ProxyDeployment.Spec.Selector.MatchLabels,
+ },
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{
+ Labels:
cr.Spec.ProxyDeployment.Spec.Template.ObjectMeta.Labels,
Review Comment:
`newDeploymentForCR` reads
`cr.Spec.ProxyDeployment.Spec.Template.ObjectMeta.Labels` without defaulting;
if omitted the resulting Deployment will have no pod labels and will not match
the selector.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
Review Comment:
Validation logs `reqLogger.Error(err, ...)` but `err` is `nil` here (the
previous Get succeeded). The logged error will have no underlying cause.
Consider logging an Info message or constructing a real error.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ err := r.client.Get(context.TODO(), request.NamespacedName, instance)
Review Comment:
Reconcile receives a `ctx context.Context` parameter but calls
`r.client.Get(context.TODO(), ...)`. It should use the supplied `ctx` so
cancellation and deadlines are respected.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
+ reqLogger.Error(err, "The value of proxyConfigPath and
proxyMode must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ if instance.Spec.BrokerConfigPath == "" && instance.Spec.ProxyMode ==
"LOCAL" {
+ reqLogger.Error(err, "ProxyMode is LOCAL, the value of
brokerConfigPath must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ proxyDeployment := newDeploymentForCR(instance)
+
+ // Set Proxy instance as the owner and controller
+ if err := controllerutil.SetControllerReference(instance,
proxyDeployment, r.scheme); err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Check if this Pod already exists
+ found := &appsv1.Deployment{}
+ err = r.client.Get(context.TODO(), types.NamespacedName{Name:
proxyDeployment.Name, Namespace: proxyDeployment.Namespace}, found)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating RocketMQ Proxy Deployment",
"Namespace", proxyDeployment, "Name", proxyDeployment.Name)
+ err = r.client.Create(context.TODO(), proxyDeployment)
+ if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // created successfully - don't requeue
+ return reconcile.Result{}, nil
+ } else if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Support proxy Deployment scaling
+ if !reflect.DeepEqual(instance.Spec.ProxyDeployment.Spec.Replicas,
found.Spec.Replicas) {
+ found.Spec.Replicas =
instance.Spec.ProxyDeployment.Spec.Replicas
+ err = r.client.Update(context.TODO(), found)
+ if err != nil {
+ reqLogger.Error(err, "Failed to update proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ } else {
+ reqLogger.Info("Successfully updated proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ }
+ }
+ // CR already exists - don't requeue
+ reqLogger.Info("Skip reconcile: RocketMQ Proxy Deployment already
exists", "Namespace", found.Namespace, "Name", found.Name)
+ return reconcile.Result{}, nil
+}
+
+// newDeploymentForCR returns a Deployment pod with modifying the ENV
+func newDeploymentForCR(cr *rocketmqv1alpha1.Proxy) *appsv1.Deployment {
+ dep := &appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: cr.Name,
+ Namespace: cr.Namespace,
+ },
+ Spec: appsv1.DeploymentSpec{
+ Replicas: cr.Spec.ProxyDeployment.Spec.Replicas,
+ Selector: &metav1.LabelSelector{
+ MatchLabels:
cr.Spec.ProxyDeployment.Spec.Selector.MatchLabels,
+ },
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{
+ Labels:
cr.Spec.ProxyDeployment.Spec.Template.ObjectMeta.Labels,
+ },
+ Spec: corev1.PodSpec{
+ ServiceAccountName:
cr.Spec.ProxyDeployment.Spec.Template.Spec.ServiceAccountName,
+ Affinity:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Affinity,
+ ImagePullSecrets:
cr.Spec.ProxyDeployment.Spec.Template.Spec.ImagePullSecrets,
+ Containers: []corev1.Container{{
+ Resources:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].Resources,
+ Image:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].Image,
+ Name:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].Name,
+ ImagePullPolicy:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].ImagePullPolicy,
+ Ports:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].Ports,
+ Env: getENV(cr),
+ VolumeMounts:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].VolumeMounts,
+ SecurityContext:
getContainerSecurityContext(cr),
+ }},
+ Volumes:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Volumes,
+ SecurityContext:
getPodSecurityContext(cr),
+ },
+ },
+ },
+ }
+
+ return dep
+}
+
+func getPodSecurityContext(proxy *rocketmqv1alpha1.Proxy)
*corev1.PodSecurityContext {
+ var securityContext = corev1.PodSecurityContext{}
+ if proxy.Spec.ProxyDeployment.Spec.Template.Spec.SecurityContext != nil
{
+ securityContext =
*proxy.Spec.ProxyDeployment.Spec.Template.Spec.SecurityContext
+ }
+ return &securityContext
+}
+
+func getContainerSecurityContext(proxy *rocketmqv1alpha1.Proxy)
*corev1.SecurityContext {
+ var securityContext = corev1.SecurityContext{}
+ if
proxy.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].SecurityContext !=
nil {
Review Comment:
`getContainerSecurityContext` also dereferences `Containers[0]` without a
length check, leading to a panic when the container list is empty.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
+ reqLogger.Error(err, "The value of proxyConfigPath and
proxyMode must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ if instance.Spec.BrokerConfigPath == "" && instance.Spec.ProxyMode ==
"LOCAL" {
+ reqLogger.Error(err, "ProxyMode is LOCAL, the value of
brokerConfigPath must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ proxyDeployment := newDeploymentForCR(instance)
+
+ // Set Proxy instance as the owner and controller
+ if err := controllerutil.SetControllerReference(instance,
proxyDeployment, r.scheme); err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Check if this Pod already exists
+ found := &appsv1.Deployment{}
+ err = r.client.Get(context.TODO(), types.NamespacedName{Name:
proxyDeployment.Name, Namespace: proxyDeployment.Namespace}, found)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating RocketMQ Proxy Deployment",
"Namespace", proxyDeployment, "Name", proxyDeployment.Name)
+ err = r.client.Create(context.TODO(), proxyDeployment)
+ if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // created successfully - don't requeue
+ return reconcile.Result{}, nil
+ } else if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Support proxy Deployment scaling
+ if !reflect.DeepEqual(instance.Spec.ProxyDeployment.Spec.Replicas,
found.Spec.Replicas) {
+ found.Spec.Replicas =
instance.Spec.ProxyDeployment.Spec.Replicas
+ err = r.client.Update(context.TODO(), found)
+ if err != nil {
+ reqLogger.Error(err, "Failed to update proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ } else {
+ reqLogger.Info("Successfully updated proxy CR ",
"Namespace", found.Namespace, "Name", found.Name)
+ }
+ }
+ // CR already exists - don't requeue
+ reqLogger.Info("Skip reconcile: RocketMQ Proxy Deployment already
exists", "Namespace", found.Namespace, "Name", found.Name)
+ return reconcile.Result{}, nil
+}
+
+// newDeploymentForCR returns a Deployment pod with modifying the ENV
+func newDeploymentForCR(cr *rocketmqv1alpha1.Proxy) *appsv1.Deployment {
+ dep := &appsv1.Deployment{
+ ObjectMeta: metav1.ObjectMeta{
+ Name: cr.Name,
+ Namespace: cr.Namespace,
+ },
+ Spec: appsv1.DeploymentSpec{
+ Replicas: cr.Spec.ProxyDeployment.Spec.Replicas,
+ Selector: &metav1.LabelSelector{
+ MatchLabels:
cr.Spec.ProxyDeployment.Spec.Selector.MatchLabels,
+ },
+ Template: corev1.PodTemplateSpec{
+ ObjectMeta: metav1.ObjectMeta{
+ Labels:
cr.Spec.ProxyDeployment.Spec.Template.ObjectMeta.Labels,
+ },
+ Spec: corev1.PodSpec{
+ ServiceAccountName:
cr.Spec.ProxyDeployment.Spec.Template.Spec.ServiceAccountName,
+ Affinity:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Affinity,
+ ImagePullSecrets:
cr.Spec.ProxyDeployment.Spec.Template.Spec.ImagePullSecrets,
+ Containers: []corev1.Container{{
+ Resources:
cr.Spec.ProxyDeployment.Spec.Template.Spec.Containers[0].Resources,
Review Comment:
`newDeploymentForCR` accesses `Spec.Template.Spec.Containers[0]` without
verifying that at least one container is provided. Omitting containers causes
an index-out-of-range panic.
##########
images/proxy/alpine/proxyStart.sh:
##########
@@ -0,0 +1,21 @@
+#!/bin/bash
+
+# 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.
+if [ $PROXY_MODE == "LOCAL" ]; then
Review Comment:
When `PROXY_MODE` is `LOCAL`, the script runs `./mqproxy -bc ... -pm LOCAL`
and then falls through to unconditionally run `./mqproxy -pc ... -pm ...` on
line 20. The second invocation lacks the broker config and will either fail or
create an unwanted cluster-mode process. An `else` branch is needed.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
Review Comment:
No unit tests or integration tests are added for the new Proxy type,
controller, or Deployment reconciliation logic. The PR introduces a new CRD and
controller that should have test coverage.
##########
pkg/controller/proxy/proxy_controller.go:
##########
@@ -0,0 +1,236 @@
+/*
+ * 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 proxy
+
+import (
+ "context"
+ rocketmqv1alpha1
"github.com/apache/rocketmq-operator/pkg/apis/rocketmq/v1alpha1"
+ cons "github.com/apache/rocketmq-operator/pkg/constants"
+ appsv1 "k8s.io/api/apps/v1"
+ 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"
+ logf "sigs.k8s.io/controller-runtime/pkg/log"
+ "sigs.k8s.io/controller-runtime/pkg/manager"
+ "sigs.k8s.io/controller-runtime/pkg/reconcile"
+ "sigs.k8s.io/controller-runtime/pkg/source"
+)
+
+var log = logf.Log.WithName("controller_proxy")
+
+/**
+* USER ACTION REQUIRED: This is a scaffold file intended for the user to
modify with their own Controller
+* business logic. Delete these comments after modifying this file.*
+ */
+
+// SetupWithManager creates a new Proxy Controller and adds it to the Manager.
The Manager will set fields on the Controller
+// and Start it when the Manager is Started.
+func SetupWithManager(mgr manager.Manager) error {
+ return add(mgr, newReconciler(mgr))
+}
+
+// newReconciler returns a new reconcile.Reconciler
+func newReconciler(mgr manager.Manager) reconcile.Reconciler {
+ return &ReconcileProxy{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 {
+ // Create a new controller
+ c, err := controller.New("proxy-controller", mgr,
controller.Options{Reconciler: r})
+ if err != nil {
+ return err
+ }
+
+ // Watch for changes to primary resource Proxy
+ err = c.Watch(&source.Kind{Type: &rocketmqv1alpha1.Proxy{}},
&handler.EnqueueRequestForObject{})
+ if err != nil {
+ return err
+ }
+
+ // TODO(user): Modify this to be the types you create that are owned by
the primary resource
+ // Watch for changes to secondary resource Pods and requeue the owner
Proxy
+ err = c.Watch(&source.Kind{Type: &corev1.Pod{}},
&handler.EnqueueRequestForOwner{
+ IsController: true,
+ OwnerType: &rocketmqv1alpha1.Proxy{},
+ })
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=rocketmq.apache.org,resources=proxys/finalizers,verbs=update
+//+kubebuilder:rbac:groups="apps",resources=Deployments,verbs=get;list;watch;create;update;patch;delete
+
+// ReconcileProxy reconciles a Proxy object
+type ReconcileProxy 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 Proxy object and makes
changes based on the state read
+// and what is in the Proxy.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 *ReconcileProxy) Reconcile(ctx context.Context, request
reconcile.Request) (reconcile.Result, error) {
+ reqLogger := log.WithValues("Request.Namespace", request.Namespace,
"Request.Name", request.Name)
+ reqLogger.Info("Reconciling Proxy")
+
+ // Fetch the Proxy instance
+ instance := &rocketmqv1alpha1.Proxy{}
+ 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
+ }
+ if instance.Spec.ProxyConfigPath == "" || instance.Spec.ProxyMode == ""
{
+ reqLogger.Error(err, "The value of proxyConfigPath and
proxyMode must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ if instance.Spec.BrokerConfigPath == "" && instance.Spec.ProxyMode ==
"LOCAL" {
+ reqLogger.Error(err, "ProxyMode is LOCAL, the value of
brokerConfigPath must be not empty.")
+ return reconcile.Result{}, nil
+ }
+ proxyDeployment := newDeploymentForCR(instance)
+
+ // Set Proxy instance as the owner and controller
+ if err := controllerutil.SetControllerReference(instance,
proxyDeployment, r.scheme); err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Check if this Pod already exists
+ found := &appsv1.Deployment{}
+ err = r.client.Get(context.TODO(), types.NamespacedName{Name:
proxyDeployment.Name, Namespace: proxyDeployment.Namespace}, found)
+ if err != nil && errors.IsNotFound(err) {
+ reqLogger.Info("Creating RocketMQ Proxy Deployment",
"Namespace", proxyDeployment, "Name", proxyDeployment.Name)
+ err = r.client.Create(context.TODO(), proxyDeployment)
+ if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // created successfully - don't requeue
+ return reconcile.Result{}, nil
+ } else if err != nil {
+ return reconcile.Result{}, err
+ }
+
+ // Support proxy Deployment scaling
+ if !reflect.DeepEqual(instance.Spec.ProxyDeployment.Spec.Replicas,
found.Spec.Replicas) {
Review Comment:
After the Deployment exists, the reconcile loop only updates `Replicas`.
Changes to image, environment variables, volumes, labels, affinity, or other
template fields are ignored, so updates to the Proxy CR will not be fully
applied.
--
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]
