Compare commits

..

16 Commits

Author SHA1 Message Date
Thomas Roehl
36dd440864 Merge branch 'develop' into main 2022-04-04 15:16:33 +02:00
Thomas Roehl
7b098e0b1b Fix for missing metrics in LikwidCollector is hwthread is inactive 2022-04-04 15:16:11 +02:00
Thomas Roehl
229a57b16a Merge branch 'develop' into main 2022-04-04 11:49:33 +02:00
Thomas Roehl
70a9530aba Set WriteFailedCallback to get some error message 2022-04-04 11:48:54 +02:00
Thomas Roehl
2f0b6057ca 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
2022-04-04 02:59:42 +02:00
Thomas Roehl
69f7c19659 InfluxAsyncSink: Add custom flush mechanism 2022-04-04 02:56:23 +02:00
Thomas Roehl
ecdb4c1bcf Add debug message when updating interval_timestep 2022-04-04 02:55:44 +02:00
Thomas Roehl
4d5b1adbc8 Fix for interval_timestamp option 2022-04-04 02:26:04 +02:00
Thomas Röhl
4763733d8d Merge branch 'develop' into main 2022-03-31 11:57:19 +02:00
Thomas Röhl
16e898ecca Merge branch 'develop' into main 2022-03-31 11:47:02 +02:00
Thomas Roehl
4851382ad7 Merge branch 'develop' into main 2022-03-16 19:08:13 +01:00
Thomas Gruber
3f76947f54 Merge latest developments into main (#67)
* Update configuration.md

Add an additional receiver to have better alignment of components

* Change default GpfsCollector command to `mmpmon` (#53)

* Set default cmd to 'mmpmon'

* Reuse looked up path

* Cast const to string

* Just download LIKWID to get the headers (#54)

* Just download LIKWID to get the headers

* Remove perl-Data-Dumper from BuildRequires, only required by LIKWID build

* Add HttpReceiver as counterpart to the HttpSink (#49)

* Use GBytes as unit for large memory numbers

* Make maxForward configurable, save old name in meta in rename metrics and make the hostname tag key configurable

* Single release action (#55)

Building all RPMs and releasing in a single workflow

* Makefile target to build binary-only Debian packages (#61)

* Add 'install' and 'DEB' make targets to build binary-only Debian packages

* Add control file for DEB builds

* Use a single line for bash loop in make clean

* Add config options for retry intervals of InfluxDB clients (#59)

* Refactoring of LikwidCollector and metric units (#62)

* Reduce complexity of LikwidCollector and allow metric units

* Add unit to LikwidCollector docu and fix some typos

* Make library path configurable

* Use old metric name in Ganglia if rename has happened in the router (#60)

* Use old metric name if rename has happened in the router

* Also check for Ganglia renames for the oldname

* Derived metrics (#57)

* Add time-based derivatived (e.g. bandwidth) to some collectors

* Add documentation

* Add comments

* Fix: Only compute rates with a valid previous state

* Only compute rates with a valid previous state

* Define const values for net/dev fields

* Set default config values

* Add comments

* Refactor: Consolidate data structures

* Refactor: Consolidate data structures

* Refactor: Avoid struct deep copy

* Refactor: Avoid redundant tag maps

* Refactor: Use int64 type for absolut values

Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com>

* Simplified iota usage

* Move unit tag to meta data tags

* Derived metrics (#65)

* Add time-based derivatived (e.g. bandwidth) to some collectors

* Add documentation

* Add comments

* Fix: Only compute rates with a valid previous state

* Only compute rates with a valid previous state

* Define const values for net/dev fields

* Set default config values

* Add comments

* Refactor: Consolidate data structures

* Refactor: Consolidate data structures

* Refactor: Avoid struct deep copy

* Refactor: Avoid redundant tag maps

* Refactor: Use int64 type for absolut values

* Update LustreCollector

Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com>

* Meta to tags list and map for sinks (#63)

* Change ccMetric->Influx functions

* Use a meta_as_tags string list in config but create a lookup map afterwards

* Add meta as tag logic to sampleSink

* Fix staticcheck warnings (#66)

Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com>
2022-03-15 16:41:11 +01:00
Thomas Roehl
3157386b3e Merge branch 'develop' into main 2022-03-04 23:34:28 +01:00
Thomas Roehl
ff08eaeb43 Set proper user for files 2022-03-04 11:49:55 +01:00
Thomas Roehl
64c41be34c Fix name for ClusterCockpit 2022-03-04 11:37:45 +01:00
Thomas Roehl
f4af520b2a Fix error print in LustreCollector 2022-03-04 11:32:39 +01:00
36 changed files with 176 additions and 712 deletions

View File

@@ -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

View File

@@ -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()

View File

@@ -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
matches map[string]string
config BeegfsMetaCollectorConfig
skipFS map[string]struct{}
statsProcessedMetrics int64
tags map[string]string
matches map[string]string
config BeegfsMetaCollectorConfig
skipFS map[string]struct{}
}
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() {

View File

@@ -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
matches map[string]string
config BeegfsStorageCollectorConfig
skipFS map[string]struct{}
statsProcessedMetrics int64
tags map[string]string
matches map[string]string
config BeegfsStorageCollectorConfig
skipFS map[string]struct{}
}
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() {

View File

@@ -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
statsProcessedMetrics int64
topology []*CPUFreqCpuInfoCollectorTopology
}
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() {

View File

@@ -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
statsProcessedMetrics int64
config struct {
topology []CPUFreqCollectorTopology
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() {

View File

@@ -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
matches map[string]int
cputags map[string]map[string]string
nodetags map[string]string
statsProcessedMetrics int64
config CpustatCollectorConfig
matches map[string]int
cputags map[string]map[string]string
nodetags map[string]string
}
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() {

View File

@@ -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
parser *influx.Parser
config CustomCmdCollectorConfig
commands []string
files []string
statsProcessedMetrics int64
statsProcessedCommands int64
statsProcessedFiles int64
handler *influx.MetricHandler
parser *influx.Parser
config CustomCmdCollectorConfig
commands []string
files []string
}
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)
}
}

View File

@@ -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
statsProcessedMetrics int64
//matches map[string]int
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
}
}

View File

@@ -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{}
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
lastState map[string]GpfsCollectorLastState
statsProcessedMetrics int64
skipFS map[string]struct{}
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
lastState map[string]GpfsCollectorLastState
}
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() {

View File

@@ -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
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
statsProcessedMetrics int64
info []*InfinibandCollectorInfo
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
}
// 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() {

View File

@@ -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
config IOstatCollectorConfig
devices map[string]IOstatCollectorEntry
statsProcessedMetrics int64
matches map[string]int
config IOstatCollectorConfig
devices map[string]IOstatCollectorEntry
}
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() {

View File

@@ -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
ipmitool string
ipmisensors string
statsProcessedMetrics int64
config IpmiCollectorConfig
ipmitool string
ipmisensors string
}
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() {

View File

@@ -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
cpu2tid map[int]int
sock2tid map[int]int
metrics map[C.int]map[string]int
groups []C.int
config LikwidCollectorConfig
gmresults map[int]map[string]float64
basefreq float64
running bool
initialized bool
likwidGroups map[C.int]LikwidEventsetConfig
lock sync.Mutex
statsMeasurements int64
statsProcessedMetrics int64
statsPublishedMetrics int64
cpulist []C.int
cpu2tid map[int]int
sock2tid map[int]int
metrics map[C.int]map[string]int
groups []C.int
config LikwidCollectorConfig
gmresults map[int]map[string]float64
basefreq float64
running bool
initialized bool
likwidGroups map[C.int]LikwidEventsetConfig
lock sync.Mutex
}
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
}
}

View File

@@ -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() {

View File

@@ -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
config LustreCollectorConfig
lctl string
sudoCmd string
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
definitions []LustreMetricDefinition // Combined list without excluded metrics
stats map[string]map[string]int64 // Data for last value per device and metric
statsProcessedMetrics int64
tags map[string]string
config LustreCollectorConfig
lctl string
sudoCmd string
lastTimestamp time.Time // Store time stamp of last tick to derive bandwidths
definitions []LustreMetricDefinition // Combined list without excluded metrics
stats map[string]map[string]int64 // Data for last value per device and metric
}
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() {

View File

@@ -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
tags map[string]string
matches map[string]string
config MemstatCollectorConfig
nodefiles map[int]MemstatCollectorNode
sendMemUsed bool
statsProcessedMetrics int64
stats map[string]int64
tags map[string]string
matches map[string]string
config MemstatCollectorConfig
nodefiles map[int]MemstatCollectorNode
sendMemUsed bool
}
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() {

View File

@@ -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
matches map[string][]NetstatCollectorMetric
lastTimestamp time.Time
statsProcessedMetrics int64
config NetstatCollectorConfig
matches map[string][]NetstatCollectorMetric
lastTimestamp time.Time
}
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() {

View File

@@ -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
statsProcessedMetrics int64
data map[string]NfsCollectorData
}
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() {

View File

@@ -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
statsProcessedMetrics int64
topology []NUMAStatsCollectorTopolgy
}
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() {

View File

@@ -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
config NvidiaCollectorConfig
gpus []NvidiaCollectorDevice
statsProcessedMetrics int64
num_gpus int
config NvidiaCollectorConfig
gpus []NvidiaCollectorDevice
}
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() {

View File

@@ -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
meta map[string]string // default meta information
tags map[string]string // default tags
statsCount int64
config SampleTimerCollectorConfig // the configuration structure
meta map[string]string // default meta information
tags map[string]string // default tags
}
// 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, ...

View File

@@ -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
statsProcessedMetrics int64
sensors []*TempCollectorSensor
}
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() {

View File

@@ -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
config TopProcsCollectorConfig
statsProcessedMetrics int64
tags map[string]string
config TopProcsCollectorConfig
}
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() {

View File

@@ -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.

View File

@@ -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
}
}

View File

@@ -40,26 +40,19 @@ type metricRouterConfig struct {
// Metric router data structure
type metricRouter struct {
hostname string // Hostname used in tags
coll_input chan lp.CCMetric // Input channel from CollectorManager
recv_input chan lp.CCMetric // Input channel from ReceiveManager
cache_input chan lp.CCMetric // Input channel from MetricCache
outputs []chan lp.CCMetric // List of all output channels
done chan bool // channel to finish / stop metric router
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector
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
config metricRouterConfig // json encoded config for metric router
cache MetricCache // pointer to MetricCache
cachewg sync.WaitGroup // wait group for MetricCache
maxForward int // number of metrics to forward maximally in one iteration
statsCollForward int64
statsRecvForward int64
statsCacheForward int64
statsTotalForward int64
statsDropped int64
statsRenamed int64
hostname string // Hostname used in tags
coll_input chan lp.CCMetric // Input channel from CollectorManager
recv_input chan lp.CCMetric // Input channel from ReceiveManager
cache_input chan lp.CCMetric // Input channel from MetricCache
outputs []chan lp.CCMetric // List of all output channels
done chan bool // channel to finish / stop metric router
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector
timestamp time.Time // timestamp periodically updated by ticker each interval
ticker mct.MultiChanTicker // periodically ticking once each interval
config metricRouterConfig // json encoded config for metric router
cache MetricCache // pointer to MetricCache
cachewg sync.WaitGroup // wait group for MetricCache
maxForward int // number of metrics to forward maximally in one iteration
}
// 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")

View File

@@ -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_config string
config GangliaSinkConfig
statsSentMetrics int64
gmetric_path string
gmetric_config string
config GangliaSinkConfig
}
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
}

View File

@@ -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
}

View File

@@ -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
statsFlushes int64
statsErrors int64
customFlushInterval time.Duration
flushTimer *time.Timer
}
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++
stats.ComponentStatInt(s.name, "flushes", s.statsFlushes)
if s.customFlushInterval != 0 && s.flushTimer != nil {
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
}

View File

@@ -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
writeApi influxdb2Api.WriteAPIBlocking
config InfluxSinkConfig
influxRetryInterval uint
influxMaxRetryTime uint
batch []*write.Point
flushTimer *time.Timer
flushDelay time.Duration
lock sync.Mutex // Flush() runs in another goroutine, so this lock has to protect the buffer
statsSentMetrics int64
statsProcessedMetrics int64
client influxdb2.Client
writeApi influxdb2Api.WriteAPIBlocking
config InfluxSinkConfig
influxRetryInterval uint
influxMaxRetryTime uint
batch []*write.Point
flushTimer *time.Timer
flushDelay time.Duration
lock sync.Mutex // Flush() runs in another goroutine, so this lock has to protect the buffer
//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
}

View File

@@ -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
global_context C.Ganglia_pool
gmond_config C.Ganglia_gmond_config
send_channels C.Ganglia_udp_send_channels
cstrCache map[string]*C.char
statsSentMetrics int64
config LibgangliaSinkConfig
global_context C.Ganglia_pool
gmond_config C.Ganglia_gmond_config
send_channels C.Ganglia_udp_send_channels
cstrCache map[string]*C.char
}
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")

View File

@@ -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
labelMetrics map[string]*prometheus.GaugeVec
nodeMetrics map[string]prometheus.Gauge
promWg sync.WaitGroup
promServer *http.Server
statsSentMetrics int64
config PrometheusSinkConfig
labelMetrics map[string]*prometheus.GaugeVec
nodeMetrics map[string]prometheus.Gauge
promWg sync.WaitGroup
promServer *http.Server
}
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
}

View File

@@ -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 {

View File

@@ -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)
go func(s Sink) {
if err := s.Write(p); err != nil {
cclog.ComponentError("SinkManager", "WRITE", s.Name(), "write failed:", err.Error())
}
wg.Done()
}(s)
if err := s.Write(p); err != nil {
cclog.ComponentError("SinkManager", "WRITE", s.Name(), "write failed:", err.Error())
}
}
wg.Wait()
}
for {

View File

@@ -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
}