Merge pull request #49 from nats-io/ms_report

add a report command for message sets

go.mod

@@ -9,5 +9,6 @@ require (
 	github.com/dustin/go-humanize v1.0.0
 	github.com/nats-io/nats-server/v2 v2.1.3-0.20200108135121-19dc3ebadc67
 	github.com/nats-io/nats.go v1.9.1
+	github.com/xlab/tablewriter v0.0.0-20160610135559-80b567a11ad5
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6
 )

go.sum

@@ -54,6 +54,8 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.1/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/xlab/tablewriter v0.0.0-20160610135559-80b567a11ad5 h1:gmD7q6cCJfBbcuobWQe/KzLsd9Cd3amS1Mq5f3uU1qo=
+github.com/xlab/tablewriter v0.0.0-20160610135559-80b567a11ad5/go.mod h1:fVwOndYN3s5IaGlMucfgxwMhqwcaJtlGejBU6zX6Yxw=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190530122614-20be4c3c3ed5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc=

jsm/jetstreammgmt.go

@@ -115,15 +115,6 @@ func (j *JetStreamMgmt) MessageSetGetItem(setName string, id int64) (msg *api.St
 
 // MessageSetInfo retrieves configuration and state of a message set
 func (j *JetStreamMgmt) MessageSetInfo(setName string) (info *api.MsgSetInfo, err error) {
-	known, err := j.IsMessageSetKnown(setName)
-	if err != nil {
-		return nil, err
-	}
-
-	if !known {
-		return nil, fmt.Errorf("message set %s is not known", setName)
-	}
-
 	info = &api.MsgSetInfo{}
 	err = j.requestAndUnMarshal(api.JetStreamMsgSetInfo, []byte(setName), info)
 

jsm/main.go

@@ -34,7 +34,6 @@ func main() {
 	jsm := kingpin.New("jsm", "JetStream Management Utility")
 	jsm.Author("NATS Authors <[email protected]>")
 	jsm.Version(version)
-
 	jsm.HelpFlag.Short('h')
 
 	jsm.Flag("server", "NATS servers").Short('s').Default("localhost:4222").Envar("SERVERS").StringVar(&servers)

jsm/ms_command.go

@@ -14,15 +14,18 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/AlecAivazis/survey/v2"
 	"github.com/dustin/go-humanize"
 	api "github.com/nats-io/nats-server/v2/server"
+	"github.com/xlab/tablewriter"
 	"gopkg.in/alecthomas/kingpin.v2"
 )
 
@@ -33,15 +36,19 @@ type msCmd struct {
 	msgID            int64
 	retentionPolicyS string
 
-	destination   string
-	subjects      []string
-	ack           bool
-	storage       string
-	maxMsgLimit   int64
-	maxBytesLimit int64
-	maxAgeLimit   string
-	maxMsgSize    int32
-	rPolicy       api.RetentionPolicy
+	destination    string
+	subjects       []string
+	ack            bool
+	storage        string
+	maxMsgLimit    int64
+	maxBytesLimit  int64
+	maxAgeLimit    string
+	maxMsgSize     int32
+	rPolicy        api.RetentionPolicy
+	reportSortObs  bool
+	reportSortMsgs bool
+	reportSortName bool
+	reportRaw      bool
 }
 
 func configureMSCommand(app *kingpin.Application) {
@@ -90,6 +97,75 @@ func configureMSCommand(app *kingpin.Application) {
 	msGet.Arg("name", "Message Set name").StringVar(&c.set)
 	msGet.Arg("id", "Message ID to retrieve").Int64Var(&c.msgID)
 	msGet.Flag("json", "Produce JSON output").Short('j').BoolVar(&c.json)
+
+	msReport := ms.Command("report", "Reports on Message Set statistics").Action(c.reportAction)
+	msReport.Flag("json", "Produce JSON output").Short('j').BoolVar(&c.json)
+	msReport.Flag("observables", "Sort by number of Observables").Short('o').BoolVar(&c.reportSortObs)
+	msReport.Flag("messages", "Sort by number of Messages").Short('m').BoolVar(&c.reportSortMsgs)
+	msReport.Flag("name", "Sort by Message Set name").Short('n').BoolVar(&c.reportSortName)
+	msReport.Flag("raw", "Show unformatted numbers").Short('r').BoolVar(&c.reportRaw)
+}
+
+func (c *msCmd) reportAction(pc *kingpin.ParseContext) error {
+	jsm, err := NewJSM(timeout, servers, natsOpts())
+	kingpin.FatalIfError(err, "setup failed")
+
+	sets, err := jsm.MessageSets()
+	kingpin.FatalIfError(err, "could not retrieve known sets")
+
+	if len(sets) == 0 {
+		return nil
+	}
+
+	type stat struct {
+		Name        string
+		Observables int
+		Msgs        int64
+		Bytes       uint64
+	}
+
+	if !c.json {
+		fmt.Printf("Obtaining Message Set stats for %d sets\n\n", len(sets))
+	}
+
+	stats := make([]stat, len(sets))
+	for i, s := range sets {
+		info, err := jsm.MessageSetInfo(s)
+		kingpin.FatalIfError(err, "could not get set info for %s", s)
+		stats[i] = stat{info.Config.Name, info.Stats.Observables, int64(info.Stats.Msgs), info.Stats.Bytes}
+	}
+
+	if c.json {
+		j, err := json.MarshalIndent(stats, "", "  ")
+		kingpin.FatalIfError(err, "could not JSON marshal stats")
+		fmt.Println(string(j))
+		return nil
+	}
+
+	if c.reportSortObs {
+		sort.Slice(stats, func(i, j int) bool { return stats[i].Observables < stats[j].Observables })
+	} else if c.reportSortMsgs {
+		sort.Slice(stats, func(i, j int) bool { return stats[i].Msgs < stats[j].Msgs })
+	} else if c.reportSortName {
+		sort.Slice(stats, func(i, j int) bool { return stats[i].Name < stats[j].Name })
+	} else {
+		sort.Slice(stats, func(i, j int) bool { return stats[i].Bytes < stats[j].Bytes })
+	}
+
+	table := tablewriter.CreateTable()
+	table.AddHeaders("Message Set", "Observables", "Messages", "Bytes")
+
+	for _, s := range stats {
+		if c.reportRaw {
+			table.AddRow(s.Name, s.Observables, s.Msgs, s.Bytes)
+		} else {
+			table.AddRow(s.Name, s.Observables, humanize.Comma(s.Msgs), humanize.IBytes(s.Bytes))
+		}
+	}
+
+	fmt.Println(table.Render())
+
+	return nil
 }
 
 func (c *msCmd) cpAction(pc *kingpin.ParseContext) error {