upgrade client-go version to 0.21.2

api/v1alpha1/flinkcluster_webhook.go

@@ -20,7 +20,7 @@ package v1alpha1
 import (
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
-	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 )
 

api/v1beta1/flinkcluster_webhook.go

@@ -20,7 +20,7 @@ package v1beta1
 import (
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
-	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 )
 

api/v1beta1/suite_test.go

@@ -48,7 +48,7 @@ func TestAPIs(t *testing.T) {
 }
 
 var _ = BeforeSuite(func(done Done) {
-	logf.SetLogger(zap.LoggerTo(GinkgoWriter, true))
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
 
 	By("bootstrapping test environment")
 	testEnv = &envtest.Environment{

controllers/batchscheduler/volcano/volcano_test.go

@@ -43,7 +43,7 @@ func TestGetClusterResource(t *testing.T) {
 				},
 			},
 			Spec: appsv1.StatefulSetSpec{
-				Replicas: &jmRep,
+				Replicas:    &jmRep,
 				ServiceName: "flinkjobcluster-sample-jobmanager",
 				Selector: &metav1.LabelSelector{
 					MatchLabels: map[string]string{
@@ -173,8 +173,8 @@ func TestGetClusterResource(t *testing.T) {
 				},
 			},
 			Spec: appsv1.StatefulSetSpec{
-				Replicas: &replicas,
-				ServiceName: "flinkjobcluster-sample-taskmanager",
+				Replicas:            &replicas,
+				ServiceName:         "flinkjobcluster-sample-taskmanager",
 				PodManagementPolicy: "Parallel",
 				Selector: &metav1.LabelSelector{
 					MatchLabels: map[string]string{

controllers/flinkcluster_controller.go

@@ -18,9 +18,10 @@ package controllers
 
 import (
 	"context"
-	"github.com/googlecloudplatform/flink-operator/controllers/history"
 	"time"
 
+	"github.com/googlecloudplatform/flink-operator/controllers/history"
+
 	"github.com/go-logr/logr"
 	v1beta1 "github.com/googlecloudplatform/flink-operator/api/v1beta1"
 	"github.com/googlecloudplatform/flink-operator/controllers/flinkclient"
@@ -64,7 +65,7 @@ type FlinkClusterReconciler struct {
 
 // Reconcile the observed state towards the desired state for a FlinkCluster custom resource.
 func (reconciler *FlinkClusterReconciler) Reconcile(
-	request ctrl.Request) (ctrl.Result, error) {
+	ctx context.Context, request ctrl.Request) (ctrl.Result, error) {
 	var log = reconciler.Log.WithValues(
 		"cluster", request.NamespacedName)
 	var handler = FlinkClusterHandler{
@@ -74,7 +75,7 @@ func (reconciler *FlinkClusterReconciler) Reconcile(
 			HTTPClient: flinkclient.HTTPClient{Log: log},
 		},
 		request:  request,
-		context:  context.Background(),
+		context:  ctx,
 		log:      log,
 		recorder: reconciler.Mgr.GetEventRecorderFor("FlinkOperator"),
 		observed: ObservedClusterState{},

controllers/flinkcluster_converter.go

@@ -68,12 +68,12 @@ func getDesiredClusterState(
 		return model.DesiredClusterState{}
 	}
 	return model.DesiredClusterState{
-		ConfigMap:    getDesiredConfigMap(cluster),
+		ConfigMap:     getDesiredConfigMap(cluster),
 		JmStatefulSet: getDesiredJobManagerStatefulSet(cluster),
-		JmService:    getDesiredJobManagerService(cluster),
-		JmIngress:    getDesiredJobManagerIngress(cluster),
+		JmService:     getDesiredJobManagerService(cluster),
+		JmIngress:     getDesiredJobManagerIngress(cluster),
 		TmStatefulSet: getDesiredTaskManagerStatefulSet(cluster),
-		Job:          getDesiredJob(observed),
+		Job:           getDesiredJob(observed),
 	}
 }
 
@@ -217,9 +217,9 @@ func getDesiredJobManagerStatefulSet(
 			Labels:          statefulSetLabels,
 		},
 		Spec: appsv1.StatefulSetSpec{
-			Replicas: jobManagerSpec.Replicas,
-			Selector: &metav1.LabelSelector{MatchLabels: podLabels},
-			ServiceName: jobManagerStatefulSetName,
+			Replicas:             jobManagerSpec.Replicas,
+			Selector:             &metav1.LabelSelector{MatchLabels: podLabels},
+			ServiceName:          jobManagerStatefulSetName,
 			VolumeClaimTemplates: jobManagerSpec.VolumeClaimTemplates,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
@@ -516,11 +516,11 @@ func getDesiredTaskManagerStatefulSet(
 			Labels: statefulSetLabels,
 		},
 		Spec: appsv1.StatefulSetSpec{
-			Replicas: &taskManagerSpec.Replicas,
-			Selector: &metav1.LabelSelector{MatchLabels: podLabels},
-			ServiceName: taskManagerStatefulSetName,
+			Replicas:             &taskManagerSpec.Replicas,
+			Selector:             &metav1.LabelSelector{MatchLabels: podLabels},
+			ServiceName:          taskManagerStatefulSetName,
 			VolumeClaimTemplates: taskManagerSpec.VolumeClaimTemplates,
-			PodManagementPolicy: "Parallel",
+			PodManagementPolicy:  "Parallel",
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels:      podLabels,

controllers/flinkcluster_converter_test.go

@@ -574,8 +574,8 @@ func TestGetDesiredClusterState(t *testing.T) {
 			},
 		},
 		Spec: appsv1.StatefulSetSpec{
-			Replicas: &replicas,
-			ServiceName: "flinkjobcluster-sample-taskmanager",
+			Replicas:            &replicas,
+			ServiceName:         "flinkjobcluster-sample-taskmanager",
 			PodManagementPolicy: "Parallel",
 			Selector: &metav1.LabelSelector{
 				MatchLabels: map[string]string{

controllers/flinkcluster_observer.go

@@ -51,10 +51,10 @@ type ObservedClusterState struct {
 	cluster           *v1beta1.FlinkCluster
 	revisions         []*appsv1.ControllerRevision
 	configMap         *corev1.ConfigMap
-	jmStatefulSet      *appsv1.StatefulSet
+	jmStatefulSet     *appsv1.StatefulSet
 	jmService         *corev1.Service
 	jmIngress         *extensionsv1beta1.Ingress
-	tmStatefulSet      *appsv1.StatefulSet
+	tmStatefulSet     *appsv1.StatefulSet
 	job               *batchv1.Job
 	jobPod            *corev1.Pod
 	flinkJobStatus    FlinkJobStatus

controllers/flinkcluster_reconciler.go

@@ -19,13 +19,13 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"reflect"
+	"time"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/retry"
-	"reflect"
-	"time"
 
 	"github.com/go-logr/logr"
 	v1beta1 "github.com/googlecloudplatform/flink-operator/api/v1beta1"
@@ -178,7 +178,7 @@ func (reconciler *ClusterReconciler) createStatefulSet(
 	return err
 }
 
-func (reconciler *ClusterReconciler) deleteOldComponent(desired runtime.Object, observed runtime.Object, component string) error {
+func (reconciler *ClusterReconciler) deleteOldComponent(desired client.Object, observed client.Object, component string) error {
 	var log = reconciler.log.WithValues("component", component)
 	if isComponentUpdated(observed, *reconciler.observed.cluster) {
 		reconciler.log.Info(fmt.Sprintf("%v is already updated, no action", component))
@@ -197,7 +197,7 @@ func (reconciler *ClusterReconciler) deleteOldComponent(desired runtime.Object,
 	return nil
 }
 
-func (reconciler *ClusterReconciler) updateComponent(desired runtime.Object, component string) error {
+func (reconciler *ClusterReconciler) updateComponent(desired client.Object, component string) error {
 	var log = reconciler.log.WithValues("component", component)
 	var context = reconciler.context
 	var k8sClient = reconciler.k8sClient

controllers/flinkcluster_updater.go

@@ -24,10 +24,11 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"k8s.io/apimachinery/pkg/types"
 	"reflect"
 	"time"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	"github.com/go-logr/logr"
 	v1beta1 "github.com/googlecloudplatform/flink-operator/api/v1beta1"
 	appsv1 "k8s.io/api/apps/v1"
@@ -933,7 +934,7 @@ func (updater *ClusterStatusUpdater) clearControlAnnotation(newControlStatus *v1
 		if err != nil {
 			return err
 		}
-		return updater.k8sClient.Patch(updater.context, updater.observed.cluster, client.ConstantPatch(types.MergePatchType, patchBytes))
+		return updater.k8sClient.Patch(updater.context, updater.observed.cluster, client.RawPatch(types.MergePatchType, patchBytes))
 	}
 
 	return nil

controllers/history/controller_history.go

@@ -21,13 +21,14 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/davecgh/go-spew/spew"
 	"hash"
 	"hash/fnv"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sort"
 	"strconv"
 
+	"github.com/davecgh/go-spew/spew"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	apps "k8s.io/api/apps/v1"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -320,14 +321,14 @@ func (rh *realHistory) AdoptControllerRevision(parent metav1.Object, parentKind
 		return nil, err
 	}
 	// Use strategic merge patch to add an owner reference indicating a controller ref
-	err = rh.Patch(rh.context, revision, client.ConstantPatch(types.StrategicMergePatchType, patchBytes))
+	err = rh.Patch(rh.context, revision, client.RawPatch(types.StrategicMergePatchType, patchBytes))
 
 	return revision, err
 }
 
 func (rh *realHistory) ReleaseControllerRevision(parent metav1.Object, revision *apps.ControllerRevision) (*apps.ControllerRevision, error) {
 	// Use strategic merge patch to add an owner reference indicating a controller ref
-	err := rh.Patch(rh.context, revision, client.ConstantPatch(types.StrategicMergePatchType, []byte(fmt.Sprintf(`{"metadata":{"ownerReferences":[{"$patch":"delete","uid":"%s"}],"uid":"%s"}}`, parent.GetUID(), revision.UID))))
+	err := rh.Patch(rh.context, revision, client.RawPatch(types.StrategicMergePatchType, []byte(fmt.Sprintf(`{"metadata":{"ownerReferences":[{"$patch":"delete","uid":"%s"}],"uid":"%s"}}`, parent.GetUID(), revision.UID))))
 
 	if err != nil {
 		if errors.IsNotFound(err) {

controllers/model/model.go

@@ -23,9 +23,9 @@ import (
 // DesiredClusterState holds desired state of a cluster.
 type DesiredClusterState struct {
 	JmStatefulSet *appsv1.StatefulSet
-	JmService    *corev1.Service
-	JmIngress    *extensionsv1beta1.Ingress
+	JmService     *corev1.Service
+	JmIngress     *extensionsv1beta1.Ingress
 	TmStatefulSet *appsv1.StatefulSet
-	ConfigMap    *corev1.ConfigMap
-	Job          *batchv1.Job
+	ConfigMap     *corev1.ConfigMap
+	Job           *batchv1.Job
 }

controllers/suite_test.go

@@ -50,7 +50,7 @@ func TestAPIs(t *testing.T) {
 }
 
 var _ = BeforeSuite(func(done Done) {
-	logf.SetLogger(zap.LoggerTo(GinkgoWriter, true))
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
 
 	By("bootstrapping test environment")
 	testEnv = &envtest.Environment{

go.mod

@@ -4,45 +4,44 @@ go 1.14
 
 require (
 	github.com/davecgh/go-spew v1.1.1
-	github.com/go-logr/logr v0.1.0
-	github.com/google/go-cmp v0.3.1
-	github.com/onsi/ginkgo v1.11.0
-	github.com/onsi/gomega v1.8.1
-	golang.org/x/net v0.0.0-20191004110552-13f9640d40b9
-	gopkg.in/yaml.v2 v2.2.8
+	github.com/go-logr/logr v0.4.0
+	github.com/google/go-cmp v0.5.5
+	github.com/onsi/ginkgo v1.16.4
+	github.com/onsi/gomega v1.13.0
+	golang.org/x/net v0.0.0-20210428140749-89ef3d95e781
+	gopkg.in/yaml.v2 v2.4.0
 	gotest.tools v2.2.0+incompatible
-	k8s.io/api v0.18.3
-	k8s.io/apiextensions-apiserver v0.18.3 // indirect
-	k8s.io/apimachinery v0.18.3
-	k8s.io/client-go v0.18.3
+	k8s.io/api v0.21.2
+	k8s.io/apimachinery v0.21.2
+	k8s.io/client-go v0.21.2
 	k8s.io/klog v1.0.0
-	sigs.k8s.io/controller-runtime v0.6.0
+	sigs.k8s.io/controller-runtime v0.9.2
 	volcano.sh/volcano v0.0.0-20200615125001-3c3aaf766110
 )
 
 replace (
-	k8s.io/api => k8s.io/api v0.18.3
-	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.18.3
-	k8s.io/apimachinery => k8s.io/apimachinery v0.18.3
-	k8s.io/apiserver => k8s.io/apiserver v0.18.3
-	k8s.io/cli-runtime => k8s.io/cli-runtime v0.18.3
-	k8s.io/client-go => k8s.io/client-go v0.18.3
-	k8s.io/cloud-provider => k8s.io/cloud-provider v0.18.3
-	k8s.io/cluster-bootstrap => k8s.io/cluster-bootstrap v0.18.3
+	k8s.io/api => k8s.io/api v0.21.2
+	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.21.2
+	k8s.io/apimachinery => k8s.io/apimachinery v0.21.2
+	k8s.io/apiserver => k8s.io/apiserver v0.21.2
+	k8s.io/cli-runtime => k8s.io/cli-runtime v0.21.2
+	k8s.io/client-go => k8s.io/client-go v0.21.2
+	k8s.io/cloud-provider => k8s.io/cloud-provider v0.21.2
+	k8s.io/cluster-bootstrap => k8s.io/cluster-bootstrap v0.21.2
 	k8s.io/code-generator => k8s.io/code-generator v0.16.10-beta.0
-	k8s.io/component-base => k8s.io/component-base v0.18.3
-	k8s.io/cri-api => k8s.io/cri-api v0.18.3
-	k8s.io/csi-translation-lib => k8s.io/csi-translation-lib v0.18.3
-	k8s.io/kube-aggregator => k8s.io/kube-aggregator v0.18.3
-	k8s.io/kube-controller-manager => k8s.io/kube-controller-manager v0.18.3
-	k8s.io/kube-proxy => k8s.io/kube-proxy v0.18.3
-	k8s.io/kube-scheduler => k8s.io/kube-scheduler v0.18.3
-	k8s.io/kubectl => k8s.io/kubectl v0.18.3
-	k8s.io/kubelet => k8s.io/kubelet v0.18.3
-	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.18.3
-	k8s.io/metrics => k8s.io/metrics v0.18.3
-	k8s.io/node-api => k8s.io/node-api v0.18.3
-	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.18.3
-	k8s.io/sample-cli-plugin => k8s.io/sample-cli-plugin v0.18.3
-	k8s.io/sample-controller => k8s.io/sample-controller v0.18.3
+	k8s.io/component-base => k8s.io/component-base v0.21.2
+	k8s.io/cri-api => k8s.io/cri-api v0.21.2
+	k8s.io/csi-translation-lib => k8s.io/csi-translation-lib v0.21.2
+	k8s.io/kube-aggregator => k8s.io/kube-aggregator v0.21.2
+	k8s.io/kube-controller-manager => k8s.io/kube-controller-manager v0.21.2
+	k8s.io/kube-proxy => k8s.io/kube-proxy v0.21.2
+	k8s.io/kube-scheduler => k8s.io/kube-scheduler v0.21.2
+	k8s.io/kubectl => k8s.io/kubectl v0.21.2
+	k8s.io/kubelet => k8s.io/kubelet v0.21.2
+	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.21.2
+	k8s.io/metrics => k8s.io/metrics v0.21.2
+	k8s.io/node-api => k8s.io/node-api v0.21.2
+	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.21.2
+	k8s.io/sample-cli-plugin => k8s.io/sample-cli-plugin v0.21.2
+	k8s.io/sample-controller => k8s.io/sample-controller v0.21.2
 )