Add metrics source info to the published file

If m.Source() gives empty string, writes "unknown"

mgmt/rest/client/client_func_test.go

@@ -180,7 +180,7 @@ func TestPulseClient(t *testing.T) {
 		Convey("invalid task (missing publisher)", func() {
 			tf := c.CreateTask(sch, wf, "baron", false)
 			So(tf.Err, ShouldNotBeNil)
-			So(tf.Err.Error(), ShouldContainSubstring, "Plugin not found: type(publisher) name(file) version(1)")
+			So(tf.Err.Error(), ShouldContainSubstring, "Plugin not found: type(publisher) name(file)")
 		})
 		Convey("plugin already loaded", func() {
 			p1 := c.LoadPlugin(DUMMY_PLUGIN_PATH1)

mgmt/rest/wmap_sample/1.json

@@ -12,7 +12,6 @@
         "publish": [
             {
                 "plugin_name": "file",
-                "plugin_version": 1,
                 "config": {
                     "file": "/tmp/rest.test"
                 }

mgmt/rest/wmap_sample/bad.json

@@ -8,7 +8,6 @@
         "publish": [
             {
                 "plugin_name": "file",
-                "plugin_version": 1,
                 "config": {
                     "file": "/tmp/rest.test"
                 }

plugin/publisher/pulse-publisher-file/file/file.go

@@ -77,7 +77,11 @@ func (f *filePublisher) Publish(contentType string, content []byte, config map[s
 	}
 	w := bufio.NewWriter(file)
 	for _, m := range metrics {
-		w.WriteString(fmt.Sprintf("%v|%v|%v\n", time.Now().Local(), m.Namespace(), m.Data()))
+		source := m.Source()
+		if source == "" {
+			source = "unknown"
+		}
+		w.WriteString(fmt.Sprintf("%v|%v|%v|%v\n", time.Now().Local(), m.Namespace(), m.Data(), source))
 	}
 	w.Flush()