[CYB - 194][UI] Git support for config pipeline. (#84)

cloudera · Nov 27, 2024 · 8748f24 · 8748f24
1 parent ef5a5ca
commit 8748f24
Show file tree

Hide file tree

Showing 75 changed files with 1,963 additions and 531 deletions.
diff --git a/flink-cyber/caracal-generator/pom.xml b/flink-cyber/caracal-generator/pom.xml
@@ -28,6 +28,12 @@
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-streaming-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-clients</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-kafka</artifactId>

diff --git a/...yber/caracal-generator/src/test/java/com/cloudera/cyber/test/TestCaracalGeneratorJob.java b/...yber/caracal-generator/src/test/java/com/cloudera/cyber/test/TestCaracalGeneratorJob.java
@@ -65,7 +65,7 @@ public void testCustomGenerator() throws Exception {
         int count = 100;
         JobTester.startTest(createPipeline(ParameterTool.fromMap(ImmutableMap.of(
                 PARAMS_RECORDS_LIMIT, String.valueOf(count),
-                PARAMS_GENERATOR_CONFIG, Objects.requireNonNull(getClass().getClassLoader().getResource("config/generator_config.json")).toExternalForm()
+                PARAMS_GENERATOR_CONFIG, Objects.requireNonNull(getClass().getClassLoader().getResource("config/generator_config_test.json")).toExternalForm()
         ))));
 
         JobTester.stopTest();

diff --git a/...st/resources/config/generator_config.json → ...sources/config/generator_config_test.json b/...st/resources/config/generator_config.json → ...sources/config/generator_config_test.json
diff --git a/flink-cyber/cyber-services/cyber-service-common/pom.xml b/flink-cyber/cyber-services/cyber-service-common/pom.xml
@@ -28,6 +28,11 @@
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-compress</artifactId>
@@ -44,10 +49,6 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-autoconfigure</artifactId>
-        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-core</artifactId>

diff --git a/...main/java/com/cloudera/service/common/config/kafka/ClouderaManagementKafkaProperties.java b/...main/java/com/cloudera/service/common/config/kafka/ClouderaManagementKafkaProperties.java
diff --git a/...src/main/java/com/cloudera/service/common/config/kafka/ClouderaWorkerKafkaProperties.java b/...src/main/java/com/cloudera/service/common/config/kafka/ClouderaWorkerKafkaProperties.java
diff --git a/...ber-service-common/src/main/java/com/cloudera/service/common/request/AbstractRequest.java b/...ber-service-common/src/main/java/com/cloudera/service/common/request/AbstractRequest.java
@@ -0,0 +1,8 @@
+package com.cloudera.service.common.request;
+
+import lombok.Data;
+
+@Data
+public abstract class AbstractRequest {
+    protected final String requestId;
+}
diff --git a/...vice-common/src/main/java/com/cloudera/service/common/request/ClusterPipelineRequest.java b/...vice-common/src/main/java/com/cloudera/service/common/request/ClusterPipelineRequest.java
@@ -0,0 +1,48 @@
+package com.cloudera.service.common.request;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+import java.util.List;
+
+
+@Getter
+@Setter
+@ToString
+@EqualsAndHashCode(callSuper = true)
+public class ClusterPipelineRequest extends AbstractRequest {
+    private final String pipelineName;
+    private final String branch;
+    private final String profileName;
+    private final String parserName;
+    private final String gitUrl;
+    private final String userName;
+    private final String password;
+    private final List<String> jobs;
+
+    @JsonCreator
+    public ClusterPipelineRequest(
+            @JsonProperty("requestId") String requestId,
+            @JsonProperty("pipelineName") String pipelineName,
+            @JsonProperty("branch") String branch,
+            @JsonProperty("profileName") String profileName,
+            @JsonProperty("parserName") String parserName,
+            @JsonProperty("gitUrl") String gitUrl,
+            @JsonProperty("userName") String userName,
+            @JsonProperty("password") String password,
+            @JsonProperty("jobs") List<String> jobs) {
+        super(requestId);
+        this.pipelineName = pipelineName;
+        this.branch = branch;
+        this.profileName = profileName;
+        this.gitUrl = gitUrl;
+        this.userName = userName;
+        this.password = password;
+        this.jobs = jobs;
+        this.parserName = parserName;
+    }
+}
diff --git a/...yber-service-common/src/main/java/com/cloudera/service/common/request/ClusterRequest.java b/...yber-service-common/src/main/java/com/cloudera/service/common/request/ClusterRequest.java
@@ -0,0 +1,23 @@
+package com.cloudera.service.common.request;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+import java.util.List;
+
+@Getter
+@Setter
+@EqualsAndHashCode(callSuper = true)
+@ToString
+public class ClusterRequest extends AbstractRequest{
+    private final String clusterId;
+    private final List<String> jobs;
+
+    public ClusterRequest(String requestId, String clusterId, List<String> jobs) {
+        super(requestId);
+        this.clusterId = clusterId;
+        this.jobs = jobs;
+    }
+}
diff --git a/...s/cyber-service-common/src/main/java/com/cloudera/service/common/request/RequestBody.java b/...s/cyber-service-common/src/main/java/com/cloudera/service/common/request/RequestBody.java
@@ -17,6 +17,7 @@ public class RequestBody {
     private String pipelineName;
     private String branch;
     private String profileName;
+    private String parserName;
     private List<String> jobs;
     private byte[] payload;
 }
diff --git a/...s/cyber-service-common/src/main/java/com/cloudera/service/common/request/RequestType.java b/...s/cyber-service-common/src/main/java/com/cloudera/service/common/request/RequestType.java
@@ -1,5 +1,5 @@
 package com.cloudera.service.common.request;
 
 public enum RequestType {
-    GET_ALL_CLUSTERS_SERVICE_REQUEST, GET_CLUSTER_SERVICE_REQUEST, START_JOB_REQUEST, RESTART_JOB_REQUEST, STOP_JOB_REQUEST, GET_JOB_CONFIG_REQUEST, CREATE_EMPTY_PIPELINE, START_ARCHIVE_PIPELINE, UPDATE_JOB_CONFIG_REQUEST
+    GET_ALL_CLUSTERS_SERVICE_REQUEST, GET_CLUSTER_SERVICE_REQUEST, START_JOB_REQUEST, RESTART_JOB_REQUEST, STOP_JOB_REQUEST, GET_JOB_CONFIG_REQUEST, CREATE_EMPTY_PIPELINE, START_PIPELINE, UPDATE_JOB_CONFIG_REQUEST
 }
diff --git a/...services/cyber-service-common/src/main/java/com/cloudera/service/common/response/Job.java b/...services/cyber-service-common/src/main/java/com/cloudera/service/common/response/Job.java
@@ -19,7 +19,7 @@ public class Job {
 
     private String jobFullName;
 
-    private String jobName;
+    private String confName;
 
     private JobStatus jobState;
 
@@ -50,7 +50,7 @@ public String[] getScript(Job job) {
             switch (this) {
                 case PROFILE:
                 case PARSER:
-                    return new String[]{scriptName, job.getJobBranch(), job.getJobPipeline(), job.getJobName()};
+                    return new String[]{scriptName, job.getJobBranch(), job.getJobPipeline(), job.getConfName()};
                 case INDEX:
                 case TRIAGE:
                     return new String[]{scriptName, job.getJobBranch(), job.getJobPipeline()};

diff --git a/...ces/cyber-service-common/src/main/java/com/cloudera/service/common/response/Pipeline.java b/...ces/cyber-service-common/src/main/java/com/cloudera/service/common/response/Pipeline.java
@@ -5,14 +5,17 @@
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.util.List;
+
 @Data
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
 public class Pipeline {
-    String id;
-    String name;
-    String clusterName;
-    String date;
-    String userName;
+    private String id;
+    private String name;
+    private String clusterName;
+    private String date;
+    private List<String> jobs;
+    private String userName;
 }
diff --git a/...cyber-service-common/src/main/java/com/cloudera/service/common/response/ResponseType.java b/...cyber-service-common/src/main/java/com/cloudera/service/common/response/ResponseType.java
@@ -1,5 +1,5 @@
 package com.cloudera.service.common.response;
 
 public enum ResponseType {
-    GET_ALL_CLUSTERS_SERVICE_RESPONSE, GET_CLUSTER_SERVICE_RESPONSE, START_JOB_RESPONSE, RESTART_JOB_RESPONSE, STOP_JOB_RESPONSE, GET_JOB_CONFIG_RESPONSE, UPDATE_JOB_CONFIG_RESPONSE, CREATE_EMPTY_PIPELINE_RESPONSE, START_ARCHIVE_PIPELINE_RESPONSE, ERROR_RESPONSE
+    GET_ALL_CLUSTERS_SERVICE_RESPONSE, GET_CLUSTER_SERVICE_RESPONSE, START_JOB_RESPONSE, RESTART_JOB_RESPONSE, STOP_JOB_RESPONSE, GET_JOB_CONFIG_RESPONSE, UPDATE_JOB_CONFIG_RESPONSE, CREATE_EMPTY_PIPELINE_RESPONSE, START_PIPELINE_RESPONSE, ERROR_RESPONSE
 }
diff --git a/...ces/cyber-service-common/src/main/java/com/cloudera/service/common/utils/ArchiveUtil.java b/...ces/cyber-service-common/src/main/java/com/cloudera/service/common/utils/ArchiveUtil.java
@@ -8,6 +8,7 @@
 import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
 import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
 import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.flink.core.fs.FileStatus;
 
 import java.io.BufferedInputStream;
@@ -37,7 +38,7 @@ public static void compressToTarGzFile(String inputPath, String outputPath) thro
 
     public static byte[] compressToTarGzInMemory(String inputPath, boolean base64) throws IOException {
         final byte[] bytes = compressToTarGzInMemory(inputPath);
-        if (base64){
+        if (base64) {
             return Base64.getEncoder().encode(bytes);
         } else {
             return bytes;
@@ -51,6 +52,35 @@ public static byte[] compressToTarGzInMemory(String inputPath) throws IOExceptio
         }
     }
 
+    public static byte[] compressToTarGzInMemory(List<Pair<String, byte[]>> files) {
+        try (ByteArrayOutputStream bos = new ByteArrayOutputStream()) {
+            if (files == null || files.isEmpty()) {
+                log.info("There are no files.");
+                return bos.toByteArray();
+            }
+            try (BufferedOutputStream buffOut = new BufferedOutputStream(bos);
+                 GzipCompressorOutputStream gzOut = new GzipCompressorOutputStream(buffOut);
+                 TarArchiveOutputStream tOut = new TarArchiveOutputStream(gzOut)) {
+                try {
+                    for (Pair<String, byte[]> file : files) {
+                        TarArchiveEntry tarEntry = new TarArchiveEntry(
+                                file.getLeft());
+                        tarEntry.setSize(file.getRight().length);
+                        tOut.putArchiveEntry(tarEntry);
+                        tOut.write(file.getRight());
+                        tOut.closeArchiveEntry();
+                    }
+                } finally {
+                    tOut.finish();
+                }
+            }
+            return bos.toByteArray();
+        } catch (IOException e) {
+            log.error("IOException occurs while processing  {}", e.getMessage());
+            return new byte[0];
+        }
+    }
+
     private static void compressToTarGz(String inputPath, OutputStream outputStream) throws IOException {
         final List<FileStatus> fileList = FileUtil.listFiles(inputPath, true);
         if (fileList == null || fileList.isEmpty()) {
@@ -111,7 +141,7 @@ public static void decompressFromTarGzInMemory(byte[] rawData, String outputPath
             throw new IOException("Provided null as .tar.gz data which is not allowed!");
         }
         final byte[] data;
-        if (base64){
+        if (base64) {
             data = Base64.getDecoder().decode(rawData);
         } else {
             data = rawData;

diff --git a/flink-cyber/cyber-services/cyber-worker-service/pom.xml b/flink-cyber/cyber-services/cyber-worker-service/pom.xml
@@ -52,15 +52,19 @@
             <artifactId>spring-boot-autoconfigure</artifactId>
         </dependency>
         <dependency>
-            <groupId>com.cloudera.cyber</groupId>
-            <artifactId>metron-common</artifactId>
-            <version>${project.parent.version}</version>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.6</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-core</artifactId>

diff --git a/...config/kafka/ClouderaKafkaProperties.java → ...onfiguration/ClouderaKafkaProperties.java b/...config/kafka/ClouderaKafkaProperties.java → ...onfiguration/ClouderaKafkaProperties.java
@@ -1,10 +1,8 @@
-package com.cloudera.service.common.config.kafka;
+package com.cloudera.cyber.restcli.configuration;
 
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
 
 @Getter
 @Setter

diff --git a/...er-worker-service/src/main/java/com/cloudera/cyber/restcli/configuration/KafkaConfig.java b/...er-worker-service/src/main/java/com/cloudera/cyber/restcli/configuration/KafkaConfig.java
@@ -1,6 +1,5 @@
 package com.cloudera.cyber.restcli.configuration;
 
-import com.cloudera.service.common.config.kafka.ClouderaKafkaProperties;
 import com.cloudera.service.common.request.RequestBody;
 import com.cloudera.service.common.response.ResponseBody;
 import lombok.AllArgsConstructor;

diff --git a/...rker-service/src/main/java/com/cloudera/cyber/restcli/configuration/WebConfiguration.java b/...rker-service/src/main/java/com/cloudera/cyber/restcli/configuration/WebConfiguration.java
diff --git a/...-service/src/main/java/com/cloudera/cyber/restcli/controller/KafkaListenerController.java b/...-service/src/main/java/com/cloudera/cyber/restcli/controller/KafkaListenerController.java
@@ -12,7 +12,6 @@
 import com.cloudera.service.common.response.ResponseType;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.Message;
@@ -88,12 +87,12 @@ public Message<ResponseBody> handleMessage(RequestBody requestBody, @Header(Kafk
                 } catch (Exception e) {
                     return handleErrorResponse(e, replyTo, correlationId);
                 }
-            case START_ARCHIVE_PIPELINE:
+            case START_PIPELINE:
                 try {
                     pipelineService.extractPipeline(requestBody.getPayload(), requestBody.getPipelineName(), requestBody.getBranch());
-                    pipelineService.startPipelineJob(requestBody.getPipelineName(), requestBody.getBranch(), requestBody.getProfileName(), requestBody.getJobs());
+                    pipelineService.startPipelineJob(requestBody.getPipelineName(), requestBody.getBranch(), requestBody.getProfileName(), requestBody.getProfileName(), requestBody.getJobs());
                     final ResponseBody responseBody = ResponseBody.builder().build();
-                    return buildResponseMessage(responseBody, ResponseType.START_ARCHIVE_PIPELINE_RESPONSE, replyTo, correlationId);
+                    return buildResponseMessage(responseBody, ResponseType.START_PIPELINE_RESPONSE, replyTo, correlationId);
                 } catch (Exception e) {
                     log.error("Exception while processing the Start All request {}", e.getMessage());
                     return handleErrorResponse(e, replyTo, correlationId);