Config based routing

control/available_plugin.go

@@ -381,9 +381,19 @@ func (ap *availablePlugins) collectMetrics(pluginKey string, metricTypes []core.
 		return metricsFromCache, nil
 	}
 
+	config := metricTypes[0].Config()
+	cfg := map[string]ctypes.ConfigValue{}
+	if config != nil {
+		cfg = config.Table()
+	}
+	opts := strategy.SelectorValues{
+		Task:   taskID,
+		Config: cfg,
+	}
+
 	pool.RLock()
 	defer pool.RUnlock()
-	p, serr := pool.SelectAP(taskID)
+	p, serr := pool.SelectAP(opts)
 	if serr != nil {
 		return nil, serr
 	}
@@ -434,7 +444,13 @@ func (ap *availablePlugins) publishMetrics(contentType string, content []byte, p
 
 	pool.RLock()
 	defer pool.RUnlock()
-	p, err := pool.SelectAP(taskID)
+
+	opts := strategy.SelectorValues{
+		Task:   taskID,
+		Config: config,
+	}
+
+	p, err := pool.SelectAP(opts)
 	if err != nil {
 		errs = append(errs, err)
 		return errs
@@ -465,10 +481,13 @@ func (ap *availablePlugins) processMetrics(contentType string, content []byte, p
 	if pool == nil {
 		return "", nil, []error{serror.New(ErrPoolNotFound, map[string]interface{}{"pool-key": key})}
 	}
-
+	opts := strategy.SelectorValues{
+		Task:   taskID,
+		Config: config,
+	}
 	pool.RLock()
 	defer pool.RUnlock()
-	p, err := pool.SelectAP(taskID)
+	p, err := pool.SelectAP(opts)
 	if err != nil {
 		errs = append(errs, err)
 		return "", nil, errs

control/strategy/config_based.go

@@ -0,0 +1,150 @@
+/*
+http://www.apache.org/licenses/LICENSE-2.0.txt
+
+
+Copyright 2015 Intel Corporation
+
+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 of the 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 specific language governing permissions and
+limitations under the License.
+*/
+
+package strategy
+
+import (
+	log "github.com/Sirupsen/logrus"
+	"github.com/intelsdi-x/snap/core"
+)
+
+import (
+	"fmt"
+	"time"
+)
+
+// config-based provides a strategy that ... concurrency count is 1
+type configBased struct {
+	plugins     map[string]AvailablePlugin
+	metricCache map[string]*cache
+	logger      *log.Entry
+	cacheTTL    time.Duration
+}
+
+func NewConfigBased(cacheTTL time.Duration) *configBased {
+	return &configBased{
+		metricCache: make(map[string]*cache),
+		plugins:     make(map[string]AvailablePlugin),
+		cacheTTL:    cacheTTL,
+		logger: log.WithFields(log.Fields{
+			"_module": "control-routing",
+		}),
+	}
+}
+
+// Select selects an available plugin using the config based plugin strategy.
+func (cb *configBased) Select(aps []AvailablePlugin, id string) (AvailablePlugin, error) {
+	if ap, ok := cb.plugins[id]; ok && ap != nil {
+		return ap, nil
+	}
+
+	// add first one in case it's new id
+	for _, ap := range aps {
+		available := true
+		for _, busyPlugin := range cb.plugins {
+			if ap == busyPlugin {
+				available = false
+			}
+		}
+		if available {
+			cb.plugins[id] = ap
+			return ap, nil
+		}
+	}
+	cb.logger.WithFields(log.Fields{
+		"_block":   "findAvailablePlugin",
+		"strategy": cb.String(),
+		"error":    fmt.Sprintf("%v of %v plugins are available", len(aps)-len(cb.plugins), len(aps)),
+	}).Error(ErrCouldNotSelect)
+	return nil, ErrCouldNotSelect
+}
+
+// Remove selects a plugin and and removes it from the cache
+func (cb *configBased) Remove(aps []AvailablePlugin, id string) (AvailablePlugin, error) {
+	ap, err := cb.Select(aps, id)
+	if err != nil {
+		return nil, err
+	}
+	delete(cb.metricCache, id)
+	delete(cb.plugins, id)
+	return ap, nil
+}
+
+// String returns the strategy name.
+func (cb *configBased) String() string {
+	return "config-based"
+}
+
+// CacheTTL returns the TTL for the cache.
+func (cb *configBased) CacheTTL(id string) (time.Duration, error) {
+	return cb.cacheTTL, nil
+}
+
+// checkCache checks the cache for metric types.
+// returns:
+//  - array of metrics that need to be collected
+//  - array of metrics that were returned from the cache
+func (cb *configBased) CheckCache(mts []core.Metric, id string) ([]core.Metric, []core.Metric) {
+	if _, ok := cb.metricCache[id]; !ok {
+		cb.metricCache[id] = NewCache(cb.cacheTTL)
+	}
+	return cb.metricCache[id].checkCache(mts)
+}
+
+// updateCache updates the cache with the given array of metrics.
+func (cb *configBased) UpdateCache(mts []core.Metric, id string) {
+	if _, ok := cb.metricCache[id]; !ok {
+		cb.metricCache[id] = NewCache(cb.cacheTTL)
+	}
+	cb.metricCache[id].updateCache(mts)
+}
+
+// AllCacheHits returns cache hits across all metrics.
+func (cb *configBased) AllCacheHits() uint64 {
+	var total uint64
+	for _, cache := range cb.metricCache {
+		total += cache.allCacheHits()
+	}
+	return total
+}
+
+// AllCacheMisses returns cache misses across all metrics.
+func (cb *configBased) AllCacheMisses() uint64 {
+	var total uint64
+	for _, cache := range cb.metricCache {
+		total += cache.allCacheMisses()
+	}
+	return total
+}
+
+// CacheHits returns the cache hits for a given metric namespace and version.
+func (cb *configBased) CacheHits(ns string, version int, id string) (uint64, error) {
+	if cache, ok := cb.metricCache[id]; ok {
+		return cache.cacheHits(ns, version)
+	}
+	return 0, ErrCacheDoesNotExist
+}
+
+// CacheMisses returns the cache misses for a given metric namespace and version.
+func (cb *configBased) CacheMisses(ns string, version int, id string) (uint64, error) {
+	if cache, ok := cb.metricCache[id]; ok {
+		return cache.cacheMisses(ns, version)
+	}
+	return 0, ErrCacheDoesNotExist
+}

control/strategy/config_based_test.go

@@ -0,0 +1,55 @@
+/*
+http://www.apache.org/licenses/LICENSE-2.0.txt
+
+Copyright 2016 Intel Corporation
+
+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 of the 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 specific language governing permissions and
+limitations under the License.
+*/
+
+package strategy
+
+import (
+	"testing"
+	"time"
+
+	. "github.com/intelsdi-x/snap/control/strategy/fixtures"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestConfigBasedRouter(t *testing.T) {
+	Convey("Given a sticky router", t, func() {
+		router := NewConfigBased(100 * time.Millisecond)
+		So(router, ShouldNotBeNil)
+		So(router.String(), ShouldResemble, "config-based")
+		Convey("Select a plugin when they are available", func() {
+			p1 := NewMockAvailablePlugin().WithName("p1")
+			p2 := NewMockAvailablePlugin().WithName("p2")
+			// select a plugin, for task1,  given a task and two available plugins
+			sp1, err := router.Select([]AvailablePlugin{p1, p2}, "cfg1")
+			So(err, ShouldBeNil)
+			So(sp1, ShouldNotBeNil)
+			So(sp1, ShouldEqual, p1)
+			// change the order of the plugins provided to the select
+			sp2, err := router.Select([]AvailablePlugin{p2, p1}, "cfg1")
+			So(err, ShouldBeNil)
+			So(sp1, ShouldNotBeNil)
+			So(sp2, ShouldEqual, p1)
+			// select the other (last) available plugin for task2
+			sp3, err := router.Select([]AvailablePlugin{p2, p1}, "cfg2")
+			So(err, ShouldBeNil)
+			So(sp3, ShouldNotBeNil)
+			So(sp3, ShouldEqual, p2)
+		})
+
+	})
+}