feature: Make deployment replicas configurable on resource level

platform-api/src/main/java/io/datacater/core/config/ConfigUtilities.java

@@ -20,6 +20,11 @@ private ConfigUtilities() {}
 
   public static Uni<List<ConfigEntity>> getMappedConfigs(
       Map<String, String> configs, DataCaterSessionFactory dsf) {
+
+    if (configs == null) {
+      return Uni.createFrom().item(new ArrayList<>());
+    }
+
     return dsf.withTransaction(
         (session, transaction) ->
             session

platform-api/src/main/java/io/datacater/core/deployment/DeploymentEndpoint.java

@@ -9,10 +9,9 @@
 import io.datacater.core.pipeline.PipelineEntity;
 import io.datacater.core.stream.StreamEntity;
 import io.datacater.core.utilities.StringUtilities;
-import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.dsl.ContainerResource;
 import io.fabric8.kubernetes.client.dsl.LogWatch;
-import io.fabric8.kubernetes.client.dsl.RollableScalableResource;
 import io.quarkus.security.Authenticated;
 import io.smallrye.mutiny.Uni;
 import io.smallrye.mutiny.unchecked.Unchecked;
@@ -66,21 +65,24 @@ public Uni<DeploymentEntity> getDeployment(@PathParam("uuid") UUID deploymentId)
   @GET
   @Path("{uuid}/logs")
   @Produces(MediaType.APPLICATION_JSON)
-  public Uni<List<String>> getLogs(@PathParam("uuid") UUID deploymentId) {
+  public Uni<List<String>> getLogs(
+      @PathParam("uuid") UUID deploymentId, @DefaultValue("1") @QueryParam("replica") int replica) {
     return dsf.withTransaction(
             ((session, transaction) -> session.find(DeploymentEntity.class, deploymentId)))
         .onItem()
         .ifNull()
         .failWith(new DeploymentNotFoundException(StaticConfig.LoggerMessages.DEPLOYMENT_NOT_FOUND))
         .onItem()
         .ifNotNull()
-        .transform(Unchecked.function(deployment -> getDeploymentLogsAsList(deployment.getId())));
+        .transform(
+            Unchecked.function(deployment -> getDeploymentLogsAsList(deployment.getId(), replica)));
   }
 
   @GET
   @Path("{uuid}/health")
   @Produces(MediaType.APPLICATION_JSON)
-  public Uni<Response> getHealth(@PathParam("uuid") UUID deploymentId) {
+  public Uni<Response> getHealth(
+      @PathParam("uuid") UUID deploymentId, @DefaultValue("1") @QueryParam("replica") int replica) {
     return dsf.withTransaction(
             ((session, transaction) -> session.find(DeploymentEntity.class, deploymentId)))
         .onItem()
@@ -94,7 +96,9 @@ public Uni<Response> getHealth(@PathParam("uuid") UUID deploymentId) {
                   HttpClient httpClient = HttpClient.newHttpClient();
                   HttpRequest req =
                       buildDeploymentServiceRequest(
-                          deploymentId, StaticConfig.EnvironmentVariables.DEPLOYMENT_HEALTH_PATH);
+                          deploymentId,
+                          StaticConfig.EnvironmentVariables.DEPLOYMENT_HEALTH_PATH,
+                          replica);
                   HttpResponse<String> response =
                       httpClient.send(req, HttpResponse.BodyHandlers.ofString());
                   return Response.ok().entity(response.body()).build();
@@ -104,7 +108,8 @@ public Uni<Response> getHealth(@PathParam("uuid") UUID deploymentId) {
   @GET
   @Path("{uuid}/metrics")
   @Produces(MediaType.TEXT_PLAIN)
-  public Uni<Response> getMetrics(@PathParam("uuid") UUID deploymentId) {
+  public Uni<Response> getMetrics(
+      @PathParam("uuid") UUID deploymentId, @DefaultValue("1") @QueryParam("replica") int replica) {
     return dsf.withTransaction(
             ((session, transaction) -> session.find(DeploymentEntity.class, deploymentId)))
         .onItem()
@@ -118,7 +123,9 @@ public Uni<Response> getMetrics(@PathParam("uuid") UUID deploymentId) {
                   HttpClient httpClient = HttpClient.newHttpClient();
                   HttpRequest req =
                       buildDeploymentServiceRequest(
-                          deploymentId, StaticConfig.EnvironmentVariables.DEPLOYMENT_METRICS_PATH);
+                          deploymentId,
+                          StaticConfig.EnvironmentVariables.DEPLOYMENT_METRICS_PATH,
+                          replica);
                   HttpResponse<String> response =
                       httpClient.send(req, HttpResponse.BodyHandlers.ofString());
                   return Response.ok().entity(response.body()).build();
@@ -129,7 +136,10 @@ public Uni<Response> getMetrics(@PathParam("uuid") UUID deploymentId) {
   @Path("{uuid}/watch-logs")
   @Produces(MediaType.SERVER_SENT_EVENTS)
   public Uni<Response> watchLogs(
-      @PathParam("uuid") UUID deploymentId, @Context Sse sse, @Context SseEventSink eventSink) {
+      @PathParam("uuid") UUID deploymentId,
+      @DefaultValue("1") @QueryParam("replica") int replica,
+      @Context Sse sse,
+      @Context SseEventSink eventSink) {
     return dsf.withTransaction(
             ((session, transaction) -> session.find(DeploymentEntity.class, deploymentId)))
         .onItem()
@@ -140,7 +150,7 @@ public Uni<Response> watchLogs(
         .transform(
             deployment -> {
               try {
-                watchLogsRunner(deployment.getId(), sse, eventSink);
+                watchLogsRunner(deployment.getId(), replica, sse, eventSink);
               } catch (IOException e) {
                 throw new DatacaterException(StringUtilities.wrapString(e.getMessage()));
               }
@@ -336,21 +346,20 @@ private Uni<StreamEntity> getStream(
                         String.format(StaticConfig.LoggerMessages.STREAM_NOT_FOUND, key))));
   }
 
-  private List<String> getDeploymentLogsAsList(UUID deploymentId) {
+  private List<String> getDeploymentLogsAsList(UUID deploymentId, int replica) {
     K8Deployment k8Deployment = new K8Deployment(client);
-    return Arrays.asList(k8Deployment.getLogs(deploymentId).split("\n"));
+    return Arrays.asList(k8Deployment.getLogs(deploymentId, replica).split("\n"));
   }
 
-  private HttpRequest buildDeploymentServiceRequest(UUID deploymentId, String path) {
+  private HttpRequest buildDeploymentServiceRequest(UUID deploymentId, String path, int replica) {
     K8Deployment k8Deployment = new K8Deployment(client);
-    String clusterIp = k8Deployment.getClusterIp(deploymentId);
+    String ip = k8Deployment.getDeploymentReplicaIp(deploymentId, replica).replace(".", "-");
+    String namespace = StaticConfig.EnvironmentVariables.NAMESPACE;
+    int port = StaticConfig.EnvironmentVariables.DEPLOYMENT_CONTAINER_PORT;
+    String protocol = StaticConfig.EnvironmentVariables.DEPLOYMENT_CONTAINER_PROTOCOL;
+
     String uriReady =
-        String.format(
-            "%s://%s:%d%s",
-            StaticConfig.EnvironmentVariables.DEPLOYMENT_CONTAINER_PROTOCOL,
-            clusterIp,
-            StaticConfig.EnvironmentVariables.DEPLOYMENT_CONTAINER_PORT,
-            path);
+        String.format("%s://%s.%s.pod.cluster.local:%d%s", protocol, ip, namespace, port, path);
 
     return HttpRequest.newBuilder()
         .GET()
@@ -359,9 +368,9 @@ private HttpRequest buildDeploymentServiceRequest(UUID deploymentId, String path
         .build();
   }
 
-  private RollableScalableResource<Deployment> watchDeploymentLogs(UUID deploymentId) {
+  private ContainerResource watchDeploymentLogs(UUID deploymentId, int replica) {
     K8Deployment k8Deployment = new K8Deployment(client);
-    return k8Deployment.watchLogs(deploymentId);
+    return k8Deployment.watchLogs(deploymentId, replica);
   }
 
   private List<DeploymentEntity> getK8Deployments(List<DeploymentEntity> deployments) {
@@ -452,12 +461,10 @@ private UUID getPipelineUUIDFromMap(Map<String, Object> map) {
     }
   }
 
-  private void watchLogsRunner(UUID deploymentId, @Context Sse sse, @Context SseEventSink eventSink)
+  private void watchLogsRunner(
+      UUID deploymentId, int replica, @Context Sse sse, @Context SseEventSink eventSink)
       throws IOException {
-    LogWatch lw =
-        watchDeploymentLogs(deploymentId)
-            .inContainer(StaticConfig.DEPLOYMENT_NAME_PREFIX + deploymentId)
-            .watchLog();
+    LogWatch lw = watchDeploymentLogs(deploymentId, replica).watchLog();
     InputStream is = lw.getOutput();
     BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(is));
     String line;