Merge pull request #1392 from runzexia/refine-job-rerun

job rerun with resourceVersion

pkg/apis/operations/v1alpha2/register.go

@@ -54,6 +54,7 @@ func addWebService(c *restful.Container) error {
 		Param(webservice.PathParameter("job", "job name")).
 		Param(webservice.PathParameter("namespace", "the name of the namespace where the job runs in")).
 		Param(webservice.QueryParameter("action", "action must be \"rerun\"")).
+		Param(webservice.QueryParameter("resourceVersion", "version of job, rerun when the version matches").Required(true)).
 		Returns(http.StatusOK, ok, errors.Error{}))
 
 	c.Add(webservice)

pkg/apis/resources/v1alpha2/register.go

@@ -82,6 +82,7 @@ func addWebService(c *restful.Container) error {
 		Param(webservice.PathParameter("job", "job name")).
 		Param(webservice.PathParameter("namespace", "the name of the namespace where the job runs in")).
 		Param(webservice.QueryParameter("action", "action must be \"rerun\"")).
+		Param(webservice.QueryParameter("resourceVersion", "version of job, rerun when the version matches")).
 		Returns(http.StatusOK, ok, errors.Error{}))
 
 	webservice.Route(webservice.GET("/{resources}").

pkg/apiserver/operations/job.go

@@ -19,6 +19,7 @@
 package operations
 
 import (
+	k8serr "k8s.io/apimachinery/pkg/api/errors"
 	"kubesphere.io/kubesphere/pkg/models/workloads"
 	"kubesphere.io/kubesphere/pkg/server/errors"
 	"net/http"
@@ -34,15 +35,20 @@ func RerunJob(req *restful.Request, resp *restful.Response) {
 	job := req.PathParameter("job")
 	namespace := req.PathParameter("namespace")
 	action := req.QueryParameter("action")
+	resourceVersion := req.QueryParameter("resourceVersion")
 
 	switch action {
 	case "rerun":
-		err = workloads.JobReRun(namespace, job)
+		err = workloads.JobReRun(namespace, job, resourceVersion)
 	default:
 		resp.WriteHeaderAndEntity(http.StatusBadRequest, errors.Wrap(fmt.Errorf("invalid operation %s", action)))
 		return
 	}
 	if err != nil {
+		if k8serr.IsConflict(err) {
+			resp.WriteHeaderAndEntity(http.StatusConflict, errors.Wrap(err))
+			return
+		}
 		resp.WriteHeaderAndEntity(http.StatusInternalServerError, errors.Wrap(err))
 		return
 	}

pkg/models/workloads/jobs.go

@@ -19,23 +19,32 @@ package workloads
 
 import (
 	"fmt"
+	"k8s.io/api/batch/v1"
+	k8serr "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/klog"
 	"kubesphere.io/kubesphere/pkg/simple/client"
 	"strings"
 	"time"
-
-	"k8s.io/api/batch/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const retryTimes = 3
 
-func JobReRun(namespace, jobName string) error {
+func JobReRun(namespace, jobName, resourceVersion string) error {
 	k8sClient := client.ClientSets().K8s().Kubernetes()
 	job, err := k8sClient.BatchV1().Jobs(namespace).Get(jobName, metav1.GetOptions{})
 	if err != nil {
 		return err
 	}
+	// do not rerun job if resourceVersion not match
+	if job.GetObjectMeta().GetResourceVersion() != resourceVersion {
+		err := k8serr.NewConflict(schema.GroupResource{
+			Group: job.GetObjectKind().GroupVersionKind().Group, Resource: "job",
+		}, jobName, fmt.Errorf("please apply your changes to the latest version and try again"))
+		klog.Warning(err)
+		return err
+	}
 
 	newJob := *job
 	newJob.ResourceVersion = ""