* Update e2e sample for NSGLogs

.gitignore

@@ -59,3 +59,4 @@ gradle/wrapper/gradle-wrapper.properties
 .vscode/settings.json
 rspec.xml
 e2e/output_file.txt
+logs.txt

docker-e2e/Filebeat-Docker

@@ -0,0 +1,2 @@
+FROM docker.elastic.co/beats/filebeat:8.7.1
+COPY --chown=go-w:filebeat ls-filebeat-config.yml /usr/share/filebeat/filebeat.yml

docker-e2e/Logstash-Docker

@@ -0,0 +1,6 @@
+FROM docker.elastic.co/logstash/logstash-oss:8.10.0
+COPY logstash-output-kusto-2.0.0-java.gem /tmp/logstash-output-kusto-2.0.0-java.gem
+RUN rm -f /usr/share/logstash/pipeline/logstash.conf && \
+  bin/logstash-plugin install /tmp/logstash-output-kusto-2.0.0-java.gem
+COPY logstash-nsg-logs.conf /usr/share/logstash/pipeline/logstash.conf
+COPY logstash.yml /usr/share/logstash/config/logstash.yml

docker-e2e/docker-compose-all.yml

@@ -0,0 +1,24 @@
+version: '3.8'
+services:
+  logstash:
+    build: # "context" and "dockerfile" fields have to be under "build"
+      context: .
+      dockerfile: Logstash-Docker
+    hostname: logstash
+    environment:
+      - "LS_JAVA_OPTS=-Xms1024m -Xmx8192m"
+    ports:
+      - "9600:9600"
+      - "5044:5044"
+    deploy:
+      restart_policy:
+        condition: on-failure
+  filebeat:
+    hostname: filebeat
+    command: filebeat -e -strict.perms=false
+    build: # "context" and "dockerfile" fields have to be under "build"
+      context: .
+      dockerfile: Filebeat-Docker
+    deploy:
+      restart_policy:
+        condition: on-failure

docker-e2e/filebeat.yml

@@ -0,0 +1,14 @@
+filebeat.inputs:
+# sends to logstash
+- type: azure-blob-storage
+  id: sdke2eteststorage
+  enabled: true
+  account_name: sdke2eteststorage
+  auth.shared_credentials.account_key: xxxxx
+  containers:
+  - name: insights-logs-networksecuritygroupflowevent
+    max_workers: 3
+    poll: true
+    poll_interval: 2m
+output.logstash:
+  hosts: ["logstash:5044"]

docker-e2e/logstash-nsg-logs.conf.template

@@ -0,0 +1,97 @@
+input { 
+  beats {
+    port => 5044
+  }
+} 
+
+
+filter { 
+    json {
+      source => "message"
+    }  
+    split { field => "[records]" } 
+    split { field => "[records][properties][flows]"} 
+    split { field => "[records][properties][flows][flows]"} 
+    split { field => "[records][properties][flows][flows][flowTuples]"} 
+
+
+  mutate { 
+    split => { "[records][resourceId]" => "/"} 
+    add_field => {
+      "Subscription" => "%{[records][resourceId][2]}" 
+      "ResourceGroup" => "%{[records][resourceId][4]}" 
+      "NetworkSecurityGroup" => "%{[records][resourceId][8]}"
+      "macAddress" => "%{[records][macAddress]}"
+      "Time" => "%{[records][time]}" 
+      "category" => "%{[records][category]}"
+      "flowLogVersion" => "%{[records][properties][Version]}"
+      "nsgResourceID" =>"%{[records][resourceId]}"
+    } 
+    convert => {"Subscription" => "string"} 
+    convert => {"Time" => "string"} 
+    convert => {"macAddress" => "string"} 
+    convert => {"flowLogVersion" => "integer"} 
+    convert => {"nsgResourceID" => "string"} 
+    convert => {"rule" => "string"} 
+    convert => {"ResourceGroup" => "string"} 
+    convert => {"NetworkSecurityGroup" => "string"} 
+
+    add_field => {
+      "rule" => "%{[records][properties][flows][rule]}" 
+    } 
+    convert => {"rule" => "string"} 
+    split => { "[records][properties][flows][flows][flowTuples]" => ","} 
+    add_field => { 
+      "Timestamp" => "%{[records][properties][flows][flows][flowTuples][0]}" 
+      "srcIP" => "%{[records][properties][flows][flows][flowTuples][1]}" 
+      "dstIP" => "%{[records][properties][flows][flows][flowTuples][2]}" 
+      "srcPort" => "%{[records][properties][flows][flows][flowTuples][3]}" 
+      "dstPort" => "%{[records][properties][flows][flows][flowTuples][4]}" 
+      "Protocol" => "%{[records][properties][flows][flows][flowTuples][5]}" 
+      "Direction" => "%{[records][properties][flows][flows][flowTuples][6]}" 
+      "Decision" => "%{[records][properties][flows][flows][flowTuples][7]}" 
+      "State" => "%{[records][properties][flows][flows][flowTuples][8]}" 
+      "PacketsSrcToDst" => "%{[records][properties][flows][flows][flowTuples][9]}" 
+      "BytesSrcToDst" => "%{[records][properties][flows][flows][flowTuples][10]}" 
+      "PacketsDstToSrc" => "%{[records][properties][flows][flows][flowTuples][11]}"  
+      "BytesDstToSrc" => "%{[records][properties][flows][flows][flowTuples][12]}"  
+      "macAddress2" => "%{[records][properties][flows][flows][mac]}" 
+    }
+  } 
+  ## replace records with -1 , these are optional fields
+  if [PacketsSrcToDst] =~ "records" { 
+      mutate { replace => { "PacketsSrcToDst" => "-1" } } 
+  }     
+  if [PacketsDstToSrc] =~ "records" { 
+      mutate { replace => { "PacketsDstToSrc" => "-1" } } 
+  }     
+  if [BytesSrcToDst] =~ "records" { 
+      mutate { replace => { "BytesSrcToDst" => "-1" } } 
+  }     
+  if [BytesDstToSrc] =~ "records" { 
+      mutate { replace => { "BytesDstToSrc" => "-1" } } 
+  }      
+  mutate { 
+    remove_field => ["message", "records","log","event","tags","host","input","agent","cloud","azure","@timestamp","ecs","@version"]
+    # Change types
+    convert => {"PacketsSrcToDst" => "integer"}
+    convert => {"PacketsDstToSrc" => "integer"}
+    convert => {"BytesSrcToDst" => "integer"}
+    convert => {"BytesDstToSrc" => "integer"}
+  } 
+  date{ 
+    match => ["Timestamp" , "UNIX"] 
+  }
+}
+output {
+    kusto { 
+        path => "/tmp/kusto/fabric-nsg-%{+YYYY-MM-dd-HH-mm}.txt" 
+        ingest_url => "" 
+        app_id => "" 
+        app_key => "" 
+        app_tenant => "" 
+        database => "" 
+        table => "flowLogs"
+        json_mapping => "flowLogsMapping" 
+    } 
+}

docker-e2e/logstash.yml

@@ -0,0 +1,6 @@
+---
+## Default Logstash configuration from Logstash base image.
+## https://github.com/elastic/logstash/blob/main/docker/data/logstash/config/logstash-full.yml
+#
+http.host: 0.0.0.0
+node.name: logstash

docker-e2e/ls-filebeat-config.yml

@@ -0,0 +1,14 @@
+filebeat.inputs:
+# sends to logstash
+- type: azure-blob-storage
+  id: nsgstorage
+  enabled: true
+  account_name: nsgstorage
+  auth.shared_credentials.account_key: xxxxxxxxxxx
+  containers:
+  - name: insights-logs-networksecuritygroupflowevent
+    max_workers: 3
+    poll: true
+    poll_interval: 2m
+output.logstash:
+  hosts: ["logstash:5044"]