diff --git a/cc-metric-collector.go b/cc-metric-collector.go index 3188f9c..30b8879 100644 --- a/cc-metric-collector.go +++ b/cc-metric-collector.go @@ -29,15 +29,17 @@ import ( ) type CentralConfigFile struct { - Interval string `json:"interval"` - Duration string `json:"duration"` + Interval string `json:"interval"` + Duration string `json:"duration"` + EnableMarkers bool `json:"enable-markers"` } type RuntimeConfig struct { - Interval time.Duration - Duration time.Duration - CliArgs map[string]string - ConfigFile CentralConfigFile + Interval time.Duration + Duration time.Duration + EnableMarkers bool + CliArgs map[string]string + ConfigFile CentralConfigFile MetricRouter mr.MetricRouter CollectManager collectors.CollectorManager @@ -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 diff --git a/collectors/collectorManager.go b/collectors/collectorManager.go index 58f9a3c..75c3835 100644 --- a/collectors/collectorManager.go +++ b/collectors/collectorManager.go @@ -56,21 +56,22 @@ var AvailableCollectors = map[string]MetricCollector{ // 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 - parallel_run bool // Flag whether the collectors are currently read in parallel + 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 + 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,7 +150,9 @@ func (cm *collectorManager) Start() { done() return case t := <-tick: - lp.NewMetric("ccmc-begin", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now()) + 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 @@ -180,7 +184,9 @@ func (cm *collectorManager) Start() { c.Read(cm.duration, cm.output) } } - lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now()) + if cm.enableMarkers { + lp.NewMetric("ccmc-end", map[string]string{"type": "node"}, nil, map[string]interface{}{"value": 0}, time.Now()) + } } } }) @@ -203,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 }