Pulsectl when task is watched prints sorted metrics

cmd/pulsectl/task.go

@@ -26,6 +26,7 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"sort"
 	"strings"
 	"syscall"
 	"text/tabwriter"
@@ -332,6 +333,7 @@ func watchTask(ctx *cli.Context) {
 		case e := <-r.EventChan:
 			switch e.EventType {
 			case "metric-event":
+				sort.Sort(e.Event)
 				for _, event := range e.Event {
 					printFields(w, false, 0,
 						event.Namespace,

mgmt/rest/rbody/task.go

@@ -238,9 +238,9 @@ func (s *ScheduledTaskWatchingEnded) ResponseBodyType() string {
 
 type StreamedTaskEvent struct {
 	// Used to describe the event
-	EventType string           `json:"type"`
-	Message   string           `json:"message"`
-	Event     []StreamedMetric `json:"event,omitempty"`
+	EventType string          `json:"type"`
+	Message   string          `json:"message"`
+	Event     StreamedMetrics `json:"event,omitempty"`
 }
 
 func (s *StreamedTaskEvent) ToJSON() string {
@@ -254,3 +254,17 @@ type StreamedMetric struct {
 	Source    string      `json:"source"`
 	Timestamp time.Time   `json:"timestamp"`
 }
+
+type StreamedMetrics []StreamedMetric
+
+func (s StreamedMetrics) Len() int {
+	return len(s)
+}
+
+func (s StreamedMetrics) Less(i, j int) bool {
+	return fmt.Sprintf("%s:%s", s[i].Source, s[i].Namespace) < fmt.Sprintf("%s:%s", s[j].Source, s[j].Namespace)
+}
+
+func (s StreamedMetrics) Swap(i, j int) {
+	s[i], s[j] = s[j], s[i]
+}