Simple conversion of the format returned by the API

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