Merge pull request #212 from takenaga/print_latency

Print latency in nanosec and millisec

app/app.go

@@ -202,6 +202,7 @@ func (a *App) InitGlobalFlags() {
 
 	a.RootCmd.PersistentFlags().BoolVarP(&a.Config.GlobalFlags.UseTunnelServer, "use-tunnel-server", "", false, "use tunnel server to dial targets")
 	a.RootCmd.PersistentFlags().StringVarP(&a.Config.GlobalFlags.AuthScheme, "auth-scheme", "", "", "authentication scheme to use for the target's username/password")
+	a.RootCmd.PersistentFlags().BoolVarP(&a.Config.GlobalFlags.CalculateLatency, "calculate-latency", "", false, "calculate the delta between each message timestamp and the receive timestamp. JSON format only")
 
 	a.RootCmd.PersistentFlags().VisitAll(func(flag *pflag.Flag) {
 		a.Config.FileConfig.BindPFlag(flag.Name, flag)
@@ -296,10 +297,11 @@ func (a *App) PrintMsg(address string, msgName string, msg proto.Message) error
 		}
 	}
 	mo := formatters.MarshalOptions{
-		Multiline:  true,
-		Indent:     "  ",
-		Format:     a.Config.Format,
-		ValuesOnly: a.Config.GetValuesOnly,
+		Multiline:        true,
+		Indent:           "  ",
+		Format:           a.Config.Format,
+		ValuesOnly:       a.Config.GetValuesOnly,
+		CalculateLatency: a.Config.CalculateLatency,
 	}
 	b, err := mo.Marshal(msg, map[string]string{"source": address})
 	if err != nil {

app/subscribe.go

@@ -224,9 +224,10 @@ func (a *App) handlePolledSubscriptions() {
 		waitChan := make(chan struct{}, 1)
 		waitChan <- struct{}{}
 		mo := &formatters.MarshalOptions{
-			Multiline: true,
-			Indent:    "  ",
-			Format:    a.Config.Format,
+			Multiline:        true,
+			Indent:           "  ",
+			Format:           a.Config.Format,
+			CalculateLatency: a.Config.GlobalFlags.CalculateLatency,
 		}
 
 		for {

config/config.go

@@ -115,6 +115,7 @@ type GlobalFlags struct {
 	Token            string        `mapstructure:"token,omitempty" json:"token,omitempty" yaml:"token,omitempty"`
 	UseTunnelServer  bool          `mapstructure:"use-tunnel-server,omitempty" json:"use-tunnel-server,omitempty" yaml:"use-tunnel-server,omitempty"`
 	AuthScheme       string        `mapstructure:"auth-scheme,omitempty" json:"auth-scheme,omitempty" yaml:"auth-scheme,omitempty"`
+	CalculateLatency bool          `mapstructure:"calculate-latency,omitempty" json:"calculate-latency,omitempty" yaml:"calculate-latency,omitempty"`
 }
 
 type LocalFlags struct {

config/outputs.go

@@ -21,9 +21,10 @@ func (c *Config) GetOutputs() (map[string]map[string]interface{}, error) {
 	outDef := c.FileConfig.GetStringMap("outputs")
 	if len(outDef) == 0 && !c.FileConfig.GetBool("subscribe-quiet") {
 		stdoutConfig := map[string]interface{}{
-			"type":      "file",
-			"file-type": "stdout",
-			"format":    c.FileConfig.GetString("format"),
+			"type":              "file",
+			"file-type":         "stdout",
+			"format":            c.FileConfig.GetString("format"),
+			"calculate-latency": c.FileConfig.GetBool("calculate-latency"),
 		}
 		outDef["default-stdout"] = stdoutConfig
 	}

docs/global_flags.md

@@ -304,3 +304,13 @@ Applied only in the case of a secure gRPC connection.
 ### username
 
 The username flag `[-u | --username]` is used to specify the target username as part of the user credentials.
+
+### calculate-latency
+
+The `calculate-latency` flag augments subscribe et get responses by calculating the delta between the message timestamp and the receive timestamp.
+The resulting message will include 4 extra fields:
+
+* `recv-timestamp`:The receive timestamp in nanoseconds.
+* `recv-time`: The receive time in ISO 8601 date and time representation, extended to include fractional seconds and a time zone offset..
+* `latency-nano`: The difference between the message timestamp and the receive time in nanoseconds.
+* `latency-milli`: The difference between the message timestamp and the receive time in milliseconds.

formatters/formats.go

@@ -22,11 +22,12 @@ import (
 )
 
 type MarshalOptions struct {
-	Multiline  bool
-	Indent     string
-	Format     string
-	OverrideTS bool
-	ValuesOnly bool
+	Multiline        bool
+	Indent           string
+	Format           string
+	OverrideTS       bool
+	ValuesOnly       bool
+	CalculateLatency bool
 }
 
 // Marshal //

formatters/json.go

@@ -93,6 +93,13 @@ func (o *MarshalOptions) formatSubscribeResponse(m *gnmi.SubscribeResponse, meta
 		}
 		t := time.Unix(0, m.Update.Timestamp)
 		msg.Time = &t
+		if o.CalculateLatency {
+			msg.RecvTimestamp = time.Now().UnixNano()
+			rt := time.Unix(0, msg.RecvTimestamp)
+			msg.RecvTime = &rt
+			msg.LatencyNano = msg.RecvTimestamp - msg.Timestamp
+			msg.LatencyMilli = msg.LatencyNano / 1000 / 1000
+		}
 		if meta == nil {
 			meta = make(map[string]string)
 		}
@@ -206,6 +213,13 @@ func (o *MarshalOptions) formatGetResponse(m *gnmi.GetResponse, meta map[string]
 		msg.Timestamp = notif.Timestamp
 		t := time.Unix(0, notif.Timestamp)
 		msg.Time = &t
+		if o.CalculateLatency && !o.ValuesOnly {
+			msg.RecvTimestamp = time.Now().UnixNano()
+			rt := time.Unix(0, msg.RecvTimestamp)
+			msg.RecvTime = &rt
+			msg.LatencyNano = msg.RecvTimestamp - msg.Timestamp
+			msg.LatencyMilli = msg.LatencyNano / 1000 / 1000
+		}
 		if meta == nil {
 			meta = make(map[string]string)
 		}
@@ -235,6 +249,7 @@ func (o *MarshalOptions) formatGetResponse(m *gnmi.GetResponse, meta map[string]
 		}
 		notifications = append(notifications, msg)
 	}
+
 	if o.ValuesOnly {
 		result := make([]interface{}, 0, len(notifications))
 		for _, n := range notifications {

formatters/msg.go

@@ -22,6 +22,10 @@ type NotificationRspMsg struct {
 	SubscriptionName string                 `json:"subscription-name,omitempty"`
 	Timestamp        int64                  `json:"timestamp,omitempty"`
 	Time             *time.Time             `json:"time,omitempty"`
+	RecvTimestamp    int64                  `json:"recv-timestamp,omitempty"`
+	RecvTime         *time.Time             `json:"recv-time,omitempty"`
+	LatencyNano      int64                  `json:"latency-nano,omitempty"`
+	LatencyMilli     int64                  `json:"latency-milli,omitempty"`
 	Prefix           string                 `json:"prefix,omitempty"`
 	Target           string                 `json:"target,omitempty"`
 	Updates          []update               `json:"updates,omitempty"`

outputs/file/file_output.go

@@ -74,6 +74,7 @@ type Config struct {
 	ConcurrencyLimit   int      `mapstructure:"concurrency-limit,omitempty"`
 	EnableMetrics      bool     `mapstructure:"enable-metrics,omitempty"`
 	Debug              bool     `mapstructure:"debug,omitempty"`
+	CalculateLatency   bool     `mapstructure:"calculate-latency,omitempty"`
 }
 
 func (f *File) String() string {
@@ -178,10 +179,11 @@ func (f *File) Init(ctx context.Context, name string, cfg map[string]interface{}
 	f.sem = semaphore.NewWeighted(int64(f.Cfg.ConcurrencyLimit))
 
 	f.mo = &formatters.MarshalOptions{
-		Multiline:  f.Cfg.Multiline,
-		Indent:     f.Cfg.Indent,
-		Format:     f.Cfg.Format,
-		OverrideTS: f.Cfg.OverrideTimestamps,
+		Multiline:        f.Cfg.Multiline,
+		Indent:           f.Cfg.Indent,
+		Format:           f.Cfg.Format,
+		OverrideTS:       f.Cfg.OverrideTimestamps,
+		CalculateLatency: f.Cfg.CalculateLatency,
 	}
 	if f.Cfg.TargetTemplate == "" {
 		f.targetTpl = outputs.DefaultTargetTemplate