Unset existing config blocks when they are missing

When a configuration block type (ie output) is not set in Central
Management, Kibana doesn't return it in the payload.

Before this change, Beats was not taking that into account os missing
block types were not reloaded in case they changed. In particular that
means that if an output configuration exists and is removed, Beats won't
apply that change.

This PR fixes that behavior by detecting missing types and applying nil
config to them.

Fixes elastic#9010

libbeat/common/reload/reload.go

@@ -110,6 +110,23 @@ func (r *Registry) MustRegisterList(name string, list ReloadableList) {
 	}
 }
 
+// GetRegisteredNames returns the list of names registered
+func (r *Registry) GetRegisteredNames() []string {
+	r.RLock()
+	defer r.RUnlock()
+	var names []string
+
+	for name := range r.confs {
+		names = append(names, name)
+	}
+
+	for name := range r.confsLists {
+		names = append(names, name)
+	}
+
+	return names
+}
+
 // GetReloadable returns the reloadable object with the given name, nil if not found
 func (r *Registry) GetReloadable(name string) Reloadable {
 	r.RLock()

libbeat/publisher/pipeline/controller.go

@@ -153,8 +153,10 @@ func makeWorkQueue() workQueue {
 func (c *outputController) Reload(cfg *reload.ConfigWithMeta) error {
 	outputCfg := common.ConfigNamespace{}
 
-	if err := cfg.Config.Unpack(&outputCfg); err != nil {
-		return err
+	if cfg != nil {
+		if err := cfg.Config.Unpack(&outputCfg); err != nil {
+			return err
+		}
 	}
 
 	output, err := loadOutput(c.beat, c.monitors, outputCfg)

x-pack/libbeat/management/manager.go

@@ -181,9 +181,24 @@ func (cm *ConfigManager) fetch() bool {
 
 func (cm *ConfigManager) apply() {
 	configOK := true
+
+	missing := map[string]bool{}
+	for _, name := range cm.registry.GetRegisteredNames() {
+		missing[name] = true
+	}
+
+	// Reload configs
 	for _, b := range cm.cache.Configs {
 		err := cm.reload(b.Type, b.Blocks)
 		configOK = configOK && err == nil
+		missing[b.Type] = false
+	}
+
+	// Unset missing configs
+	for name := range missing {
+		if missing[name] {
+			cm.reload(name, []*api.ConfigBlock{})
+		}
 	}
 
 	if !configOK {
@@ -199,15 +214,20 @@ func (cm *ConfigManager) reload(t string, blocks []*api.ConfigBlock) error {
 
 	if obj := cm.registry.GetReloadable(t); obj != nil {
 		// Single object
-		if len(blocks) != 1 {
+		if len(blocks) > 1 {
 			err := fmt.Errorf("got an invalid number of configs for %s: %d, expected: 1", t, len(blocks))
 			cm.logger.Error(err)
 			return err
 		}
-		config, err := blocks[0].ConfigWithMeta()
-		if err != nil {
-			cm.logger.Error(err)
-			return err
+
+		var config *reload.ConfigWithMeta
+		var err error
+		if len(blocks) == 1 {
+			config, err = blocks[0].ConfigWithMeta()
+			if err != nil {
+				cm.logger.Error(err)
+				return err
+			}
 		}
 
 		if err := obj.Reload(config); err != nil {