collectorManager: Make markers opt-in

These markers are sent out at the beginning and end of a collection run.
This can be used to reconstruct, which metrics belong together and were
obtained during the same tick.

cc-metric-collector.go

@@ -31,11 +31,13 @@ import (
 type CentralConfigFile struct {
 	Interval      string `json:"interval"`
 	Duration      string `json:"duration"`
+	EnableMarkers bool   `json:"enable-markers"`
 }
 
 type RuntimeConfig struct {
 	Interval      time.Duration
 	Duration      time.Duration
+	EnableMarkers bool
 	CliArgs       map[string]string
 	ConfigFile    CentralConfigFile
 
@@ -157,6 +159,7 @@ func mainFunc() int {
 		cclog.Error("The interval should be greater than duration")
 		return 1
 	}
+	rcfg.EnableMarkers = rcfg.ConfigFile.EnableMarkers
 
 	routerConf := ccconf.GetPackageConfig("router")
 	if len(routerConf) == 0 {
@@ -199,7 +202,7 @@ func mainFunc() int {
 	rcfg.MetricRouter.AddOutput(RouterToSinksChannel)
 
 	// Create new collector manager
-	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, &rcfg.Sync, collectorConf)
+	rcfg.CollectManager, err = collectors.New(rcfg.MultiChanTicker, rcfg.Duration, rcfg.EnableMarkers, &rcfg.Sync, collectorConf)
 	if err != nil {
 		cclog.Error(err.Error())
 		return 1

collectors/collectorManager.go

@@ -66,11 +66,12 @@ type collectorManager struct {
 	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
 	parallel_run  bool                       // Flag whether the collectors are currently read in parallel
+	enableMarkers bool                       // Send ccmc-{begin,end} metrics
 }
 
 // Metric collector manager access functions
 type CollectorManager interface {
-	Init(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) error
+	Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error
 	AddOutput(output chan lp.CCMessage)
 	Start()
 	Close()
@@ -83,7 +84,7 @@ type CollectorManager interface {
 // * 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 {
+func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) error {
 	cm.collectors = make([]MetricCollector, 0)
 	cm.serial = make([]MetricCollector, 0)
 	cm.output = nil
@@ -91,6 +92,7 @@ func (cm *collectorManager) Init(ticker mct.MultiChanTicker, duration time.Durat
 	cm.wg = wg
 	cm.ticker = ticker
 	cm.duration = duration
+	cm.enableMarkers = enableMarkers
 
 	d := json.NewDecoder(bytes.NewReader(collectConfig))
 	d.DisallowUnknownFields()
@@ -148,6 +150,9 @@ func (cm *collectorManager) Start() {
 				done()
 				return
 			case t := <-tick:
+				if cm.enableMarkers {
+					lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now())
+				}
 				cm.parallel_run = true
 				for _, c := range cm.collectors {
 					// Wait for done signal or execute the collector
@@ -179,6 +184,9 @@ func (cm *collectorManager) Start() {
 						c.Read(cm.duration, cm.output)
 					}
 				}
+				if cm.enableMarkers {
+					lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now())
+				}
 			}
 		}
 	})
@@ -201,9 +209,9 @@ func (cm *collectorManager) Close() {
 }
 
 // New creates a new initialized metric collector manager
-func New(ticker mct.MultiChanTicker, duration time.Duration, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
+func New(ticker mct.MultiChanTicker, duration time.Duration, enableMarkers bool, wg *sync.WaitGroup, collectConfig json.RawMessage) (CollectorManager, error) {
 	cm := new(collectorManager)
-	err := cm.Init(ticker, duration, wg, collectConfig)
+	err := cm.Init(ticker, duration, enableMarkers, wg, collectConfig)
 	if err != nil {
 		return nil, err
 	}

collectors/megwareEureka.go

@@ -65,7 +65,7 @@ func (m *MegwareEurekaCollector) Init(config json.RawMessage) error {
 		return nil
 	}
 
-	m.name = "MegwareEureka"
+	m.name = "MegwareEurekaCollector"
 	if err := m.setup(); err != nil {
 		return fmt.Errorf("%s Init(): setup() call failed: %w", m.name, err)
 	}
@@ -127,9 +127,6 @@ func (m *MegwareEurekaCollector) readMpsData() (*mpsData, error) {
 		return nil, fmt.Errorf("unable to decode u20 JSON output: %w (stdout=%s)", err, stdout.String())
 	}
 
-	fmt.Printf("string: %+v\n", stdout.String())
-	fmt.Printf("obj:    %+v\n", u20output)
-
 	u20output.GetMpsPollValues.Timestamp = time.Now()
 
 	return &u20output.GetMpsPollValues, nil