Cache Netatmo data in memory

collector.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"sync"
 	"time"
 
 	netatmo "github.com/exzz/netatmo-api-go"
@@ -11,7 +12,18 @@ import (
 var (
 	prefix        = "netatmo_"
 	netatmoUpDesc = prometheus.NewDesc(prefix+"up",
-		"Zero if there was an error scraping the Netatmo API.",
+		"Zero if there was an error during the last refresh try.",
+		nil, nil)
+
+	refreshPrefix        = prefix + "last_refresh"
+	refreshTimestampDesc = prometheus.NewDesc(
+		refreshPrefix+"_time",
+		"Contains the time of the last refresh try, successful or not.",
+		nil, nil)
+
+	cacheTimestampDesc = prometheus.NewDesc(
+		prefix+"cache_updated_time",
+		"Contains the time of the cached data.",
 		nil, nil)
 
 	varLabels = []string{
@@ -93,9 +105,15 @@ var (
 )
 
 type netatmoCollector struct {
-	log            logrus.FieldLogger
+	log              logrus.FieldLogger
-	staleThreshold time.Duration
+	refreshInterval  time.Duration
-	client         *netatmo.Client
+	staleThreshold   time.Duration
+	client           *netatmo.Client
+	lastRefresh      time.Time
+	lastRefreshError error
+	cacheLock        sync.RWMutex
+	cacheTimestamp   time.Time
+	cachedData       *netatmo.DeviceCollection
 }
 
 func (c *netatmoCollector) Describe(dChan chan<- *prometheus.Desc) {
@@ -106,25 +124,51 @@ func (c *netatmoCollector) Describe(dChan chan<- *prometheus.Desc) {
 }
 
 func (c *netatmoCollector) Collect(mChan chan<- prometheus.Metric) {
-	devices, err := c.client.Read()
+	now := time.Now()
-	if err != nil {
+	if now.Sub(c.lastRefresh) >= c.refreshInterval {
-		c.log.Errorf("Error getting data: %s", err)
+		go c.refreshData(now)
-
-		c.sendMetric(mChan, netatmoUpDesc, prometheus.GaugeValue, 0.0)
-		return
 	}
-	c.sendMetric(mChan, netatmoUpDesc, prometheus.GaugeValue, 1.0)
 
-	for _, dev := range devices.Devices() {
+	upValue := 1.0
-		stationName := dev.StationName
+	if c.lastRefresh.IsZero() || c.lastRefreshError != nil {
-		c.collectData(mChan, dev, stationName)
+		upValue = 0
+	}
+	c.sendMetric(mChan, netatmoUpDesc, prometheus.GaugeValue, upValue)
+	c.sendMetric(mChan, refreshTimestampDesc, prometheus.GaugeValue, convertTime(c.lastRefresh))
 
-		for _, module := range dev.LinkedModules {
+	c.cacheLock.RLock()
-			c.collectData(mChan, module, stationName)
+	defer c.cacheLock.RUnlock()
+
+	c.sendMetric(mChan, cacheTimestampDesc, prometheus.GaugeValue, convertTime(c.cacheTimestamp))
+	if c.cachedData != nil {
+		for _, dev := range c.cachedData.Devices() {
+			stationName := dev.StationName
+			c.collectData(mChan, dev, stationName)
+
+			for _, module := range dev.LinkedModules {
+				c.collectData(mChan, module, stationName)
+			}
 		}
 	}
 }
 
+func (c *netatmoCollector) refreshData(now time.Time) {
+	c.log.Debugf("Refresh interval elapsed: %s > %s", now.Sub(c.lastRefresh), c.refreshInterval)
+	c.lastRefresh = now
+
+	devices, err := c.client.Read()
+	if err != nil {
+		c.log.Errorf("Error during refresh: %s", err)
+		c.lastRefreshError = err
+		return
+	}
+
+	c.cacheLock.Lock()
+	defer c.cacheLock.Unlock()
+	c.cacheTimestamp = now
+	c.cachedData = devices
+}
+
 func (c *netatmoCollector) collectData(ch chan<- prometheus.Metric, device *netatmo.Device, stationName string) {
 	moduleName := device.ModuleName
 	data := device.DashboardData
@@ -193,3 +237,11 @@ func (c *netatmoCollector) sendMetric(ch chan<- prometheus.Metric, desc *prometh
 	}
 	ch <- m
 }
+
+func convertTime(t time.Time) float64 {
+	if t.IsZero() {
+		return 0.0
+	}
+
+	return float64(t.Unix())
+}

main.go

@@ -36,9 +36,10 @@ func main() {
 	}
 
 	metrics := &netatmoCollector{
-		log:            log,
+		log:             log,
-		client:         client,
+		client:          client,
-		staleThreshold: cfg.StaleDuration,
+		refreshInterval: cfg.RefreshInterval,
+		staleThreshold:  cfg.StaleDuration,
 	}
 	prometheus.MustRegister(metrics)