[CYB-4;CYB-137] TableAPI Flink Indexer + Parser UI updates

flink-cyber/cyber-jobs/src/main/resources/conf/templates/index/index.properties.template

@@ -2,6 +2,16 @@ hive.catalog=hive
 hive.schema=cyber
 hive.table=events
 
+#flink.writer=tableapi
+
+#Uncomment one of the following lines in case flink.writer=tableapi is specified
+#flink.output-connector=hive
+#flink.output-connector=kafka
+
+#Uncomment following two lines in case flink.writer=tableapi is specified
+#flink.tables-init-file=PIPELINE/index/table-config.json
+#flink.mapping-file=PIPELINE/index/mapping-config.json
+
 #hive.confdir=/etc/hive/conf/
 
 #hive.transaction.messages=500

flink-cyber/cyber-jobs/src/main/resources/conf/templates/index/mapping-config.json

@@ -0,0 +1,21 @@
+{
+  "squid": {
+    "table_name": "hive_table",
+    "ignore_fields": [
+      "code"
+    ],
+    "column_mapping": [
+      {
+        "name": "full_hostname"
+      },
+      {
+        "name": "action"
+      }
+    ]
+  },
+  "test": {
+    "table_name": "another_hive_table",
+    "column_mapping": [
+    ]
+  }
+}

flink-cyber/cyber-jobs/src/main/resources/conf/templates/index/table-config.json

@@ -0,0 +1,14 @@
+{
+  "hive_table": [
+    {
+      "name": "full_hostname",
+      "type": "string"
+    },
+    {
+      "name": "action",
+      "type": "string"
+    }
+  ],
+  "another_hive_table": [
+  ]
+}

flink-cyber/cyber-jobs/src/main/resources/scripts/cs-add-index

@@ -35,7 +35,7 @@ fi
 pipeline=$1
 index_dir=$pipeline/index
 
-mkdir -p $index_dir
+mkdir -p $index_dir/conf
 for FILE in $TEMPLATES_DIR/index/*; do
     sed -e s/PIPELINE/$pipeline/g $FILE > $index_dir/${FILE##*/}
 done

flink-cyber/flink-alert-scoring-api/src/main/java/com/cloudera/cyber/scoring/ScoredMessage.java

@@ -1,12 +1,12 @@
 /*
  * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
  *
- * This file is licensed under the Apache License Version 2.0 (the "License"). You may not use this file 
- * except in compliance with the License. You may obtain a copy of the License at 
+ * This file is licensed under the Apache License Version 2.0 (the "License"). You may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
  * http://www.apache.org/licenses/LICENSE-2.0.
  *
- * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, 
- * either express or implied. Refer to the License for the specific permissions and 
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
  * limitations governing your use of the file.
  */
 
@@ -25,6 +25,9 @@
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.specific.SpecificRecord;
 import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
 
 import java.util.List;
 
@@ -59,6 +62,18 @@ public long getTs() {
             .optionalDouble("cyberScore")
             .endRecord();
 
+    public static final TypeInformation<Row> FLINK_TYPE_INFO = Types.ROW_NAMED(
+            new String[]{"message", "cyberScoresDetails", "cyberScore"},
+            Message.FLINK_TYPE_INFO, Types.OBJECT_ARRAY(Scores.FLINK_TYPE_INFO), Types.DOUBLE);
+
+    public Row toRow() {
+        return Row.of(message == null ? null : message.toRow(),
+                cyberScoresDetails == null ? null : cyberScoresDetails.stream()
+                        .map(Scores::toRow)
+                        .toArray(Row[]::new),
+                cyberScore);
+    }
+
     @Override
     public Schema getSchema() {
         return SCHEMA$;

flink-cyber/flink-alert-scoring-api/src/main/java/com/cloudera/cyber/scoring/Scores.java

@@ -1,12 +1,12 @@
 /*
  * Copyright 2020 - 2022 Cloudera. All Rights Reserved.
  *
- * This file is licensed under the Apache License Version 2.0 (the "License"). You may not use this file 
- * except in compliance with the License. You may obtain a copy of the License at 
+ * This file is licensed under the Apache License Version 2.0 (the "License"). You may not use this file
+ * except in compliance with the License. You may obtain a copy of the License at
  * http://www.apache.org/licenses/LICENSE-2.0.
  *
- * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, 
- * either express or implied. Refer to the License for the specific permissions and 
+ * This file is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied. Refer to the License for the specific permissions and
  * limitations governing your use of the file.
  */
 
@@ -22,8 +22,9 @@
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.specific.SpecificRecord;
 import org.apache.avro.specific.SpecificRecordBase;
-
-import java.util.UUID;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
 
 @Data
 @Builder
@@ -42,6 +43,14 @@ public class Scores extends SpecificRecordBase implements SpecificRecord {
             .requiredString("reason")
             .endRecord();
 
+    public static final TypeInformation<Row> FLINK_TYPE_INFO = Types.ROW_NAMED(
+            new String[]{"ruleId", "score", "reason"},
+            Types.STRING, Types.DOUBLE, Types.STRING);
+
+    public Row toRow() {
+        return Row.of(ruleId, score, reason);
+    }
+
     @Override
     public Schema getSchema() {
         return SCHEMA$;
@@ -50,20 +59,31 @@ public Schema getSchema() {
     @Override
     public Object get(int field$) {
         switch (field$) {
-            case 0: return ruleId;
-            case 1: return score;
-            case 2: return reason;
-            default: throw new AvroRuntimeException("Bad index");
+            case 0:
+                return ruleId;
+            case 1:
+                return score;
+            case 2:
+                return reason;
+            default:
+                throw new AvroRuntimeException("Bad index");
         }
     }
 
     @Override
     public void put(int field$, Object value$) {
         switch (field$) {
-            case 0: ruleId = value$.toString(); break;
-            case 1: score = (Double)value$; break;
-            case 2: reason = value$.toString(); break;
-            default: throw new AvroRuntimeException("Bad Index");
+            case 0:
+                ruleId = value$.toString();
+                break;
+            case 1:
+                score = (Double) value$;
+                break;
+            case 2:
+                reason = value$.toString();
+                break;
+            default:
+                throw new AvroRuntimeException("Bad Index");
         }
     }
 }

flink-cyber/flink-common/src/main/java/com/cloudera/cyber/flink/Utils.java

@@ -12,6 +12,9 @@
 
 package com.cloudera.cyber.flink;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.io.Resources;
 import com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient;
 
@@ -34,8 +37,6 @@
 import java.io.IOException;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Paths;
 import java.security.InvalidKeyException;
 import java.security.NoSuchAlgorithmException;
 import java.security.PrivateKey;
@@ -67,6 +68,8 @@ public class Utils {
     public static final String K_KEYSTORE_PASSWORD = "keyStorePassword";
     public static final String PASSWORD = "password";
 
+    public static final ObjectMapper MAPPER = new ObjectMapper();
+
     public static Properties readProperties(Properties properties, String prefix) {
         Properties targetProperties = new Properties();
         for (String key : properties.stringPropertyNames()) {
@@ -162,6 +165,13 @@ public static Map<String, Object> readSchemaRegistryProperties(ParameterTool par
         return schemaRegistryConf;
     }
 
+    private static <T> T jsonToObject(String json, TypeReference<T> typeReference) throws JsonProcessingException {
+        return MAPPER.readValue(json, typeReference);
+    }
+
+    public static <T> T readResourceFile(String resourceLocation, Class<?> cls, TypeReference<T> typeReference) throws IOException {
+        return jsonToObject(readResourceFile(resourceLocation, cls), typeReference);
+    }
 
     public static String readResourceFile(String resourceLocation, Class<?> cls) {
         try {
@@ -271,8 +281,18 @@ public static ParameterTool getParamToolsFromProperties(String[] pathToPropertyF
     }
 
 
+    public static <T> T readFile(String path, TypeReference<T> typeReference) throws IOException {
+        return jsonToObject(readFile(path), typeReference);
+    }
+
     public static String readFile(String path) throws IOException {
-        return new String(Files.readAllBytes(Paths.get(path)), StandardCharsets.UTF_8);
+        final Path filePath = new Path(path);
+        final FileSystem fileSystem = filePath.getFileSystem();
+        try (FSDataInputStream fsDataInputStream = fileSystem.open(filePath)) {
+            return IOUtils.toString(fsDataInputStream, StandardCharsets.UTF_8);
+        } catch (Exception e) {
+            throw new RuntimeException(String.format("Wasn't able to read file [%s]!", filePath), e);
+        }
     }
 
 

flink-cyber/flink-cyber-api/pom.xml

@@ -50,6 +50,11 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-table-common</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>

flink-cyber/flink-cyber-api/src/main/java/com/cloudera/cyber/DataQualityMessage.java

@@ -22,6 +22,9 @@
 import org.apache.avro.specific.SpecificRecord;
 import org.apache.avro.specific.SpecificRecordBase;
 import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
 
 @Data
 @Builder(toBuilder = true)
@@ -45,6 +48,14 @@ public class DataQualityMessage extends SpecificRecordBase implements SpecificRe
             .requiredString("message")
             .endRecord();
 
+    public static final TypeInformation<Row> FLINK_TYPE_INFO = Types.ROW_NAMED(
+            new String[]{"level", "feature", "field", "message"},
+            Types.STRING, Types.STRING, Types.STRING, Types.STRING);
+
+    public Row toRow() {
+        return Row.of(level, feature, field, message);
+    }
+
     @Override
     public Schema getSchema() {
         return SCHEMA$;
@@ -71,5 +82,4 @@ public void put(int field$, Object value$) {
             default: throw new AvroRuntimeException("Bad index");
         }
     }
-
 }

flink-cyber/flink-cyber-api/src/main/java/com/cloudera/cyber/Message.java

@@ -12,17 +12,25 @@
 
 package com.cloudera.cyber;
 
-import lombok.*;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
 import org.apache.avro.Schema;
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.specific.SpecificRecord;
 import org.apache.avro.specific.SpecificRecordBase;
 import org.apache.avro.util.Utf8;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
 
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 import static com.cloudera.cyber.AvroTypes.toListOf;
 import static com.cloudera.cyber.AvroTypes.utf8toStringMap;
@@ -56,6 +64,31 @@ public class Message extends SpecificRecordBase implements SpecificRecord, Ident
             .name("dataQualityMessages").type().optional().type(Schema.createArray(DataQualityMessage.SCHEMA$))
             .endRecord();
 
+    public static final TypeInformation<Row> FLINK_TYPE_INFO = Types.ROW_NAMED(
+            new String[]{"id", "ts", "originalSource", "message", "threats", "extensions", "source", "dataQualityMessages"},
+            Types.STRING, Types.LONG, SignedSourceKey.FLINK_TYPE_INFO,Types.STRING,
+            Types.MAP(Types.STRING, Types.OBJECT_ARRAY(ThreatIntelligence.FLINK_TYPE_INFO)),
+            Types.MAP(Types.STRING, Types.STRING), Types.STRING, Types.OBJECT_ARRAY(DataQualityMessage.FLINK_TYPE_INFO));
+
+    public Row toRow() {
+        return Row.of(id,
+                ts,
+                originalSource.toRow(),
+                message,
+                threats == null ? null : threats.entrySet().stream()
+                        .collect(Collectors.toMap(
+                                Map.Entry::getKey,
+                                e -> e.getValue().stream()
+                                        .map(ThreatIntelligence::toRow)
+                                        .toArray(Row[]::new))),
+                extensions,
+                source,
+                dataQualityMessages == null ? null : dataQualityMessages.stream()
+                        .map(DataQualityMessage::toRow)
+                        .toArray(Row[]::new)
+        );
+    }
+
     public static Schema getClassSchema() {
         return SCHEMA$;
     }

flink-cyber/flink-cyber-api/src/main/java/com/cloudera/cyber/SignedSourceKey.java

@@ -21,6 +21,9 @@
 import org.apache.avro.SchemaBuilder;
 import org.apache.avro.specific.SpecificRecord;
 import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.types.Row;
 
 import java.nio.ByteBuffer;
 
@@ -34,6 +37,7 @@ public class SignedSourceKey extends SpecificRecordBase implements SpecificRecor
     private int partition;
     private long offset;
     private byte[] signature;
+
     public static final Schema SCHEMA$ = SchemaBuilder
             .record(SignedSourceKey.class.getName())
             .namespace(SignedSourceKey.class.getPackage().getName())
@@ -44,6 +48,15 @@ public class SignedSourceKey extends SpecificRecordBase implements SpecificRecor
             .name("signature").type().bytesBuilder().endBytes().noDefault()
             .endRecord();
 
+    public static final TypeInformation<Row> FLINK_TYPE_INFO = Types.ROW_NAMED(
+            new String[]{"topic", "partition", "offset", "signature"},
+            Types.STRING, Types.INT, Types.LONG, Types.PRIMITIVE_ARRAY(Types.BYTE));
+
+
+    public Row toRow() {
+        return Row.of(topic, partition, offset, signature);
+    }
+
 //    static public class SignedSourceKeyBuilder {
 //        public SignedSourceKeyBuilder signature(byte[] bytes) {
 //            this.signature = ByteBuffer.wrap(bytes);