mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2024-11-10 04:27:25 +01:00
Wait for close of done channel, to ensure manager finished.
This commit is contained in:
parent
8df58c051f
commit
9e99e47d73
@ -115,6 +115,7 @@ func (cm *collectorManager) Start() {
|
||||
for _, c := range cm.collectors {
|
||||
c.Close()
|
||||
}
|
||||
close(cm.done)
|
||||
cclog.ComponentDebug("CollectorManager", "DONE")
|
||||
}
|
||||
|
||||
@ -154,6 +155,8 @@ func (cm *collectorManager) AddOutput(output chan lp.CCMetric) {
|
||||
func (cm *collectorManager) Close() {
|
||||
cclog.ComponentDebug("CollectorManager", "CLOSE")
|
||||
cm.done <- true
|
||||
// wait for close of channel cm.done
|
||||
<-cm.done
|
||||
}
|
||||
|
||||
// New creates a new initialized metric collector manager
|
||||
|
@ -101,6 +101,7 @@ func (r *metricRouter) StartTimer() {
|
||||
for {
|
||||
select {
|
||||
case <-r.timerdone:
|
||||
close(r.timerdone)
|
||||
cclog.ComponentDebug("MetricRouter", "TIMER DONE")
|
||||
return
|
||||
case t := <-m:
|
||||
@ -195,6 +196,7 @@ func (r *metricRouter) Start() {
|
||||
|
||||
// Router manager is done
|
||||
done := func() {
|
||||
close(r.done)
|
||||
cclog.ComponentDebug("MetricRouter", "DONE")
|
||||
}
|
||||
|
||||
@ -257,9 +259,13 @@ func (r *metricRouter) AddOutput(output chan lp.CCMetric) {
|
||||
func (r *metricRouter) Close() {
|
||||
cclog.ComponentDebug("MetricRouter", "CLOSE")
|
||||
r.done <- true
|
||||
// wait for close of channel r.done
|
||||
<-r.done
|
||||
if r.config.IntervalStamp {
|
||||
cclog.ComponentDebug("MetricRouter", "TIMER CLOSE")
|
||||
r.timerdone <- true
|
||||
// wait for close of channel r.timerdone
|
||||
<-r.timerdone
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -23,6 +23,7 @@ func (t *multiChanTicker) Init(duration time.Duration) {
|
||||
t.done = make(chan bool)
|
||||
go func() {
|
||||
done := func() {
|
||||
close(t.done)
|
||||
cclog.ComponentDebug("MultiChanTicker", "DONE")
|
||||
}
|
||||
for {
|
||||
@ -52,6 +53,8 @@ func (t *multiChanTicker) AddChannel(channel chan time.Time) {
|
||||
func (t *multiChanTicker) Close() {
|
||||
cclog.ComponentDebug("MultiChanTicker", "CLOSE")
|
||||
t.done <- true
|
||||
// wait for close of channel t.done
|
||||
<-t.done
|
||||
}
|
||||
|
||||
func NewTicker(duration time.Duration) MultiChanTicker {
|
||||
|
@ -81,6 +81,7 @@ func (sm *sinkManager) Start() {
|
||||
s.Close()
|
||||
}
|
||||
|
||||
close(sm.done)
|
||||
cclog.ComponentDebug("SinkManager", "DONE")
|
||||
}
|
||||
|
||||
@ -149,6 +150,8 @@ func (sm *sinkManager) AddOutput(rawConfig json.RawMessage) error {
|
||||
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
|
||||
|
Loading…
Reference in New Issue
Block a user