control-service: fix data job image building

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

@@ -183,7 +183,7 @@ public void updateDeployment(
               jobDeployment.getDataJobName(), jobDeployment.getGitCommitSha());
 
       if (jobImageBuilder.buildImage(
-          imageName, dataJob, toDesiredDataJobDeployment(jobDeployment), sendNotification)) {
+          imageName, dataJob, toDesiredDataJobDeployment(jobDeployment), null, sendNotification)) {
         log.info(
             "Image {} has been built. Will now schedule job {} for execution",
             imageName,

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

@@ -160,7 +160,8 @@ public void updateDeployment(
           dockerRegistryService.dataJobImage(
               desiredJobDeployment.getDataJobName(), desiredJobDeployment.getGitCommitSha());
 
-      if (jobImageBuilder.buildImage(imageName, dataJob, desiredJobDeployment, sendNotification)) {
+      if (jobImageBuilder.buildImage(
+          imageName, dataJob, desiredJobDeployment, actualJobDeployment, sendNotification)) {
         ActualDataJobDeployment actualJobDeploymentResult =
             jobImageDeployer.scheduleJob(
                 dataJob,

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

@@ -11,6 +11,7 @@
 import com.vmware.taurus.exception.KubernetesException;
 import com.vmware.taurus.service.credentials.AWSCredentialsService;
 import com.vmware.taurus.service.kubernetes.ControlKubernetesService;
+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;
@@ -113,7 +114,8 @@ public JobImageBuilder(
    *
    * @param imageName Full name of the image to build.
    * @param dataJob Information about the data job.
-   * @param jobDeployment Information about the data job deployment.
+   * @param desiredDataJobDeployment Information about the desired data job deployment.
+   * @param actualDataJobDeployment Information about the actual data job deployment.
    * @param sendNotification
    * @return True if build and push was successful. False otherwise.
    * @throws ApiException
@@ -123,7 +125,8 @@ public JobImageBuilder(
   public boolean buildImage(
       String imageName,
       DataJob dataJob,
-      DesiredDataJobDeployment jobDeployment,
+      DesiredDataJobDeployment desiredDataJobDeployment,
+      ActualDataJobDeployment actualDataJobDeployment,
       Boolean sendNotification)
       throws ApiException, IOException, InterruptedException {
     // TODO: refactor and hide AWS details behind DockerRegistryService?
@@ -145,17 +148,22 @@ public boolean buildImage(
       }
     }
 
-    if (jobDeployment.getPythonVersion() == null) {
+    if (desiredDataJobDeployment.getPythonVersion() == null) {
       log.warn("Missing pythonVersion. Data Job cannot be deployed.");
       return false;
     }
 
-    if (dockerRegistryService.dataJobImageExists(imageName, credentials)) {
+    // Rebuild the image if the Python version changes but the gitCommitSha remains the same.
+    if ((actualDataJobDeployment == null
+            || desiredDataJobDeployment
+                .getPythonVersion()
+                .equals(actualDataJobDeployment.getPythonVersion()))
+        && dockerRegistryService.dataJobImageExists(imageName, credentials)) {
       log.trace("Data Job image {} already exists and nothing else to do.", imageName);
       return true;
     }
 
-    String builderJobName = getBuilderJobName(jobDeployment.getDataJobName());
+    String builderJobName = getBuilderJobName(desiredDataJobDeployment.getDataJobName());
 
     log.debug("Check if old builder job {} exists", builderJobName);
     if (controlKubernetesService.listJobs().contains(builderJobName)) {
@@ -187,13 +195,14 @@ public boolean buildImage(
             registryUsername,
             registryPassword,
             builderAwsSessionToken);
-    var envs = getBuildParameters(dataJob, jobDeployment);
-    String builderImage = supportedPythonVersions.getBuilderImage(jobDeployment.getPythonVersion());
+    var envs = getBuildParameters(dataJob, desiredDataJobDeployment);
+    String builderImage =
+        supportedPythonVersions.getBuilderImage(desiredDataJobDeployment.getPythonVersion());
 
     log.info(
         "Creating builder job {} for data job version {}",
         builderJobName,
-        jobDeployment.getGitCommitSha());
+        desiredDataJobDeployment.getGitCommitSha());
     controlKubernetesService.createJob(
         builderJobName,
         builderImage,
@@ -215,7 +224,7 @@ public boolean buildImage(
     log.debug(
         "Waiting for builder job {} for data job version {}",
         builderJobName,
-        jobDeployment.getGitCommitSha());
+        desiredDataJobDeployment.getGitCommitSha());
 
     var condition =
         controlKubernetesService.watchJob(
@@ -234,7 +243,7 @@ public boolean buildImage(
     }
     if (!condition.isSuccess()) {
       notificationHelper.verifyBuilderResult(
-          builderJobName, dataJob, jobDeployment, condition, logs, sendNotification);
+          builderJobName, dataJob, desiredDataJobDeployment, condition, logs, sendNotification);
     } else {
       log.info("Builder job {} finished successfully. Will delete it now", builderJobName);
       log.info(

projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/DeploymentServiceTest.java

@@ -158,6 +158,7 @@ public void updateDeployment_newDeploymentCreated()
             TEST_JOB_IMAGE_NAME,
             testDataJob,
             DeploymentModelConverter.toDesiredDataJobDeployment(jobDeployment),
+            null,
             true))
         .thenReturn(true);
 
@@ -170,6 +171,7 @@ public void updateDeployment_newDeploymentCreated()
             TEST_JOB_IMAGE_NAME,
             testDataJob,
             DeploymentModelConverter.toDesiredDataJobDeployment(jobDeployment),
+            null,
             true);
     verify(kubernetesService)
         .createCronJob(
@@ -207,6 +209,7 @@ public void updateDeployment_existingDeploymentUpdated()
             TEST_JOB_IMAGE_NAME,
             testDataJob,
             DeploymentModelConverter.toDesiredDataJobDeployment(jobDeployment),
+            null,
             true))
         .thenReturn(true);
     when(kubernetesService.listCronJobs()).thenReturn(Set.of(TEST_CRONJOB_NAME));
@@ -220,6 +223,7 @@ public void updateDeployment_existingDeploymentUpdated()
             TEST_JOB_IMAGE_NAME,
             testDataJob,
             DeploymentModelConverter.toDesiredDataJobDeployment(jobDeployment),
+            null,
             true);
     verify(kubernetesService)
         .updateCronJob(
@@ -254,6 +258,7 @@ public void updateDeployment_failedToBuildImage_deploymentSkipped()
             TEST_JOB_IMAGE_NAME,
             testDataJob,
             DeploymentModelConverter.toDesiredDataJobDeployment(jobDeployment),
+            null,
             true))
         .thenReturn(false);
 
@@ -266,6 +271,7 @@ public void updateDeployment_failedToBuildImage_deploymentSkipped()
             TEST_JOB_IMAGE_NAME,
             testDataJob,
             DeploymentModelConverter.toDesiredDataJobDeployment(jobDeployment),
+            null,
             true);
     verify(kubernetesService, never())
         .updateCronJob(

projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/DeploymentServiceV2TestIT.java

@@ -56,7 +56,7 @@ public void updateDeployment_withDesiredDeploymentStatusNone_shouldStartDeployme
     updateDeployment(DeploymentStatus.NONE, 1, true);
 
     Mockito.verify(jobImageBuilder, Mockito.times(1))
-        .buildImage(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.eq(true));
+        .buildImage(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.eq(true));
   }
 
   @Test
@@ -66,7 +66,7 @@ public void updateDeployment_withDesiredDeploymentStatusNone_shouldStartDeployme
     updateDeployment(DeploymentStatus.NONE, 1, false);
 
     Mockito.verify(jobImageBuilder, Mockito.times(1))
-        .buildImage(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.eq(false));
+        .buildImage(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.eq(false));
   }
 
   private void updateDeployment(
@@ -86,7 +86,8 @@ private void updateDeployment(
     DataJob dataJob = new DataJob();
     dataJob.setJobConfig(new JobConfig());
     Mockito.when(
-            jobImageBuilder.buildImage(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()))
+            jobImageBuilder.buildImage(
+                Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any()))
         .thenReturn(false);
 
     deploymentService.updateDeployment(

projects/control-service/projects/pipelines_control_service/src/test/java/com/vmware/taurus/service/deploy/JobImageBuilderTest.java

@@ -20,6 +20,7 @@
 import com.vmware.taurus.service.credentials.AWSCredentialsService;
 import com.vmware.taurus.service.credentials.AWSCredentialsService.AWSCredentialsDTO;
 import com.vmware.taurus.service.kubernetes.ControlKubernetesService;
+import com.vmware.taurus.service.model.ActualDataJobDeployment;
 import com.vmware.taurus.service.model.DataJob;
 import com.vmware.taurus.service.model.DesiredDataJobDeployment;
 import com.vmware.taurus.service.model.JobConfig;
@@ -99,7 +100,7 @@ public void buildImage_notExist_success() throws InterruptedException, ApiExcept
     jobDeployment.setEnabled(true);
     jobDeployment.setPythonVersion("3.7");
 
-    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, true);
+    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, null, true);
 
     verify(kubernetesService)
         .createJob(
@@ -143,7 +144,8 @@ public void buildImage_builderRunning_oldBuilderDeleted()
     jobDeployment.setEnabled(true);
     jobDeployment.setPythonVersion("3.7");
 
-    var result = jobImageBuilder.buildImage(TEST_IMAGE_NAME, testDataJob, jobDeployment, true);
+    var result =
+        jobImageBuilder.buildImage(TEST_IMAGE_NAME, testDataJob, jobDeployment, null, true);
 
     verify(kubernetesService, times(2)).deleteJob(TEST_BUILDER_IMAGE_NAME);
     verify(kubernetesService)
@@ -179,7 +181,8 @@ public void buildImage_imageExists_buildSkipped()
     jobDeployment.setEnabled(true);
     jobDeployment.setPythonVersion("3.7");
 
-    var result = jobImageBuilder.buildImage(TEST_IMAGE_NAME, testDataJob, jobDeployment, true);
+    var result =
+        jobImageBuilder.buildImage(TEST_IMAGE_NAME, testDataJob, jobDeployment, null, true);
 
     verify(kubernetesService, never())
         .createJob(
@@ -221,7 +224,7 @@ public void buildImage_jobFailed_failure()
     jobDeployment.setEnabled(true);
     jobDeployment.setPythonVersion("3.7");
 
-    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, true);
+    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, null, true);
 
     verify(kubernetesService)
         .createJob(
@@ -273,7 +276,7 @@ public void buildImage_jobFailed_failure()
 
     ArgumentCaptor<Map<String, String>> captor = ArgumentCaptor.forClass(Map.class);
 
-    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, true);
+    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, null, true);
 
     verify(kubernetesService)
         .createJob(
@@ -310,7 +313,7 @@ public void buildImage_PythonVersionNull_shouldNotCreateCronjob()
     jobDeployment.setGitCommitSha("test-commit");
     jobDeployment.setEnabled(true);
 
-    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, true);
+    var result = jobImageBuilder.buildImage("test-image", testDataJob, jobDeployment, null, true);
 
     verify(supportedPythonVersions, never()).isPythonVersionSupported("3.11");
     verify(supportedPythonVersions, never()).getJobBaseImage("3.11");
@@ -337,6 +340,94 @@ public void buildImage_PythonVersionNull_shouldNotCreateCronjob()
     Assertions.assertFalse(result);
   }
 
+  @Test
+  public void buildImage_imageExistsAndEqualPythonVersions_shouldSkipBuild()
+      throws InterruptedException, ApiException, IOException {
+    when(dockerRegistryService.dataJobImageExists(eq(TEST_IMAGE_NAME), Mockito.any()))
+        .thenReturn(true);
+
+    DesiredDataJobDeployment jobDeployment = new DesiredDataJobDeployment();
+    jobDeployment.setDataJobName(TEST_JOB_NAME);
+    jobDeployment.setGitCommitSha("test-commit");
+    jobDeployment.setEnabled(true);
+    jobDeployment.setPythonVersion("3.7");
+
+    ActualDataJobDeployment actualDataJobDeployment = new ActualDataJobDeployment();
+    actualDataJobDeployment.setPythonVersion("3.7");
+
+    var result =
+        jobImageBuilder.buildImage(
+            TEST_IMAGE_NAME, testDataJob, jobDeployment, actualDataJobDeployment, true);
+
+    verify(kubernetesService, never())
+        .createJob(
+            anyString(),
+            anyString(),
+            anyBoolean(),
+            anyBoolean(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            anyLong(),
+            anyLong(),
+            anyLong(),
+            anyString(),
+            anyString());
+    verify(notificationHelper, never())
+        .verifyBuilderResult(anyString(), any(), any(), any(), anyString(), anyBoolean());
+    Assertions.assertTrue(result);
+  }
+
+  @Test
+  public void buildImage_imageExistsAndDifferentPythonVersions_shouldSucceed()
+      throws InterruptedException, ApiException, IOException {
+    when(kubernetesService.listJobs()).thenReturn(Collections.emptySet());
+    var builderJobResult =
+        new KubernetesService.JobStatusCondition(true, "type", "test-reason", "test-message", 0);
+    when(kubernetesService.watchJob(any(), anyInt(), any())).thenReturn(builderJobResult);
+    when(supportedPythonVersions.getJobBaseImage(any())).thenReturn("python:3.7-slim");
+    when(supportedPythonVersions.getBuilderImage(any())).thenReturn(TEST_BUILDER_IMAGE_NAME);
+
+    DesiredDataJobDeployment jobDeployment = new DesiredDataJobDeployment();
+    jobDeployment.setDataJobName(TEST_JOB_NAME);
+    jobDeployment.setGitCommitSha("test-commit");
+    jobDeployment.setEnabled(true);
+    jobDeployment.setPythonVersion("3.8");
+
+    ActualDataJobDeployment actualDataJobDeployment = new ActualDataJobDeployment();
+    actualDataJobDeployment.setPythonVersion("3.7");
+
+    var result =
+        jobImageBuilder.buildImage(
+            TEST_IMAGE_NAME, testDataJob, jobDeployment, actualDataJobDeployment, true);
+
+    verify(kubernetesService)
+        .createJob(
+            eq(TEST_BUILDER_JOB_NAME),
+            eq(TEST_BUILDER_IMAGE_NAME),
+            eq(false),
+            eq(false),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            anyLong(),
+            anyLong(),
+            anyLong(),
+            any(),
+            any());
+
+    verify(kubernetesService).deleteJob(TEST_BUILDER_JOB_NAME);
+    Assertions.assertTrue(result);
+  }
+
   private static Map<String, Map<String, String>> generateSupportedPythonVersionsConf() {
     return Map.of(
         "3.10", Map.of("baseImage", "python:3.10-slim", "vdkImage", "test_vdk_image_3.10"),