mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2025-07-20 03:41:42 +02:00
Compare commits
16 Commits
http_stats
...
v0.5.1
Author | SHA1 | Date | |
---|---|---|---|
|
36dd440864 | ||
|
7b098e0b1b | ||
|
229a57b16a | ||
|
70a9530aba | ||
|
2f0b6057ca | ||
|
69f7c19659 | ||
|
ecdb4c1bcf | ||
|
4d5b1adbc8 | ||
|
4763733d8d | ||
|
16e898ecca | ||
|
4851382ad7 | ||
|
3f76947f54 | ||
|
3157386b3e | ||
|
ff08eaeb43 | ||
|
64c41be34c | ||
|
f4af520b2a |
@@ -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
|
||||
|
@@ -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()
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -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
|
||||
}
|
||||
}
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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
|
||||
}
|
||||
}
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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, ...
|
||||
|
@@ -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() {
|
||||
|
@@ -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() {
|
||||
|
@@ -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.
|
@@ -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
|
||||
}
|
||||
}
|
@@ -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")
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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
|
||||
}
|
||||
|
@@ -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")
|
||||
|
@@ -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
|
||||
}
|
||||
|
||||
|
@@ -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 {
|
||||
|
@@ -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 {
|
||||
|
@@ -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
|
||||
}
|
||||
|
Reference in New Issue
Block a user