Cleanup/fix program termination sequence.

Change-Id: I2bc58a2583fb079c9ef383cfc7a5e0fbe613f1cd

main.go

@@ -85,6 +85,7 @@ type prometheus struct {
 	unwrittenSamples chan *extraction.Result
 
 	ruleManager   rules.RuleManager
+	targetManager retrieval.TargetManager
 	notifications chan notification.NotificationReqs
 	storage       *metric.TieredStorage
 
@@ -174,11 +175,15 @@ func (p *prometheus) close() {
 		p.deletionTimer.Stop()
 	}
 
+	// Stop any currently active curation (deletion or compaction).
 	if len(p.stopBackgroundOperations) == 0 {
 		p.stopBackgroundOperations <- true
 	}
 
 	p.ruleManager.Stop()
+	p.targetManager.Stop()
+	close(p.unwrittenSamples)
+
 	p.storage.Close()
 
 	close(p.notifications)
@@ -294,6 +299,7 @@ func main() {
 		stopBackgroundOperations: make(chan bool, 1),
 
 		ruleManager:   ruleManager,
+		targetManager: targetManager,
 		notifications: notifications,
 		storage:       ts,
 	}

retrieval/targetmanager.go

@@ -29,6 +29,7 @@ type TargetManager interface {
 	ReplaceTargets(job config.JobConfig, newTargets []Target)
 	Remove(t Target)
 	AddTargetsFromConfig(config config.Config)
+	Stop()
 	Pools() map[string]*TargetPool
 }
 
@@ -115,6 +116,13 @@ func (m *targetManager) AddTargetsFromConfig(config config.Config) {
 	}
 }
 
+func (m *targetManager) Stop() {
+	glog.Info("Target manager exiting...")
+	for _, p := range m.poolsByJob {
+		p.Stop()
+	}
+}
+
 // XXX: Not really thread-safe. Only used in /status page for now.
 func (m *targetManager) Pools() map[string]*TargetPool {
 	return m.poolsByJob

retrieval/targetpool.go

@@ -47,6 +47,7 @@ func NewTargetPool(m TargetManager, p TargetProvider) *TargetPool {
 		addTargetQueue:      make(chan Target, targetAddQueueSize),
 		replaceTargetsQueue: make(chan targets, targetReplaceQueueSize),
 		targetProvider:      p,
+		done:                make(chan bool),
 	}
 }
 

rules/manager.go

@@ -91,8 +91,8 @@ func (m *ruleManager) Run() {
 			m.runIteration(m.results)
 			evalDurations.Add(map[string]string{intervalKey: m.interval.String()}, float64(time.Since(start)/time.Millisecond))
 		case <-m.done:
-			glog.Info("RuleManager exiting...")
+			glog.Info("Rule manager exiting...")
-			break
+			return
 		}
 	}
 }