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

…leted 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. Testing done Integration tests. Signed-off-by: Miroslav Ivanov miroslavi@vmware.com
vmware · Oct 19, 2023 · 9c8baa8 · 9c8baa8
1 parent db51a9f
commit 9c8baa8
Show file tree

Hide file tree

Showing 11 changed files with 123 additions and 70 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,24 @@ 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 =
-        StreamSupport.stream(jobsService.findAllDataJobs().spliterator(), false)
-            .collect(Collectors.toList());
+    Map<String, DataJob> dataJobsFromDBMap =
+            StreamSupport.stream(jobsService.findAllDataJobs().spliterator(), false)
+                    .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));
+              + " from Kubernetes.", e);
       return;
     }
 
@@ -108,28 +96,38 @@ public void synchronizeDataJobs() {
     Map<String, ActualDataJobDeployment> actualDataJobDeploymentsFromDBMap =
         deploymentService.findAllActualDataJobDeployments();
 
-    CountDownLatch countDownLatch = new CountDownLatch(dataJobsFromDB.size());
+    // 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());
 
-    dataJobsFromDB.forEach(
-        dataJob ->
-            executeSynchronizationTask(
-                () ->
-                    synchronizeDataJob(
-                        dataJob,
-                        desiredDataJobDeploymentsFromDBMap.get(dataJob.getName()),
-                        actualDataJobDeploymentsFromDBMap.get(dataJob.getName()),
-                        finalDataJobDeploymentNamesFromKubernetes.contains(dataJob.getName())),
-                countDownLatch));
+    CountDownLatch countDownLatch = new CountDownLatch(dataJobsFromDBMap.size() + actualDataJobDeploymentsThatShouldBeDeleted.size());
 
-    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);
-    }
+    // 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);
+  }
+
+  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 +143,58 @@ 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/...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,7 @@ void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbTrue_shouldSkipS
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbFalse_shouldSkipSynchronization()
-      throws ApiException {
+  void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbFalse_shouldSkipSynchronization() {
     initSynchronizationProcessConfig(false, false);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -89,8 +86,7 @@ void synchronizeDataJobs_synchronizationEnabledFalseAndWriteToDbFalse_shouldSkip
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbTrue_shouldFinishSynchronization()
-      throws ApiException {
+  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbTrue_shouldFinishSynchronization() {
     initSynchronizationProcessConfig(true, true);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -100,8 +96,7 @@ void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbTrue_shouldFinish
   }
 
   @Test
-  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbFalse_shouldSkipSynchronization()
-      throws ApiException {
+  void synchronizeDataJobs_synchronizationEnabledTrueAndWriteToDbFalse_shouldSkipSynchronization() {
     initSynchronizationProcessConfig(true, false);
 
     dataJobsSynchronizer.synchronizeDataJobs();
@@ -141,6 +136,7 @@ void synchronizeDataJob_desiredDeploymentNullAndActualDeploymentNull_shouldSkipS
     boolean isDeploymentPresentInKubernetes = true;
     DesiredDataJobDeployment desiredDataJobDeployment = null;
     ActualDataJobDeployment actualDataJobDeployment = new ActualDataJobDeployment();
+    actualDataJobDeployment.setDataJobName(dataJob.getName());
 
     dataJobsSynchronizer.synchronizeDataJob(
         dataJob,

diff --git a/...e/src/test/java/com/vmware/taurus/service/repository/ActualJobDeploymentRepositoryIT.java b/...e/src/test/java/com/vmware/taurus/service/repository/ActualJobDeploymentRepositoryIT.java
@@ -83,7 +83,6 @@ private ActualDataJobDeployment createDataJobDeployment() {
 
     ActualDataJobDeployment expectedDataJobDeployment = new ActualDataJobDeployment();
     expectedDataJobDeployment.setDataJobName(actualDataJob.getName());
-    expectedDataJobDeployment.setDataJob(actualDataJob);
     expectedDataJobDeployment.setPythonVersion("3.9-secure");
     expectedDataJobDeployment.setGitCommitSha("commit");
 

diff --git a/..._control_service/src/test/java/com/vmware/taurus/service/repository/JobsRepositoryIT.java b/..._control_service/src/test/java/com/vmware/taurus/service/repository/JobsRepositoryIT.java
@@ -122,7 +122,6 @@ void testDeleteDataJob_withAssociatedDeployment_dataJobAndDeploymentShouldBeDele
 
     ActualDataJobDeployment expectedDataJobDeployment = new ActualDataJobDeployment();
     expectedDataJobDeployment.setDataJobName(dataJob.getName());
-    expectedDataJobDeployment.setDataJob(dataJob);
     expectedDataJobDeployment.setPythonVersion("3.9-secure");
     expectedDataJobDeployment.setGitCommitSha("commit");
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;