Fixes #1576 - flakey tests scheduler

- Adds event for when plugins are unsubscribed
- Listens for unsubscription event before asserting

core/scheduler_event/scheduler_event.go

@@ -24,6 +24,7 @@ import (
 )
 
 const (
+	PluginsUnsubscribed    = "Scheduler.PluginUnsubscribed"
 	TaskCreated            = "Scheduler.TaskCreated"
 	TaskDeleted            = "Scheduler.TaskDeleted"
 	TaskStarted            = "Scheduler.TaskStarted"
@@ -34,6 +35,15 @@ const (
 	MetricCollectionFailed = "Scheduler.MetricCollectionFailed"
 )
 
+type PluginsUnsubscribedEvent struct {
+	TaskID  string
+	Plugins []core.SubscribedPlugin
+}
+
+func (e PluginsUnsubscribedEvent) Namespace() string {
+	return PluginsUnsubscribed
+}
+
 type TaskStartedEvent struct {
 	TaskID string
 	Source string

scheduler/distributed_task_test.go

@@ -320,20 +320,19 @@ func TestDistributedSubscriptions(t *testing.T) {
 				schTask.RemoteManagers.Add(fmt.Sprintf("127.0.0.1:%v", port1), remoteMockManager)
 				terrs := s.StartTask(t.ID())
 				So(terrs, ShouldBeNil)
+				// wait for the task to stop and plugins to be unsubscribed (or timeout)
+				select {
+				case event := <-lse.UnsubscribedPluginEvents:
+					So(event.TaskID, ShouldEqual, t.ID())
+				case <-time.After(time.Duration(int64(count)*interval.Nanoseconds()) + 1*time.Second):
+				}
 
 				Convey("So all dependencies should have been subscribed to", func() {
 					So(localMockManager.SubscribeCallCount, ShouldBeGreaterThan, 0)
 					So(remoteMockManager.SubscribeCallCount, ShouldBeGreaterThan, 0)
 				})
 				Convey("Task should be ended after one interval", func() {
-					// wait for the end of the task (or timeout)
-					select {
-					case <-lse.Ended:
-					case <-time.After(time.Duration(int64(count)*interval.Nanoseconds()) + 1*time.Second):
-					}
-
 					So(t.State(), ShouldEqual, core.TaskEnded)
-
 					Convey("So all dependencies should have been usubscribed", func() {
 						So(remoteMockManager.UnsubscribeCallCount, ShouldEqual, remoteMockManager.SubscribeCallCount)
 						So(localMockManager.UnsubscribeCallCount, ShouldEqual, localMockManager.SubscribeCallCount)
@@ -370,9 +369,10 @@ func TestDistributedSubscriptions(t *testing.T) {
 					So(remoteMockManager.SubscribeCallCount, ShouldBeGreaterThan, 0)
 				})
 				Convey("Task should have been ended after reaching the end of window", func() {
-					// wait for the end of the task (or timeout)
+					// wait for the task to stop and plugins to be unsubscribed (or timeout)
 					select {
-					case <-lse.Ended:
+					case event := <-lse.UnsubscribedPluginEvents:
+						So(event.TaskID, ShouldEqual, t.ID())
 					case <-time.After(stop.Add(interval + 1*time.Second).Sub(start)):
 					}
 

scheduler/fixtures/fixtures.go

@@ -25,19 +25,23 @@ import "github.com/intelsdi-x/gomit"
 import "github.com/intelsdi-x/snap/core/scheduler_event"
 
 type listenToSchedulerEvent struct {
-	Ended chan struct{}
+	Ended                    chan struct{}
+	UnsubscribedPluginEvents chan *scheduler_event.PluginsUnsubscribedEvent
 }
 
 // NewListenToSchedulerEvent
 func NewListenToSchedulerEvent() *listenToSchedulerEvent {
 	return &listenToSchedulerEvent{
 		Ended: make(chan struct{}),
+		UnsubscribedPluginEvents: make(chan *scheduler_event.PluginsUnsubscribedEvent),
 	}
 }
 
 func (l *listenToSchedulerEvent) HandleGomitEvent(e gomit.Event) {
-	switch e.Body.(type) {
+	switch msg := e.Body.(type) {
 	case *scheduler_event.TaskEndedEvent:
 		l.Ended <- struct{}{}
+	case *scheduler_event.PluginsUnsubscribedEvent:
+		l.UnsubscribedPluginEvents <- msg
 	}
 }

scheduler/task.go

@@ -366,6 +366,11 @@ func (t *task) UnsubscribePlugins() []serror.SnapError {
 	depGroups := getWorkflowPlugins(t.workflow.processNodes, t.workflow.publishNodes, t.workflow.metrics)
 	var errs []serror.SnapError
 	for k := range depGroups {
+		event := scheduler_event.PluginsUnsubscribedEvent{
+			TaskID:  t.ID(),
+			Plugins: depGroups[k].subscribedPlugins,
+		}
+		defer t.eventEmitter.Emit(event)
 		mgr, err := t.RemoteManagers.Get(k)
 		if err != nil {
 			errs = append(errs, serror.New(err))