cmd/observe: add a new flag to allow specifying different time formats for timestamps

cmd/observe/observe.go

@@ -23,6 +23,7 @@ import (
 	"os"
 	"os/signal"
 	"strings"
+	"time"
 
 	pb "github.com/cilium/cilium/api/v1/flow"
 	"github.com/cilium/cilium/api/v1/observer"
@@ -56,6 +57,8 @@ var (
 		dictOutput    bool
 		output        string
 
+		timeFormat string
+
 		enableIPTranslation bool
 		nodeName            bool
 		numeric             bool
@@ -81,6 +84,25 @@ func eventTypes() (l []string) {
 	return
 }
 
+func timeFormatNameToLayout(name string) string {
+	switch strings.ToLower(name) {
+	case "rfc3339":
+		return time.RFC3339
+	case "rfc3339milli":
+		return hubtime.RFC3339Milli
+	case "rfc3339micro":
+		return hubtime.RFC3339Micro
+	case "rfc3339nano":
+		return time.RFC3339Nano
+	case "rfc1123z":
+		return time.RFC1123Z
+	case "stampmilli":
+		fallthrough
+	default:
+		return time.StampMilli
+	}
+}
+
 // New observer command.
 func New(vp *viper.Viper) *cobra.Command {
 	return newObserveCmd(vp, newObserveFilter())
@@ -256,6 +278,7 @@ more.`,
 		"Show all flows terminating at an endpoint with the given security identity"))
 	observeCmd.Flags().AddFlagSet(filterFlags)
 
+	// formatting flags
 	formattingFlags := pflag.NewFlagSet("Formatting", pflag.ContinueOnError)
 	formattingFlags.BoolVarP(
 		&formattingOpts.jsonOutput, "json", "j", false, "Deprecated. Use '--output json' instead.",
@@ -290,6 +313,17 @@ more.`,
 		"Translate IP addresses to logical names such as pod name, FQDN, ...",
 	)
 	formattingFlags.BoolVarP(&formattingOpts.nodeName, "print-node-name", "", false, "Print node name in output")
+	formattingFlags.StringVar(
+		&formattingOpts.timeFormat, "time-format", "StampMilli",
+		fmt.Sprintf(`Specify the time format for printing. This option does not apply to the json and jsonpb output type. One of:
+  StampMilli:   %s
+  RFC3339:      %s
+  RFC3339Milli: %s
+  RFC3339Micro: %s
+  RFC3339Nano:  %s
+  RFC1123Z:     %s
+ `, time.StampMilli, time.RFC3339, hubtime.RFC3339Milli, hubtime.RFC3339Micro, time.RFC3339Nano, time.RFC1123Z),
+	)
 	observeCmd.Flags().AddFlagSet(formattingFlags)
 
 	// other flags
@@ -346,6 +380,16 @@ more.`,
 			"table",
 		}, cobra.ShellCompDirectiveDefault
 	})
+	observeCmd.RegisterFlagCompletionFunc("time-format", func(_ *cobra.Command, _ []string, _ string) ([]string, cobra.ShellCompDirective) {
+		return []string{
+			"StampMilli",
+			"RFC3339",
+			"RFC3339Milli",
+			"RFC3339Micro",
+			"RFC3339Nano",
+			"RFC1123Z",
+		}, cobra.ShellCompDirectiveDefault
+	})
 
 	// default value for when the flag is on the command line without any options
 	observeCmd.Flags().Lookup("not").NoOptDefVal = "true"
@@ -361,7 +405,9 @@ func handleArgs(ofilter *observeFilter, debug bool) (err error) {
 	}
 
 	// initialize the printer with any options that were passed in
-	var opts []hubprinter.Option
+	var opts = []hubprinter.Option{
+		hubprinter.WithTimeFormat(timeFormatNameToLayout(formattingOpts.timeFormat)),
+	}
 
 	if formattingOpts.output == "" { // support deprecated output flags if provided
 		if formattingOpts.jsonOutput {

pkg/printer/options.go

@@ -43,6 +43,7 @@ type Options struct {
 	enableDebug         bool
 	enableIPTranslation bool
 	nodeName            bool
+	timeFormat          string
 }
 
 // Option ...
@@ -117,3 +118,13 @@ func WithNodeName() Option {
 		opts.nodeName = true
 	}
 }
+
+// WithTimeFormat specifies the time format layout to use when printing out
+// timestamps. This option has no effect if JSON or JSONPB option is used.
+// The layout must be a time format layout as specified in the standard
+// library's time package.
+func WithTimeFormat(layout string) Option {
+	return func(opts *Options) {
+		opts.timeFormat = layout
+	}
+}

pkg/printer/printer.go

@@ -59,9 +59,10 @@ func (ew *errWriter) write(a ...interface{}) {
 func New(fopts ...Option) *Printer {
 	// default options
 	opts := Options{
-		output: TabOutput,
-		w:      os.Stdout,
-		werr:   os.Stderr,
+		output:     TabOutput,
+		w:          os.Stdout,
+		werr:       os.Stderr,
+		timeFormat: time.StampMilli,
 	}
 
 	// apply optional parameters
@@ -161,12 +162,11 @@ func (p *Printer) GetHostNames(f *pb.Flow) (string, string) {
 	return src, dst
 }
 
-func fmtTimestamp(ts *timestamppb.Timestamp) string {
+func fmtTimestamp(layout string, ts *timestamppb.Timestamp) string {
 	if !ts.IsValid() {
 		return "N/A"
 	}
-	// TODO: support more date formats through options to `hubble observe`
-	return ts.AsTime().Format(time.StampMilli)
+	return ts.AsTime().Format(layout)
 }
 
 // GetFlowType returns the type of a flow as a string.
@@ -226,7 +226,7 @@ func (p *Printer) WriteProtoFlow(res *observerpb.GetFlowsResponse) error {
 				"SUMMARY", newline,
 			)
 		}
-		ew.write(fmtTimestamp(f.GetTime()), tab)
+		ew.write(fmtTimestamp(p.opts.timeFormat, f.GetTime()), tab)
 		if p.opts.nodeName {
 			ew.write(f.GetNodeName(), tab)
 		}
@@ -251,7 +251,7 @@ func (p *Printer) WriteProtoFlow(res *observerpb.GetFlowsResponse) error {
 
 		// this is a little crude, but will do for now. should probably find the
 		// longest header and auto-format the keys
-		ew.write("  TIMESTAMP: ", fmtTimestamp(f.GetTime()), newline)
+		ew.write("  TIMESTAMP: ", fmtTimestamp(p.opts.timeFormat, f.GetTime()), newline)
 		if p.opts.nodeName {
 			ew.write("       NODE: ", f.GetNodeName(), newline)
 		}
@@ -274,7 +274,7 @@ func (p *Printer) WriteProtoFlow(res *observerpb.GetFlowsResponse) error {
 		}
 		_, err := fmt.Fprintf(p.opts.w,
 			"%s%s: %s -> %s %s %s (%s)\n",
-			fmtTimestamp(f.GetTime()),
+			fmtTimestamp(p.opts.timeFormat, f.GetTime()),
 			node,
 			src,
 			dst,
@@ -356,7 +356,7 @@ func (p *Printer) WriteProtoNodeStatusEvent(r *observerpb.GetFlowsResponse) erro
 			message = strconv.Quote(m)
 		}
 		_, err := fmt.Fprint(p.opts.werr,
-			"  TIMESTAMP: ", fmtTimestamp(r.GetTime()), newline,
+			"  TIMESTAMP: ", fmtTimestamp(p.opts.timeFormat, r.GetTime()), newline,
 			"      STATE: ", s.StateChange.String(), newline,
 			"      NODES: ", nodeNames, newline,
 			"    MESSAGE: ", message, newline,
@@ -367,7 +367,7 @@ func (p *Printer) WriteProtoNodeStatusEvent(r *observerpb.GetFlowsResponse) erro
 	case TabOutput, CompactOutput:
 		numNodes := len(s.NodeNames)
 		nodeNames := joinWithCutOff(s.NodeNames, ", ", nodeNamesCutOff)
-		prefix := fmt.Sprintf("%s [%s]", fmtTimestamp(r.GetTime()), r.GetNodeName())
+		prefix := fmt.Sprintf("%s [%s]", fmtTimestamp(p.opts.timeFormat, r.GetTime()), r.GetNodeName())
 		msg := fmt.Sprintf("%s: unknown node status event: %+v", prefix, s)
 		switch s.StateChange {
 		case relaypb.NodeState_NODE_CONNECTED:
@@ -390,15 +390,15 @@ func formatServiceAddr(a *pb.ServiceUpsertNotificationAddr) string {
 	return net.JoinHostPort(a.Ip, strconv.Itoa(int(a.Port)))
 }
 
-func getAgentEventDetails(e *pb.AgentEvent) string {
+func getAgentEventDetails(e *pb.AgentEvent, timeLayout string) string {
 	switch e.GetType() {
 	case pb.AgentEventType_AGENT_EVENT_UNKNOWN:
 		if u := e.GetUnknown(); u != nil {
 			return fmt.Sprintf("type: %s, notification: %s", u.Type, u.Notification)
 		}
 	case pb.AgentEventType_AGENT_STARTED:
 		if a := e.GetAgentStart(); a != nil {
-			return fmt.Sprintf("start time: %s", fmtTimestamp(a.Time))
+			return fmt.Sprintf("start time: %s", fmtTimestamp(timeLayout, a.Time))
 		}
 	case pb.AgentEventType_POLICY_UPDATED, pb.AgentEventType_POLICY_DELETED:
 		if p := e.GetPolicyUpdate(); p != nil {
@@ -497,13 +497,13 @@ func (p *Printer) WriteProtoAgentEvent(r *observerpb.GetFlowsResponse) error {
 			ew.write(dictSeparator)
 		}
 
-		ew.write("  TIMESTAMP: ", fmtTimestamp(r.GetTime()), newline)
+		ew.write("  TIMESTAMP: ", fmtTimestamp(p.opts.timeFormat, r.GetTime()), newline)
 		if p.opts.nodeName {
 			ew.write("       NODE: ", r.GetNodeName(), newline)
 		}
 		ew.write(
 			"       TYPE: ", e.GetType(), newline,
-			"    DETAILS: ", getAgentEventDetails(e), newline,
+			"    DETAILS: ", getAgentEventDetails(e, p.opts.timeFormat), newline,
 		)
 		if ew.err != nil {
 			return fmt.Errorf("failed to write out agent event: %v", ew.err)
@@ -520,13 +520,13 @@ func (p *Printer) WriteProtoAgentEvent(r *observerpb.GetFlowsResponse) error {
 				"DETAILS", newline,
 			)
 		}
-		ew.write(fmtTimestamp(r.GetTime()), tab)
+		ew.write(fmtTimestamp(p.opts.timeFormat, r.GetTime()), tab)
 		if p.opts.nodeName {
 			ew.write(r.GetNodeName(), tab)
 		}
 		ew.write(
 			e.GetType(), tab,
-			getAgentEventDetails(e), newline,
+			getAgentEventDetails(e, p.opts.timeFormat), newline,
 		)
 		if ew.err != nil {
 			return fmt.Errorf("failed to write out agent event: %v", ew.err)
@@ -539,10 +539,10 @@ func (p *Printer) WriteProtoAgentEvent(r *observerpb.GetFlowsResponse) error {
 		}
 		_, err := fmt.Fprintf(p.opts.w,
 			"%s%s: %s (%s)\n",
-			fmtTimestamp(r.GetTime()),
+			fmtTimestamp(p.opts.timeFormat, r.GetTime()),
 			node,
 			e.GetType(),
-			getAgentEventDetails(e))
+			getAgentEventDetails(e, p.opts.timeFormat))
 		if err != nil {
 			return fmt.Errorf("failed to write out agent event: %v", err)
 		}
@@ -599,7 +599,7 @@ func (p *Printer) WriteProtoDebugEvent(r *observerpb.GetFlowsResponse) error {
 			ew.write(dictSeparator)
 		}
 
-		ew.write("  TIMESTAMP: ", fmtTimestamp(r.GetTime()), newline)
+		ew.write("  TIMESTAMP: ", fmtTimestamp(p.opts.timeFormat, r.GetTime()), newline)
 		if p.opts.nodeName {
 			ew.write("       NODE: ", r.GetNodeName(), newline)
 		}
@@ -628,7 +628,7 @@ func (p *Printer) WriteProtoDebugEvent(r *observerpb.GetFlowsResponse) error {
 				"MESSAGE", newline,
 			)
 		}
-		ew.write(fmtTimestamp(r.GetTime()), tab)
+		ew.write(fmtTimestamp(p.opts.timeFormat, r.GetTime()), tab)
 		if p.opts.nodeName {
 			ew.write(r.GetNodeName(), tab)
 		}
@@ -648,7 +648,7 @@ func (p *Printer) WriteProtoDebugEvent(r *observerpb.GetFlowsResponse) error {
 		}
 		_, err := fmt.Fprintf(p.opts.w,
 			"%s%s: %s %s MARK: %s CPU: %s (%s)\n",
-			fmtTimestamp(r.GetTime()),
+			fmtTimestamp(p.opts.timeFormat, r.GetTime()),
 			node,
 			fmtEndpointShort(e.GetSource()),
 			e.GetType(),

pkg/printer/printer_test.go

@@ -428,7 +428,8 @@ func Test_getHostNames(t *testing.T) {
 
 func Test_fmtTimestamp(t *testing.T) {
 	type args struct {
-		t *timestamppb.Timestamp
+		layout string
+		t      *timestamppb.Timestamp
 	}
 	tests := []struct {
 		name string
@@ -438,6 +439,7 @@ func Test_fmtTimestamp(t *testing.T) {
 		{
 			name: "valid",
 			args: args{
+				layout: time.StampMilli,
 				t: &timestamppb.Timestamp{
 					Seconds: 0,
 					Nanos:   0,
@@ -448,6 +450,7 @@ func Test_fmtTimestamp(t *testing.T) {
 		{
 			name: "valid non-zero",
 			args: args{
+				layout: time.StampMilli,
 				t: &timestamppb.Timestamp{
 					Seconds: 1530984600,
 					Nanos:   123000000,
@@ -458,6 +461,7 @@ func Test_fmtTimestamp(t *testing.T) {
 		{
 			name: "invalid",
 			args: args{
+				layout: time.StampMilli,
 				t: &timestamppb.Timestamp{
 					Seconds: -1,
 					Nanos:   -1,
@@ -473,7 +477,7 @@ func Test_fmtTimestamp(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := fmtTimestamp(tt.args.t); got != tt.want {
+			if got := fmtTimestamp(tt.args.layout, tt.args.t); got != tt.want {
 				t.Errorf("getTimestamp() = %v, want %v", got, tt.want)
 			}
 		})
@@ -688,7 +692,7 @@ func TestPrinter_AgentEventDetails(t *testing.T) {
 					},
 				},
 			},
-			want: "start time: " + fmtTimestamp(startTS),
+			want: "start time: " + fmtTimestamp(time.StampMilli, startTS),
 		},
 		{
 			name: "policy update",
@@ -836,7 +840,7 @@ func TestPrinter_AgentEventDetails(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := getAgentEventDetails(tt.ev); got != tt.want {
+			if got := getAgentEventDetails(tt.ev, time.StampMilli); got != tt.want {
 				t.Errorf("getAgentEventDetails()\ngot:  %v,\nwant: %v", got, tt.want)
 			}
 		})

pkg/time/time.go

@@ -19,12 +19,31 @@ import (
 	"time"
 )
 
+const (
+	// RFC3339Milli is a time format layout for use in time.Format and
+	// time.Parse. It follows the RFC3339 format with millisecond precision.
+	RFC3339Milli = "2006-01-02T15:04:05.999Z07:00"
+	// RFC3339Micro is a time format layout for use in time.Format and
+	// time.Parse. It follows the RFC3339 format with microsecond precision.
+	RFC3339Micro = "2006-01-02T15:04:05.999999Z07:00"
+)
+
 var (
 	// Now is a hijackable function for time.Now() that makes unit testing a lot
 	// easier for stuff that relies on relative time.
 	Now = time.Now
 )
 
+// layouts is a set of supported time format layouts. Format that only apply to
+// local times should not be added to this list.
+var layouts = []string{
+	time.RFC3339,
+	time.RFC3339Nano,
+	RFC3339Milli,
+	RFC3339Micro,
+	time.RFC1123Z,
+}
+
 // FromString takes as input a string in either RFC3339 or time.Duration
 // format in the past and converts it to a time.Time.
 func FromString(input string) (time.Time, error) {
@@ -34,10 +53,11 @@ func FromString(input string) (time.Time, error) {
 		return Now().Add(-d), nil
 	}
 
-	// try as rfc3339
-	t, err := time.Parse(time.RFC3339, input)
-	if err == nil {
-		return t, nil
+	for _, layout := range layouts {
+		t, err := time.Parse(layout, input)
+		if err == nil {
+			return t, nil
+		}
 	}
 
 	return time.Time{}, fmt.Errorf(