cc-metric-collector/sinks/sinkManager.go
Thomas Gruber e28c1fb30b
Ganglia sink using libganglia.so directly (#35)
* Add sink directly using libganglia.so

* Remove unneeded confuse header

* add submodule init to build action

* add submodule init to runonce action

* add installation og ganglia to runonce

* add installation of ganglia to runonce

* add installation of ganglia to runonce

* libconfuse not required

* Remove ganglia submodule

* Remove ganglia.h

* Add Makefile to help creating the libganglia.so link

* Fix cgo header

* Rename new Ganglia sink to 'libgangliaSink'

* Add documentation for libgangliaSink

* Extend make buildsystem with find&symlink helper for libgangliaSink

* Add metric renaming function

* Add build tag 'ganglia' and create corresponding files
2022-02-16 18:33:46 +01:00

180 lines
4.4 KiB
Go

package sinks
import (
"encoding/json"
"fmt"
"os"
"sync"
cclog "github.com/ClusterCockpit/cc-metric-collector/internal/ccLogger"
lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
)
const SINK_MAX_FORWARD = 50
// Map of all available sinks
var AvailableSinks = map[string]Sink{
"influxdb": new(InfluxSink),
"stdout": new(StdoutSink),
"nats": new(NatsSink),
"http": new(HttpSink),
"ganglia": new(GangliaSink),
"influxasync": new(InfluxAsyncSink),
"libganglia": new(LibgangliaSink),
}
// Metric collector manager data structure
type sinkManager struct {
input chan lp.CCMetric // input channel
done chan bool // channel to finish / stop metric sink manager
wg *sync.WaitGroup // wait group for all goroutines in cc-metric-collector
sinks map[string]Sink // Mapping sink name to sink
maxForward int // number of metrics to write maximally in one iteration
}
// Sink manager access functions
type SinkManager interface {
Init(wg *sync.WaitGroup, sinkConfigFile string) error
AddInput(input chan lp.CCMetric)
AddOutput(name string, config json.RawMessage) error
Start()
Close()
}
// Init initializes the sink manager by:
// * Reading its configuration file
// * Adding the configured sinks and providing them with the corresponding config
func (sm *sinkManager) Init(wg *sync.WaitGroup, sinkConfigFile string) error {
sm.input = nil
sm.done = make(chan bool)
sm.wg = wg
sm.sinks = make(map[string]Sink, 0)
sm.maxForward = SINK_MAX_FORWARD
if len(sinkConfigFile) == 0 {
return nil
}
// Read sink config file
configFile, err := os.Open(sinkConfigFile)
if err != nil {
cclog.ComponentError("SinkManager", err.Error())
return err
}
defer configFile.Close()
// Parse config
jsonParser := json.NewDecoder(configFile)
var rawConfigs map[string]json.RawMessage
err = jsonParser.Decode(&rawConfigs)
if err != nil {
cclog.ComponentError("SinkManager", err.Error())
return err
}
// Start sinks
for name, raw := range rawConfigs {
err = sm.AddOutput(name, raw)
if err != nil {
cclog.ComponentError("SinkManager", err.Error())
continue
}
}
return nil
}
// Start starts the sink managers background task, which
// distributes received metrics to the sinks
func (sm *sinkManager) Start() {
sm.wg.Add(1)
go func() {
defer sm.wg.Done()
// Sink manager is done
done := func() {
for _, s := range sm.sinks {
s.Close()
}
close(sm.done)
cclog.ComponentDebug("SinkManager", "DONE")
}
toTheSinks := func(p lp.CCMetric) {
// Send received metric to all outputs
cclog.ComponentDebug("SinkManager", "WRITE", p)
for _, s := range sm.sinks {
if err := s.Write(p); err != nil {
cclog.ComponentError("SinkManager", "WRITE", s.Name(), "write failed:", err.Error())
}
}
}
for {
select {
case <-sm.done:
done()
return
case p := <-sm.input:
toTheSinks(p)
for i := 0; len(sm.input) > 0 && i < sm.maxForward; i++ {
p := <-sm.input
toTheSinks(p)
}
}
}
}()
// Sink manager is started
cclog.ComponentDebug("SinkManager", "STARTED")
}
// AddInput adds the input channel to the sink manager
func (sm *sinkManager) AddInput(input chan lp.CCMetric) {
sm.input = input
}
func (sm *sinkManager) AddOutput(name string, rawConfig json.RawMessage) error {
var err error
var sinkConfig defaultSinkConfig
if len(rawConfig) > 0 {
err := json.Unmarshal(rawConfig, &sinkConfig)
if err != nil {
return err
}
}
if _, found := AvailableSinks[sinkConfig.Type]; !found {
cclog.ComponentError("SinkManager", "SKIP", name, "unknown sink:", sinkConfig.Type)
return err
}
s := AvailableSinks[sinkConfig.Type]
err = s.Init(rawConfig)
if err != nil {
cclog.ComponentError("SinkManager", "SKIP", s.Name(), "initialization failed:", err.Error())
return err
}
sm.sinks[name] = s
cclog.ComponentDebug("SinkManager", "ADD SINK", s.Name(), "with name", fmt.Sprintf("'%s'", name))
return nil
}
// Close finishes / stops the sink manager
func (sm *sinkManager) Close() {
cclog.ComponentDebug("SinkManager", "CLOSE")
sm.done <- true
// wait for close of channel sm.done
<-sm.done
}
// New creates a new initialized sink manager
func New(wg *sync.WaitGroup, sinkConfigFile string) (SinkManager, error) {
sm := new(sinkManager)
err := sm.Init(wg, sinkConfigFile)
if err != nil {
return nil, err
}
return sm, err
}