Fixes #1222 Applies config defaults

* Adds ApplyDefaults(map[string]ctypes.ConfigValue) to ConfigDataNode * Renames ReverseMerge to ReverseMergeInPlace on ConfigDataNode * Adds ReverseMerge to ConfigDataNode returning a copy of the merged node ReverseMerge, which now returns a copy, is called when validating deps (task creation) and when loading plugins.
intelsdi-x · Oct 5, 2016 · 068def9 · 068def9
1 parent 36fd32f
commit 068def9
Show file tree

Hide file tree

Showing 7 changed files with 71 additions and 12 deletions.
diff --git a/control/config.go b/control/config.go
@@ -482,7 +482,6 @@ func (p *pluginConfig) getPluginConfigDataNode(pluginType core.PluginType, name
 		}
 	}
 
-	//todo change to debug
 	log.WithFields(log.Fields{
 		"_block_":            "getPluginConfigDataNode",
 		"_module":            "config",

diff --git a/control/control.go b/control/control.go
@@ -713,6 +713,12 @@ func (p *pluginControl) getMetricsAndCollectors(requested []core.RequestedMetric
 			// set config to metric
 			mt.config = cfg
 
+			// apply defaults to the metric that may be present in the plugins
+			// configpolicy
+			if pluginCfg := mt.Plugin.ConfigPolicy.Get(mt.Namespace().Strings()); pluginCfg != nil {
+				mt.config.ApplyDefaults(pluginCfg.Defaults())
+			}
+
 			// loaded plugin which exposes the metric
 			lp := mt.Plugin
 			key := lp.Key()
@@ -911,6 +917,12 @@ func (p *pluginControl) CollectMetrics(id string, allTags map[string]map[string]
 
 	// For each available plugin call available plugin using RPC client and wait for response (goroutines)
 	for pluginKey, pmt := range pluginToMetricMap {
+		// merge global plugin config into the config for the metric
+		for _, mt := range pmt.metricTypes {
+			if mt.Config() != nil {
+				mt.Config().ReverseMergeInPlace(p.Config.Plugins.getPluginConfigDataNode(core.CollectorPluginType, pmt.plugin.Name(), pmt.plugin.Version()))
+			}
+		}
 
 		wg.Add(1)
 

diff --git a/control/control_test.go b/control/control_test.go
@@ -816,9 +816,6 @@ func TestMetricConfig(t *testing.T) {
 			errs := c.subscriptionGroups.validateMetric(m1)
 			So(errs, ShouldBeNil)
 		})
-		Convey("So mock should have name: bob config from defaults", func() {
-			So(c.Config.Plugins.pluginCache["0"+core.Separator+"mock"+core.Separator+"1"].Table()["name"], ShouldResemble, ctypes.ConfigValueStr{Value: "bob"})
-		})
 
 		c.Stop()
 		time.Sleep(100 * time.Millisecond)

diff --git a/control/plugin/cpolicy/node.go b/control/plugin/cpolicy/node.go
@@ -153,6 +153,18 @@ func (c *ConfigPolicyNode) HasRules() bool {
 	return false
 }
 
+// Defaults returns a map[string]ctypes.ConfigValue for all of the rules that
+// have defaults.
+func (c *ConfigPolicyNode) Defaults() map[string]ctypes.ConfigValue {
+	defaults := map[string]ctypes.ConfigValue{}
+	for name, rule := range c.rules {
+		if def := rule.Default(); def != nil {
+			defaults[name] = def
+		}
+	}
+	return defaults
+}
+
 // Validates and returns a processed policy node or nil and error if validation has failed
 func (c *ConfigPolicyNode) Process(m map[string]ctypes.ConfigValue) (*map[string]ctypes.ConfigValue, *ProcessingErrors) {
 	c.mutex.Lock()

diff --git a/control/plugin_manager.go b/control/plugin_manager.go
@@ -380,7 +380,7 @@ func (p *pluginManager) LoadPlugin(details *pluginDetails, emitter gomit.Emitter
 			}
 
 			// Update config policy with defaults
-			cfgNode.ReverseMerge(defaults)
+			cfgNode = cfgNode.ReverseMerge(defaults)
 			cp, err = c.GetConfigPolicy()
 			if err != nil {
 				pmLogger.WithFields(log.Fields{

diff --git a/control/subscription_group.go b/control/subscription_group.go
@@ -227,10 +227,10 @@ func (s *subscriptionGroups) ValidateDeps(requested []core.RequestedMetric,
 		if err != nil {
 			return []serror.SnapError{serror.New(err)}
 		}
-		plg.Config().ReverseMerge(
+		mergedConfig := plg.Config().ReverseMerge(
 			s.Config.Plugins.getPluginConfigDataNode(
 				typ, plg.Name(), plg.Version()))
-		errs := s.validatePluginSubscription(plg)
+		errs := s.validatePluginSubscription(plg, mergedConfig)
 		if len(errs) > 0 {
 			serrs = append(serrs, errs...)
 			return serrs
@@ -239,7 +239,7 @@ func (s *subscriptionGroups) ValidateDeps(requested []core.RequestedMetric,
 	return
 }
 
-func (p *subscriptionGroups) validatePluginSubscription(pl core.SubscribedPlugin) []serror.SnapError {
+func (p *subscriptionGroups) validatePluginSubscription(pl core.SubscribedPlugin, mergedConfig *cdata.ConfigDataNode) []serror.SnapError {
 	var serrs = []serror.SnapError{}
 	controlLogger.WithFields(log.Fields{
 		"_block": "validate-plugin-subscription",
@@ -259,7 +259,7 @@ func (p *subscriptionGroups) validatePluginSubscription(pl core.SubscribedPlugin
 
 	if lp.ConfigPolicy != nil {
 		ncd := lp.ConfigPolicy.Get([]string{""})
-		_, errs := ncd.Process(pl.Config().Table())
+		_, errs := ncd.Process(mergedConfig.Table())
 		if errs != nil && errs.HasErrors() {
 			for _, e := range errs.Errors() {
 				se := serror.New(e)
@@ -300,7 +300,7 @@ func (s *subscriptionGroups) validateMetric(
 
 	// merge global plugin config
 	if m.config != nil {
-		m.config.ReverseMerge(
+		m.config.ReverseMergeInPlace(
 			s.Config.Plugins.getPluginConfigDataNode(typ,
 				m.Plugin.Name(), m.Plugin.Version()))
 	} else {
@@ -342,10 +342,20 @@ func (s *subscriptionGroup) process(id string) (serrs []serror.SnapError) {
 		"metrics":    fmt.Sprintf("%+v", s.requestedMetrics),
 	}).Debug("gathered collectors")
 
-	//add processors and publishers to collectors just gathered
 	for _, plugin := range s.requestedPlugins {
+		//add processors and publishers to collectors just gathered
 		if plugin.TypeName() != core.CollectorPluginType.String() {
 			plugins = append(plugins, plugin)
+			// add defaults to plugins (exposed in a plugins ConfigPolicy)
+			if lp, err := s.pluginManager.get(
+				fmt.Sprintf("%s:%s:%d",
+					plugin.TypeName(),
+					plugin.Name(),
+					plugin.Version())); err == nil && lp.ConfigPolicy != nil {
+				if policy := lp.ConfigPolicy.Get([]string{""}); policy != nil && len(policy.Defaults()) > 0 {
+					plugin.Config().ApplyDefaults(policy.Defaults())
+				}
+			}
 		}
 	}
 

diff --git a/core/cdata/node.go b/core/cdata/node.go
@@ -142,7 +142,7 @@ func (c ConfigDataNode) Merge(n ctree.Node) ctree.Node {
 
 // Merges a ConfigDataNode with this one but does not overwrite any
 // conflicting values. Any conflicts are decided by the callers value.
-func (c *ConfigDataNode) ReverseMerge(n ctree.Node) ctree.Node {
+func (c *ConfigDataNode) ReverseMergeInPlace(n ctree.Node) ctree.Node {
 	cd := n.(*ConfigDataNode)
 	new_table := make(map[string]ctypes.ConfigValue)
 	// Lock here since we are modifying c.table
@@ -160,6 +160,35 @@ func (c *ConfigDataNode) ReverseMerge(n ctree.Node) ctree.Node {
 	return c
 }
 
+// Merges a ConfigDataNode with a copy of the current ConfigDataNode and returns
+// the copy.  The merge does not overwrite any conflicting values.
+// Any conflicts are decided by the callers value.
+func (c *ConfigDataNode) ReverseMerge(n ctree.Node) *ConfigDataNode {
+	cd := n.(*ConfigDataNode)
+	copy := NewNode()
+	t2 := c.table
+	for k, v := range cd.Table() {
+		copy.table[k] = v
+	}
+	for k, v := range t2 {
+		copy.table[k] = v
+	}
+	return copy
+}
+
+// ApplyDefaults will set default values if the given ConfigDataNode doesn't
+// already have a value for the given configuration.
+func (c *ConfigDataNode) ApplyDefaults(defaults map[string]ctypes.ConfigValue) {
+	// Lock here since we are modifying c.table
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+	for name, def := range defaults {
+		if _, ok := c.table[name]; !ok {
+			c.table[name] = def
+		}
+	}
+}
+
 // Deletes a field in ConfigDataNode. If the field does not exist Delete is
 // considered a no-op
 func (c ConfigDataNode) DeleteItem(k string) {