contrib/exporters/secadvisor: Encode flows in a JSON object

During the refactoring to pipelines the export format for flows was
changed to JSON array, whereas Security Advisor expects the objects in
COS/S3 to be a JSON object with a top-level "data" key whose value is an
array of flow objects.

This change adds an `sa` encode type which adds this top-level element
(and uses the core JSON encoder to perform the actual encoding, so
`pretty: true` still works).

To use this encoder, the secadvisor exporter should have this config:

    pipeline:
      encode:
        type: sa
        json:
          pretty: false   # or true

Author: dubek

contrib/exporters/allinone/main.go

@@ -28,6 +28,7 @@ func main() {
 }
 
 func init() {
+	core.EncoderHandlers.Register("sa", sa.NewEncode, false)
 	core.TransformerHandlers.Register("awsflowlogs", aws.NewTransform, false)
 	core.TransformerHandlers.Register("vpclogs", sa.NewTransform, false)
 }

contrib/exporters/core/encode.go

@@ -25,12 +25,13 @@ import (
 	"github.com/spf13/viper"
 )
 
-type encodeJSON struct {
+// EncodeJSON encoder encodes flows as a JSON array
+type EncodeJSON struct {
 	pretty bool
 }
 
-// Encode explements Encounter interface
-func (e *encodeJSON) Encode(in interface{}) ([]byte, error) {
+// Encode implements Encoder interface
+func (e *EncodeJSON) Encode(in interface{}) ([]byte, error) {
 	buf := new(bytes.Buffer)
 	encoder := json.NewEncoder(buf)
 
@@ -45,22 +46,23 @@ func (e *encodeJSON) Encode(in interface{}) ([]byte, error) {
 	return buf.Bytes(), nil
 }
 
-// NewEncodeJSON create an encode object
+// NewEncodeJSON creates an encode object
 func NewEncodeJSON(cfg *viper.Viper) (interface{}, error) {
-	return &encodeJSON{
+	return &EncodeJSON{
 		pretty: cfg.GetBool(CfgRoot + "encode.json.pretty"),
 	}, nil
 }
 
-type encodeCSV struct {
+// EncodeCSV encoder encodes flows as CSV rows
+type EncodeCSV struct {
 }
 
-// Encode explements Encounter interface
-func (e *encodeCSV) Encode(in interface{}) ([]byte, error) {
+// Encode implements Encoder interface
+func (e *EncodeCSV) Encode(in interface{}) ([]byte, error) {
 	return gocsv.MarshalBytes(in)
 }
 
-// NewEncodeCSV create an encode object
+// NewEncodeCSV creates an encode object
 func NewEncodeCSV(cfg *viper.Viper) (interface{}, error) {
-	return &encodeCSV{}, nil
+	return &EncodeCSV{}, nil
 }

contrib/exporters/secadvisor/main.go

@@ -27,5 +27,6 @@ func main() {
 }
 
 func init() {
-	core.TransformerHandlers.Register("sa", mod.NewTransform, false)
+	core.EncoderHandlers.Register("sa", mod.NewEncode, true)
+	core.TransformerHandlers.Register("sa", mod.NewTransform, true)
 }

contrib/exporters/secadvisor/mod/encode.go

@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2019 IBM, Inc.
+ *
+ * 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 ofthe License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specificlanguage governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package mod
+
+import (
+	"github.com/spf13/viper"
+
+	"github.com/skydive-project/skydive/contrib/exporters/core"
+)
+
+type encode struct {
+	*core.EncodeJSON
+}
+
+type topLevelObject struct {
+	Data []interface{} `json:"data"`
+}
+
+// Encode the incoming flows array as a JSON object with key "data" whose value
+// holds the array
+func (e *encode) Encode(in interface{}) ([]byte, error) {
+	return e.EncodeJSON.Encode(topLevelObject{Data: in.([]interface{})})
+}
+
+// NewEncode creates an encode object for Secadvisor format
+func NewEncode(cfg *viper.Viper) (interface{}, error) {
+	jsonEncoder, err := core.NewEncodeJSON(cfg)
+	if err != nil {
+		return nil, err
+	}
+	return &encode{jsonEncoder.(*core.EncodeJSON)}, nil
+}

contrib/exporters/secadvisor/mod/encode_test.go

@@ -0,0 +1,113 @@
+/*
+ * Copyright (C) 2019 IBM, Inc.
+ *
+ * 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 ofthe License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specificlanguage governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package mod
+
+import (
+	"encoding/json"
+	"reflect"
+	"testing"
+
+	"github.com/spf13/viper"
+
+	"github.com/skydive-project/skydive/contrib/exporters/core"
+)
+
+func areEqualJSON(buf1, buf2 []byte) (bool, error) {
+	var o1 interface{}
+	var o2 interface{}
+
+	var err error
+	err = json.Unmarshal(buf1, &o1)
+	if err != nil {
+		return false, err
+	}
+	err = json.Unmarshal(buf2, &o2)
+	if err != nil {
+		return false, err
+	}
+	return reflect.DeepEqual(o1, o2), nil
+}
+
+func getEncoder(t *testing.T) core.Encoder {
+	cfg := viper.New()
+	encoder, err := NewEncode(cfg)
+	if err != nil {
+		t.Fatalf("NewEncode returned unexpected error: %v", err)
+	}
+	return encoder.(core.Encoder)
+}
+
+func Test_Encode_empty_flows_array(t *testing.T) {
+	in := make([]interface{}, 0)
+	result, err := getEncoder(t).Encode(in)
+	if err != nil {
+		t.Fatalf("Encode returned unexpected error: %v", err)
+	}
+	expected := []byte(
+		`{
+			"data": []
+		}`)
+	equal, err := areEqualJSON(expected, result)
+	if err != nil {
+		t.Fatalf("Error parsing JSON: %v", err)
+	}
+	if !equal {
+		t.Fatalf("Objects not identical")
+	}
+}
+
+func Test_Encode_flows_array_with_objects(t *testing.T) {
+	in := []interface{}{
+		&SecurityAdvisorFlow{
+			Status: "STARTED",
+			Start:  1234,
+		},
+		&SecurityAdvisorFlow{
+			Status: "ENDED",
+			Last:   5678,
+		},
+	}
+	result, err := getEncoder(t).Encode(in)
+	if err != nil {
+		t.Fatalf("Encode returned unexpected error: %v", err)
+	}
+	expected := []byte(
+		`{
+			"data": [
+				{
+					"Status": "STARTED",
+					"Start": 1234,
+					"Last": 0,
+					"UpdateCount": 0
+				},
+				{
+					"Status": "ENDED",
+					"Start": 0,
+					"Last": 5678,
+					"UpdateCount": 0
+				}
+			]
+		}`)
+	equal, err := areEqualJSON(expected, result)
+	if err != nil {
+		t.Fatalf("Error parsing JSON: %v", err)
+	}
+	if !equal {
+		t.Fatalf("Objects not identical")
+	}
+}

contrib/exporters/secadvisor/secadvisor.yml.default

@@ -31,7 +31,7 @@ pipeline:
     sa:
       exclude_started_flows: true
   encode:
-    type: json
+    type: sa
     json:
       pretty: true
   compress: