control-service: Adopt use of the V1CronJob API (#767)

To allow for the automatic job clean up of manually ran data jobs
and builder jobs, the VDK control service must use the V1CronJob
API. However, V1CronJobs are available from Kubernetes 1.21 onwards,
and VDK must be able to run on older versions of Kubernetes.
This change manages this by duplicating relevant methods, and relying
on a feature flag to determine whether the Kubernetes cluster supports
V1CronJobs, and defaulting back to V1beta1CronJobs in the case where
it does not.

Testing done: all existing unit and intergration tests pass, added unit tests

Signed-off-by: Gabriel Georgiev <gageorgiev@vmware.com>

Author: gabrielgeorgiev1

projects/control-service/projects/base/src/main/java/com/vmware/taurus/base/FeatureFlags.java

@@ -26,6 +26,7 @@ public class FeatureFlags {
    @Value("${featureflag.authorization.enabled:false}")
    boolean authorizationEnabled = false;
 
+
    @Value("${datajobs.security.kerberos.enabled:false}")
    boolean krbAuthEnabled = false;
 

projects/control-service/projects/pipelines_control_service/src/integration-test/resources/application-test.properties

@@ -14,6 +14,7 @@ datajobs.vdk.image=registry.hub.docker.com/versatiledatakit/quickstart-vdk:relea
 
 datajobs.deployment.k8s.kubeconfig=${DEPLOYMENT_K8S_KUBECONFIG}
 datajobs.deployment.k8s.namespace=${DEPLOYMENT_K8S_NAMESPACE}
+datajobs.control.k8s.k8sSupportsV1CronJob=false
 
 datajobs.control.k8s.kubeconfig=${CONTROL_K8S_KUBECONFIG}
 datajobs.control.k8s.namespace=${CONTROL_K8S_NAMESPACE}

projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/KubernetesService.java

@@ -21,7 +21,8 @@ public class ControlKubernetesService extends KubernetesService {
 
    // those should be null/empty when Control service is deployed in k8s hence default is empty
    public ControlKubernetesService(@Value("${datajobs.control.k8s.namespace:}") String namespace,
-                                   @Value("${datajobs.control.k8s.kubeconfig:}") String kubeconfig) {
-      super(namespace, kubeconfig, log);
+                                   @Value("${datajobs.control.k8s.kubeconfig:}") String kubeconfig,
+                                   @Value("${datajobs.control.k8s.k8sSupportsV1CronJob}") boolean k8sSupportsV1CronJob) {
+      super(namespace, kubeconfig, k8sSupportsV1CronJob, log);
    }
 }

projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/kubernetes/ControlKubernetesService.java

@@ -23,8 +23,9 @@
 public class DataJobsKubernetesService extends KubernetesService {
 
    public DataJobsKubernetesService(@Value("${datajobs.deployment.k8s.namespace:}") String namespace,
-                                    @Value("${datajobs.deployment.k8s.kubeconfig:}") String kubeconfig) {
-      super(namespace, kubeconfig, log);
+                                    @Value("${datajobs.deployment.k8s.kubeconfig:}") String kubeconfig,
+                                    @Value("${datajobs.control.k8s.k8sSupportsV1CronJob}") boolean k8sSupportsV1CronJob) {
+      super(namespace, kubeconfig, k8sSupportsV1CronJob, log);
       if (StringUtils.isBlank(kubeconfig) && !new File(kubeconfig).isFile()) {
          log.warn("Data Jobs (Deployment) Kubernetes service may not have been correctly bootstrapped. {} file is missing " +
                          "Will try to use same cluster as control Plane. But this is not recommended in production.",

projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/kubernetes/DataJobsKubernetesService.java

@@ -43,6 +43,7 @@ featureflag.authorization.enabled=false
 datajobs.authorization.webhook.endpoint=
 datajobs.authorization.jwt.claim.username=username
 
+
 # Data Jobs post webhook settings (Create and Delete)
 datajobs.post.create.webhook.endpoint=
 datajobs.post.create.webhook.internal.errors.retries=3
@@ -77,6 +78,7 @@ logging.level.io.swagger.models.parameters.AbstractSerializableParameter=ERROR
 
 datajobs.control.k8s.namespace=
 datajobs.control.k8s.kubeconfig=${HOME}/.kube/config
+datajobs.control.k8s.k8sSupportsV1CronJob=false
 # Location to a K8s cronjob yaml file which will be used as a template
 # for all data jobs. If the location is missing, the default internal
 # cronjob yaml resource will be used (see k8s-data-job-template.yaml).

projects/control-service/projects/pipelines_control_service/src/main/resources/application.properties

@@ -32,6 +32,7 @@ public class KubernetesServiceStartJobWithArgumentsIT {
     public void getMockKubernetesServiceForVdkRunExtraArgsTests() throws Exception {
 
         kubernetesService = Mockito.mock(KubernetesService.class);
+        Mockito.when(kubernetesService.getK8sSupportsV1CronJob()).thenReturn(false);
         V1beta1CronJob internalCronjobTemplate = getValidCronJobForVdkRunExtraArgsTests();
         BatchV1beta1Api mockBatch = Mockito.mock(BatchV1beta1Api.class);
         Mockito.when(kubernetesService.initBatchV1beta1Api()).thenReturn(mockBatch);
@@ -131,15 +132,15 @@ public void testStartCronJobWithEmptyArgumentsForVdkRun() {
     }
 
     private V1beta1CronJob getValidCronJobForVdkRunExtraArgsTests() throws Exception {
-        KubernetesService service = new DataJobsKubernetesService("default", "someConfig");
+        KubernetesService service = new DataJobsKubernetesService("default", "someConfig", false);
         // V1betaCronJob initializing snippet copied from tests above, using reflection
         service.afterPropertiesSet();
-        Method loadInternalCronjobTemplate = KubernetesService.class.getDeclaredMethod("loadInternalCronjobTemplate");
-        if (loadInternalCronjobTemplate == null) {
-            Assertions.fail("The method 'loadInternalCronjobTemplate' does not exist.");
+        Method loadInternalV1beta1CronjobTemplate = KubernetesService.class.getDeclaredMethod("loadInternalV1beta1CronjobTemplate");
+        if (loadInternalV1beta1CronjobTemplate == null) {
+            Assertions.fail("The method 'loadInternalV1beta1CronjobTemplate' does not exist.");
         }
-        loadInternalCronjobTemplate.setAccessible(true);
-        V1beta1CronJob internalCronjobTemplate = (V1beta1CronJob) loadInternalCronjobTemplate.invoke(service);
+        loadInternalV1beta1CronjobTemplate.setAccessible(true);
+        V1beta1CronJob internalCronjobTemplate = (V1beta1CronJob) loadInternalV1beta1CronjobTemplate.invoke(service);
         var container = internalCronjobTemplate.getSpec()
                 .getJobTemplate()
                 .getSpec()