control-service: handle deployment deletion in case of a job being de…

…leted (#2816) # Why When the data job is deleted, both desired deployment records are deleted, but the actual deployment records and Kubernetes cron jobs are still present. # What Extended the deployment synchronization to cover this scenario. Added an additional flyway script as part of the integration tests because of the different H2 constraint naming. # Testing done Integration tests. Signed-off-by: Miroslav Ivanov miroslavi@vmware.com --------- Signed-off-by: Miroslav Ivanov miroslavi@vmware.com Co-authored-by: github-actions <>
vmware · Oct 19, 2023 · cab16e7 · cab16e7
1 parent e17ead2
commit cab16e7
Show file tree

Hide file tree

Showing 13 changed files with 141 additions and 71 deletions.
diff --git a/...src/integration-test/java/com/vmware/taurus/service/deploy/DataJobDeploymentCrudITV2.java b/...src/integration-test/java/com/vmware/taurus/service/deploy/DataJobDeploymentCrudITV2.java
@@ -200,6 +200,19 @@ public void testSynchronizeDataJobs() throws Exception {
     OffsetDateTime lastDeployedDateInitial = actualDataJobDeployment.getLastDeployedDate();
     Assertions.assertNotNull(deploymentVersionShaInitial);
     Assertions.assertNotNull(lastDeployedDateInitial);
+
+    jobsRepository.deleteById(testJobName);
+
+    // Re-deploys data job
+    dataJobsSynchronizer.synchronizeDataJobs();
+
+    // Wait for the job deployment to complete, polling every 15 seconds
+    // See: https://github.com/awaitility/awaitility/wiki/Usage
+    await()
+        .atMost(10, TimeUnit.MINUTES)
+        .with()
+        .pollInterval(15, TimeUnit.SECONDS)
+        .until(() -> deploymentService.readDeployment(testJobName).isEmpty());
   }
 
   private ActualDataJobDeployment verifyDeploymentStatus(boolean enabled) {
@@ -225,7 +238,9 @@ public void cleanUp() throws Exception {
             delete(String.format("/data-jobs/for-team/%s/jobs/%s", TEST_TEAM_NAME, testJobName))
                 .with(user(TEST_USERNAME))
                 .contentType(MediaType.APPLICATION_JSON));
-    if (perform.andReturn().getResponse().getStatus() != 200) {
+    int responseStatus = perform.andReturn().getResponse().getStatus();
+
+    if (responseStatus != 200 && responseStatus != 404) {
       throw new Exception(
           "status is "
               + perform.andReturn().getResponse().getStatus()

diff --git a/...ects/pipelines_control_service/src/integration-test/resources/application-test.properties b/...ects/pipelines_control_service/src/integration-test/resources/application-test.properties
@@ -85,3 +85,5 @@ datajobs.aws.serviceAccountSecretAccessKey=${DATAJOBS_AWS_SERVICE_ACCOUNT_SECRET
 #Write to values for integration tests.
 datajobs.deployment.configuration.persistence.writeTos=${DATAJOBS_DEPLOYMENT_CONFIGURATION_PERSISTENCE_WRITE_TOS:"K8S"}
 datajobs.deployment.configuration.persistence.readDataSource=${DATAJOBS_DEPLOYMENT_CONFIGURATION_PERSISTENCE_READ_DATA_SOURCE:"K8S"}
+
+spring.flyway.locations=classpath:/db/migration,classpath:/db/migration/test
diff --git a/...V20231018150000__drop_constraint_constraint_907_from_actual_data_job_deployment_table.sql b/...V20231018150000__drop_constraint_constraint_907_from_actual_data_job_deployment_table.sql
@@ -0,0 +1,2 @@
+alter table if exists actual_data_job_deployment
+    drop constraint if exists CONSTRAINT_907;
diff --git a/...es_control_service/src/main/java/com/vmware/taurus/datajobs/DeploymentModelConverter.java b/...es_control_service/src/main/java/com/vmware/taurus/datajobs/DeploymentModelConverter.java
@@ -71,7 +71,6 @@ public static ActualDataJobDeployment toActualJobDeployment(
       OffsetDateTime lastDeployedDate) {
     ActualDataJobDeployment deployment = new ActualDataJobDeployment();
     deployment.setDataJobName(desiredDataJobDeployment.getDataJobName());
-    deployment.setDataJob(desiredDataJobDeployment.getDataJob());
     deployment.setEnabled(desiredDataJobDeployment.getEnabled());
 
     DataJobDeploymentResources desiredDataJobDeploymentResources =

diff --git a/..._control_service/src/main/java/com/vmware/taurus/service/deploy/DataJobsSynchronizer.java b/..._control_service/src/main/java/com/vmware/taurus/service/deploy/DataJobsSynchronizer.java
@@ -10,7 +10,6 @@
 import com.vmware.taurus.service.model.ActualDataJobDeployment;
 import com.vmware.taurus.service.model.DataJob;
 import com.vmware.taurus.service.model.DesiredDataJobDeployment;
-import io.kubernetes.client.openapi.ApiException;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import net.javacrumbs.shedlock.spring.annotation.SchedulerLock;
@@ -19,10 +18,10 @@
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
@@ -68,35 +67,25 @@ public class DataJobsSynchronizer {
           "${datajobs.deployment.configuration.synchronization.task.initial.delay.ms:10000}")
   @SchedulerLock(name = "synchronizeDataJobsTask")
   public void synchronizeDataJobs() {
-    if (!synchronizationEnabled) {
-      log.debug("Skipping the synchronization of data job deployments since it is disabled.");
-      return;
-    }
-
-    if (!dataJobDeploymentPropertiesConfig
-        .getWriteTos()
-        .contains(DataJobDeploymentPropertiesConfig.WriteTo.DB)) {
-      log.debug(
-          "Skipping data job deployments' synchronization due to the disabled writes to the"
-              + " database.");
+    if (!validateConfiguration()) {
       return;
     }
 
     log.info("Data job deployments synchronization has started.");
 
-    List<DataJob> dataJobsFromDB =
+    Map<String, DataJob> dataJobsFromDBMap =
         StreamSupport.stream(jobsService.findAllDataJobs().spliterator(), false)
-            .collect(Collectors.toList());
+            .collect(Collectors.toMap(DataJob::getName, Function.identity()));
     Set<String> dataJobDeploymentNamesFromKubernetes;
 
     try {
       dataJobDeploymentNamesFromKubernetes =
           deploymentService.findAllActualDeploymentNamesFromKubernetes();
-    } catch (ApiException e) {
+    } catch (KubernetesException e) {
       log.error(
           "Skipping data job deployment synchronization because deployment names cannot be loaded"
               + " from Kubernetes.",
-          new KubernetesException("Cannot load cron jobs", e));
+          e);
       return;
     }
 
@@ -108,28 +97,53 @@ public void synchronizeDataJobs() {
     Map<String, ActualDataJobDeployment> actualDataJobDeploymentsFromDBMap =
         deploymentService.findAllActualDataJobDeployments();
 
-    CountDownLatch countDownLatch = new CountDownLatch(dataJobsFromDB.size());
-
-    dataJobsFromDB.forEach(
-        dataJob ->
-            executeSynchronizationTask(
-                () ->
-                    synchronizeDataJob(
-                        dataJob,
-                        desiredDataJobDeploymentsFromDBMap.get(dataJob.getName()),
-                        actualDataJobDeploymentsFromDBMap.get(dataJob.getName()),
-                        finalDataJobDeploymentNamesFromKubernetes.contains(dataJob.getName())),
-                countDownLatch));
+    // Actual deployments that do not have an associated existing data jobs with them.
+    Set<String> actualDataJobDeploymentsThatShouldBeDeleted =
+        actualDataJobDeploymentsFromDBMap.keySet().stream()
+            .filter(dataJobName -> !dataJobsFromDBMap.containsKey(dataJobName))
+            .collect(Collectors.toSet());
+
+    CountDownLatch countDownLatch =
+        new CountDownLatch(
+            dataJobsFromDBMap.size() + actualDataJobDeploymentsThatShouldBeDeleted.size());
+
+    // Synchronizes deployments that have associated existing data jobs with them.
+    // In this scenario, the deployment creation or updating has been requested.
+    synchronizeDataJobs(
+        dataJobsFromDBMap.keySet(),
+        dataJobsFromDBMap,
+        desiredDataJobDeploymentsFromDBMap,
+        actualDataJobDeploymentsFromDBMap,
+        finalDataJobDeploymentNamesFromKubernetes,
+        countDownLatch);
+    // Synchronizes deployments that do not have an associated existing data jobs with them.
+    // In this scenario, the deployment deletion has been requested.
+    synchronizeDataJobs(
+        actualDataJobDeploymentsThatShouldBeDeleted,
+        dataJobsFromDBMap,
+        desiredDataJobDeploymentsFromDBMap,
+        actualDataJobDeploymentsFromDBMap,
+        finalDataJobDeploymentNamesFromKubernetes,
+        countDownLatch);
+
+    waitForSynchronizationCompletion(countDownLatch);
+  }
 
-    try {
-      log.debug(
-          "Waiting for data job deployments' synchronization to complete. This process may take"
-              + " some time...");
-      countDownLatch.await();
-      log.info("Data job deployments synchronization has successfully completed.");
-    } catch (InterruptedException e) {
-      log.error("An error occurred during the data job deployments' synchronization", e);
-    }
+  private void synchronizeDataJobs(
+      Set<String> dataJobsToBeSynchronized,
+      Map<String, DataJob> dataJobsFromDBMap,
+      Map<String, DesiredDataJobDeployment> desiredDataJobDeploymentsFromDBMap,
+      Map<String, ActualDataJobDeployment> actualDataJobDeploymentsFromDBMap,
+      Set<String> finalDataJobDeploymentNamesFromKubernetes,
+      CountDownLatch countDownLatch) {
+    dataJobsToBeSynchronized.forEach(
+        dataJobName ->
+            executeDataJobSynchronizationTask(
+                dataJobsFromDBMap.get(dataJobName),
+                desiredDataJobDeploymentsFromDBMap.get(dataJobName),
+                actualDataJobDeploymentsFromDBMap.get(dataJobName),
+                finalDataJobDeploymentNamesFromKubernetes.contains(dataJobName),
+                countDownLatch));
   }
 
   // Default for testing purposes
@@ -145,18 +159,59 @@ void synchronizeDataJob(
           actualDataJobDeployment,
           isDeploymentPresentInKubernetes);
     } else if (actualDataJobDeployment != null) {
-      deploymentService.deleteActualDeployment(dataJob.getName());
+      deploymentService.deleteActualDeployment(actualDataJobDeployment.getDataJobName());
     }
   }
 
-  private void executeSynchronizationTask(Runnable runnable, CountDownLatch countDownLatch) {
+  private void executeDataJobSynchronizationTask(
+      DataJob dataJob,
+      DesiredDataJobDeployment desiredDataJobDeployment,
+      ActualDataJobDeployment actualDataJobDeployment,
+      boolean isDeploymentPresentInKubernetes,
+      CountDownLatch countDownLatch) {
     dataJobsSynchronizerTaskExecutor.execute(
         () -> {
           try {
-            runnable.run();
+            synchronizeDataJob(
+                dataJob,
+                desiredDataJobDeployment,
+                actualDataJobDeployment,
+                isDeploymentPresentInKubernetes);
           } finally {
             countDownLatch.countDown();
           }
         });
   }
+
+  private boolean validateConfiguration() {
+    boolean valid = true;
+
+    if (!synchronizationEnabled) {
+      log.debug("Skipping the synchronization of data job deployments since it is disabled.");
+      valid = false;
+    }
+
+    if (!dataJobDeploymentPropertiesConfig
+        .getWriteTos()
+        .contains(DataJobDeploymentPropertiesConfig.WriteTo.DB)) {
+      log.debug(
+          "Skipping data job deployments' synchronization due to the disabled writes to the"
+              + " database.");
+      valid = false;
+    }
+
+    return valid;
+  }
+
+  private void waitForSynchronizationCompletion(CountDownLatch countDownLatch) {
+    try {
+      log.debug(
+          "Waiting for data job deployments' synchronization to complete. This process may take"
+              + " some time...");
+      countDownLatch.await();
+      log.info("Data job deployments synchronization has successfully completed.");
+    } catch (InterruptedException e) {
+      log.error("An error occurred during the data job deployments' synchronization", e);
+    }
+  }
 }
diff --git a/...s_control_service/src/main/java/com/vmware/taurus/service/deploy/DeploymentServiceV2.java b/...s_control_service/src/main/java/com/vmware/taurus/service/deploy/DeploymentServiceV2.java
@@ -228,8 +228,12 @@ public Map<String, ActualDataJobDeployment> findAllActualDataJobDeployments() {
         .collect(Collectors.toMap(ActualDataJobDeployment::getDataJobName, Function.identity()));
   }
 
-  public Set<String> findAllActualDeploymentNamesFromKubernetes() throws ApiException {
-    return dataJobsKubernetesService.listCronJobs();
+  public Set<String> findAllActualDeploymentNamesFromKubernetes() throws KubernetesException {
+    try {
+      return dataJobsKubernetesService.listCronJobs();
+    } catch (ApiException e) {
+      throw new KubernetesException("Cannot load cron jobs", e);
+    }
   }
 
   private void handleException(

diff --git a/..._control_service/src/main/java/com/vmware/taurus/service/model/BaseDataJobDeployment.java b/..._control_service/src/main/java/com/vmware/taurus/service/model/BaseDataJobDeployment.java
@@ -23,13 +23,6 @@ public abstract class BaseDataJobDeployment {
   @Column(name = "data_job_name")
   private String dataJobName;
 
-  @MapsId
-  @OneToOne(cascade = CascadeType.MERGE)
-  @JoinColumn(name = "data_job_name")
-  @ToString.Exclude
-  @EqualsAndHashCode.Exclude
-  private DataJob dataJob;
-
   private String pythonVersion;
 
   private String gitCommitSha;

diff --git a/...ntrol_service/src/main/java/com/vmware/taurus/service/model/DesiredDataJobDeployment.java b/...ntrol_service/src/main/java/com/vmware/taurus/service/model/DesiredDataJobDeployment.java
@@ -10,8 +10,7 @@
 import lombok.Setter;
 import lombok.ToString;
 
-import javax.persistence.Column;
-import javax.persistence.Entity;
+import javax.persistence.*;
 
 @Getter
 @Setter
@@ -20,6 +19,13 @@
 @Entity
 public class DesiredDataJobDeployment extends BaseDataJobDeployment {
 
+  @MapsId
+  @OneToOne(cascade = CascadeType.MERGE)
+  @JoinColumn(name = "data_job_name")
+  @ToString.Exclude
+  @EqualsAndHashCode.Exclude
+  private DataJob dataJob;
+
   private DeploymentStatus status;
 
   @Column(name = "is_user_initiated")

diff --git a/...lines_control_service/src/test/java/com/vmware/taurus/datajobs/ModelApiConverterTest.java b/...lines_control_service/src/test/java/com/vmware/taurus/datajobs/ModelApiConverterTest.java
@@ -54,7 +54,6 @@ private DataJob createTestJob(String jobName, String teamName) {
   private ActualDataJobDeployment createActualJobDeployment(DataJob dataJob) {
     var deployment = new ActualDataJobDeployment();
     deployment.setGitCommitSha("actualSha");
-    deployment.setDataJob(dataJob);
     deployment.setDataJobName(dataJob.getName());
     deployment.setPythonVersion("3.9-secure");
     deployment.setEnabled(true);

diff --git a/...e/src/test/java/com/vmware/taurus/service/deploy/DataJobDeploymentControllerReadTest.java b/...e/src/test/java/com/vmware/taurus/service/deploy/DataJobDeploymentControllerReadTest.java
@@ -96,7 +96,6 @@ private DataJob createTestJob(String jobName, String teamName) {
   private ActualDataJobDeployment createActualJobDeployment(DataJob dataJob) {
     var deployment = new ActualDataJobDeployment();
     deployment.setGitCommitSha("actualSha");
-    deployment.setDataJob(dataJob);
     deployment.setDataJobName(dataJob.getName());
     deployment.setPythonVersion("3.9-secure");
     deployment.setEnabled(true);

diff --git a/...trol_service/src/test/java/com/vmware/taurus/service/deploy/DataJobsSynchronizerTest.java b/...trol_service/src/test/java/com/vmware/taurus/service/deploy/DataJobsSynchronizerTest.java
@@ -6,6 +6,7 @@
 package com.vmware.taurus.service.deploy;
 
 import com.vmware.taurus.ServiceApp;
+import com.vmware.taurus.exception.KubernetesException;
 import com.vmware.taurus.service.model.ActualDataJobDeployment;
 import com.vmware.taurus.service.model.DataJob;
 import com.vmware.taurus.service.model.DesiredDataJobDeployment;
@@ -34,8 +35,7 @@ public class DataJobsSynchronizerTest {
 
   @Test
   void
-      synchronizeDataJobs_loadDeploymentNamesFromKubernetesReturnsValue_shouldFinishSynchronization()
-          throws ApiException {
+      synchronizeDataJobs_loadDeploymentNamesFromKubernetesReturnsValue_shouldFinishSynchronization() {
     enableSynchronizationProcess();
 
     Mockito.when(deploymentService.findAllActualDeploymentNamesFromKubernetes())
@@ -51,12 +51,11 @@ public class DataJobsSynchronizerTest {
 
   @Test
   void
-      synchronizeDataJobs_loadDeploymentNamesFromKubernetesThrowsApiException_shouldSkipSynchronization()
-          throws ApiException {
+      synchronizeDataJobs_loadDeploymentNamesFromKubernetesThrowsApiException_shouldSkipSynchronization() {
     enableSynchronizationProcess();
 
     Mockito.when(deploymentService.findAllActualDeploymentNamesFromKubernetes())
-        .thenThrow(new ApiException());
+        .thenThrow(new KubernetesException("Test exception", new ApiException()));
 
     dataJobsSynchronizer.synchronizeDataJobs();
 
@@ -67,8 +66,7 @@ public class DataJobsSynchronizerTest {
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbTrue_shouldSkipSynchronization()
-      throws ApiException {
+  void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbTrue_shouldSkipSynchronization() {
     initSynchronizationProcessConfig(false, true);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -78,8 +76,8 @@ void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbTrue_shouldSkipS
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbFalse_shouldSkipSynchronization()
-      throws ApiException {
+  void
+      synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbFalse_shouldSkipSynchronization() {
     initSynchronizationProcessConfig(false, false);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -89,8 +87,8 @@ void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbFalse_shouldSkip
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbTrue_shouldFinishSynchronization()
-      throws ApiException {
+  void
+      synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbTrue_shouldFinishSynchronization() {
     initSynchronizationProcessConfig(true, true);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -100,8 +98,7 @@ void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbTrue_shouldFinish
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbFalse_shouldSkipSynchronization()
-      throws ApiException {
+  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbFalse_shouldSkipSynchronization() {
     initSynchronizationProcessConfig(true, false);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -141,6 +138,7 @@ void synchronizeDataJob_desiredDeploymentNullAndActualDeploymentNull_shouldSkipS
     boolean isDeploymentPresentInKubernetes = true;
     DesiredDataJobDeployment desiredDataJobDeployment = null;
     ActualDataJobDeployment actualDataJobDeployment = new ActualDataJobDeployment();
+    actualDataJobDeployment.setDataJobName(dataJob.getName());
 
     dataJobsSynchronizer.synchronizeDataJob(
         dataJob,
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1,2 @@
		alter table if exists actual_data_job_deployment
		drop constraint if exists CONSTRAINT_907;