Merge pull request #25 from newrelic/gsanchez/fix/cache_update

Improve performance

accumulators/README.md

@@ -11,6 +11,5 @@ All metrics include the PCF meta data. Only PCFContainerMetric and PCFLogMessage
 | PCFContainerMetric | ContainerMetric | Application specific metrics | [`accumulators/container/container.go`](container/container.go)
 | PCFValueMetric | ValueMetric | PCF System metrics of multiple metric types | [`accumulators/value/value.go`](value/value.go)
 | PCFCounterEvent | CounterEvent | PCF System metrics as counter types only | [`accumulators/counter/counter.go`](counter/counter.go)
-| PCFCapacity | ValueMetric | PCF System metric, derived from Total and Remaining samples in order to provide percent used. | [`accumulators/capacity/capacity.go`](capacity/capacity.go)
 | PCFLogMessage | LogMessage | PCF Logs | [`accumulators/logmessage/logmessage.go`](logmessage/logmessage.go)
 | PCFHttpStartStop | HttpStartStop | PCF HTTP request details | [`accumulators/http/http.go`](http/http.go)

cfclient/cfapps/cache.go

@@ -15,13 +15,14 @@ type Cache struct {
 	Collection  map[string]*CFApp
 	WriteBuffer chan *CFApp
 	sync        *sync.RWMutex
+	isUpdating  bool
 }
 
 // NewCache ...
 func NewCache() *Cache {
 	cache := &Cache{
 		Collection:  map[string]*CFApp{},
-		WriteBuffer: make(chan *CFApp, 1024),
+		WriteBuffer: make(chan *CFApp, app.Get().Config.GetDuration("FIREHOSE_CACHE_WRITE_BUFFER_SIZE")),
 		sync:        &sync.RWMutex{},
 	}
 	cache.Start()
@@ -32,7 +33,8 @@ func NewCache() *Cache {
 func (c *Cache) Start() {
 	go func() {
 		cacheDuration := app.Get().Config.GetDuration("FIREHOSE_CACHE_DURATION_MINS")
-		update := time.NewTicker(30 * time.Second).C
+		cacheUpdate := app.Get().Config.GetDuration("FIREHOSE_CACHE_UPDATE_INTERVAL_SECS")
+		update := time.NewTicker(cacheUpdate * time.Second).C
 		timeoutCache := time.NewTicker(cacheDuration * time.Minute).C
 
 		for {
@@ -53,12 +55,9 @@ func (c *Cache) Start() {
 				c.sync.Unlock()
 
 			case <-update:
-				GetInstance().app.Log.Debug("Updating status of applications")
-				c.sync.RLock()
-				for _, v := range c.Collection {
-					v.UpdateInstances()
+				if !c.isUpdating {
+					go c.updateInstances()
 				}
-				c.sync.RUnlock()
 
 			case app := <-c.WriteBuffer:
 				c.sync.Lock()
@@ -69,6 +68,26 @@ func (c *Cache) Start() {
 	}()
 }
 
+func (c *Cache) updateInstances() {
+	c.isUpdating = true
+	GetInstance().app.Log.Debug("Updating status of applications")
+	now := time.Now()
+
+	var apps []*CFApp
+	c.sync.RLock()
+	for _, v := range c.Collection {
+		apps = append(apps, v)
+	}
+	c.sync.RUnlock()
+
+	for _, a := range apps {
+		a.UpdateInstances()
+	}
+
+	c.isUpdating = false
+	GetInstance().app.Log.Debugf("Finish cache updating %s", time.Since(now))
+}
+
 // Get ...
 func (c *Cache) Get(id string) (app *CFApp, found bool) {
 	c.sync.RLock()

config/config.go

@@ -74,6 +74,10 @@ func set() *Config {
 
 	// Cache purge threshold in minutes
 	v.SetDefault("FIREHOSE_CACHE_DURATION_MINS", 30)
+	// Cache instance update in seconds
+	v.SetDefault("FIREHOSE_CACHE_UPDATE_INTERVAL_SECS", 60)
+	// Cache instance update in seconds
+	v.SetDefault("FIREHOSE_CACHE_WRITE_BUFFER_SIZE", 2048)
 	// Rate limiter burst limit
 	v.SetDefault("FIREHOSE_RATE_BURST", 5)
 	// Rate limiter timeout in seconds.
@@ -92,8 +96,6 @@ func set() *Config {
 	v.SetDefault("NEWRELIC_DRAIN_INTERVAL", "59s")
 	v.SetDefault("NEWRELIC_ENQUEUE_TIMEOUT", "1s")
 
-	v.SetDefault("CAPACITY_ENTITY_AGE_MINS", 5)
-
 	v.SetDefault(NewRelicEventTypeContainer, "PCFContainerMetric")
 	v.SetDefault(NewRelicEventTypeValueMetric, "PCFValueMetric")
 	v.SetDefault(NewRelicEventTypeCounterEvent, "PCFCounterEvent")
@@ -128,7 +130,6 @@ func set() *Config {
 
 	// Filtering capabilities for envelope types - | separated values.
 	// By default, all message types are enabled.  User configurations will override this behavior.
-	// Capacity accumulator is enabled by default when ValueMetric is enabled.
 	v.SetDefault("ENABLED_ENVELOPE_TYPES", "ContainerMetric|CounterEvent|HttpStartStop|LogMessage|ValueMetric")
 
 	// Default account location will be US unless set to EU by cf push or tile.

dashboard/dashboard.json

@@ -19,7 +19,7 @@
           },
           "data": [
             {
-              "nrql": "SELECT count(*) AS 'Event Count' FROM PCFCapacity, PCFContainerMetric, PCFCounterEvent, PCFHttpStartStop, PCFLogMessage, PCFValueMetric facet pcf.domain"
+              "nrql": "SELECT count(*) AS 'Event Count' FROM PCFContainerMetric, PCFCounterEvent, PCFHttpStartStop, PCFLogMessage, PCFValueMetric facet pcf.domain"
             }
           ],
           "presentation": {
@@ -365,7 +365,7 @@
           },
           "data": [
             {
-              "nrql": "SELECT count(*) as 'All Events' FROM PCFCapacity, PCFContainerMetric, PCFCounterEvent, PCFHttpStartStop, PCFLogMessage, PCFValueMetric TIMESERIES"
+              "nrql": "SELECT count(*) as 'All Events' FROM PCFContainerMetric, PCFCounterEvent, PCFHttpStartStop, PCFLogMessage, PCFValueMetric TIMESERIES"
             }
           ],
           "presentation": {
@@ -376,7 +376,6 @@
       ],
       "filter": {
         "event_types": [
-          "PCFCapacity",
           "PCFContainerMetric",
           "PCFCounterEvent",
           "PCFHttpStartStop",

newrelic/registry/registry.go

@@ -4,7 +4,6 @@
 package registry
 
 import (
-	"github.com/newrelic/newrelic-pcf-nozzle-tile/accumulators/capacity"
 	"github.com/newrelic/newrelic-pcf-nozzle-tile/accumulators/container"
 	"github.com/newrelic/newrelic-pcf-nozzle-tile/accumulators/counter"
 	"github.com/newrelic/newrelic-pcf-nozzle-tile/accumulators/http"
@@ -21,7 +20,6 @@ var Register = &Accumulators{
 	counter.Metrics{},
 	container.Metrics{},
 	value.Metrics{},
-	capacity.Metrics{},
 	logmessage.Nrevents{},
 	http.Nrevents{},
 }