6671: Initial push for uploading rebublished docker images

java/clients/src/main/java/sleeper/clients/admin/AdminClient.java

@@ -33,7 +33,6 @@
 import sleeper.clients.admin.screen.IngestStatusReportScreen;
 import sleeper.clients.admin.screen.InstanceConfigurationScreen;
 import sleeper.clients.admin.screen.PartitionsStatusReportScreen;
-import sleeper.clients.deploy.container.CheckVersionExistsInEcr;
 import sleeper.clients.deploy.container.DockerImageConfiguration;
 import sleeper.clients.deploy.container.UploadDockerImages;
 import sleeper.clients.deploy.container.UploadDockerImagesToEcr;
@@ -102,8 +101,7 @@ public static void main(String[] args) throws IOException, InterruptedException
                 StsClient stsClient = AwsV2ClientHelper.buildAwsV2Client(StsClient.builder())) {
             AwsRegionProvider regionProvider = DefaultAwsRegionProviderChain.builder().build();
             UploadDockerImagesToEcr uploadDockerImages = new UploadDockerImagesToEcr(
-                    UploadDockerImages.fromScriptsDirectory(scriptsDir),
-                    CheckVersionExistsInEcr.withEcrClient(ecrClient),
+                    UploadDockerImages.fromScriptsDirectory(scriptsDir), //6671 TODO do we need to pass in an ecr client
                     stsClient.getCallerIdentity().account(), regionProvider.getRegion().id());
             errorCode = start(instanceId, s3Client, dynamoClient, cdk, generatedDir,
                     uploadDockerImages, DockerImageConfiguration.getDefault(), out, in,

java/clients/src/main/java/sleeper/clients/deploy/DeployExistingInstance.java

@@ -25,7 +25,7 @@
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.sts.StsClient;
 
-import sleeper.clients.deploy.container.CheckVersionExistsInEcr;
+import sleeper.clients.deploy.container.CheckDigestExistsInEcr;
 import sleeper.clients.deploy.container.UploadDockerImages;
 import sleeper.clients.deploy.container.UploadDockerImagesToEcr;
 import sleeper.clients.deploy.jar.SyncJars;
@@ -107,10 +107,11 @@ public void update() throws IOException, InterruptedException {
                         UploadDockerImages.builder()
                                 .scriptsDirectory(scriptsDirectory)
                                 .deployConfig(DeployConfiguration.fromScriptsDirectory(scriptsDirectory))
+                                .repositoryChecker(CheckDigestExistsInEcr.withEcrClient(ecr))
                                 .commandRunner(runCommand)
                                 .createMultiplatformBuilder(createMultiPlatformBuilder)
                                 .build(),
-                        CheckVersionExistsInEcr.withEcrClient(ecr), sts.getCallerIdentity().account(), regionProvider.getRegion().id()),
+                        sts.getCallerIdentity().account(), regionProvider.getRegion().id()),
                 DeployInstance.WriteLocalProperties.underScriptsDirectory(scriptsDirectory),
                 InvokeCdk.builder().scriptsDirectory(scriptsDirectory).runCommand(runCommand).build());
 

java/clients/src/main/java/sleeper/clients/deploy/DeployNewInstance.java

@@ -24,7 +24,7 @@
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.sts.StsClient;
 
-import sleeper.clients.deploy.container.CheckVersionExistsInEcr;
+import sleeper.clients.deploy.container.CheckDigestExistsInEcr;
 import sleeper.clients.deploy.container.StackDockerImage;
 import sleeper.clients.deploy.container.UploadDockerImages;
 import sleeper.clients.deploy.container.UploadDockerImagesToEcr;
@@ -122,10 +122,11 @@ public void deploy() throws IOException, InterruptedException {
                         UploadDockerImages.builder()
                                 .scriptsDirectory(scriptsDirectory)
                                 .deployConfig(DeployConfiguration.fromScriptsDirectory(scriptsDirectory))
+                                .repositoryChecker(CheckDigestExistsInEcr.withEcrClient(ecrClient))
                                 .commandRunner(runCommand)
                                 .createMultiplatformBuilder(createMultiPlatformBuilder)
                                 .build(),
-                        CheckVersionExistsInEcr.withEcrClient(ecrClient), stsClient.getCallerIdentity().account(), regionProvider.getRegion().id()),
+                        stsClient.getCallerIdentity().account(), regionProvider.getRegion().id()),
                 DeployInstance.WriteLocalProperties.underScriptsDirectory(scriptsDirectory),
                 InvokeCdk.builder().scriptsDirectory(scriptsDirectory).runCommand(runCommand).build());
 

java/clients/src/main/java/sleeper/clients/deploy/UploadArtefacts.java

@@ -21,7 +21,7 @@
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.sts.StsClient;
 
-import sleeper.clients.deploy.container.CheckVersionExistsInEcr;
+import sleeper.clients.deploy.container.CheckDigestExistsInEcr;
 import sleeper.clients.deploy.container.DockerImageConfiguration;
 import sleeper.clients.deploy.container.StackDockerImage;
 import sleeper.clients.deploy.container.UploadDockerImages;
@@ -153,9 +153,10 @@ public static void main(String[] rawArgs) throws IOException, InterruptedExcepti
                     UploadDockerImages.builder()
                             .scriptsDirectory(args.scriptsDir())
                             .deployConfig(DeployConfiguration.fromScriptsDirectory(args.scriptsDir()))
+                            .repositoryChecker(CheckDigestExistsInEcr.withEcrClient(ecrClient))
                             .createMultiplatformBuilder(args.createMultiplatformBuilder())
                             .build(),
-                    CheckVersionExistsInEcr.withEcrClient(ecrClient), account, region);
+                    account, region);
 
             if (args.createDeployment()) {
                 InvokeCdk.fromScriptsDirectory(args.scriptsDir())

java/clients/src/main/java/sleeper/clients/deploy/container/CheckVersionExistsInEcr.java → java/clients/src/main/java/sleeper/clients/deploy/container/CheckDigestExistsInEcr.java

@@ -21,18 +21,18 @@
 
 import java.util.Objects;
 
-public interface CheckVersionExistsInEcr {
+public interface CheckDigestExistsInEcr {
 
-    boolean versionExistsInRepository(String repository, String version);
+    boolean digestExistsInRepository(String repository, String digest);
 
-    static CheckVersionExistsInEcr withEcrClient(EcrClient ecrClient) {
-        return (repository, version) -> {
+    static CheckDigestExistsInEcr withEcrClient(EcrClient ecrClient) {
+        return (repository, digest) -> {
             try {
                 ListImagesResponse response = ecrClient.listImages(request -> request.repositoryName(repository));
                 // Note that the image tag can be null. In particular, a multiplatform image has multiple images but
-                // only the image index is tagged.
+                // only the image index is tagged. 6671 TODO - Is this comment still relevant
                 return response.imageIds().stream()
-                        .anyMatch(imageIdentifier -> Objects.equals(version, imageIdentifier.imageTag()));
+                        .anyMatch(imageIdentifier -> Objects.equals(digest, imageIdentifier.imageDigest()));
             } catch (RepositoryNotFoundException e) {
                 return false;
             }

java/clients/src/main/java/sleeper/clients/deploy/container/UploadDockerImages.java

@@ -29,6 +29,7 @@
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
 
 import static java.util.Objects.requireNonNull;
 
@@ -37,6 +38,7 @@ public class UploadDockerImages {
     private final Path baseDockerDirectory;
     private final Path jarsDirectory;
     private final DeployConfiguration deployConfig;
+    private final CheckDigestExistsInEcr repositoryChecker;
     private final CommandPipelineRunner commandRunner;
     private final CopyFile copyFile;
     private final String version;
@@ -46,6 +48,7 @@ private UploadDockerImages(Builder builder) {
         baseDockerDirectory = requireNonNull(builder.baseDockerDirectory, "baseDockerDirectory must not be null");
         jarsDirectory = requireNonNull(builder.jarsDirectory, "jarsDirectory must not be null");
         deployConfig = requireNonNull(builder.deployConfig, "deployConfig must not be null");
+        repositoryChecker = requireNonNull(builder.repositoryChecker, "repositoryChecker must not be null");
         commandRunner = requireNonNull(builder.commandRunner, "commandRunner must not be null");
         copyFile = requireNonNull(builder.copyFile, "copyFile must not be null");
         version = requireNonNull(builder.version, "version must not be null");
@@ -63,7 +66,7 @@ public static UploadDockerImages fromScriptsDirectory(Path scriptsDirectory) thr
                 .build();
     }
 
-    public void upload(String repositoryPrefix, List<StackDockerImage> imagesToUpload) throws IOException, InterruptedException {
+    public void upload(String repositoryPrefix, List<StackDockerImage> imagesToUpload, boolean overwriteExisting) throws IOException, InterruptedException {
         if (imagesToUpload.isEmpty()) {
             LOGGER.info("No images need to be built and uploaded, skipping");
             return;
@@ -107,21 +110,55 @@ private void buildAndPushImage(String tag, StackDockerImage image) throws IOExce
             } else {
                 commandRunner.runOrThrow("docker", "build", "-t", tag, dockerfileDirectory.toString());
             }
+
             commandRunner.runOrThrow("docker", "push", tag);
         }
     }
 
     private void pullAndPushImage(String tag, StackDockerImage image) throws IOException, InterruptedException {
         String sourceTag = buildTag(deployConfig.dockerRepositoryPrefix(), image);
         commandRunner.runOrThrow("docker", "pull", sourceTag);
-        commandRunner.runOrThrow("docker", "tag", sourceTag, tag);
-        commandRunner.runOrThrow("docker", "push", tag);
+        String digest = getDigestForImage(image);
+        if (imageDoesNotExistInRepositoryWithDigest(image, null, digest)) {
+            commandRunner.runOrThrow("docker", "tag", sourceTag, tag);
+            commandRunner.runOrThrow("docker", "push", tag);
+        }
     }
 
     private String buildTag(String repositoryPrefix, StackDockerImage image) {
         return repositoryPrefix + "/" + image.getImageName() + ":" + version;
     }
 
+    private String getDigestForImage(StackDockerImage image) {
+        try {
+            return CommandUtils.runCommandReturnOutput("docker", "images",
+                    "--filter", "reference=" + image.getDirectoryName(), //6671 TODO - directory name might not be the correct one
+                    "--format", "{{.Repository}}: {{.Digest}}").get(0);
+        } catch (IOException | InterruptedException | ExecutionException e) {
+            // 6671 TODO - This exception handling could be moved into CommandUtils
+            // and/or we could just not check ECR for the digest if not found. Doing so would probably help 5852
+            return "";
+        }
+    }
+
+    private boolean imageDoesNotExistInRepositoryWithDigest(
+            StackDockerImage stackDockerImage, UploadDockerImagesToEcrRequest request, String digest) {
+        if (request.isOverwriteExistingTag()) {
+            return true;
+        }
+
+        String repository = request.getEcrPrefix() + "/" + stackDockerImage.getImageName();
+        if (repositoryChecker.digestExistsInRepository(repository, digest)) {
+            LOGGER.info("ECR repository {} already contains digest {}",
+                    repository, digest);
+            return false;
+        } else {
+            LOGGER.info("ECR repository {} does not contain version {}",
+                    repository, digest);
+            return true;
+        }
+    }
+
     public CommandPipelineRunner getCommandRunner() {
         return commandRunner;
     }
@@ -134,6 +171,7 @@ public static final class Builder {
         private Path baseDockerDirectory;
         private Path jarsDirectory;
         private DeployConfiguration deployConfig;
+        private CheckDigestExistsInEcr repositoryChecker;
         private CommandPipelineRunner commandRunner = CommandUtils::runCommandInheritIO;
         private CopyFile copyFile = (source, target) -> Files.copy(source, target, StandardCopyOption.REPLACE_EXISTING);
         private String version = SleeperVersion.getVersion();
@@ -162,6 +200,11 @@ public Builder deployConfig(DeployConfiguration deployConfig) {
             return this;
         }
 
+        public Builder repositoryChecker(CheckDigestExistsInEcr repositoryChecker) {
+            this.repositoryChecker = repositoryChecker;
+            return this;
+        }
+
         public Builder commandRunner(CommandPipelineRunner commandRunner) {
             this.commandRunner = commandRunner;
             return this;

java/clients/src/main/java/sleeper/clients/deploy/container/UploadDockerImagesToEcr.java

@@ -20,7 +20,6 @@
 
 import java.io.IOException;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import static sleeper.clients.util.command.Command.command;
 import static sleeper.clients.util.command.CommandPipeline.pipeline;
@@ -33,14 +32,12 @@ public class UploadDockerImagesToEcr {
     public static final Logger LOGGER = LoggerFactory.getLogger(UploadDockerImagesToEcr.class);
 
     private final UploadDockerImages uploader;
-    private final CheckVersionExistsInEcr repositoryChecker;
     private final String account;
     private final String region;
     private final String repositoryHost;
 
-    public UploadDockerImagesToEcr(UploadDockerImages uploader, CheckVersionExistsInEcr repositoryChecker, String account, String region) {
+    public UploadDockerImagesToEcr(UploadDockerImages uploader, String account, String region) {
         this.uploader = uploader;
-        this.repositoryChecker = repositoryChecker;
         this.account = account;
         this.region = region;
         this.repositoryHost = String.format("%s.dkr.ecr.%s.amazonaws.com", this.account, this.region);
@@ -49,33 +46,12 @@ public UploadDockerImagesToEcr(UploadDockerImages uploader, CheckVersionExistsIn
     public void upload(UploadDockerImagesToEcrRequest request) throws IOException, InterruptedException {
         List<StackDockerImage> requestedImages = request.getImages();
         LOGGER.info("Images expected: {}", requestedImages);
-        List<StackDockerImage> imagesToUpload = requestedImages.stream()
-                .filter(image -> imageDoesNotExistInRepositoryWithVersion(image, request))
-                .collect(Collectors.toUnmodifiableList());
         String repositoryPrefix = repositoryHost + "/" + request.getEcrPrefix();
-        if (!imagesToUpload.isEmpty()) {
+        if (requestedImages.isEmpty()) {
             uploader.getCommandRunner().runOrThrow(pipeline(
                     command("aws", "ecr", "get-login-password", "--region", region),
                     command("docker", "login", "--username", "AWS", "--password-stdin", repositoryHost)));
         }
-        uploader.upload(repositoryPrefix, imagesToUpload);
+        uploader.upload(repositoryPrefix, requestedImages, request.isOverwriteExistingTag());
     }
-
-    private boolean imageDoesNotExistInRepositoryWithVersion(
-            StackDockerImage stackDockerImage, UploadDockerImagesToEcrRequest request) {
-        if (request.isOverwriteExistingTag()) {
-            return true;
-        }
-        String repository = request.getEcrPrefix() + "/" + stackDockerImage.getImageName();
-        if (repositoryChecker.versionExistsInRepository(repository, uploader.getVersion())) {
-            LOGGER.info("ECR repository {} already contains version {}",
-                    repository, uploader.getVersion());
-            return false;
-        } else {
-            LOGGER.info("ECR repository {} does not contain version {}",
-                    repository, uploader.getVersion());
-            return true;
-        }
-    }
-
 }

java/clients/src/main/java/sleeper/clients/deploy/container/UploadDockerImagesToRepository.java

@@ -41,13 +41,14 @@ public static void main(String[] args) throws Exception {
         UploadDockerImages uploader = UploadDockerImages.builder()
                 .scriptsDirectory(scriptsDirectory)
                 .deployConfig(DeployConfiguration.fromLocalBuild())
+                //6671 TODO - We need an ecrClient here
                 .createMultiplatformBuilder(createMultiplatformBuilder)
                 .build();
         uploadAllImages(DockerImageConfiguration.getDefault(), uploader, repositoryPrefix);
     }
 
     public static void uploadAllImages(DockerImageConfiguration imageConfig, UploadDockerImages uploader, String repositoryPrefix) throws IOException, InterruptedException {
-        uploader.upload(repositoryPrefix, imageConfig.getAllImagesToUpload());
+        uploader.upload(repositoryPrefix, imageConfig.getAllImagesToUpload(), true); //6671 TODO - We may want to have the user pass in this
     }
 
 }

java/clients/src/main/java/sleeper/clients/util/command/CommandPipeline.java

@@ -50,6 +50,20 @@ public List<Process> startProcesses() throws IOException {
         }
     }
 
+    public List<Process> startMultipleProcesses() throws IOException {
+        if (commands.size() == 1) {
+            return startProcesses();
+        } else {
+            List<Process> processes = new ArrayList<>(commands.size());
+            for (Command command : commands) {
+                ProcessBuilder processBuilder = command.toProcessBuilder();
+                Process process = processBuilder.start();
+                processes.add(process);
+            }
+            return processes;
+        }
+    }
+
     public List<Process> startProcessesInheritIO() throws IOException {
         int size = commands.size();
         if (size == 1) {

java/clients/src/main/java/sleeper/clients/util/command/CommandUtils.java

@@ -27,9 +27,13 @@
 import java.io.InputStreamReader;
 import java.io.UncheckedIOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 
 import static sleeper.clients.util.command.Command.command;
 import static sleeper.clients.util.command.CommandPipeline.pipeline;
@@ -95,6 +99,47 @@ private static void logTo(InputStream stream, Consumer<String> logLine) {
         }
     }
 
+    public static List<String> runCommandReturnOutput(String... commands) throws IOException, InterruptedException, ExecutionException {
+        return runCommandReturnOutput(pipeline(command(commands)));
+    }
+
+    public static List<String> runCommandReturnOutput(CommandPipeline pipeline) throws IOException, InterruptedException, ExecutionException {
+        LOGGER.info("Running command: {}", pipeline);
+        return pipeline.startProcesses().stream()
+                .map(CommandUtils::returnOutput)
+                .map(CompletableFuture::join)
+                .flatMap(Collection::stream)
+                .collect(Collectors.toList());
+    }
+
+    public static List<String> runMultipleCommandReturnOutput(CommandPipeline pipeline) throws IOException, InterruptedException, ExecutionException {
+        LOGGER.info("Running commands: {}", pipeline);
+        return pipeline.startMultipleProcesses().stream()
+                .map(CommandUtils::returnOutput)
+                .map(CompletableFuture::join)
+                .flatMap(Collection::stream)
+                .collect(Collectors.toList());
+    }
+
+    private static CompletableFuture<List<String>> returnOutput(Process process) {
+        return CompletableFuture.supplyAsync(() -> returnOutput(process.getInputStream()));
+    }
+
+    private static List<String> returnOutput(InputStream stream) {
+        List<String> output = new ArrayList<>();
+        try {
+            BufferedReader reader = new BufferedReader(new InputStreamReader(stream, StandardCharsets.UTF_8));
+            String line = reader.readLine();
+            while (line != null) {
+                output.add(line);
+                line = reader.readLine();
+            }
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        return output;
+    }
+
     public static int runCommandInheritIO(String... command) throws IOException, InterruptedException {
         return runCommandInheritIO(pipeline(command(command))).getLastExitCode();
     }