mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2024-11-14 14:07:26 +01:00
be20f956c2
* InfiniBandCollector: Scale raw readings from octets to bytes * Fix clock frequency coming from LikwidCollector and update docs * Build DEB package for Ubuntu 20.04 for releases * Fix memstat collector with numa_stats option * Remove useless prints from MemstatCollector * Replace ioutils with os and io (#87) * Use lower case for error strings in RocmSmiCollector * move maybe-usable-by-other-cc-components to pkg. Fix all files to use the new paths (#88) * Add collector for monitoring the execution of cc-metric-collector itself (#81) * Add collector to monitor execution of cc-metric-collector itself * Register SelfCollector * Fix import paths for moved packages
116 lines
2.9 KiB
Go
116 lines
2.9 KiB
Go
package receivers
|
|
|
|
import (
|
|
"encoding/json"
|
|
"os"
|
|
"sync"
|
|
|
|
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
|
lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
|
|
)
|
|
|
|
var AvailableReceivers = map[string]func(name string, config json.RawMessage) (Receiver, error){
|
|
"nats": NewNatsReceiver,
|
|
"redfish": NewRedfishReceiver,
|
|
}
|
|
|
|
type receiveManager struct {
|
|
inputs []Receiver
|
|
output chan lp.CCMetric
|
|
config []json.RawMessage
|
|
}
|
|
|
|
type ReceiveManager interface {
|
|
Init(wg *sync.WaitGroup, receiverConfigFile string) error
|
|
AddInput(name string, rawConfig json.RawMessage) error
|
|
AddOutput(output chan lp.CCMetric)
|
|
Start()
|
|
Close()
|
|
}
|
|
|
|
func (rm *receiveManager) Init(wg *sync.WaitGroup, receiverConfigFile string) error {
|
|
// Initialize struct fields
|
|
rm.inputs = make([]Receiver, 0)
|
|
rm.output = nil
|
|
rm.config = make([]json.RawMessage, 0)
|
|
|
|
configFile, err := os.Open(receiverConfigFile)
|
|
if err != nil {
|
|
cclog.ComponentError("ReceiveManager", err.Error())
|
|
return err
|
|
}
|
|
defer configFile.Close()
|
|
jsonParser := json.NewDecoder(configFile)
|
|
var rawConfigs map[string]json.RawMessage
|
|
err = jsonParser.Decode(&rawConfigs)
|
|
if err != nil {
|
|
cclog.ComponentError("ReceiveManager", err.Error())
|
|
return err
|
|
}
|
|
for name, raw := range rawConfigs {
|
|
rm.AddInput(name, raw)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (rm *receiveManager) Start() {
|
|
cclog.ComponentDebug("ReceiveManager", "START")
|
|
|
|
for _, r := range rm.inputs {
|
|
cclog.ComponentDebug("ReceiveManager", "START", r.Name())
|
|
r.Start()
|
|
}
|
|
cclog.ComponentDebug("ReceiveManager", "STARTED")
|
|
}
|
|
|
|
func (rm *receiveManager) AddInput(name string, rawConfig json.RawMessage) error {
|
|
var config defaultReceiverConfig
|
|
err := json.Unmarshal(rawConfig, &config)
|
|
if err != nil {
|
|
cclog.ComponentError("ReceiveManager", "SKIP", config.Type, "JSON config error:", err.Error())
|
|
return err
|
|
}
|
|
if _, found := AvailableReceivers[config.Type]; !found {
|
|
cclog.ComponentError("ReceiveManager", "SKIP", config.Type, "unknown receiver:", err.Error())
|
|
return err
|
|
}
|
|
r, err := AvailableReceivers[config.Type](name, rawConfig)
|
|
if err != nil {
|
|
cclog.ComponentError("ReceiveManager", "SKIP", name, "initialization failed:", err.Error())
|
|
return err
|
|
}
|
|
rm.inputs = append(rm.inputs, r)
|
|
rm.config = append(rm.config, rawConfig)
|
|
cclog.ComponentDebug("ReceiveManager", "ADD RECEIVER", r.Name())
|
|
return nil
|
|
}
|
|
|
|
func (rm *receiveManager) AddOutput(output chan lp.CCMetric) {
|
|
rm.output = output
|
|
for _, r := range rm.inputs {
|
|
r.SetSink(rm.output)
|
|
}
|
|
}
|
|
|
|
func (rm *receiveManager) Close() {
|
|
cclog.ComponentDebug("ReceiveManager", "CLOSE")
|
|
|
|
// Close all receivers
|
|
for _, r := range rm.inputs {
|
|
cclog.ComponentDebug("ReceiveManager", "CLOSE", r.Name())
|
|
r.Close()
|
|
}
|
|
|
|
cclog.ComponentDebug("ReceiveManager", "DONE")
|
|
}
|
|
|
|
func New(wg *sync.WaitGroup, receiverConfigFile string) (ReceiveManager, error) {
|
|
r := new(receiveManager)
|
|
err := r.Init(wg, receiverConfigFile)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return r, err
|
|
}
|