Fix publishers and processors subscription

Author: Mateusz Kleina

control/available_plugin.go

@@ -625,15 +625,15 @@ func (ap *availablePlugins) findLatestPool(pType, name string) (strategy.Pool, s
 	var latest strategy.Pool
 	for key, pool := range ap.table {
 		tnv := strings.Split(key, core.Separator)
-		if tnv[0] == pType && tnv[1] == name {
+		if tnv[0] == pType && tnv[1] == name && pool.Count() > 0 {
 			latest = pool
 			break
 		}
 	}
 	if latest != nil {
 		for key, pool := range ap.table {
 			tnv := strings.Split(key, core.Separator)
-			if tnv[0] == pType && tnv[1] == name && pool.Version() > latest.Version() {
+			if tnv[0] == pType && tnv[1] == name && pool.Version() > latest.Version() && pool.Count() > 0 {
 				latest = pool
 			}
 		}

control/subscription_group.go

@@ -433,8 +433,6 @@ func (s *subscriptionGroup) process(id string) (serrs []serror.SnapError) {
 
 	// notice that requested plugins contains only processors and publishers
 	for _, plugin := range s.requestedPlugins {
-		// add processors and publishers to collectors just gathered
-		plugins = append(plugins, plugin)
 		// add defaults to plugins (exposed in a plugins ConfigPolicy)
 		if lp, err := s.pluginManager.get(
 			fmt.Sprintf("%s"+core.Separator+"%s"+core.Separator+"%d",
@@ -445,6 +443,21 @@ func (s *subscriptionGroup) process(id string) (serrs []serror.SnapError) {
 				// set defaults to plugin config
 				plugin.Config().ApplyDefaults(policy.Defaults())
 			}
+
+			// update version info for subscribed processor or publisher
+			version := plugin.Version()
+			if version < 1 {
+				version = lp.Version()
+			}
+			s := subscribedPlugin{
+				name:     plugin.Name(),
+				typeName: plugin.TypeName(),
+				version:  version,
+				config:   plugin.Config(),
+			}
+
+			// add processors and publishers to collectors just gathered
+			plugins = append(plugins, s)
 		}
 	}
 	// calculates those plugins that need to be subscribed and unsubscribed to

scheduler/workflow_test.go

@@ -143,7 +143,7 @@ func TestCollectPublishWorkflow(t *testing.T) {
 			So(err, ShouldBeNil)
 			rp2, err := core.NewRequestedPlugin(snap_publisher_file_path, c.GetTempDir(), nil)
 			So(err, ShouldBeNil)
-			_, err = c.Load(rp2)
+			plugPublisher, err := c.Load(rp2)
 			So(err, ShouldBeNil)
 			rp3, err := core.NewRequestedPlugin(snap_processor_passthru_path, c.GetTempDir(), nil)
 			So(err, ShouldBeNil)
@@ -189,6 +189,12 @@ func TestCollectPublishWorkflow(t *testing.T) {
 					So(t.LastFailureMessage(), ShouldBeEmpty)
 					So(t.FailedCount(), ShouldEqual, 0)
 					So(t.HitCount(), ShouldBeGreaterThan, metricsToCollect)
+
+					// check if task fails after unloading publisher
+					c.Unload(plugPublisher)
+					<-el.done
+					So(t.LastFailureMessage(), ShouldNotBeEmpty)
+					So(t.FailedCount(), ShouldBeGreaterThan, 0)
 				})
 			})
 		})