mirror of
				https://github.com/ClusterCockpit/cc-metric-collector.git
				synced 2025-11-04 10:45:06 +01:00 
			
		
		
		
	Provide info to CollectorManager whether the collector can be executed in parallel with others
This commit is contained in:
		@@ -55,6 +55,7 @@ func (m *BeegfsMetaCollector) Init(config json.RawMessage) error {
 | 
			
		||||
 | 
			
		||||
	m.name = "BeegfsMetaCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	// Set default beegfs-ctl binary
 | 
			
		||||
 | 
			
		||||
	m.config.Beegfs = DEFAULT_BEEGFS_CMD
 | 
			
		||||
 
 | 
			
		||||
@@ -48,6 +48,7 @@ func (m *BeegfsStorageCollector) Init(config json.RawMessage) error {
 | 
			
		||||
 | 
			
		||||
	m.name = "BeegfsStorageCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	// Set default beegfs-ctl binary
 | 
			
		||||
 | 
			
		||||
	m.config.Beegfs = DEFAULT_BEEGFS_CMD
 | 
			
		||||
 
 | 
			
		||||
@@ -48,6 +48,7 @@ func (m *CPUFreqCpuInfoCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.setup()
 | 
			
		||||
 | 
			
		||||
	m.name = "CPUFreqCpuInfoCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{
 | 
			
		||||
		"source": m.name,
 | 
			
		||||
		"group":  "CPU",
 | 
			
		||||
 
 | 
			
		||||
@@ -53,6 +53,7 @@ func (m *CPUFreqCollector) Init(config json.RawMessage) error {
 | 
			
		||||
 | 
			
		||||
	m.name = "CPUFreqCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
		err := json.Unmarshal(config, &m.config)
 | 
			
		||||
		if err != nil {
 | 
			
		||||
 
 | 
			
		||||
@@ -30,6 +30,7 @@ type CpustatCollector struct {
 | 
			
		||||
func (m *CpustatCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "CpustatCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "CPU", "unit": "Percent"}
 | 
			
		||||
	m.nodetags = map[string]string{"type": "node"}
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
 
 | 
			
		||||
@@ -33,6 +33,7 @@ type CustomCmdCollector struct {
 | 
			
		||||
func (m *CustomCmdCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "CustomCmdCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "Custom"}
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
		err = json.Unmarshal(config, &m.config)
 | 
			
		||||
 
 | 
			
		||||
@@ -29,6 +29,7 @@ type DiskstatCollector struct {
 | 
			
		||||
 | 
			
		||||
func (m *DiskstatCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "DiskstatCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "Disk"}
 | 
			
		||||
	m.setup()
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
@@ -77,11 +78,18 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMetric
 | 
			
		||||
			continue
 | 
			
		||||
		}
 | 
			
		||||
		path := strings.Replace(linefields[1], `\040`, " ", -1)
 | 
			
		||||
		stat := syscall.Statfs_t{}
 | 
			
		||||
		stat := syscall.Statfs_t{
 | 
			
		||||
			Blocks: 0,
 | 
			
		||||
			Bsize:  0,
 | 
			
		||||
			Bfree:  0,
 | 
			
		||||
		}
 | 
			
		||||
		err := syscall.Statfs(path, &stat)
 | 
			
		||||
		if err != nil {
 | 
			
		||||
			continue
 | 
			
		||||
		}
 | 
			
		||||
		if stat.Blocks == 0 || stat.Bsize == 0 {
 | 
			
		||||
			continue
 | 
			
		||||
		}
 | 
			
		||||
		tags := map[string]string{"type": "node", "device": linefields[0]}
 | 
			
		||||
		total := (stat.Blocks * uint64(stat.Bsize)) / uint64(1000000000)
 | 
			
		||||
		y, err := lp.New("disk_total", tags, m.meta, map[string]interface{}{"value": total}, time.Now())
 | 
			
		||||
@@ -95,9 +103,11 @@ func (m *DiskstatCollector) Read(interval time.Duration, output chan lp.CCMetric
 | 
			
		||||
			y.AddMeta("unit", "GBytes")
 | 
			
		||||
			output <- y
 | 
			
		||||
		}
 | 
			
		||||
		perc := (100 * (total - free)) / total
 | 
			
		||||
		if perc > part_max_used {
 | 
			
		||||
			part_max_used = perc
 | 
			
		||||
		if total > 0 {
 | 
			
		||||
			perc := (100 * (total - free)) / total
 | 
			
		||||
			if perc > part_max_used {
 | 
			
		||||
				part_max_used = perc
 | 
			
		||||
			}
 | 
			
		||||
		}
 | 
			
		||||
	}
 | 
			
		||||
	y, err := lp.New("part_max_used", map[string]string{"type": "node"}, m.meta, map[string]interface{}{"value": int(part_max_used)}, time.Now())
 | 
			
		||||
 
 | 
			
		||||
@@ -46,6 +46,7 @@ func (m *GpfsCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "GpfsCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
 | 
			
		||||
	// Set default mmpmon binary
 | 
			
		||||
	m.config.Mmpmon = DEFAULT_GPFS_CMD
 | 
			
		||||
 
 | 
			
		||||
@@ -54,6 +54,7 @@ func (m *InfinibandCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "InfinibandCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{
 | 
			
		||||
		"source": m.name,
 | 
			
		||||
		"group":  "Network",
 | 
			
		||||
 
 | 
			
		||||
@@ -37,6 +37,7 @@ type IOstatCollector struct {
 | 
			
		||||
func (m *IOstatCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "IOstatCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "Disk"}
 | 
			
		||||
	m.setup()
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
 
 | 
			
		||||
@@ -34,6 +34,7 @@ type IpmiCollector struct {
 | 
			
		||||
func (m *IpmiCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "IpmiCollector"
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "IPMI"}
 | 
			
		||||
	m.config.IpmitoolPath = string(IPMITOOL_PATH)
 | 
			
		||||
	m.config.IpmisensorsPath = string(IPMISENSORS_PATH)
 | 
			
		||||
 
 | 
			
		||||
@@ -177,6 +177,7 @@ func getBaseFreq() float64 {
 | 
			
		||||
 | 
			
		||||
func (m *LikwidCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "LikwidCollector"
 | 
			
		||||
	m.parallel = false
 | 
			
		||||
	m.initialized = false
 | 
			
		||||
	m.running = false
 | 
			
		||||
	m.config.AccessMode = LIKWID_DEF_ACCESSMODE
 | 
			
		||||
 
 | 
			
		||||
@@ -36,6 +36,7 @@ type LoadavgCollector struct {
 | 
			
		||||
 | 
			
		||||
func (m *LoadavgCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "LoadavgCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.setup()
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
		err := json.Unmarshal(config, &m.config)
 | 
			
		||||
 
 | 
			
		||||
@@ -288,6 +288,7 @@ var LustreDeriveMetrics = []LustreMetricDefinition{
 | 
			
		||||
func (m *LustreCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "LustreCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
		err = json.Unmarshal(config, &m.config)
 | 
			
		||||
		if err != nil {
 | 
			
		||||
 
 | 
			
		||||
@@ -81,6 +81,7 @@ func getStats(filename string) map[string]MemstatStats {
 | 
			
		||||
func (m *MemstatCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "MemstatCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.config.NodeStats = true
 | 
			
		||||
	m.config.NumaStats = false
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
@@ -159,8 +160,10 @@ func (m *MemstatCollector) Init(config json.RawMessage) error {
 | 
			
		||||
 | 
			
		||||
func (m *MemstatCollector) Read(interval time.Duration, output chan lp.CCMetric) {
 | 
			
		||||
	if !m.init {
 | 
			
		||||
		cclog.ComponentPrint(m.name, "Here")
 | 
			
		||||
		return
 | 
			
		||||
	}
 | 
			
		||||
	cclog.ComponentPrint(m.name, time.Now())
 | 
			
		||||
 | 
			
		||||
	sendStats := func(stats map[string]MemstatStats, tags map[string]string) {
 | 
			
		||||
		for match, name := range m.matches {
 | 
			
		||||
 
 | 
			
		||||
@@ -13,17 +13,19 @@ import (
 | 
			
		||||
)
 | 
			
		||||
 | 
			
		||||
type MetricCollector interface {
 | 
			
		||||
	Name() string                                         // Name of the metric collector
 | 
			
		||||
	Init(config json.RawMessage) error                    // Initialize metric collector
 | 
			
		||||
	Initialized() bool                                    // Is metric collector initialized?
 | 
			
		||||
	Name() string                      // Name of the metric collector
 | 
			
		||||
	Init(config json.RawMessage) error // Initialize metric collector
 | 
			
		||||
	Initialized() bool                 // Is metric collector initialized?
 | 
			
		||||
	Parallel() bool
 | 
			
		||||
	Read(duration time.Duration, output chan lp.CCMetric) // Read metrics from metric collector
 | 
			
		||||
	Close()                                               // Close / finish metric collector
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
type metricCollector struct {
 | 
			
		||||
	name string            // name of the metric
 | 
			
		||||
	init bool              // is metric collector initialized?
 | 
			
		||||
	meta map[string]string // static meta data tags
 | 
			
		||||
	name     string            // name of the metric
 | 
			
		||||
	init     bool              // is metric collector initialized?
 | 
			
		||||
	parallel bool              // can the metric collector be executed in parallel with others
 | 
			
		||||
	meta     map[string]string // static meta data tags
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Name returns the name of the metric collector
 | 
			
		||||
@@ -31,6 +33,11 @@ func (c *metricCollector) Name() string {
 | 
			
		||||
	return c.name
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Name returns the name of the metric collector
 | 
			
		||||
func (c *metricCollector) Parallel() bool {
 | 
			
		||||
	return c.parallel
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
// Setup is for future use
 | 
			
		||||
func (c *metricCollector) setup() error {
 | 
			
		||||
	return nil
 | 
			
		||||
 
 | 
			
		||||
@@ -39,6 +39,7 @@ type NetstatCollector struct {
 | 
			
		||||
 | 
			
		||||
func (m *NetstatCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "NetstatCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.lastTimestamp = time.Now()
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
@@ -114,6 +114,7 @@ func (m *nfsCollector) MainInit(config json.RawMessage) error {
 | 
			
		||||
	m.data = make(map[string]NfsCollectorData)
 | 
			
		||||
	m.initStats()
 | 
			
		||||
	m.init = true
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	return nil
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
@@ -54,6 +54,7 @@ func (m *NUMAStatsCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	m.name = "NUMAStatsCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.setup()
 | 
			
		||||
	m.meta = map[string]string{
 | 
			
		||||
		"source": m.name,
 | 
			
		||||
 
 | 
			
		||||
@@ -35,6 +35,10 @@ func (m *SampleCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	m.name = "InternalCollector"
 | 
			
		||||
	// This is for later use, also call it early
 | 
			
		||||
	m.setup()
 | 
			
		||||
	// Tell whether the collector should be run in parallel with others (reading files, ...)
 | 
			
		||||
	// or it should be run serially, mostly for collectors acutally doing measurements
 | 
			
		||||
	// because they should not measure the execution of the other collectors
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	// Define meta information sent with each metric
 | 
			
		||||
	// (Can also be dynamic or this is the basic set with extension through AddMeta())
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "SAMPLE"}
 | 
			
		||||
 
 | 
			
		||||
@@ -50,6 +50,7 @@ func (m *TempCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	}
 | 
			
		||||
 | 
			
		||||
	m.name = "TempCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.setup()
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
		err := json.Unmarshal(config, &m.config)
 | 
			
		||||
@@ -116,6 +117,10 @@ func (m *TempCollector) Init(config json.RawMessage) error {
 | 
			
		||||
		}
 | 
			
		||||
 | 
			
		||||
		// Sensor file
 | 
			
		||||
		_, err = ioutil.ReadFile(file)
 | 
			
		||||
		if err != nil {
 | 
			
		||||
			continue
 | 
			
		||||
		}
 | 
			
		||||
		sensor.file = file
 | 
			
		||||
 | 
			
		||||
		// Sensor tags
 | 
			
		||||
 
 | 
			
		||||
@@ -28,6 +28,7 @@ type TopProcsCollector struct {
 | 
			
		||||
func (m *TopProcsCollector) Init(config json.RawMessage) error {
 | 
			
		||||
	var err error
 | 
			
		||||
	m.name = "TopProcsCollector"
 | 
			
		||||
	m.parallel = true
 | 
			
		||||
	m.tags = map[string]string{"type": "node"}
 | 
			
		||||
	m.meta = map[string]string{"source": m.name, "group": "TopProcs"}
 | 
			
		||||
	if len(config) > 0 {
 | 
			
		||||
 
 | 
			
		||||
		Reference in New Issue
	
	Block a user