googlecloud/vpcflow fileset: Populate additional log fields

filebeat/docs/modules/googlecloud.asciidoc

@@ -40,6 +40,7 @@ Example config:
     var.topic: googlecloud-vpc-flowlogs
     var.subscription_name: filebeat-googlecloud-vpc-flowlogs-sub
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
+    var.keep_original_message: false
 ----
 
 include::../include/var-paths.asciidoc[]
@@ -61,6 +62,11 @@ exist it will be created.
 
 Path to a JSON file containing the credentials and key used to subscribe.
 
+*`var.keep_original_message`*::
+
+Flag to control whether the original message is stored in the `log.original`
+field. Defaults to `false`, meaning the original message is not saved.
+
 :fileset_ex!:
 
 :fileset_ex: firewall

x-pack/filebeat/module/googlecloud/_meta/docs.asciidoc

@@ -35,6 +35,7 @@ Example config:
     var.topic: googlecloud-vpc-flowlogs
     var.subscription_name: filebeat-googlecloud-vpc-flowlogs-sub
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
+    var.keep_original_message: false
 ----
 
 include::../include/var-paths.asciidoc[]
@@ -56,6 +57,11 @@ exist it will be created.
 
 Path to a JSON file containing the credentials and key used to subscribe.
 
+*`var.keep_original_message`*::
+
+Flag to control whether the original message is stored in the `log.original`
+field. Defaults to `false`, meaning the original message is not saved.
+
 :fileset_ex!:
 
 :fileset_ex: firewall

x-pack/filebeat/module/googlecloud/vpcflow/config/input.yml

@@ -22,3 +22,5 @@ processors:
       lang: javascript
       id: googlecloud_vpcflow_script
       file: ${path.home}/module/googlecloud/vpcflow/config/pipeline.js
+      params:
+        keep_original_message: {{ .keep_original_message }}

x-pack/filebeat/module/googlecloud/vpcflow/config/pipeline.js

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
-var vpcflow = (function () {
+function VPCFlow(keep_original_message) {
     var processor = require("processor");
 
     // The pub/sub input writes the Stackdriver LogEntry object into the message
@@ -21,6 +21,16 @@ var vpcflow = (function () {
         ignore_missing: true,
     });
 
+    var saveOriginalMessage = function(evt) {};
+    if (keep_original_message) {
+        saveOriginalMessage = new processor.Convert({
+            fields: [
+                {from: "message", to: "event.original"}
+            ],
+            mode: "rename"
+        });
+    }
+
     var dropPubSubFields = function(evt) {
         evt.Delete("message");
         evt.Delete("labels");
@@ -34,6 +44,14 @@ var vpcflow = (function () {
         },
     });
 
+
+    var saveMetadata = new processor.Convert({
+        fields: [
+            {from: "json.logName", to: "log.logger"},
+        ],
+        ignore_missing: true
+    });
+
     // Use the LogEntry object's timestamp. VPC flow logs are structured so the
     // LogEntry includes a jsonPayload field.
     // https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry
@@ -205,8 +223,10 @@ var vpcflow = (function () {
     var pipeline = new processor.Chain()
         .Add(decodeJson)
         .Add(parseTimestamp)
+        .Add(saveOriginalMessage)
         .Add(dropPubSubFields)
         .Add(categorizeEvent)
+        .Add(saveMetadata)
         .Add(convertLogEntry)
         .Add(convertJsonPayload)
         .Add(dropEmptyObjects)
@@ -223,7 +243,14 @@ var vpcflow = (function () {
     return {
         process: pipeline.Run,
     };
-})();
+}
+
+var vpcflow;
+
+// Register params from configuration.
+function register(params) {
+    vpcflow = new VPCFlow(params.keep_original_message);
+}
 
 function process(evt) {
     return vpcflow.process(evt);

x-pack/filebeat/module/googlecloud/vpcflow/manifest.yml

@@ -11,7 +11,8 @@ var:
     default: filebeat-googlecloud-vpcflow
   - name: credentials_file
     default: googlecloud-vpcflow-reader-service-identity.json
-
+  - name: keep_original_message
+    default: false
 ingest_pipeline: ingest/pipeline.yml
 input: config/input.yml