Simple conversion of the format returned by the API (elastic#10019)

* Simple conversion conversion of the format returned by the API

The Kibana API for CM is undergoing internal changes to better allow
future scalability of the configurations, one of the requirements for
this move was to flatten as much as possible the configuration file so
the format would be the same independant of the key that need to
received ot the configuration. It will be up to the manager to take the
JSON format and convert it into something that our Configuration can
understand.

ref: elastic/kibana#27717

CHANGELOG.next.asciidoc

@@ -21,6 +21,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Rename `process.exe` to `process.executable` in add_process_metadata to align with ECS. {pull}9949[9949]
 - Import ECS change https://github.com/elastic/ecs/pull/308[ecs#308]:
   leaf field `user.group` is now the `group` field set. {pull}10275[10275]
+- Update the code of Central Management to align with the new returned format. {pull}10019[10019]
 - Docker and Kubernetes labels/annotations will be "dedoted" by default. {pull}10338[10338]
 
 *Auditbeat*

x-pack/libbeat/management/api/configuration.go

@@ -5,6 +5,7 @@
 package api
 
 import (
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"reflect"
@@ -50,16 +51,40 @@ func (c *ConfigBlock) ConfigWithMeta() (*reload.ConfigWithMeta, error) {
 	}, nil
 }
 
+type configResponse struct {
+	Type string
+	Raw  map[string]interface{}
+}
+
+func (c *configResponse) UnmarshalJSON(b []byte) error {
+	var resp = struct {
+		Type string                 `json:"type"`
+		Raw  map[string]interface{} `json:"config"`
+	}{}
+
+	if err := json.Unmarshal(b, &resp); err != nil {
+		return err
+	}
+
+	converter := selectConverter(resp.Type)
+	newMap, err := converter(resp.Raw)
+	if err != nil {
+		return err
+	}
+	*c = configResponse{
+		Type: resp.Type,
+		Raw:  newMap,
+	}
+	return nil
+}
+
 // Configuration retrieves the list of configuration blocks from Kibana
 func (c *Client) Configuration(accessToken string, beatUUID uuid.UUID, configOK bool) (ConfigBlocks, error) {
 	headers := http.Header{}
 	headers.Set("kbn-beats-access-token", accessToken)
 
 	resp := struct {
-		ConfigBlocks []*struct {
-			Type string                 `json:"type"`
-			Raw  map[string]interface{} `json:"config"`
-		} `json:"configuration_blocks"`
+		ConfigBlocks []*configResponse `json:"configuration_blocks"`
 	}{}
 	url := fmt.Sprintf("/api/beats/agent/%s/configuration?validSetting=%t", beatUUID, configOK)
 	statusCode, err := c.request("GET", url, nil, headers, &resp)

x-pack/libbeat/management/api/configuration_test.go

@@ -28,7 +28,7 @@ func TestConfiguration(t *testing.T) {
 
 		assert.Equal(t, "false", r.URL.Query().Get("validSetting"))
 
-		fmt.Fprintf(w, `{"configuration_blocks":[{"type":"filebeat.modules","config":{"module":"apache2"}},{"type":"metricbeat.modules","config":{"module":"system","period":"10s"}}]}`)
+		fmt.Fprintf(w, `{"configuration_blocks":[{"type":"filebeat.modules","config":{"_sub_type":"apache2"}},{"type":"metricbeat.modules","config":{"_sub_type":"system","period":"10s"}}]}`)
 	}))
 	defer server.Close()
 

x-pack/libbeat/management/api/convert.go

@@ -0,0 +1,79 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"fmt"
+	"strings"
+)
+
+type converter func(map[string]interface{}) (map[string]interface{}, error)
+
+var mapper = map[string]converter{
+	".inputs":  noopConvert,
+	".modules": convertMultiple,
+	"output":   convertSingle,
+}
+
+var errSubTypeNotFound = fmt.Errorf("'%s' key not found", subTypeKey)
+
+var (
+	subTypeKey = "_sub_type"
+	moduleKey  = "module"
+)
+
+func selectConverter(t string) converter {
+	for k, v := range mapper {
+		if strings.Index(t, k) > -1 {
+			return v
+		}
+	}
+	return noopConvert
+}
+
+func convertSingle(m map[string]interface{}) (map[string]interface{}, error) {
+	subType, err := extractSubType(m)
+	if err != nil {
+		return nil, err
+	}
+
+	delete(m, subTypeKey)
+	newMap := map[string]interface{}{subType: m}
+	return newMap, nil
+}
+
+func convertMultiple(m map[string]interface{}) (map[string]interface{}, error) {
+	subType, err := extractSubType(m)
+	if err != nil {
+		return nil, err
+	}
+
+	v, ok := m[moduleKey]
+
+	if ok && v != subType {
+		return nil, fmt.Errorf("module key already exist in the raw document and doesn't match the 'sub_type', expecting '%s' and received '%s", subType, v)
+	}
+
+	m[moduleKey] = subType
+	delete(m, subTypeKey)
+	return m, nil
+}
+
+func noopConvert(m map[string]interface{}) (map[string]interface{}, error) {
+	return m, nil
+}
+
+func extractSubType(m map[string]interface{}) (string, error) {
+	subType, ok := m[subTypeKey]
+	if !ok {
+		return "", errSubTypeNotFound
+	}
+
+	k, ok := subType.(string)
+	if !ok {
+		return "", fmt.Errorf("invalid type for `sub_type`, expecting a string received %T", subType)
+	}
+	return k, nil
+}

x-pack/libbeat/management/api/convert_test.go

@@ -0,0 +1,111 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package api
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestConvertAPI(t *testing.T) {
+	tests := map[string]struct {
+		t        string
+		config   map[string]interface{}
+		expected map[string]interface{}
+		err      bool
+	}{
+		"output": {
+			t: "output",
+			config: map[string]interface{}{
+				"_sub_type": "elasticsearch",
+				"username":  "foobar",
+			},
+			expected: map[string]interface{}{
+				"elasticsearch": map[string]interface{}{
+					"username": "foobar",
+				},
+			},
+		},
+		"filebeat inputs": {
+			t: "filebeat.inputs",
+			config: map[string]interface{}{
+				"type": "log",
+				"paths": []string{
+					"/var/log/message.log",
+					"/var/log/system.log",
+				},
+			},
+			expected: map[string]interface{}{
+				"type": "log",
+				"paths": []string{
+					"/var/log/message.log",
+					"/var/log/system.log",
+				},
+			},
+		},
+		"filebeat modules": {
+			t: "filebeat.modules",
+			config: map[string]interface{}{
+				"_sub_type": "system",
+			},
+			expected: map[string]interface{}{
+				"module": "system",
+			},
+		},
+		"metricbeat modules": {
+			t: "metricbeat.modules",
+			config: map[string]interface{}{
+				"_sub_type": "logstash",
+			},
+			expected: map[string]interface{}{
+				"module": "logstash",
+			},
+		},
+		"badly formed output": {
+			err: true,
+			t:   "output",
+			config: map[string]interface{}{
+				"nosubtype": "logstash",
+			},
+		},
+		"badly formed filebeat module": {
+			err: true,
+			t:   "filebeat.modules",
+			config: map[string]interface{}{
+				"nosubtype": "logstash",
+			},
+		},
+		"badly formed metricbeat module": {
+			err: true,
+			t:   "metricbeat.modules",
+			config: map[string]interface{}{
+				"nosubtype": "logstash",
+			},
+		},
+		"unknown type is passthrough": {
+			t: "unkown",
+			config: map[string]interface{}{
+				"nosubtype": "logstash",
+			},
+			expected: map[string]interface{}{
+				"nosubtype": "logstash",
+			},
+		},
+	}
+
+	for name, test := range tests {
+		test := test
+		t.Run(name, func(t *testing.T) {
+			converter := selectConverter(test.t)
+			newMap, err := converter(test.config)
+			if !assert.Equal(t, test.err, err != nil) {
+				return
+			}
+			assert.True(t, reflect.DeepEqual(newMap, test.expected))
+		})
+	}
+}

x-pack/libbeat/management/api/doc.go

@@ -0,0 +1,69 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+/*
+The Kibana CM Api returns a configuration format which cannot be ingested directly by our
+configuration parser, it need to be transformed from the generic format into an adapted format
+which is dependant on the type of configuration.
+
+
+Translations:
+
+Type: output
+
+{
+    "configuration_blocks": [
+
+        {
+            "config": {
+              "_sub_type": "elasticsearch"
+              "_id": "12312341231231"
+              "hosts": [ "localhost" ],
+              "password": "foobar"
+              "username": "elastic"
+            },
+            "type": "output"
+        }
+    ]
+}
+
+YAML representation:
+
+{
+	"elasticsearch": {
+		"hosts": [ "localhost" ],
+		"password": "foobar"
+		"username": "elastic"
+	}
+}
+
+
+Type: *.modules
+
+{
+    "configuration_blocks": [
+
+        {
+            "config": {
+              "_sub_type": "system"
+              "_id": "12312341231231"
+							"path" "foobar"
+            },
+            "type": "filebeat.module"
+        }
+    ]
+}
+
+YAML representation:
+
+[
+{
+	"module": "system"
+	"path": "foobar"
+}
+]
+
+*/
+
+package api