Merge branch 'develop' into main

- MetricRouter: Fix interval_timestamp option
- InfluxSink & InfluxAsyncSink: Add own flush mechanism
- InfluxSink & InfluxAsyncSink: Use default options, overwrite if
configured otherwise
- InfinibandCollector: Add units to metrics
- LikwidCollector: Fix for dying accessdaemon

README.md

@@ -20,7 +20,6 @@ There is a main configuration file with basic settings that point to the other c
   "collectors" : "collectors.json",
   "receivers" : "receivers.json",
   "router" : "router.json",
-  "stats_api" : "api.json",
   "interval": 10,
   "duration": 1
 }
@@ -33,7 +32,6 @@ See the component READMEs for their configuration:
 * [`sinks`](./sinks/README.md)
 * [`receivers`](./receivers/README.md)
 * [`router`](./internal/metricRouter/README.md)
-* [`stats_api`](./internal/metricRouter/StatsApi.md)
 
 
 # Installation

cc-metric-collector.go

@@ -28,7 +28,6 @@ type CentralConfigFile struct {
 	RouterConfigFile    string `json:"router"`
 	SinkConfigFile      string `json:"sinks"`
 	ReceiverConfigFile  string `json:"receivers,omitempty"`
-	StatsApiConfigFile  string `json:"stats_api,omitempty"`
 }
 
 func LoadCentralConfiguration(file string, config *CentralConfigFile) error {
@@ -53,7 +52,6 @@ type RuntimeConfig struct {
 	CollectManager  collectors.CollectorManager
 	SinkManager     sinks.SinkManager
 	ReceiveManager  receivers.ReceiveManager
-	StatsApi        mr.StatsApi
 	MultiChanTicker mct.MultiChanTicker
 
 	Channels []chan lp.CCMetric
@@ -154,16 +152,11 @@ func shutdownHandler(config *RuntimeConfig, shutdownSignal chan os.Signal) {
 		cclog.Debug("Shutdown SinkManager...")
 		config.SinkManager.Close()
 	}
-	if config.StatsApi != nil {
-		cclog.Debug("Shutdown StatsApi...")
-		config.StatsApi.Close()
-	}
 }
 
 func mainFunc() int {
 	var err error
 	use_recv := false
-	use_api := false
 
 	// Initialize runtime configuration
 	rcfg := RuntimeConfig{
@@ -171,7 +164,6 @@ func mainFunc() int {
 		CollectManager: nil,
 		SinkManager:    nil,
 		ReceiveManager: nil,
-		StatsApi:       nil,
 		CliArgs:        ReadCli(),
 	}
 
@@ -261,16 +253,6 @@ func mainFunc() int {
 		use_recv = true
 	}
 
-	// Create new statistics API manager
-	if len(rcfg.ConfigFile.StatsApiConfigFile) > 0 {
-		rcfg.StatsApi, err = mr.NewStatsApi(rcfg.MultiChanTicker, &rcfg.Sync, rcfg.ConfigFile.StatsApiConfigFile)
-		if err != nil {
-			cclog.Error(err.Error())
-			return 1
-		}
-		use_api = true
-	}
-
 	// Create shutdown handler
 	shutdownSignal := make(chan os.Signal, 1)
 	signal.Notify(shutdownSignal, os.Interrupt)
@@ -278,11 +260,6 @@ func mainFunc() int {
 	rcfg.Sync.Add(1)
 	go shutdownHandler(&rcfg, shutdownSignal)
 
-	// Start the stats api early to be prepared for init settings
-	if use_api {
-		rcfg.StatsApi.Start()
-	}
-
 	// Start the managers
 	rcfg.MetricRouter.Start()
 	rcfg.SinkManager.Start()

collectors/beegfsmetaMetric.go

@@ -16,7 +16,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const DEFAULT_BEEGFS_CMD = "beegfs-ctl"
@@ -30,11 +29,10 @@ type BeegfsMetaCollectorConfig struct {
 
 type BeegfsMetaCollector struct {
 	metricCollector
-	tags                  map[string]string
+	tags    map[string]string
-	matches               map[string]string
+	matches map[string]string
-	config                BeegfsMetaCollectorConfig
+	config  BeegfsMetaCollectorConfig
-	skipFS                map[string]struct{}
+	skipFS  map[string]struct{}
-	statsProcessedMetrics int64
 }
 
 func (m *BeegfsMetaCollector) Init(config json.RawMessage) error {
@@ -107,7 +105,6 @@ func (m *BeegfsMetaCollector) Init(config json.RawMessage) error {
 	if err != nil {
 		return fmt.Errorf("BeegfsMetaCollector.Init(): Failed to find beegfs-ctl binary '%s': %v", m.config.Beegfs, err)
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -221,12 +218,10 @@ func (m *BeegfsMetaCollector) Read(interval time.Duration, output chan lp.CCMetr
 				y, err := lp.New(key, m.tags, m.meta, map[string]interface{}{"value": value}, time.Now())
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *BeegfsMetaCollector) Close() {

collectors/beegfsstorageMetric.go

@@ -16,7 +16,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 // Struct for the collector-specific JSON config
@@ -28,11 +27,10 @@ type BeegfsStorageCollectorConfig struct {
 
 type BeegfsStorageCollector struct {
 	metricCollector
-	tags                  map[string]string
+	tags    map[string]string
-	matches               map[string]string
+	matches map[string]string
-	config                BeegfsStorageCollectorConfig
+	config  BeegfsStorageCollectorConfig
-	skipFS                map[string]struct{}
+	skipFS  map[string]struct{}
-	statsProcessedMetrics int64
 }
 
 func (m *BeegfsStorageCollector) Init(config json.RawMessage) error {
@@ -100,7 +98,6 @@ func (m *BeegfsStorageCollector) Init(config json.RawMessage) error {
 	if err != nil {
 		return fmt.Errorf("BeegfsStorageCollector.Init(): Failed to find beegfs-ctl binary '%s': %v", m.config.Beegfs, err)
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -213,12 +210,10 @@ func (m *BeegfsStorageCollector) Read(interval time.Duration, output chan lp.CCM
 				y, err := lp.New(key, m.tags, m.meta, map[string]interface{}{"value": value}, time.Now())
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *BeegfsStorageCollector) Close() {

collectors/cpufreqCpuinfoMetric.go

@@ -12,7 +12,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 //
@@ -37,8 +36,7 @@ type CPUFreqCpuInfoCollectorTopology struct {
 
 type CPUFreqCpuInfoCollector struct {
 	metricCollector
-	topology              []*CPUFreqCpuInfoCollectorTopology
+	topology []*CPUFreqCpuInfoCollectorTopology
-	statsProcessedMetrics int64
 }
 
 func (m *CPUFreqCpuInfoCollector) Init(config json.RawMessage) error {
@@ -157,7 +155,7 @@ func (m *CPUFreqCpuInfoCollector) Init(config json.RawMessage) error {
 			"package_id": t.physicalPackageID,
 		}
 	}
-	m.statsProcessedMetrics = 0
+
 	m.init = true
 	return nil
 }
@@ -198,7 +196,6 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
 						return
 					}
 					if y, err := lp.New("cpufreq", t.tagSet, m.meta, map[string]interface{}{"value": value}, now); err == nil {
-						m.statsProcessedMetrics++
 						output <- y
 					}
 				}
@@ -206,7 +203,6 @@ func (m *CPUFreqCpuInfoCollector) Read(interval time.Duration, output chan lp.CC
 			}
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *CPUFreqCpuInfoCollector) Close() {

collectors/cpufreqMetric.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	"golang.org/x/sys/unix"
 )
 
@@ -40,9 +39,8 @@ type CPUFreqCollectorTopology struct {
 //
 type CPUFreqCollector struct {
 	metricCollector
-	topology              []CPUFreqCollectorTopology
+	topology []CPUFreqCollectorTopology
-	statsProcessedMetrics int64
+	config   struct {
-	config                struct {
 		ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
 	}
 }
@@ -168,7 +166,7 @@ func (m *CPUFreqCollector) Init(config json.RawMessage) error {
 			"package_id": t.physicalPackageID,
 		}
 	}
-	m.statsProcessedMetrics = 0
+
 	m.init = true
 	return nil
 }
@@ -205,11 +203,9 @@ func (m *CPUFreqCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		}
 
 		if y, err := lp.New("cpufreq", t.tagSet, m.meta, map[string]interface{}{"value": cpuFreq}, now); err == nil {
-			m.statsProcessedMetrics++
 			output <- y
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *CPUFreqCollector) Close() {

collectors/cpustatMetric.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const CPUSTATFILE = `/proc/stat`
@@ -22,11 +21,10 @@ type CpustatCollectorConfig struct {
 
 type CpustatCollector struct {
 	metricCollector
-	config                CpustatCollectorConfig
+	config   CpustatCollectorConfig
-	matches               map[string]int
+	matches  map[string]int
-	cputags               map[string]map[string]string
+	cputags  map[string]map[string]string
-	nodetags              map[string]string
+	nodetags map[string]string
-	statsProcessedMetrics int64
 }
 
 func (m *CpustatCollector) Init(config json.RawMessage) error {
@@ -88,7 +86,6 @@ func (m *CpustatCollector) Init(config json.RawMessage) error {
 			num_cpus++
 		}
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -109,7 +106,6 @@ func (m *CpustatCollector) parseStatLine(linefields []string, tags map[string]st
 	for name, value := range values {
 		y, err := lp.New(name, tags, m.meta, map[string]interface{}{"value": (value * 100.0) / total}, t)
 		if err == nil {
-			m.statsProcessedMetrics++
 			output <- y
 		}
 	}
@@ -145,10 +141,8 @@ func (m *CpustatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		time.Now(),
 	)
 	if err == nil {
-		m.statsProcessedMetrics++
 		output <- num_cpus_metric
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *CpustatCollector) Close() {

collectors/customCmdMetric.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	influx "github.com/influxdata/line-protocol"
 )
 
@@ -24,14 +23,11 @@ type CustomCmdCollectorConfig struct {
 
 type CustomCmdCollector struct {
 	metricCollector
-	handler                *influx.MetricHandler
+	handler  *influx.MetricHandler
-	parser                 *influx.Parser
+	parser   *influx.Parser
-	config                 CustomCmdCollectorConfig
+	config   CustomCmdCollectorConfig
-	commands               []string
+	commands []string
-	files                  []string
+	files    []string
-	statsProcessedMetrics  int64
-	statsProcessedCommands int64
-	statsProcessedFiles    int64
 }
 
 func (m *CustomCmdCollector) Init(config json.RawMessage) error {
@@ -70,9 +66,6 @@ func (m *CustomCmdCollector) Init(config json.RawMessage) error {
 	m.handler = influx.NewMetricHandler()
 	m.parser = influx.NewParser(m.handler)
 	m.parser.SetTimeFunc(DefaultTime)
-	m.statsProcessedMetrics = 0
-	m.statsProcessedFiles = 0
-	m.statsProcessedCommands = 0
 	m.init = true
 	return nil
 }
@@ -107,13 +100,9 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMetri
 
 			y := lp.FromInfluxMetric(c)
 			if err == nil {
-				m.statsProcessedMetrics++
-				stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 				output <- y
 			}
 		}
-		m.statsProcessedCommands++
-		stats.ComponentStatInt(m.name, "processed_commands", m.statsProcessedCommands)
 	}
 	for _, file := range m.files {
 		buffer, err := ioutil.ReadFile(file)
@@ -133,13 +122,9 @@ func (m *CustomCmdCollector) Read(interval time.Duration, output chan lp.CCMetri
 			}
 			y := lp.FromInfluxMetric(f)
 			if err == nil {
-				m.statsProcessedMetrics++
-				stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 				output <- y
 			}
 		}
-		m.statsProcessedFiles++
-		stats.ComponentStatInt(m.name, "processed_files", m.statsProcessedFiles)
 	}
 }
 

collectors/diskstatMetric.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 //	"log"
@@ -24,8 +23,9 @@ type DiskstatCollectorConfig struct {
 
 type DiskstatCollector struct {
 	metricCollector
-	config                DiskstatCollectorConfig
+	//matches map[string]int
-	statsProcessedMetrics int64
+	config IOstatCollectorConfig
+	//devices map[string]IOstatCollectorEntry
 }
 
 func (m *DiskstatCollector) Init(config json.RawMessage) error {
@@ -44,7 +44,6 @@ func (m *DiskstatCollector) Init(config json.RawMessage) error {
 		return err
 	}
 	defer file.Close()
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -90,16 +89,12 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMetric
 		y, err := lp.New("disk_total", tags, m.meta, map[string]interface{}{"value": total}, time.Now())
 		if err == nil {
 			y.AddMeta("unit", "GBytes")
-			m.statsProcessedMetrics++
-			stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 			output <- y
 		}
 		free := (stat.Bfree * uint64(stat.Bsize)) / uint64(1000000000)
 		y, err = lp.New("disk_free", tags, m.meta, map[string]interface{}{"value": free}, time.Now())
 		if err == nil {
 			y.AddMeta("unit", "GBytes")
-			m.statsProcessedMetrics++
-			stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 			output <- y
 		}
 		perc := (100 * (total - free)) / total
@@ -110,8 +105,6 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMetric
 	y, err := lp.New("part_max_used", map[string]string{"type": "node"}, m.meta, map[string]interface{}{"value": int(part_max_used)}, time.Now())
 	if err == nil {
 		y.AddMeta("unit", "percent")
-		m.statsProcessedMetrics++
-		stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 		output <- y
 	}
 }

collectors/gpfsMetric.go

@@ -15,7 +15,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const DEFAULT_GPFS_CMD = "mmpmon"
@@ -33,10 +32,9 @@ type GpfsCollector struct {
 		ExcludeFilesystem []string `json:"exclude_filesystem,omitempty"`
 		SendBandwidths    bool     `json:"send_bandwidths"`
 	}
-	skipFS                map[string]struct{}
+	skipFS        map[string]struct{}
-	lastTimestamp         time.Time // Store time stamp of last tick to derive bandwidths
+	lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
-	lastState             map[string]GpfsCollectorLastState
+	lastState     map[string]GpfsCollectorLastState
-	statsProcessedMetrics int64
 }
 
 func (m *GpfsCollector) Init(config json.RawMessage) error {
@@ -88,7 +86,7 @@ func (m *GpfsCollector) Init(config json.RawMessage) error {
 		return fmt.Errorf("failed to find mmpmon binary '%s': %v", m.config.Mmpmon, err)
 	}
 	m.config.Mmpmon = p
-	m.statsProcessedMetrics = 0
+
 	m.init = true
 	return nil
 }
@@ -213,14 +211,12 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_bytes_read", m.tags, m.meta, map[string]interface{}{"value": bytesRead}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 		if m.config.SendBandwidths {
 			if lastBytesRead := m.lastState[filesystem].bytesRead; lastBytesRead >= 0 {
 				bwRead := float64(bytesRead-lastBytesRead) / timeDiff
 				if y, err := lp.New("gpfs_bw_read", m.tags, m.meta, map[string]interface{}{"value": bwRead}, timestamp); err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -235,14 +231,12 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_bytes_written", m.tags, m.meta, map[string]interface{}{"value": bytesWritten}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 		if m.config.SendBandwidths {
 			if lastBytesWritten := m.lastState[filesystem].bytesRead; lastBytesWritten >= 0 {
 				bwWrite := float64(bytesWritten-lastBytesWritten) / timeDiff
 				if y, err := lp.New("gpfs_bw_write", m.tags, m.meta, map[string]interface{}{"value": bwWrite}, timestamp); err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -264,7 +258,6 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_num_opens", m.tags, m.meta, map[string]interface{}{"value": numOpens}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 
 		// number of closes
@@ -277,7 +270,6 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_num_closes", m.tags, m.meta, map[string]interface{}{"value": numCloses}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 
 		// number of reads
@@ -290,7 +282,6 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_num_reads", m.tags, m.meta, map[string]interface{}{"value": numReads}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 
 		// number of writes
@@ -303,7 +294,6 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_num_writes", m.tags, m.meta, map[string]interface{}{"value": numWrites}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 
 		// number of read directories
@@ -316,7 +306,6 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		}
 		if y, err := lp.New("gpfs_num_readdirs", m.tags, m.meta, map[string]interface{}{"value": numReaddirs}, timestamp); err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 
 		// Number of inode updates
@@ -328,11 +317,9 @@ func (m *GpfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 			continue
 		}
 		if y, err := lp.New("gpfs_num_inode_updates", m.tags, m.meta, map[string]interface{}{"value": numInodeUpdates}, timestamp); err == nil {
-			m.statsProcessedMetrics++
 			output <- y
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *GpfsCollector) Close() {

collectors/infinibandMetric.go

@@ -7,7 +7,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	"golang.org/x/sys/unix"
 
 	"encoding/json"
@@ -40,9 +39,8 @@ type InfinibandCollector struct {
 		SendAbsoluteValues bool     `json:"send_abs_values"`           // Send absolut values as read from sys filesystem
 		SendDerivedValues  bool     `json:"send_derived_values"`       // Send derived values e.g. rates
 	}
-	info                  []*InfinibandCollectorInfo
+	info          []*InfinibandCollectorInfo
-	lastTimestamp         time.Time // Store time stamp of last tick to derive bandwidths
+	lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
-	statsProcessedMetrics int64
 }
 
 // Init initializes the Infiniband collector by walking through files below IB_BASEPATH
@@ -151,7 +149,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
 	if len(m.info) == 0 {
 		return fmt.Errorf("found no IB devices")
 	}
-	m.statsProcessedMetrics = 0
+
 	m.init = true
 	return nil
 }
@@ -198,7 +196,6 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMetr
 				if y, err := lp.New(counterName, info.tagSet, m.meta, map[string]interface{}{"value": v}, now); err == nil {
 					y.AddMeta("unit", counterDef.unit)
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 
@@ -209,7 +206,6 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMetr
 					if y, err := lp.New(counterName+"_bw", info.tagSet, m.meta, map[string]interface{}{"value": rate}, now); err == nil {
 						y.AddMeta("unit", counterDef.unit+"/sec")
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 				// Save current state
@@ -218,7 +214,6 @@ func (m *InfinibandCollector) Read(interval time.Duration, output chan lp.CCMetr
 		}
 
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *InfinibandCollector) Close() {

collectors/iostatMetric.go

@@ -6,7 +6,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 
 	//	"log"
 	"encoding/json"
@@ -30,10 +29,9 @@ type IOstatCollectorEntry struct {
 
 type IOstatCollector struct {
 	metricCollector
-	matches               map[string]int
+	matches map[string]int
-	config                IOstatCollectorConfig
+	config  IOstatCollectorConfig
-	devices               map[string]IOstatCollectorEntry
+	devices map[string]IOstatCollectorEntry
-	statsProcessedMetrics int64
 }
 
 func (m *IOstatCollector) Init(config json.RawMessage) error {
@@ -104,7 +102,6 @@ func (m *IOstatCollector) Init(config json.RawMessage) error {
 			lastValues: values,
 		}
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return err
 }
@@ -144,7 +141,6 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 					y, err := lp.New(name, entry.tags, m.meta, map[string]interface{}{"value": int(diff)}, time.Now())
 					if err == nil {
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 				entry.lastValues[name] = x
@@ -152,7 +148,6 @@ func (m *IOstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		}
 		m.devices[device] = entry
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *IOstatCollector) Close() {

collectors/ipmiMetric.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const IPMITOOL_PATH = `ipmitool`
@@ -27,10 +26,9 @@ type IpmiCollector struct {
 	metricCollector
 	//tags        map[string]string
 	//matches     map[string]string
-	config                IpmiCollectorConfig
+	config      IpmiCollectorConfig
-	ipmitool              string
+	ipmitool    string
-	ipmisensors           string
+	ipmisensors string
-	statsProcessedMetrics int64
 }
 
 func (m *IpmiCollector) Init(config json.RawMessage) error {
@@ -58,7 +56,6 @@ func (m *IpmiCollector) Init(config json.RawMessage) error {
 	if len(m.ipmitool) == 0 && len(m.ipmisensors) == 0 {
 		return errors.New("no IPMI reader found")
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -97,7 +94,6 @@ func (m *IpmiCollector) readIpmiTool(cmd string, output chan lp.CCMetric) {
 			if err == nil {
 				y.AddMeta("unit", unit)
 				output <- y
-				m.statsProcessedMetrics++
 			}
 		}
 	}
@@ -127,7 +123,6 @@ func (m *IpmiCollector) readIpmiSensors(cmd string, output chan lp.CCMetric) {
 						y.AddMeta("unit", lv[4])
 					}
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -146,7 +141,6 @@ func (m *IpmiCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 			m.readIpmiSensors(m.config.IpmisensorsPath, output)
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *IpmiCollector) Close() {

collectors/likwidMetric.go

@@ -28,7 +28,6 @@ import (
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
 	topo "github.com/ClusterCockpit/cc-metric-collector/internal/ccTopology"
 	agg "github.com/ClusterCockpit/cc-metric-collector/internal/metricAggregator"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	"github.com/NVIDIA/go-nvml/pkg/dl"
 )
 
@@ -73,21 +72,18 @@ type LikwidCollectorConfig struct {
 
 type LikwidCollector struct {
 	metricCollector
-	cpulist               []C.int
+	cpulist      []C.int
-	cpu2tid               map[int]int
+	cpu2tid      map[int]int
-	sock2tid              map[int]int
+	sock2tid     map[int]int
-	metrics               map[C.int]map[string]int
+	metrics      map[C.int]map[string]int
-	groups                []C.int
+	groups       []C.int
-	config                LikwidCollectorConfig
+	config       LikwidCollectorConfig
-	gmresults             map[int]map[string]float64
+	gmresults    map[int]map[string]float64
-	basefreq              float64
+	basefreq     float64
-	running               bool
+	running      bool
-	initialized           bool
+	initialized  bool
-	likwidGroups          map[C.int]LikwidEventsetConfig
+	likwidGroups map[C.int]LikwidEventsetConfig
-	lock                  sync.Mutex
+	lock         sync.Mutex
-	statsMeasurements     int64
-	statsProcessedMetrics int64
-	statsPublishedMetrics int64
 }
 
 type LikwidMetric struct {
@@ -271,9 +267,6 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 		cclog.ComponentError(m.name, err.Error())
 		return err
 	}
-	m.statsMeasurements = 0
-	m.statsProcessedMetrics = 0
-	m.statsPublishedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -281,7 +274,6 @@ func (m *LikwidCollector) Init(config json.RawMessage) error {
 // take a measurement for 'interval' seconds of event set index 'group'
 func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval time.Duration) (bool, error) {
 	var ret C.int
-
 	m.lock.Lock()
 	if m.initialized {
 		ret = C.perfmon_setupCounters(evset.gid)
@@ -325,8 +317,6 @@ func (m *LikwidCollector) takeMeasurement(evset LikwidEventsetConfig, interval t
 		}
 	}
 	m.lock.Unlock()
-	m.statsMeasurements++
-	stats.ComponentStatInt(m.name, "measurements", m.statsMeasurements)
 	return false, nil
 }
 
@@ -339,7 +329,11 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
 		gctr := C.GoString(counter)
 		for _, tid := range m.cpu2tid {
 			res := C.perfmon_getLastResult(evset.gid, C.int(eidx), C.int(tid))
-			evset.results[tid][gctr] = float64(res)
+			fres := float64(res)
+			if m.config.InvalidToZero && (math.IsNaN(fres) || math.IsInf(fres, 0)) {
+				fres = 0.0
+			}
+			evset.results[tid][gctr] = fres
 			evset.results[tid]["time"] = interval.Seconds()
 			evset.results[tid]["inverseClock"] = invClock
 		}
@@ -358,17 +352,12 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
 				value, err := agg.EvalFloat64Condition(metric.Calc, evset.results[tid])
 				if err != nil {
 					cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
-					continue
+					value = 0.0
+				}
+				if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
+					value = 0.0
 				}
 				evset.metrics[tid][metric.Name] = value
-				if m.config.InvalidToZero && math.IsNaN(value) {
-					value = 0.0
-				}
-				if m.config.InvalidToZero && math.IsInf(value, 0) {
-					value = 0.0
-				}
-				m.statsProcessedMetrics++
-				stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 				// Now we have the result, send it with the proper tags
 				if !math.IsNaN(value) {
 					if metric.Publish {
@@ -381,8 +370,6 @@ func (m *LikwidCollector) calcEventsetMetrics(evset LikwidEventsetConfig, interv
 							if len(metric.Unit) > 0 {
 								y.AddMeta("unit", metric.Unit)
 							}
-							m.statsPublishedMetrics++
-							stats.ComponentStatInt(m.name, "published_metrics", m.statsPublishedMetrics)
 							output <- y
 						}
 					}
@@ -414,17 +401,12 @@ func (m *LikwidCollector) calcGlobalMetrics(interval time.Duration, output chan
 				value, err := agg.EvalFloat64Condition(metric.Calc, params)
 				if err != nil {
 					cclog.ComponentError(m.name, "Calculation for metric", metric.Name, "failed:", err.Error())
-					continue
+					value = 0.0
+				}
+				if m.config.InvalidToZero && (math.IsNaN(value) || math.IsInf(value, 0)) {
+					value = 0.0
 				}
 				m.gmresults[tid][metric.Name] = value
-				if m.config.InvalidToZero && math.IsNaN(value) {
-					value = 0.0
-				}
-				if m.config.InvalidToZero && math.IsInf(value, 0) {
-					value = 0.0
-				}
-				m.statsProcessedMetrics++
-				stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 				// Now we have the result, send it with the proper tags
 				if !math.IsNaN(value) {
 					if metric.Publish {
@@ -438,8 +420,6 @@ func (m *LikwidCollector) calcGlobalMetrics(interval time.Duration, output chan
 							if len(metric.Unit) > 0 {
 								y.AddMeta("unit", metric.Unit)
 							}
-							m.statsPublishedMetrics++
-							stats.ComponentStatInt(m.name, "published_metrics", m.statsPublishedMetrics)
 							output <- y
 						}
 					}

collectors/loadavgMetric.go

@@ -10,7 +10,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 //
@@ -33,7 +32,6 @@ type LoadavgCollector struct {
 	config       struct {
 		ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
 	}
-	statsProcessedMetrics int64
 }
 
 func (m *LoadavgCollector) Init(config json.RawMessage) error {
@@ -65,7 +63,6 @@ func (m *LoadavgCollector) Init(config json.RawMessage) error {
 	for i, name := range m.proc_matches {
 		_, m.proc_skips[i] = stringArrayContains(m.config.ExcludeMetrics, name)
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -101,7 +98,6 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		y, err := lp.New(name, m.tags, m.meta, map[string]interface{}{"value": x}, now)
 		if err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 	}
 
@@ -121,10 +117,9 @@ func (m *LoadavgCollector) Read(interval time.Duration, output chan lp.CCMetric)
 		y, err := lp.New(name, m.tags, m.meta, map[string]interface{}{"value": x}, now)
 		if err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
+
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *LoadavgCollector) Close() {

collectors/lustreMetric.go

@@ -12,7 +12,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const LUSTRE_SYSFS = `/sys/fs/lustre`
@@ -38,14 +37,13 @@ type LustreMetricDefinition struct {
 
 type LustreCollector struct {
 	metricCollector
-	tags                  map[string]string
+	tags          map[string]string
-	config                LustreCollectorConfig
+	config        LustreCollectorConfig
-	lctl                  string
+	lctl          string
-	sudoCmd               string
+	sudoCmd       string
-	lastTimestamp         time.Time                   // Store time stamp of last tick to derive bandwidths
+	lastTimestamp time.Time                   // Store time stamp of last tick to derive bandwidths
-	definitions           []LustreMetricDefinition    // Combined list without excluded metrics
+	definitions   []LustreMetricDefinition    // Combined list without excluded metrics
-	stats                 map[string]map[string]int64 // Data for last value per device and metric
+	stats         map[string]map[string]int64 // Data for last value per device and metric
-	statsProcessedMetrics int64
 }
 
 func (m *LustreCollector) getDeviceDataCommand(device string) []string {
@@ -374,7 +372,6 @@ func (m *LustreCollector) Init(config json.RawMessage) error {
 		}
 	}
 	m.lastTimestamp = time.Now()
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -421,13 +418,11 @@ func (m *LustreCollector) Read(interval time.Duration, output chan lp.CCMetric)
 					y.AddMeta("unit", def.unit)
 				}
 				output <- y
-				m.statsProcessedMetrics++
 			}
 			devData[def.name] = use_x
 		}
 	}
 	m.lastTimestamp = now
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *LustreCollector) Close() {

collectors/memstatMetric.go

@@ -14,7 +14,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const MEMSTATFILE = "/proc/meminfo"
@@ -33,13 +32,12 @@ type MemstatCollectorNode struct {
 
 type MemstatCollector struct {
 	metricCollector
-	stats                 map[string]int64
+	stats       map[string]int64
-	tags                  map[string]string
+	tags        map[string]string
-	matches               map[string]string
+	matches     map[string]string
-	config                MemstatCollectorConfig
+	config      MemstatCollectorConfig
-	nodefiles             map[int]MemstatCollectorNode
+	nodefiles   map[int]MemstatCollectorNode
-	sendMemUsed           bool
+	sendMemUsed bool
-	statsProcessedMetrics int64
 }
 
 type MemstatStats struct {
@@ -155,7 +153,6 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
 			}
 		}
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return err
 }
@@ -181,7 +178,6 @@ func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if len(unit) > 0 {
 					y.AddMeta("unit", unit)
 				}
-				m.statsProcessedMetrics++
 				output <- y
 			}
 		}
@@ -211,7 +207,6 @@ func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if len(unit) > 0 {
 					y.AddMeta("unit", unit)
 				}
-				m.statsProcessedMetrics++
 				output <- y
 			}
 		}
@@ -228,7 +223,6 @@ func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 			sendStats(stats, nodeConf.tags)
 		}
 	}
-	stats.ComponentStatInt(m.name, "collected_metrics", m.statsProcessedMetrics)
 }
 
 func (m *MemstatCollector) Close() {

collectors/netstatMetric.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const NETSTATFILE = "/proc/net/dev"
@@ -33,10 +32,9 @@ type NetstatCollectorMetric struct {
 
 type NetstatCollector struct {
 	metricCollector
-	config                NetstatCollectorConfig
+	config        NetstatCollectorConfig
-	matches               map[string][]NetstatCollectorMetric
+	matches       map[string][]NetstatCollectorMetric
-	lastTimestamp         time.Time
+	lastTimestamp time.Time
-	statsProcessedMetrics int64
 }
 
 func (m *NetstatCollector) Init(config json.RawMessage) error {
@@ -150,7 +148,6 @@ func (m *NetstatCollector) Init(config json.RawMessage) error {
 	if len(m.matches) == 0 {
 		return errors.New("no devices to collector metrics found")
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -201,7 +198,6 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if m.config.SendAbsoluteValues {
 					if y, err := lp.New(metric.name, metric.tags, metric.meta, map[string]interface{}{"value": v}, now); err == nil {
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 				if m.config.SendDerivedValues {
@@ -209,7 +205,6 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 						rate := float64(v-metric.lastValue) / timeDiff
 						if y, err := lp.New(metric.name+"_bw", metric.tags, metric.meta_rates, map[string]interface{}{"value": rate}, now); err == nil {
 							output <- y
-							m.statsProcessedMetrics++
 						}
 					}
 					metric.lastValue = v
@@ -217,7 +212,6 @@ func (m *NetstatCollector) Read(interval time.Duration, output chan lp.CCMetric)
 			}
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *NetstatCollector) Close() {

collectors/nfsMetric.go

@@ -12,7 +12,6 @@ import (
 	"time"
 
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 // First part contains the code for the general NfsCollector.
@@ -33,8 +32,7 @@ type nfsCollector struct {
 		Nfsstats       string   `json:"nfsstat"`
 		ExcludeMetrics []string `json:"exclude_metrics,omitempty"`
 	}
-	data                  map[string]NfsCollectorData
+	data map[string]NfsCollectorData
-	statsProcessedMetrics int64
 }
 
 func (m *nfsCollector) initStats() error {
@@ -115,7 +113,6 @@ func (m *nfsCollector) MainInit(config json.RawMessage) error {
 	}
 	m.data = make(map[string]NfsCollectorData)
 	m.initStats()
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -146,10 +143,8 @@ func (m *nfsCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		if err == nil {
 			y.AddMeta("version", m.version)
 			output <- y
-			m.statsProcessedMetrics++
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *nfsCollector) Close() {

collectors/numastatsMetric.go

@@ -12,7 +12,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 //
@@ -45,8 +44,7 @@ type NUMAStatsCollectorTopolgy struct {
 
 type NUMAStatsCollector struct {
 	metricCollector
-	topology              []NUMAStatsCollectorTopolgy
+	topology []NUMAStatsCollectorTopolgy
-	statsProcessedMetrics int64
 }
 
 func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
@@ -82,7 +80,7 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
 				tagSet: map[string]string{"memoryDomain": node},
 			})
 	}
-	m.statsProcessedMetrics = 0
+
 	m.init = true
 	return nil
 }
@@ -129,13 +127,11 @@ func (m *NUMAStatsCollector) Read(interval time.Duration, output chan lp.CCMetri
 			)
 			if err == nil {
 				output <- y
-				m.statsProcessedMetrics++
 			}
 		}
 
 		file.Close()
 	}
-	stats.ComponentStatInt(m.name, "collected_metrics", m.statsProcessedMetrics)
 }
 
 func (m *NUMAStatsCollector) Close() {

collectors/nvidiaMetric.go

@@ -9,7 +9,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	"github.com/NVIDIA/go-nvml/pkg/nvml"
 )
 
@@ -27,10 +26,9 @@ type NvidiaCollectorDevice struct {
 
 type NvidiaCollector struct {
 	metricCollector
-	num_gpus              int
+	num_gpus int
-	config                NvidiaCollectorConfig
+	config   NvidiaCollectorConfig
-	gpus                  []NvidiaCollectorDevice
+	gpus     []NvidiaCollectorDevice
-	statsProcessedMetrics int64
 }
 
 func (m *NvidiaCollector) CatchPanic() {
@@ -122,7 +120,7 @@ func (m *NvidiaCollector) Init(config json.RawMessage) error {
 				pciInfo.Device)
 		}
 	}
-	m.statsProcessedMetrics = 0
+
 	m.init = true
 	return nil
 }
@@ -153,7 +151,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 					if err == nil {
 						y.AddMeta("unit", "%")
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 				if !device.excludeMetrics["nv_mem_util"] {
@@ -161,7 +158,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 					if err == nil {
 						y.AddMeta("unit", "%")
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 			}
@@ -190,7 +186,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 					if err == nil {
 						y.AddMeta("unit", "MByte")
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 
@@ -200,7 +195,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 					if err == nil {
 						y.AddMeta("unit", "MByte")
 						output <- y
-						m.statsProcessedMetrics++
 					}
 				}
 			}
@@ -218,7 +212,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "degC")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -239,7 +232,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "%")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -266,13 +258,11 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				}
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			} else if ret == nvml.ERROR_NOT_SUPPORTED {
 				y, err := lp.New("nv_ecc_mode", device.tags, m.meta, map[string]interface{}{"value": "N/A"}, time.Now())
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -290,7 +280,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				y, err := lp.New("nv_perf_state", device.tags, m.meta, map[string]interface{}{"value": fmt.Sprintf("P%d", int(pState))}, time.Now())
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -307,7 +296,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "watts")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -325,7 +313,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "MHz")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -337,7 +324,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "MHz")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -349,7 +335,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "MHz")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -372,7 +357,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "MHz")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -384,7 +368,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "MHz")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -396,7 +379,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "MHz")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -416,7 +398,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				y, err := lp.New("nv_ecc_db_error", device.tags, m.meta, map[string]interface{}{"value": float64(ecc_db)}, time.Now())
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -427,7 +408,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				y, err := lp.New("nv_ecc_sb_error", device.tags, m.meta, map[string]interface{}{"value": float64(ecc_sb)}, time.Now())
 				if err == nil {
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -445,7 +425,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "watts")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -462,7 +441,6 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "%")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
@@ -479,12 +457,11 @@ func (m *NvidiaCollector) Read(interval time.Duration, output chan lp.CCMetric)
 				if err == nil {
 					y.AddMeta("unit", "%")
 					output <- y
-					m.statsProcessedMetrics++
 				}
 			}
 		}
 	}
-	stats.ComponentStatInt(m.name, "collected_metrics", m.statsProcessedMetrics)
+
 }
 
 func (m *NvidiaCollector) Close() {

collectors/sampleMetric.go

@@ -6,7 +6,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 // These are the fields we read from the JSON configuration
@@ -18,10 +17,9 @@ type SampleCollectorConfig struct {
 // defined by metricCollector (name, init, ...)
 type SampleCollector struct {
 	metricCollector
-	config     SampleTimerCollectorConfig // the configuration structure
+	config SampleTimerCollectorConfig // the configuration structure
-	meta       map[string]string          // default meta information
+	meta   map[string]string          // default meta information
-	tags       map[string]string          // default tags
+	tags   map[string]string          // default tags
-	statsCount int64
 }
 
 // Functions to implement MetricCollector interface
@@ -60,9 +58,6 @@ func (m *SampleCollector) Init(config json.RawMessage) error {
 	// for all topological entities (sockets, NUMA domains, ...)
 	// Return some useful error message in case of any failures
 
-	// Initialize counts for statistics
-	m.statsCount = 0
-
 	// Set this flag only if everything is initialized properly, all required files exist, ...
 	m.init = true
 	return err
@@ -85,11 +80,8 @@ func (m *SampleCollector) Read(interval time.Duration, output chan lp.CCMetric)
 	if err == nil {
 		// Send it to output channel
 		output <- y
-		// increment count for each sent metric or any other operation
-		m.statsCount++
 	}
-	// Set stats for the component
+
-	stats.ComponentStatInt(m.name, "count", m.statsCount)
 }
 
 // Close metric collector: close network connection, close files, close libraries, ...

collectors/tempMetric.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 // See: https://www.kernel.org/doc/html/latest/hwmon/sysfs-interface.html
@@ -41,8 +40,7 @@ type TempCollector struct {
 		ReportMaxTemp      bool                         `json:"report_max_temperature"`
 		ReportCriticalTemp bool                         `json:"report_critical_temperature"`
 	}
-	sensors               []*TempCollectorSensor
+	sensors []*TempCollectorSensor
-	statsProcessedMetrics int64
 }
 
 func (m *TempCollector) Init(config json.RawMessage) error {
@@ -164,7 +162,6 @@ func (m *TempCollector) Init(config json.RawMessage) error {
 	}
 
 	// Finished initialization
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -197,7 +194,6 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 		)
 		if err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 
 		// max temperature
@@ -211,7 +207,6 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 			)
 			if err == nil {
 				output <- y
-				m.statsProcessedMetrics++
 			}
 		}
 
@@ -226,11 +221,10 @@ func (m *TempCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 			)
 			if err == nil {
 				output <- y
-				m.statsProcessedMetrics++
 			}
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
+
 }
 
 func (m *TempCollector) Close() {

collectors/topprocsMetric.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const MAX_NUM_PROCS = 10
@@ -22,9 +21,8 @@ type TopProcsCollectorConfig struct {
 
 type TopProcsCollector struct {
 	metricCollector
-	tags                  map[string]string
+	tags   map[string]string
-	config                TopProcsCollectorConfig
+	config TopProcsCollectorConfig
-	statsProcessedMetrics int64
 }
 
 func (m *TopProcsCollector) Init(config json.RawMessage) error {
@@ -50,7 +48,6 @@ func (m *TopProcsCollector) Init(config json.RawMessage) error {
 	if err != nil {
 		return errors.New("failed to execute command")
 	}
-	m.statsProcessedMetrics = 0
 	m.init = true
 	return nil
 }
@@ -73,10 +70,8 @@ func (m *TopProcsCollector) Read(interval time.Duration, output chan lp.CCMetric
 		y, err := lp.New(name, m.tags, m.meta, map[string]interface{}{"value": string(lines[i])}, time.Now())
 		if err == nil {
 			output <- y
-			m.statsProcessedMetrics++
 		}
 	}
-	stats.ComponentStatInt(m.name, "processed_metrics", m.statsProcessedMetrics)
 }
 
 func (m *TopProcsCollector) Close() {

internal/metricRouter/StatsApi.md

@@ -1,17 +0,0 @@
-# Stats API
-
-The Stats API can be used for debugging. It publishes counts at an HTTP endpoint as JSON from different componenets of the CC Metric Collector.
-
-# Configuration
-
-The Stats API has an own configuration file to specify the listen host and port. The defaults are `localhost` and `8080`.
-
-```json
-{
-  "bindhost" : "",
-  "port" : "8080",
-  "publish_collectorstate" : true
-}
-```
-
-The `bindhost` and `port` can be used to specify the listen host and port. The `publish_collectorstate` needs to be `true`, otherwise nothing is presented. This option is for future use if we need to publish more infos using different domains.

internal/metricRouter/metricApi.go

@@ -1,232 +0,0 @@
-package metricRouter
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"io"
-	"net/http"
-	"os"
-	"sync"
-	"time"
-
-	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
-	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	mct "github.com/ClusterCockpit/cc-metric-collector/internal/multiChanTicker"
-	"github.com/gorilla/mux"
-)
-
-type statsApiConfig struct {
-	PublishCollectorState bool   `json:"publish_collectorstate"`
-	Host                  string `json:"bindhost"`
-	Port                  string `json:"port"`
-}
-
-// Metric cache data structure
-type statsApi struct {
-	name     string
-	input    chan lp.CCMetric
-	indone   chan bool
-	outdone  chan bool
-	config   statsApiConfig
-	wg       *sync.WaitGroup
-	statsWg  sync.WaitGroup
-	ticker   mct.MultiChanTicker
-	tickchan chan time.Time
-	server   *http.Server
-	router   *mux.Router
-	lock     sync.Mutex
-	baseurl  string
-	stats    map[string]map[string]int64
-	outStats map[string]map[string]int64
-}
-
-type StatsApi interface {
-	Start()
-	Close()
-	StatsFunc(w http.ResponseWriter, r *http.Request)
-}
-
-var statsApiServer *statsApi = nil
-
-func (a *statsApi) updateStats(point lp.CCMetric) {
-	switch point.Name() {
-	case "_stats":
-		if name, nok := point.GetMeta("source"); nok {
-			var compStats map[string]int64
-			var ok bool
-
-			if compStats, ok = a.stats[name]; !ok {
-				a.stats[name] = make(map[string]int64)
-				compStats = a.stats[name]
-			}
-			for k, v := range point.Fields() {
-				switch value := v.(type) {
-				case int:
-					compStats[k] = int64(value)
-				case uint:
-					compStats[k] = int64(value)
-				case int32:
-					compStats[k] = int64(value)
-				case uint32:
-					compStats[k] = int64(value)
-				case int64:
-					compStats[k] = int64(value)
-				case uint64:
-					compStats[k] = int64(value)
-				default:
-					cclog.ComponentDebug(a.name, "Unusable stats for", k, ". Values should be int64")
-				}
-			}
-			a.stats[name] = compStats
-		}
-	}
-}
-
-func (a *statsApi) Start() {
-	a.ticker.AddChannel(a.tickchan)
-	a.wg.Add(1)
-	a.statsWg.Add(1)
-	go func() {
-		a.stats = make(map[string]map[string]int64)
-		defer a.statsWg.Done()
-		for {
-			select {
-			case <-a.indone:
-				cclog.ComponentDebug(a.name, "INPUT DONE")
-				close(a.indone)
-				return
-			case p := <-a.input:
-				a.lock.Lock()
-				a.updateStats(p)
-				a.lock.Unlock()
-			}
-		}
-	}()
-	a.statsWg.Add(1)
-	go func() {
-		a.outStats = make(map[string]map[string]int64)
-		defer a.statsWg.Done()
-		a.lock.Lock()
-		for comp, compData := range a.stats {
-			var outData map[string]int64
-			var ok bool
-			if outData, ok = a.outStats[comp]; !ok {
-				outData = make(map[string]int64)
-			}
-			for k, v := range compData {
-				outData[k] = v
-			}
-			a.outStats[comp] = outData
-		}
-		a.lock.Unlock()
-		for {
-			select {
-			case <-a.outdone:
-				cclog.ComponentDebug(a.name, "OUTPUT DONE")
-				close(a.outdone)
-				return
-			case <-a.tickchan:
-				a.lock.Lock()
-				for comp, compData := range a.stats {
-					var outData map[string]int64
-					var ok bool
-					if outData, ok = a.outStats[comp]; !ok {
-						outData = make(map[string]int64)
-					}
-					for k, v := range compData {
-						outData[k] = v
-					}
-					a.outStats[comp] = outData
-				}
-				a.lock.Unlock()
-			}
-		}
-	}()
-	a.statsWg.Add(1)
-	go func() {
-		defer a.statsWg.Done()
-		err := a.server.ListenAndServe()
-		if err != nil && err.Error() != "http: Server closed" {
-			cclog.ComponentError(a.name, err.Error())
-		}
-		cclog.ComponentDebug(a.name, "SERVER DONE")
-	}()
-	cclog.ComponentDebug(a.name, "STARTED")
-}
-
-func (a *statsApi) StatsFunc(w http.ResponseWriter, r *http.Request) {
-	data, err := json.Marshal(a.outStats)
-	if err == nil {
-		w.Header().Set("Content-Type", "application/json")
-		io.WriteString(w, string(data))
-	}
-}
-
-// Close finishes / stops the metric cache
-func (a *statsApi) Close() {
-	cclog.ComponentDebug(a.name, "CLOSE")
-	a.indone <- true
-	a.outdone <- true
-	a.server.Shutdown(context.Background())
-	// wait for close of channel r.done
-	<-a.indone
-	<-a.outdone
-	a.statsWg.Wait()
-	a.wg.Done()
-
-	//a.wg.Wait()
-}
-
-func NewStatsApi(ticker mct.MultiChanTicker, wg *sync.WaitGroup, statsApiConfigfile string) (StatsApi, error) {
-	a := new(statsApi)
-	a.name = "StatsApi"
-	a.config.Host = "localhost"
-	a.config.Port = "8080"
-	configFile, err := os.Open(statsApiConfigfile)
-	if err != nil {
-		cclog.ComponentError(a.name, err.Error())
-		return nil, err
-	}
-	defer configFile.Close()
-	jsonParser := json.NewDecoder(configFile)
-	err = jsonParser.Decode(&a.config)
-	if err != nil {
-		cclog.ComponentError(a.name, err.Error())
-		return nil, err
-	}
-	a.input = make(chan lp.CCMetric)
-	a.ticker = ticker
-	a.tickchan = make(chan time.Time)
-	a.wg = wg
-	a.indone = make(chan bool)
-	a.outdone = make(chan bool)
-	a.router = mux.NewRouter()
-	a.baseurl = fmt.Sprintf("%s:%s", a.config.Host, a.config.Port)
-	a.server = &http.Server{Addr: a.baseurl, Handler: a.router}
-	if a.config.PublishCollectorState {
-		a.router.HandleFunc("/", a.StatsFunc)
-	}
-	statsApiServer = a
-	return a, nil
-}
-
-func ComponentStatInt(component string, key string, value int64) {
-	if statsApiServer == nil {
-		return
-	}
-	y, err := lp.New("_stats", map[string]string{}, map[string]string{"source": component}, map[string]interface{}{key: value}, time.Now())
-	if err == nil {
-		statsApiServer.input <- y
-	}
-}
-
-func ComponentStatString(component string, key string, value int64) {
-	if statsApiServer == nil {
-		return
-	}
-	y, err := lp.New("_stats", map[string]string{}, map[string]string{"source": component}, map[string]interface{}{key: value}, time.Now())
-	if err == nil {
-		statsApiServer.input <- y
-	}
-}

internal/metricRouter/metricRouter.go

@@ -40,26 +40,19 @@ type metricRouterConfig struct {
 
 // Metric router data structure
 type metricRouter struct {
-	hostname          string              // Hostname used in tags
+	hostname    string              // Hostname used in tags
-	coll_input        chan lp.CCMetric    // Input channel from CollectorManager
+	coll_input  chan lp.CCMetric    // Input channel from CollectorManager
-	recv_input        chan lp.CCMetric    // Input channel from ReceiveManager
+	recv_input  chan lp.CCMetric    // Input channel from ReceiveManager
-	cache_input       chan lp.CCMetric    // Input channel from MetricCache
+	cache_input chan lp.CCMetric    // Input channel from MetricCache
-	outputs           []chan lp.CCMetric  // List of all output channels
+	outputs     []chan lp.CCMetric  // List of all output channels
-	done              chan bool           // channel to finish / stop metric router
+	done        chan bool           // channel to finish / stop metric router
-	wg                *sync.WaitGroup     // wait group for all goroutines in cc-metric-collector
+	wg          *sync.WaitGroup     // wait group for all goroutines in cc-metric-collector
-	timestamp         time.Time           // timestamp periodically updated by ticker each interval
+	timestamp   time.Time           // timestamp periodically updated by ticker each interval
-	timerdone         chan bool           // channel to finish / stop timestamp updater
+	ticker      mct.MultiChanTicker // periodically ticking once each interval
-	ticker            mct.MultiChanTicker // periodically ticking once each interval
+	config      metricRouterConfig  // json encoded config for metric router
-	config            metricRouterConfig  // json encoded config for metric router
+	cache       MetricCache         // pointer to MetricCache
-	cache             MetricCache         // pointer to MetricCache
+	cachewg     sync.WaitGroup      // wait group for MetricCache
-	cachewg           sync.WaitGroup      // wait group for MetricCache
+	maxForward  int                 // number of metrics to forward maximally in one iteration
-	maxForward        int                 // number of metrics to forward maximally in one iteration
-	statsCollForward  int64
-	statsRecvForward  int64
-	statsCacheForward int64
-	statsTotalForward int64
-	statsDropped      int64
-	statsRenamed      int64
 }
 
 // MetricRouter access functions
@@ -127,39 +120,9 @@ func (r *metricRouter) Init(ticker mct.MultiChanTicker, wg *sync.WaitGroup, rout
 	for _, mname := range r.config.DropMetrics {
 		r.config.dropMetrics[mname] = true
 	}
-	r.statsCollForward = 0
-	r.statsRecvForward = 0
-	r.statsCacheForward = 0
-	r.statsTotalForward = 0
-	r.statsDropped = 0
-	r.statsRenamed = 0
 	return nil
 }
 
-// StartTimer starts a timer which updates timestamp periodically
-func (r *metricRouter) StartTimer() {
-	m := make(chan time.Time)
-	r.ticker.AddChannel(m)
-	r.timerdone = make(chan bool)
-
-	r.wg.Add(1)
-	go func() {
-		defer r.wg.Done()
-		for {
-			select {
-			case <-r.timerdone:
-				close(r.timerdone)
-				cclog.ComponentDebug("MetricRouter", "TIMER DONE")
-				return
-			case t := <-m:
-				cclog.ComponentDebug("MetricRouter", "INTERVAL_TICK", t.Unix())
-				r.timestamp = t
-			}
-		}
-	}()
-	cclog.ComponentDebug("MetricRouter", "TIMER START")
-}
-
 func getParamMap(point lp.CCMetric) map[string]interface{} {
 	params := make(map[string]interface{})
 	params["metric"] = point
@@ -248,8 +211,9 @@ func (r *metricRouter) dropMetric(point lp.CCMetric) bool {
 func (r *metricRouter) Start() {
 	// start timer if configured
 	r.timestamp = time.Now()
+	timeChan := make(chan time.Time)
 	if r.config.IntervalStamp {
-		r.StartTimer()
+		r.ticker.AddChannel(timeChan)
 	}
 
 	// Router manager is done
@@ -266,8 +230,6 @@ func (r *metricRouter) Start() {
 		r.DoDelTags(point)
 		name := point.Name()
 		if new, ok := r.config.RenameMetrics[name]; ok {
-			r.statsRenamed++
-			ComponentStatInt("MetricRouter", "renamed", r.statsRenamed)
 			point.SetName(new)
 			point.AddMeta("oldname", name)
 		}
@@ -287,14 +249,7 @@ func (r *metricRouter) Start() {
 			p.SetTime(r.timestamp)
 		}
 		if !r.dropMetric(p) {
-			r.statsCollForward++
-			r.statsTotalForward++
-			ComponentStatInt("MetricRouter", "collector_forward", r.statsCollForward)
-			ComponentStatInt("MetricRouter", "total_forward", r.statsTotalForward)
 			forward(p)
-		} else {
-			r.statsDropped++
-			ComponentStatInt("MetricRouter", "dropped", r.statsDropped)
 		}
 		// even if the metric is dropped, it is stored in the cache for
 		// aggregations
@@ -310,14 +265,7 @@ func (r *metricRouter) Start() {
 			p.SetTime(r.timestamp)
 		}
 		if !r.dropMetric(p) {
-			r.statsRecvForward++
-			r.statsTotalForward++
-			ComponentStatInt("MetricRouter", "receiver_forward", r.statsRecvForward)
-			ComponentStatInt("MetricRouter", "total_forward", r.statsTotalForward)
 			forward(p)
-		} else {
-			r.statsDropped++
-			ComponentStatInt("MetricRouter", "dropped", r.statsDropped)
 		}
 	}
 
@@ -326,14 +274,7 @@ func (r *metricRouter) Start() {
 		// receive from metric collector
 		if !r.dropMetric(p) {
 			p.AddTag(r.config.HostnameTagName, r.hostname)
-			r.statsCacheForward++
-			r.statsTotalForward++
-			ComponentStatInt("MetricRouter", "cache_forward", r.statsCacheForward)
-			ComponentStatInt("MetricRouter", "total_forward", r.statsTotalForward)
 			forward(p)
-		} else {
-			r.statsDropped++
-			ComponentStatInt("MetricRouter", "dropped", r.statsDropped)
 		}
 	}
 
@@ -352,6 +293,10 @@ func (r *metricRouter) Start() {
 				done()
 				return
 
+			case timestamp := <-timeChan:
+				r.timestamp = timestamp
+				cclog.ComponentDebug("MetricRouter", "Update timestamp", r.timestamp.UnixNano())
+
 			case p := <-r.coll_input:
 				coll_forward(p)
 				for i := 0; len(r.coll_input) > 0 && i < (r.maxForward-1); i++ {
@@ -397,14 +342,6 @@ func (r *metricRouter) Close() {
 	// wait for close of channel r.done
 	<-r.done
 
-	// stop timer
-	if r.config.IntervalStamp {
-		cclog.ComponentDebug("MetricRouter", "TIMER CLOSE")
-		r.timerdone <- true
-		// wait for close of channel r.timerdone
-		<-r.timerdone
-	}
-
 	// stop metric cache
 	if r.config.NumCacheIntervals > 0 {
 		cclog.ComponentDebug("MetricRouter", "CACHE CLOSE")

sinks/gangliaSink.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 const GMETRIC_EXEC = `gmetric`
@@ -30,10 +29,9 @@ type GangliaSinkConfig struct {
 
 type GangliaSink struct {
 	sink
-	gmetric_path     string
+	gmetric_path   string
-	gmetric_config   string
+	gmetric_config string
-	config           GangliaSinkConfig
+	config         GangliaSinkConfig
-	statsSentMetrics int64
 }
 
 func (s *GangliaSink) Write(point lp.CCMetric) error {
@@ -80,8 +78,6 @@ func (s *GangliaSink) Write(point lp.CCMetric) error {
 	command := exec.Command(s.gmetric_path, argstr...)
 	command.Wait()
 	_, err = command.Output()
-	s.statsSentMetrics++
-	stats.ComponentStatInt(s.name, "sent_metrics", s.statsSentMetrics)
 	return err
 }
 
@@ -124,6 +120,5 @@ func NewGangliaSink(name string, config json.RawMessage) (Sink, error) {
 	if len(s.config.GmetricConfig) > 0 {
 		s.gmetric_config = s.config.GmetricConfig
 	}
-	s.statsSentMetrics = 0
 	return s, nil
 }

sinks/httpSink.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	influx "github.com/influxdata/line-protocol"
 )
 
@@ -37,8 +36,6 @@ type HttpSink struct {
 	idleConnTimeout time.Duration
 	timeout         time.Duration
 	flushDelay      time.Duration
-	statsProcessed  int64
-	statsFlushes    int64
 }
 
 func (s *HttpSink) Write(m lp.CCMetric) error {
@@ -66,8 +63,6 @@ func (s *HttpSink) Write(m lp.CCMetric) error {
 		cclog.ComponentError(s.name, "encoding failed:", err.Error())
 		return err
 	}
-	s.statsProcessed++
-	stats.ComponentStatInt(s.name, "processed_metrics", s.statsProcessed)
 
 	// Flush synchronously if "flush_delay" is zero
 	if s.flushDelay == 0 {
@@ -117,8 +112,6 @@ func (s *HttpSink) Flush() error {
 		cclog.ComponentError(s.name, "application error:", err.Error())
 		return err
 	}
-	s.statsFlushes++
-	stats.ComponentStatInt(s.name, "flushes", s.statsFlushes)
 
 	return nil
 }
@@ -184,7 +177,5 @@ func NewHttpSink(name string, config json.RawMessage) (Sink, error) {
 	s.buffer = &bytes.Buffer{}
 	s.encoder = influx.NewEncoder(s.buffer)
 	s.encoder.SetPrecision(time.Second)
-	s.statsFlushes = 0
-	s.statsProcessed = 0
 	return s, nil
 }

sinks/influxAsyncSink.go

@@ -6,13 +6,14 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
+	influxdb2ApiHttp "github.com/influxdata/influxdb-client-go/v2/api/http"
 )
 
 type InfluxAsyncSinkConfig struct {
@@ -33,6 +34,8 @@ type InfluxAsyncSinkConfig struct {
 	InfluxExponentialBase uint   `json:"retry_exponential_base,omitempty"`
 	InfluxMaxRetries      uint   `json:"max_retries,omitempty"`
 	InfluxMaxRetryTime    string `json:"max_retry_time,omitempty"`
+	CustomFlushInterval   string `json:"custom_flush_interval,omitempty"`
+	MaxRetryAttempts      uint   `json:"max_retry_attempts,omitempty"`
 }
 
 type InfluxAsyncSink struct {
@@ -43,9 +46,8 @@ type InfluxAsyncSink struct {
 	config              InfluxAsyncSinkConfig
 	influxRetryInterval uint
 	influxMaxRetryTime  uint
-	sentMetrics         int64
+	customFlushInterval time.Duration
-	statsFlushes        int64
+	flushTimer          *time.Timer
-	statsErrors         int64
 }
 
 func (s *InfluxAsyncSink) connect() error {
@@ -102,22 +104,35 @@ func (s *InfluxAsyncSink) connect() error {
 	if !ok {
 		return fmt.Errorf("connection to %s not healthy", uri)
 	}
+	s.writeApi.SetWriteFailedCallback(func(batch string, err influxdb2ApiHttp.Error, retryAttempts uint) bool {
+		mlist := strings.Split(batch, "\n")
+		cclog.ComponentError(s.name, fmt.Sprintf("Failed to write batch with %d metrics %d times (max: %d): %s", len(mlist), retryAttempts, s.config.MaxRetryAttempts, err.Error()))
+		return retryAttempts <= s.config.MaxRetryAttempts
+	})
 	return nil
 }
 
 func (s *InfluxAsyncSink) Write(m lp.CCMetric) error {
+	if s.customFlushInterval != 0 && s.flushTimer == nil {
+		// Run a batched flush for all lines that have arrived in the defined interval
+		s.flushTimer = time.AfterFunc(s.customFlushInterval, func() {
+			if err := s.Flush(); err != nil {
+				cclog.ComponentError(s.name, "flush failed:", err.Error())
+			}
+		})
+	}
 	s.writeApi.WritePoint(
 		m.ToPoint(s.meta_as_tags),
 	)
-	s.sentMetrics++
-	stats.ComponentStatInt(s.name, "send_metrics", s.sentMetrics)
 	return nil
 }
 
 func (s *InfluxAsyncSink) Flush() error {
+	cclog.ComponentDebug(s.name, "Flushing")
 	s.writeApi.Flush()
-	s.statsFlushes++
+	if s.customFlushInterval != 0 && s.flushTimer != nil {
-	stats.ComponentStatInt(s.name, "flushes", s.statsFlushes)
+		s.flushTimer = nil
+	}
 	return nil
 }
 
@@ -140,6 +155,9 @@ func NewInfluxAsyncSink(name string, config json.RawMessage) (Sink, error) {
 	s.config.InfluxMaxRetries = 0
 	s.config.InfluxExponentialBase = 0
 	s.config.FlushInterval = 0
+	s.config.CustomFlushInterval = ""
+	s.customFlushInterval = time.Duration(0)
+	s.config.MaxRetryAttempts = 1
 
 	// Default retry intervals (in seconds)
 	// 1 2
@@ -191,23 +209,27 @@ func NewInfluxAsyncSink(name string, config json.RawMessage) (Sink, error) {
 	s.influxRetryInterval = toUint(s.config.InfluxRetryInterval, s.influxRetryInterval)
 	s.influxMaxRetryTime = toUint(s.config.InfluxMaxRetryTime, s.influxMaxRetryTime)
 
+	// Use a own timer for calling Flush()
+	if len(s.config.CustomFlushInterval) > 0 {
+		t, err := time.ParseDuration(s.config.CustomFlushInterval)
+		if err != nil {
+			return nil, fmt.Errorf("invalid duration in 'custom_flush_interval': %v", err)
+		}
+		s.customFlushInterval = t
+	}
+
 	// Connect to InfluxDB server
 	if err := s.connect(); err != nil {
 		return nil, fmt.Errorf("unable to connect: %v", err)
 	}
 
 	// Start background: Read from error channel
-	s.statsErrors = 0
 	s.errors = s.writeApi.Errors()
 	go func() {
 		for err := range s.errors {
-			s.statsErrors++
-			stats.ComponentStatInt(s.name, "errors", s.statsErrors)
 			cclog.ComponentError(s.name, err.Error())
 		}
 	}()
 
-	s.sentMetrics = 0
-	s.statsFlushes = 0
 	return s, nil
 }

sinks/influxSink.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
 	"github.com/influxdata/influxdb-client-go/v2/api/write"
@@ -38,17 +37,15 @@ type InfluxSinkConfig struct {
 
 type InfluxSink struct {
 	sink
-	client                influxdb2.Client
+	client              influxdb2.Client
-	writeApi              influxdb2Api.WriteAPIBlocking
+	writeApi            influxdb2Api.WriteAPIBlocking
-	config                InfluxSinkConfig
+	config              InfluxSinkConfig
-	influxRetryInterval   uint
+	influxRetryInterval uint
-	influxMaxRetryTime    uint
+	influxMaxRetryTime  uint
-	batch                 []*write.Point
+	batch               []*write.Point
-	flushTimer            *time.Timer
+	flushTimer          *time.Timer
-	flushDelay            time.Duration
+	flushDelay          time.Duration
-	lock                  sync.Mutex // Flush() runs in another goroutine, so this lock has to protect the buffer
+	lock                sync.Mutex // Flush() runs in another goroutine, so this lock has to protect the buffer
-	statsSentMetrics      int64
-	statsProcessedMetrics int64
 	//influxMaxRetryDelay uint
 }
 
@@ -126,10 +123,8 @@ func (s *InfluxSink) Write(m lp.CCMetric) error {
 	}
 	p := m.ToPoint(s.meta_as_tags)
 	s.lock.Lock()
-	s.statsProcessedMetrics++
 	s.batch = append(s.batch, p)
 	s.lock.Unlock()
-	stats.ComponentStatInt(s.name, "processed_metrics", s.statsProcessedMetrics)
 
 	// Flush synchronously if "flush_delay" is zero
 	if s.flushDelay == 0 {
@@ -150,8 +145,6 @@ func (s *InfluxSink) Flush() error {
 		cclog.ComponentError(s.name, "flush failed:", err.Error())
 		return err
 	}
-	s.statsSentMetrics += int64(len(s.batch))
-	stats.ComponentStatInt(s.name, "sent_metrics", s.statsSentMetrics)
 	s.batch = s.batch[:0]
 	return nil
 }
@@ -218,7 +211,5 @@ func NewInfluxSink(name string, config json.RawMessage) (Sink, error) {
 	if err := s.connect(); err != nil {
 		return nil, fmt.Errorf("unable to connect: %v", err)
 	}
-	s.statsSentMetrics = 0
-	s.statsProcessedMetrics = 0
 	return s, nil
 }

sinks/libgangliaSink.go

@@ -73,7 +73,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	"github.com/NVIDIA/go-nvml/pkg/dl"
 )
 
@@ -103,12 +102,11 @@ type LibgangliaSinkConfig struct {
 
 type LibgangliaSink struct {
 	sink
-	config           LibgangliaSinkConfig
+	config         LibgangliaSinkConfig
-	global_context   C.Ganglia_pool
+	global_context C.Ganglia_pool
-	gmond_config     C.Ganglia_gmond_config
+	gmond_config   C.Ganglia_gmond_config
-	send_channels    C.Ganglia_udp_send_channels
+	send_channels  C.Ganglia_udp_send_channels
-	cstrCache        map[string]*C.char
+	cstrCache      map[string]*C.char
-	statsSentMetrics int64
 }
 
 func (s *LibgangliaSink) Write(point lp.CCMetric) error {
@@ -204,8 +202,6 @@ func (s *LibgangliaSink) Write(point lp.CCMetric) error {
 	C.Ganglia_metric_destroy(gmetric)
 	// Free the value C string, the only one not stored in the cache
 	C.free(unsafe.Pointer(c_value))
-	s.statsSentMetrics++
-	stats.ComponentStatInt(s.name, "sent_metrics", s.statsSentMetrics)
 	return err
 }
 
@@ -251,7 +247,7 @@ func NewLibgangliaSink(name string, config json.RawMessage) (Sink, error) {
 	if err != nil {
 		return nil, fmt.Errorf("error opening %s: %v", s.config.GangliaLib, err)
 	}
-	s.statsSentMetrics = 0
+
 	// Set up cache for the C strings
 	s.cstrCache = make(map[string]*C.char)
 	// s.cstrCache["globals"] = C.CString("globals")

sinks/prometheusSink.go

@@ -11,7 +11,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -30,12 +29,11 @@ type PrometheusSinkConfig struct {
 
 type PrometheusSink struct {
 	sink
-	config           PrometheusSinkConfig
+	config       PrometheusSinkConfig
-	labelMetrics     map[string]*prometheus.GaugeVec
+	labelMetrics map[string]*prometheus.GaugeVec
-	nodeMetrics      map[string]prometheus.Gauge
+	nodeMetrics  map[string]prometheus.Gauge
-	promWg           sync.WaitGroup
+	promWg       sync.WaitGroup
-	promServer       *http.Server
+	promServer   *http.Server
-	statsSentMetrics int64
 }
 
 func intToFloat64(input interface{}) (float64, error) {
@@ -115,8 +113,6 @@ func (s *PrometheusSink) newMetric(metric lp.CCMetric) error {
 		s.nodeMetrics[name] = new
 		prometheus.Register(new)
 	}
-	s.statsSentMetrics++
-	stats.ComponentStatInt(s.name, "sent_metrics", s.statsSentMetrics)
 	return nil
 }
 
@@ -150,8 +146,6 @@ func (s *PrometheusSink) updateMetric(metric lp.CCMetric) error {
 		}
 		s.nodeMetrics[name].Set(value)
 	}
-	s.statsSentMetrics++
-	stats.ComponentStatInt(s.name, "sent_metrics", s.statsSentMetrics)
 	return nil
 }
 

sinks/sampleSink.go

@@ -7,7 +7,6 @@ import (
 
 	cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 type SampleSinkConfig struct {
@@ -15,15 +14,12 @@ type SampleSinkConfig struct {
 	// See: metricSink.go
 	defaultSinkConfig
 	// Additional config options, for SampleSink
-
 }
 
 type SampleSink struct {
 	// declares elements 	'name' and 'meta_as_tags' (string to bool map!)
 	sink
 	config SampleSinkConfig // entry point to the SampleSinkConfig
-	// Stats counters
-	statsSentMetrics int64
 }
 
 // Implement functions required for Sink interface
@@ -34,8 +30,6 @@ type SampleSink struct {
 func (s *SampleSink) Write(point lp.CCMetric) error {
 	// based on s.meta_as_tags use meta infos as tags
 	log.Print(point)
-	s.statsSentMetrics++
-	stats.ComponentStatInt(s.name, "sent_metrics", s.statsSentMetrics)
 	return nil
 }
 
@@ -69,9 +63,6 @@ func NewSampleSink(name string, config json.RawMessage) (Sink, error) {
 		}
 	}
 
-	// Initalize stats counters
-	s.statsSentMetrics = 0
-
 	// Create lookup map to use meta infos as tags in the output metric
 	s.meta_as_tags = make(map[string]bool)
 	for _, k := range s.config.MetaAsTags {

sinks/sinkManager.go

@@ -102,19 +102,13 @@ func (sm *sinkManager) Start() {
 		}
 
 		toTheSinks := func(p lp.CCMetric) {
-			var wg sync.WaitGroup
 			// Send received metric to all outputs
 			cclog.ComponentDebug("SinkManager", "WRITE", p)
 			for _, s := range sm.sinks {
-				wg.Add(1)
+				if err := s.Write(p); err != nil {
-				go func(s Sink) {
+					cclog.ComponentError("SinkManager", "WRITE", s.Name(), "write failed:", err.Error())
-					if err := s.Write(p); err != nil {
+				}
-						cclog.ComponentError("SinkManager", "WRITE", s.Name(), "write failed:", err.Error())
-					}
-					wg.Done()
-				}(s)
 			}
-			wg.Wait()
 		}
 
 		for {

sinks/stdoutSink.go

@@ -8,7 +8,6 @@ import (
 
 	//	"time"
 	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
-	stats "github.com/ClusterCockpit/cc-metric-collector/internal/metricRouter"
 )
 
 type StdoutSink struct {
@@ -18,7 +17,6 @@ type StdoutSink struct {
 		defaultSinkConfig
 		Output string `json:"output_file,omitempty"`
 	}
-	sentMetrics int64
 }
 
 func (s *StdoutSink) Write(m lp.CCMetric) error {
@@ -26,8 +24,6 @@ func (s *StdoutSink) Write(m lp.CCMetric) error {
 		s.output,
 		m.ToLineProtocol(s.meta_as_tags),
 	)
-	s.sentMetrics++
-	stats.ComponentStatInt(s.name, "sent_metrics", s.sentMetrics)
 	return nil
 }
 
@@ -72,7 +68,6 @@ func NewStdoutSink(name string, config json.RawMessage) (Sink, error) {
 	for _, k := range s.config.MetaAsTags {
 		s.meta_as_tags[k] = true
 	}
-	s.sentMetrics = 0
 
 	return s, nil
 }