cc-metric-collector/collectors/collectorManager.go

// Copyright (C) NHR@FAU, University Erlangen-Nuremberg.
// All rights reserved. This file is part of cc-lib.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
// additional authors:
// Holger Obermaier (NHR@KIT)

package collectors

import (
	"bytes"
	"encoding/json"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	cclog "github.com/ClusterCockpit/cc-lib/v2/ccLogger"
	lp "github.com/ClusterCockpit/cc-lib/v2/ccMessage"
	mct "github.com/ClusterCockpit/cc-metric-collector/pkg/multiChanTicker"
)

// Map of all available metric collectors
var AvailableCollectors = map[string]MetricCollector{
	"likwid":             new(LikwidCollector),
	"loadavg":            new(LoadavgCollector),
	"memstat":            new(MemstatCollector),
	"netstat":            new(NetstatCollector),
	"ibstat":             new(InfinibandCollector),
	"lustrestat":         new(LustreCollector),
	"cpustat":            new(CpustatCollector),
	"topprocs":           new(TopProcsCollector),
	"nvidia":             new(NvidiaCollector),
	"customcmd":          new(CustomCmdCollector),
	"iostat":             new(IOstatCollector),
	"diskstat":           new(DiskstatCollector),
	"tempstat":           new(TempCollector),
	"ipmistat":           new(IpmiCollector),
	"gpfs":               new(GpfsCollector),
	"cpufreq":            new(CPUFreqCollector),
	"cpufreq_cpuinfo":    new(CPUFreqCpuInfoCollector),
	"nfs3stat":           new(Nfs3Collector),
	"nfs4stat":           new(Nfs4Collector),
	"numastats":          new(NUMAStatsCollector),
	"beegfs_meta":        new(BeegfsMetaCollector),
	"beegfs_storage":     new(BeegfsStorageCollector),
	"rapl":               new(RAPLCollector),
	"rocm_smi":           new(RocmSmiCollector),
	"self":               new(SelfCollector),
	"schedstat":          new(SchedstatCollector),
	"nfsiostat":          new(NfsIOStatCollector),
	"slurm_cgroup":       new(SlurmCgroupCollector),
	"smartmon":           new(SmartMonCollector),
	"lenovo_dense_power": new(LenovoDensePowerCollector),
	"megware_eureka":     new(MegwareEurekaCollector),
}

// Metric collector manager data structure
type collectorManager struct {
	collectors    []MetricCollector          // List of metric collectors to read in parallel
	serial        []MetricCollector          // List of metric collectors to read serially
	output        chan lp.CCMessage          // Output channels
	done          chan bool                  // channel to finish / stop metric collector manager
	ticker        mct.MultiChanTicker        // periodically ticking once each interval
	duration      time.Duration              // duration (for metrics that measure over a given duration)
	wg            *sync.WaitGroup            // wait group for all goroutines in cc-metric-collector
	config        map[string]json.RawMessage // json encoded config for collector manager
	collector_wg  sync.WaitGroup             // internally used wait group for the parallel reading of collector
	round_wg      sync.WaitGroup             // wait group for the currently running collection round
	round_running atomic.Bool                // Flag whether a collection round is currently running
}

// Metric collector manager access functions
type CollectorManager interface {
	Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error
	AddOutput(output chan lp.CCMessage)
	Start()
	Close()
}

// Init initializes a new metric collector manager by setting up:
// * output channel
// * done channel
// * wait group synchronization for goroutines (from variable wg)
// * ticker (from variable ticker)
// * configuration (read from config file in variable collectConfigFile)
// Initialization is done for all configured collectors
func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
	cm.collectors = make([]MetricCollector, 0)
	cm.serial = make([]MetricCollector, 0)
	cm.output = nil
	cm.done = make(chan bool)
	cm.wg = wg
	cm.ticker = ticker
	cm.duration = duration

	d := json.NewDecoder(bytes.NewReader(collectConfig))
	d.DisallowUnknownFields()
	if err := d.Decode(&cm.config); err != nil {
		return fmt.Errorf("%s Init(): Error decoding collector manager config: %w", "CollectorManager", err)
	}

	// Initialize configured collectors
	for collectorName, collectorCfg := range cm.config {
		if _, found := AvailableCollectors[collectorName]; !found {
			cclog.ComponentError("CollectorManager", "SKIP unknown collector", collectorName)
			continue
		}
		collector := AvailableCollectors[collectorName]

		err := collector.Init(collectorCfg)
		if err != nil {
			cclog.ComponentError("CollectorManager", fmt.Sprintf("Collector %s initialization failed: %v", collectorName, err))
			continue
		}
		cclog.ComponentDebug("CollectorManager", "ADD COLLECTOR", collector.Name())
		if collector.Parallel() {
			cm.collectors = append(cm.collectors, collector)
		} else {
			cm.serial = append(cm.serial, collector)
		}
	}
	return nil
}

// runRound executes one collection round: first all parallel collectors
// concurrently, then the serial collectors one by one
func (cm *collectorManager) runRound(t time.Time) {
	roundStart := time.Now()
	for _, c := range cm.collectors {
		// Read metrics from collector c via goroutine
		cclog.ComponentDebug("CollectorManager", c.Name(), t)
		cm.collector_wg.Add(1)
		go func(myc MetricCollector) {
			start := time.Now()
			myc.Read(cm.duration, cm.output)
			cclog.ComponentDebug("CollectorManager", myc.Name(), "took", time.Since(start))
			cm.collector_wg.Done()
		}(c)
	}
	cm.collector_wg.Wait()
	for _, c := range cm.serial {
		// Read metrics from collector c
		cclog.ComponentDebug("CollectorManager", c.Name(), t)
		start := time.Now()
		c.Read(cm.duration, cm.output)
		cclog.ComponentDebug("CollectorManager", c.Name(), "took", time.Since(start))
	}
	cclog.ComponentDebug("CollectorManager", "collection round took", time.Since(roundStart))
}

// Start starts the metric collector manager
func (cm *collectorManager) Start() {
	tick := make(chan time.Time, 1)
	cm.ticker.AddChannel(tick)

	cm.wg.Go(func() {
		// Collector manager is done
		done := func() {
			// wait for a still running collection round, then close all metric collectors
			cm.round_wg.Wait()
			for _, c := range cm.collectors {
				c.Close()
			}
			for _, c := range cm.serial {
				c.Close()
			}
			close(cm.done)
			cclog.ComponentDebug("CollectorManager", "DONE")
		}

		var roundStart time.Time

		// Wait for done signal or timer event
		for {
			select {
			case <-cm.done:
				done()
				return
			case t := <-tick:
				// The round runs detached from this loop, so the tick channel
				// stays drained even when a round takes longer than the interval
				if cm.round_running.Load() {
					cclog.ComponentWarn("CollectorManager", "collection round still running after", time.Since(roundStart), "- skipping tick")
					continue
				}
				cm.round_running.Store(true)
				roundStart = time.Now()
				cm.round_wg.Add(1)
				go func() {
					defer cm.round_wg.Done()
					defer cm.round_running.Store(false)
					cm.runRound(t)
				}()
			}
		}
	})

	// Collector manager is started
	cclog.ComponentDebug("CollectorManager", "STARTED")
}

// AddOutput adds the output channel to the metric collector manager
func (cm *collectorManager) AddOutput(output chan lp.CCMessage) {
	cm.output = output
}

// Close finishes / stops the metric collector manager
func (cm *collectorManager) Close() {
	cclog.ComponentDebug("CollectorManager", "CLOSE")
	cm.done <- true
	// wait for close of channel cm.done
	<-cm.done
}

// New creates a new initialized metric collector manager
func New(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
	cm := new(collectorManager)
	err := cm.Init(ticker, duration, wg, collectConfig)
	if err != nil {
		return nil, err
	}
	return cm, err
}