Modularize the whole thing (#16)

* Use channels, add a metric router, split up configuration and use extended version of Influx line protocol internally * Use central timer for collectors and router. Add expressions to router * Add expression to router config * Update entry points * Start with README * Update README for CCMetric * Formatting * Update README.md * Add README for MultiChanTicker * Add README for MultiChanTicker * Update README.md * Add README to metric router * Update main README * Remove SinkEntity type * Update README for sinks * Update go files * Update README for receivers * Update collectors README * Update collectors README * Use seperate page per collector * Fix for tempstat page * Add docs for customcmd collector * Add docs for ipmistat collector * Add docs for topprocs collector * Update customCmdMetric.md * Use seconds when calculating LIKWID metrics * Add IB metrics ib_recv_pkts and ib_xmit_pkts * Drop domain part of host name * Updated to latest stable version of likwid * Define source code dependencies in Makefile * Add GPFS / IBM Spectrum Scale collector * Add vet and staticcheck make targets * Add vet and staticcheck make targets * Avoid go vet warning: struct field tag `json:"..., omitempty"` not compatible with reflect.StructTag.Get: suspicious space in struct tag value struct field tag `json:"...", omitempty` not compatible with reflect.StructTag.Get: key:"value" pairs not separated by spaces * Add sample collector to README.md * Add CPU frequency collector * Avoid staticcheck warning: redundant return statement * Avoid staticcheck warning: unnecessary assignment to the blank identifier * Simplified code * Add CPUFreqCollectorCpuinfo a metric collector to measure the current frequency of the CPUs as obtained from /proc/cpuinfo Only measure on the first hyperthread * Add collector for NFS clients * Move publication of metrics into Flush() for NatsSink * Update GitHub actions * Refactoring * Avoid vet warning: Println arg list ends with redundant newline * Avoid vet warning struct field commands has json tag but is not exported * Avoid vet warning: return copies lock value. * Corrected typo * Refactoring * Add go sources in internal/... * Bad separator in Makefile * Fix Infiniband collector Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com>
2025-08-20 09:22:59 +02:00 · 2022-01-25 15:37:43 +01:00
parent 222862af32
commit 200af84c54
60 changed files with 2596 additions and 1105 deletions
--- a/sinks/sinkManager.go
+++ b/sinks/sinkManager.go
@@ -0,0 +1,141 @@
+package sinks
+
+import (
+	"encoding/json"
+	"log"
+	"os"
+	"sync"
+
+	lp "github.com/ClusterCockpit/cc-metric-collector/internal/ccMetric"
+)
+
+var AvailableSinks = map[string]Sink{
+	"influxdb": &InfluxSink{},
+	"stdout":   &StdoutSink{},
+	"nats":     &NatsSink{},
+	"http":     &HttpSink{},
+}
+
+type sinkManager struct {
+	input   chan lp.CCMetric
+	outputs []Sink
+	done    chan bool
+	wg      *sync.WaitGroup
+	config  []sinkConfig
+}
+
+type SinkManager interface {
+	Init(wg *sync.WaitGroup, sinkConfigFile string) error
+	AddInput(input chan lp.CCMetric)
+	AddOutput(config json.RawMessage) error
+	Start()
+	Close()
+}
+
+func (sm *sinkManager) Init(wg *sync.WaitGroup, sinkConfigFile string) error {
+	sm.input = nil
+	sm.outputs = make([]Sink, 0)
+	sm.done = make(chan bool)
+	sm.wg = wg
+	sm.config = make([]sinkConfig, 0)
+	if len(sinkConfigFile) > 0 {
+		configFile, err := os.Open(sinkConfigFile)
+		if err != nil {
+			log.Print("[SinkManager] ", err.Error())
+			return err
+		}
+		defer configFile.Close()
+		jsonParser := json.NewDecoder(configFile)
+		var rawConfigs []json.RawMessage
+		err = jsonParser.Decode(&rawConfigs)
+		if err != nil {
+			log.Print("[SinkManager] ", err.Error())
+			return err
+		}
+		for _, raw := range rawConfigs {
+			err = sm.AddOutput(raw)
+			if err != nil {
+				continue
+			}
+		}
+	}
+	return nil
+}
+
+func (sm *sinkManager) Start() {
+	sm.wg.Add(1)
+	batchcount := 20
+	go func() {
+		for {
+		SinkManagerLoop:
+			select {
+			case <-sm.done:
+				for _, s := range sm.outputs {
+					s.Close()
+				}
+				log.Print("[SinkManager] DONE\n")
+				sm.wg.Done()
+				break SinkManagerLoop
+			case p := <-sm.input:
+				log.Print("[SinkManager] WRITE ", p)
+				for _, s := range sm.outputs {
+					s.Write(p)
+				}
+				if batchcount == 0 {
+					log.Print("[SinkManager] FLUSH")
+					for _, s := range sm.outputs {
+						s.Flush()
+					}
+					batchcount = 20
+				}
+				batchcount--
+			default:
+			}
+		}
+		log.Print("[SinkManager] EXIT\n")
+	}()
+	log.Print("[SinkManager] STARTED\n")
+}
+
+func (sm *sinkManager) AddInput(input chan lp.CCMetric) {
+	sm.input = input
+}
+
+func (sm *sinkManager) AddOutput(rawConfig json.RawMessage) error {
+	var err error
+	var config sinkConfig
+	if len(rawConfig) > 3 {
+		err = json.Unmarshal(rawConfig, &config)
+		if err != nil {
+			log.Print("[SinkManager] SKIP ", config.Type, " JSON config error: ", err.Error())
+			return err
+		}
+	}
+	if _, found := AvailableSinks[config.Type]; !found {
+		log.Print("[SinkManager] SKIP ", config.Type, " unknown sink: ", err.Error())
+		return err
+	}
+	s := AvailableSinks[config.Type]
+	err = s.Init(config)
+	if err != nil {
+		log.Print("[SinkManager] SKIP ", s.Name(), " initialization failed: ", err.Error())
+		return err
+	}
+	sm.outputs = append(sm.outputs, s)
+	sm.config = append(sm.config, config)
+	return nil
+}
+
+func (sm *sinkManager) Close() {
+	sm.done <- true
+	log.Print("[SinkManager] CLOSE")
+}
+
+func New(wg *sync.WaitGroup, sinkConfigFile string) (SinkManager, error) {
+	sm := &sinkManager{}
+	err := sm.Init(wg, sinkConfigFile)
+	if err != nil {
+		return nil, err
+	}
+	return sm, err
+}