Stop appender go funcs on close (#427)

Author: pavius

Diff for: pkg/appender/appender.go

@@ -130,6 +130,8 @@ type MetricsCache struct {
 
 	lastError           error
 	performanceReporter *performance.MetricReporter
+
+	stopChan      chan int
 }
 
 func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config.V3ioConfig,
@@ -146,6 +148,7 @@ func NewMetricsCache(container v3io.Container, logger logger.Logger, cfg *config
 	newCache.metricQueue = NewElasticQueue()
 	newCache.updatesComplete = make(chan int, 100)
 	newCache.newUpdates = make(chan int, 1000)
+	newCache.stopChan = make(chan int, 3)
 
 	newCache.NameLabelMap = map[string]bool{}
 	newCache.performanceReporter = performance.ReporterInstanceFromConfig(cfg)
@@ -285,6 +288,12 @@ func verifyTimeValid(t int64) error {
 	}
 	return nil
 }
+func (mc *MetricsCache) Close() {
+	//for 3 go funcs
+	mc.stopChan <- 0
+	mc.stopChan <- 0
+	mc.stopChan <- 0
+}
 
 func (mc *MetricsCache) WaitForCompletion(timeout time.Duration) (int, error) {
 	waitChan := make(chan int, 2)

Diff for: pkg/appender/ingest.go

@@ -53,6 +53,8 @@ func (mc *MetricsCache) metricFeed(index int) {
 
 		for {
 			select {
+			case _ = <-mc.stopChan:
+				return
 			case inFlight = <-mc.updatesComplete:
 				// Handle completion notifications from the update loop
 				length := mc.metricQueue.Length()
@@ -149,6 +151,8 @@ func (mc *MetricsCache) metricsUpdateLoop(index int) {
 		counter := 0
 		for {
 			select {
+			case _ = <-mc.stopChan:
+				return
 			case _ = <-mc.newUpdates:
 				// Handle new metric notifications (from metricFeed)
 				for mc.updatesInFlight < mc.cfg.Workers*2 { //&& newMetrics > 0{
@@ -347,24 +351,27 @@ func (mc *MetricsCache) nameUpdateRespLoop() {
 
 	go func() {
 		for {
-			resp := <-mc.nameUpdateChan
-			// Handle V3IO PutItem in names table
-
-			metric, ok := resp.Context.(*MetricState)
-			if ok {
-				metric.Lock()
-				if resp.Error != nil {
-					// Count errors
-					mc.performanceReporter.IncrementCounter("UpdateNameError", 1)
-
-					mc.logger.ErrorWith("Update-name process failed", "id", resp.ID, "name", metric.name)
-				} else {
-					mc.logger.DebugWith("Update-name process response", "id", resp.ID, "name", metric.name)
+			select {
+			case _ = <-mc.stopChan:
+				return
+			case resp := <-mc.nameUpdateChan:
+				// Handle V3IO PutItem in names table
+				metric, ok := resp.Context.(*MetricState)
+				if ok {
+					metric.Lock()
+					if resp.Error != nil {
+						// Count errors
+						mc.performanceReporter.IncrementCounter("UpdateNameError", 1)
+
+						mc.logger.ErrorWith("Update-name process failed", "id", resp.ID, "name", metric.name)
+					} else {
+						mc.logger.DebugWith("Update-name process response", "id", resp.ID, "name", metric.name)
+					}
+					metric.Unlock()
 				}
-				metric.Unlock()
-			}
 
-			resp.Release()
+				resp.Release()
+			}
 		}
 	}()
 }

Diff for: pkg/tsdb/v3iotsdb.go

@@ -343,6 +343,10 @@ func (a v3ioAppender) WaitForCompletion(timeout time.Duration) (int, error) {
 	return a.metricsCache.WaitForCompletion(timeout)
 }
 
+func (a v3ioAppender) Close() {
+	a.metricsCache.Close()
+}
+
 // In V3IO, all operations are committed (no client cache)
 func (a v3ioAppender) Commit() error   { return nil }
 func (a v3ioAppender) Rollback() error { return nil }
@@ -354,4 +358,5 @@ type Appender interface {
 	WaitForCompletion(timeout time.Duration) (int, error)
 	Commit() error
 	Rollback() error
+	Close()
 }

Diff for: pkg/tsdbctl/add.go

@@ -181,6 +181,7 @@ func (ac *addCommandeer) add() error {
 		return errors.Wrap(err, "Operation timed out.")
 	}
 
+	appender.Close()
 	ac.rootCommandeer.logger.Info("Done!")
 	return nil
 }