Bump k8s dependency to 1.19 and add context support (#126)

* Bump k8s dependency to 1.19 and add context support

1. Upgrade k8s dependency to 1.19
2. 1.18 client-go add context and extra options. Adapt to the new changes.
3. Upgrade golang to 1.15 to match golang used in 1.19

Signed-off-by: Jiaxin Shan <[email protected]>

* Regenerate objects using latest codegen and openapi

* Update CI golang version to 1.15

.travis.yml

@@ -1,7 +1,7 @@
 language: go
 
 go:
-  - "1.13"
+  - "1.15"
 
 go_import_path: github.com/kubeflow/common
 

go.mod

@@ -1,26 +1,43 @@
 module github.com/kubeflow/common
 
-go 1.13
+go 1.15
 
 require (
-	github.com/go-openapi/spec v0.19.2
-	github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef // indirect
-	github.com/googleapis/gnostic v0.2.0 // indirect
-	github.com/imdario/mergo v0.3.7 // indirect
-	github.com/prometheus/client_golang v1.5.1
-	github.com/sirupsen/logrus v1.4.2
-	github.com/stretchr/testify v1.4.0
-	k8s.io/api v0.16.15
-	k8s.io/apimachinery v0.16.15
-	k8s.io/client-go v0.16.15
-	k8s.io/code-generator v0.16.15
-	k8s.io/kube-openapi v0.0.0-20200410163147-594e756bea31
-	volcano.sh/apis v1.2.0-k8s1.16.15
+	github.com/go-openapi/spec v0.19.3
+	github.com/prometheus/client_golang v1.7.1
+	github.com/sirupsen/logrus v1.6.0
+	k8s.io/api v0.19.9
+	k8s.io/apimachinery v0.19.9
+	k8s.io/client-go v0.19.9
+	k8s.io/code-generator v0.19.9
+	k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6
+	volcano.sh/apis v1.2.0-k8s1.19.6
 )
 
+
 replace (
-	k8s.io/api => k8s.io/api v0.16.9
-	k8s.io/apimachinery => k8s.io/apimachinery v0.16.10-beta.0
-	k8s.io/client-go => k8s.io/client-go v0.16.9
-	k8s.io/code-generator => k8s.io/code-generator v0.16.10-beta.0
+	k8s.io/api => k8s.io/api v0.19.9
+	k8s.io/apiextensions-apiserver => k8s.io/apiextensions-apiserver v0.19.9
+	k8s.io/apimachinery => k8s.io/apimachinery v0.19.9
+	k8s.io/apiserver => k8s.io/apiserver v0.19.9
+	k8s.io/cli-runtime => k8s.io/cli-runtime v0.19.9
+	k8s.io/client-go => k8s.io/client-go v0.19.9
+	k8s.io/cloud-provider => k8s.io/cloud-provider v0.19.9
+	k8s.io/cluster-bootstrap => k8s.io/cluster-bootstrap v0.19.9
+	k8s.io/code-generator => k8s.io/code-generator v0.19.9
+	k8s.io/component-base => k8s.io/component-base v0.19.9
+	k8s.io/cri-api => k8s.io/cri-api v0.16.10-beta.0
+	k8s.io/csi-translation-lib => k8s.io/csi-translation-lib v0.19.9
+	k8s.io/kube-aggregator => k8s.io/kube-aggregator v0.19.9
+	k8s.io/kube-controller-manager => k8s.io/kube-controller-manager v0.19.9
+	k8s.io/kube-proxy => k8s.io/kube-proxy v0.19.9
+	k8s.io/kube-scheduler => k8s.io/kube-scheduler v0.19.9
+	k8s.io/kubectl => k8s.io/kubectl v0.19.9
+	k8s.io/kubelet => k8s.io/kubelet v0.19.9
+	k8s.io/legacy-cloud-providers => k8s.io/legacy-cloud-providers v0.19.9
+	k8s.io/metrics => k8s.io/metrics v0.19.9
+	k8s.io/node-api => k8s.io/node-api v0.19.9
+	k8s.io/sample-apiserver => k8s.io/sample-apiserver v0.19.9
+	k8s.io/sample-cli-plugin => k8s.io/sample-cli-plugin v0.19.9
+	k8s.io/sample-controller => k8s.io/sample-controller v0.19.9
 )

hack/update-codegen.sh

@@ -29,7 +29,13 @@ CODEGEN_VERSION=$(grep 'k8s.io/code-generator' go.sum | awk '{print $2}' | head
 CODEGEN_PKG=$(echo `go env GOPATH`"/pkg/mod/k8s.io/code-generator@${CODEGEN_VERSION}")
 
 # Grab kube-openapi version from go.sum
+# Due to dependency issue, we use go.mod to pin right version
 OPENAPI_VERSION=$(grep 'k8s.io/kube-openapi' go.mod | awk '{print $2}' | head -1)
+# remove /go.mod if it happens to match the version
+if [[ $OPENAPI_VERSION == */go.mod ]]; then
+    OPENAPI_VERSION=${OPENAPI_VERSION%/*}
+fi
+
 OPENAPI_PKG=$(echo `go env GOPATH`"/pkg/mod/k8s.io/kube-openapi@${OPENAPI_VERSION}")
 
 if [[ ! -d ${CODEGEN_PKG} || ! -d ${OPENAPI_PKG} ]]; then
@@ -104,4 +110,3 @@ ${GOPATH}/bin/openapi-gen --input-dirs github.com/kubeflow/common/test_job/apis/
 
 ## Copy everything back.
 cp -a "${TEMP_DIR}/${ROOT_PKG}/." "${SCRIPT_ROOT}/"
-

pkg/controller.v1/common/job_controller.go

@@ -1,6 +1,7 @@
 package common
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
@@ -219,7 +220,7 @@ func (jc *JobController) SyncPodGroup(job metav1.Object, pgSpec v1beta1.PodGroup
 
 	volcanoClientSet := jc.VolcanoClientSet
 	// Check whether podGroup exists or not
-	podGroup, err := volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Get(job.GetName(), metav1.GetOptions{})
+	podGroup, err := volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Get(context.TODO(), job.GetName(), metav1.GetOptions{})
 	if err == nil {
 		return podGroup, nil
 	}
@@ -236,7 +237,7 @@ func (jc *JobController) SyncPodGroup(job metav1.Object, pgSpec v1beta1.PodGroup
 		},
 		Spec: pgSpec,
 	}
-	createdPodGroup, err := volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Create(createPodGroup)
+	createdPodGroup, err := volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Create(context.TODO(), createPodGroup, metav1.CreateOptions{})
 	if err != nil {
 		return createdPodGroup, fmt.Errorf("unable to create PodGroup: %v", err)
 	}
@@ -249,7 +250,7 @@ func (jc *JobController) SyncPdb(job metav1.Object, minAvailableReplicas int32)
 	labelJobName := apiv1.JobNameLabel
 
 	// Check the pdb exist or not
-	pdb, err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Get(job.GetName(), metav1.GetOptions{})
+	pdb, err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Get(context.TODO(), job.GetName(), metav1.GetOptions{})
 	if err == nil || !k8serrors.IsNotFound(err) {
 		if err == nil {
 			err = errors.New(string(metav1.StatusReasonAlreadyExists))
@@ -275,7 +276,7 @@ func (jc *JobController) SyncPdb(job metav1.Object, minAvailableReplicas int32)
 			},
 		},
 	}
-	createdPdb, err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Create(createPdb)
+	createdPdb, err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Create(context.TODO(), createPdb, metav1.CreateOptions{})
 	if err != nil {
 		return createdPdb, fmt.Errorf("unable to create pdb: %v", err)
 	}
@@ -287,15 +288,15 @@ func (jc *JobController) DeletePodGroup(job metav1.Object) error {
 	volcanoClientSet := jc.VolcanoClientSet
 
 	// Check whether podGroup exists or not
-	_, err := volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Get(job.GetName(), metav1.GetOptions{})
+	_, err := volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Get(context.TODO(), job.GetName(), metav1.GetOptions{})
 	if err != nil && k8serrors.IsNotFound(err) {
 		return nil
 	}
 
 	log.Infof("Deleting PodGroup %s", job.GetName())
 
 	// Delete podGroup
-	err = volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Delete(job.GetName(), &metav1.DeleteOptions{})
+	err = volcanoClientSet.SchedulingV1beta1().PodGroups(job.GetNamespace()).Delete(context.TODO(), job.GetName(), metav1.DeleteOptions{})
 	if err != nil {
 		return fmt.Errorf("unable to delete PodGroup: %v", err)
 	}
@@ -306,15 +307,15 @@ func (jc *JobController) DeletePodGroup(job metav1.Object) error {
 func (jc *JobController) DeletePdb(job metav1.Object) error {
 
 	// Check whether pdb exists or not
-	_, err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Get(job.GetName(), metav1.GetOptions{})
+	_, err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Get(context.TODO(), job.GetName(), metav1.GetOptions{})
 	if err != nil && k8serrors.IsNotFound(err) {
 		return nil
 	}
 
 	msg := fmt.Sprintf("Deleting pdb %s", job.GetName())
 	log.Info(msg)
 
-	if err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Delete(job.GetName(), &metav1.DeleteOptions{}); err != nil {
+	if err := jc.KubeClientSet.PolicyV1beta1().PodDisruptionBudgets(job.GetNamespace()).Delete(context.TODO(), job.GetName(), metav1.DeleteOptions{}); err != nil {
 		return fmt.Errorf("unable to delete pdb: %v", err)
 	}
 	deletedPDBCount.Inc()

pkg/controller.v1/control/pod_control.go

@@ -15,6 +15,7 @@
 package control
 
 import (
+	"context"
 	"fmt"
 	commonutil "github.com/kubeflow/common/pkg/util"
 	"k8s.io/api/core/v1"
@@ -110,7 +111,7 @@ func (r RealPodControl) CreatePodsOnNode(nodeName, namespace string, template *v
 }
 
 func (r RealPodControl) PatchPod(namespace, name string, data []byte) error {
-	_, err := r.KubeClient.CoreV1().Pods(namespace).Patch(name, types.StrategicMergePatchType, data)
+	_, err := r.KubeClient.CoreV1().Pods(namespace).Patch(context.TODO(), name, types.StrategicMergePatchType, data, metav1.PatchOptions{})
 	return err
 }
 
@@ -146,7 +147,7 @@ func (r RealPodControl) createPods(nodeName, namespace string, template *v1.PodT
 		return fmt.Errorf("unable to create pods, no labels")
 	}
 	logger := commonutil.LoggerForPod(pod, object.GetObjectKind().GroupVersionKind().Kind)
-	if newPod, err := r.KubeClient.CoreV1().Pods(namespace).Create(pod); err != nil {
+	if newPod, err := r.KubeClient.CoreV1().Pods(namespace).Create(context.TODO(), pod, metav1.CreateOptions{}); err != nil {
 		r.Recorder.Eventf(object, v1.EventTypeWarning, FailedCreatePodReason, "Error creating: %v", err)
 		return err
 	} else {
@@ -167,7 +168,7 @@ func (r RealPodControl) DeletePod(namespace string, podID string, object runtime
 		return fmt.Errorf("object does not have ObjectMeta, %v", err)
 	}
 	logger := commonutil.LoggerForJob(accessor)
-	pod, err := r.KubeClient.CoreV1().Pods(namespace).Get(podID, metav1.GetOptions{})
+	pod, err := r.KubeClient.CoreV1().Pods(namespace).Get(context.TODO(), podID, metav1.GetOptions{})
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return nil
@@ -179,7 +180,8 @@ func (r RealPodControl) DeletePod(namespace string, podID string, object runtime
 		return nil
 	}
 	logger.Infof("Controller %v deleting pod %v/%v", accessor.GetName(), namespace, podID)
-	if err := r.KubeClient.CoreV1().Pods(namespace).Delete(podID, nil); err != nil {
+	// delete options
+	if err := r.KubeClient.CoreV1().Pods(namespace).Delete(context.TODO(), podID, metav1.DeleteOptions{}); err != nil {
 		r.Recorder.Eventf(object, v1.EventTypeWarning, FailedDeletePodReason, "Error deleting: %v", err)
 		return fmt.Errorf("unable to delete pods: %v", err)
 	} else {

pkg/controller.v1/control/service_control.go

@@ -15,6 +15,7 @@
 package control
 
 import (
+	"context"
 	"fmt"
 	"sync"
 
@@ -65,7 +66,7 @@ type RealServiceControl struct {
 }
 
 func (r RealServiceControl) PatchService(namespace, name string, data []byte) error {
-	_, err := r.KubeClient.CoreV1().Services(namespace).Patch(name, types.StrategicMergePatchType, data)
+	_, err := r.KubeClient.CoreV1().Services(namespace).Patch(context.TODO(), name, types.StrategicMergePatchType, data, metav1.PatchOptions{})
 	return err
 }
 
@@ -90,7 +91,7 @@ func (r RealServiceControl) createServices(namespace string, service *v1.Service
 		return fmt.Errorf("unable to create services: %v", err)
 	}
 
-	newService, err := r.KubeClient.CoreV1().Services(namespace).Create(serviceWithOwner)
+	newService, err := r.KubeClient.CoreV1().Services(namespace).Create(context.TODO(), serviceWithOwner, metav1.CreateOptions{})
 	if err != nil {
 		r.Recorder.Eventf(object, v1.EventTypeWarning, FailedCreateServiceReason, "Error creating: %v", err)
 		return fmt.Errorf("unable to create services: %v", err)
@@ -113,7 +114,7 @@ func (r RealServiceControl) DeleteService(namespace, serviceID string, object ru
 	if err != nil {
 		return fmt.Errorf("object does not have ObjectMeta, %v", err)
 	}
-	service, err := r.KubeClient.CoreV1().Services(namespace).Get(serviceID, metav1.GetOptions{})
+	service, err := r.KubeClient.CoreV1().Services(namespace).Get(context.TODO(), serviceID, metav1.GetOptions{})
 	if err != nil {
 		if errors.IsNotFound(err) {
 			return nil
@@ -125,7 +126,7 @@ func (r RealServiceControl) DeleteService(namespace, serviceID string, object ru
 		return nil
 	}
 	log.Infof("Controller %v deleting service %v/%v", accessor.GetName(), namespace, serviceID)
-	if err := r.KubeClient.CoreV1().Services(namespace).Delete(serviceID, nil); err != nil {
+	if err := r.KubeClient.CoreV1().Services(namespace).Delete(context.TODO(), serviceID, metav1.DeleteOptions{}); err != nil {
 		r.Recorder.Eventf(object, v1.EventTypeWarning, FailedDeleteServiceReason, "Error deleting: %v", err)
 		return fmt.Errorf("unable to delete service: %v", err)
 	} else {

pkg/util/k8sutil/client.go

@@ -13,6 +13,7 @@
 package k8sutil
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 
@@ -74,7 +75,7 @@ func (c *CRDRestClient) Update(obj *metav1unstructured.Unstructured, plural stri
 		return fmt.Errorf("plural must be set")
 	}
 	r := c.restcli.Put().Resource(plural).Namespace(obj.GetNamespace()).Name(obj.GetName()).Body(obj)
-	_, err := r.DoRaw()
+	_, err := r.DoRaw(context.TODO())
 	if err != nil {
 		logger.Errorf("Could not issue update using URL: %v; error; %v", r.URL().String(), err)
 	}
@@ -88,7 +89,7 @@ func (c *CRDRestClient) UpdateStatus(obj *metav1unstructured.Unstructured, plura
 		return fmt.Errorf("plural must be set")
 	}
 	r := c.restcli.Put().Resource(plural).Namespace(obj.GetNamespace()).Name(obj.GetName()).SubResource("status").Body(obj)
-	_, err := r.DoRaw()
+	_, err := r.DoRaw(context.TODO())
 	if err != nil {
 		logger.Errorf("Could not issue update using URL: %v; error; %v", r.URL().String(), err)
 	}