TUN-7132 TUN-7136: Add filter support for streaming logs

Additionally adds similar support in cloudflared tail to provide
filters for events and log level.

Author: DevinCarr

cmd/cloudflared/tail/cmd.go

@@ -39,6 +39,17 @@ func Command() *cli.Command {
 				Value:   "",
 				EnvVars: []string{"TUNNEL_MANAGEMENT_CONNECTOR"},
 			},
+			&cli.StringSliceFlag{
+				Name:    "event",
+				Usage:   "Filter by specific Events (cloudflared, http, tcp, udp) otherwise, defaults to send all events",
+				EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_EVENTS"},
+			},
+			&cli.StringFlag{
+				Name:    "level",
+				Usage:   "Filter by specific log levels (debug, info, warn, error)",
+				EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_LEVEL"},
+				Value:   "debug",
+			},
 			&cli.StringFlag{
 				Name:    "token",
 				Usage:   "Access token for a specific tunnel",
@@ -61,7 +72,7 @@ func Command() *cli.Command {
 			&cli.StringFlag{
 				Name:    logger.LogLevelFlag,
 				Value:   "info",
-				Usage:   "Application logging level {debug, info, warn, error, fatal}. ",
+				Usage:   "Application logging level {debug, info, warn, error, fatal}",
 				EnvVars: []string{"TUNNEL_LOGLEVEL"},
 			},
 		},
@@ -113,6 +124,41 @@ func createLogger(c *cli.Context) *zerolog.Logger {
 	return &log
 }
 
+// parseFilters will attempt to parse provided filters to send to with the EventStartStreaming
+func parseFilters(c *cli.Context) (*management.StreamingFilters, error) {
+	var level *management.LogLevel
+	var events []management.LogEventType
+
+	argLevel := c.String("level")
+	argEvents := c.StringSlice("event")
+
+	if argLevel != "" {
+		l, ok := management.ParseLogLevel(argLevel)
+		if !ok {
+			return nil, fmt.Errorf("invalid --level filter provided, please use one of the following Log Levels: debug, info, warn, error")
+		}
+		level = &l
+	}
+
+	for _, v := range argEvents {
+		t, ok := management.ParseLogEventType(v)
+		if !ok {
+			return nil, fmt.Errorf("invalid --event filter provided, please use one of the following EventTypes: cloudflared, http, tcp, udp")
+		}
+		events = append(events, t)
+	}
+
+	if level == nil && len(events) == 0 {
+		// When no filters are provided, do not return a StreamingFilters struct
+		return nil, nil
+	}
+
+	return &management.StreamingFilters{
+		Level:  level,
+		Events: events,
+	}, nil
+}
+
 // Run implements a foreground runner
 func Run(c *cli.Context) error {
 	log := createLogger(c)
@@ -121,6 +167,12 @@ func Run(c *cli.Context) error {
 	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
 	defer signal.Stop(signals)
 
+	filters, err := parseFilters(c)
+	if err != nil {
+		log.Error().Err(err).Msgf("invalid filters provided")
+		return nil
+	}
+
 	managementHostname := c.String("management-hostname")
 	token := c.String("token")
 	u := url.URL{Scheme: "wss", Host: managementHostname, Path: "/logs", RawQuery: "access_token=" + token}
@@ -148,6 +200,7 @@ func Run(c *cli.Context) error {
 	// Once connection is established, send start_streaming event to begin receiving logs
 	err = management.WriteEvent(conn, ctx, &management.EventStartStreaming{
 		ClientEvent: management.ClientEvent{Type: management.StartStreaming},
+		Filters:     filters,
 	})
 	if err != nil {
 		log.Error().Err(err).Msg("unable to request logs from management tunnel")

management/events.go

@@ -48,7 +48,12 @@ type ClientEvent struct {
 // Additional filters can be provided to augment the log events requested.
 type EventStartStreaming struct {
 	ClientEvent
-	Filters []string `json:"filters"`
+	Filters *StreamingFilters `json:"filters,omitempty"`
+}
+
+type StreamingFilters struct {
+	Events []LogEventType `json:"events,omitempty"`
+	Level  *LogLevel      `json:"level,omitempty"`
 }
 
 // EventStopStreaming signifies that the client wishes to halt receiving log events.
@@ -65,7 +70,7 @@ type EventLog struct {
 // LogEventType is the way that logging messages are able to be filtered.
 // Example: assigning LogEventType.Cloudflared to a zerolog event will allow the client to filter for only
 // the Cloudflared-related events.
-type LogEventType int
+type LogEventType int8
 
 const (
 	// Cloudflared events are signficant to cloudflared operations like connection state changes.
@@ -76,6 +81,20 @@ const (
 	UDP
 )
 
+func ParseLogEventType(s string) (LogEventType, bool) {
+	switch s {
+	case "cloudflared":
+		return Cloudflared, true
+	case "http":
+		return HTTP, true
+	case "tcp":
+		return TCP, true
+	case "udp":
+		return UDP, true
+	}
+	return -1, false
+}
+
 func (l LogEventType) String() string {
 	switch l {
 	case Cloudflared:
@@ -91,18 +110,79 @@ func (l LogEventType) String() string {
 	}
 }
 
+func (l LogEventType) MarshalJSON() ([]byte, error) {
+	return json.Marshal(l.String())
+}
+
+func (e *LogEventType) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return errors.New("unable to unmarshal LogEventType string")
+	}
+	if event, ok := ParseLogEventType(s); ok {
+		*e = event
+		return nil
+	}
+	return errors.New("unable to unmarshal LogEventType")
+}
+
 // LogLevel corresponds to the zerolog logging levels
 // "panic", "fatal", and "trace" are exempt from this list as they are rarely used and, at least
 // the the first two are limited to failure conditions that lead to cloudflared shutting down.
-type LogLevel string
+type LogLevel int8
 
 const (
-	Debug LogLevel = "debug"
-	Info  LogLevel = "info"
-	Warn  LogLevel = "warn"
-	Error LogLevel = "error"
+	Debug LogLevel = 0
+	Info  LogLevel = 1
+	Warn  LogLevel = 2
+	Error LogLevel = 3
 )
 
+func ParseLogLevel(l string) (LogLevel, bool) {
+	switch l {
+	case "debug":
+		return Debug, true
+	case "info":
+		return Info, true
+	case "warn":
+		return Warn, true
+	case "error":
+		return Error, true
+	}
+	return -1, false
+}
+
+func (l LogLevel) String() string {
+	switch l {
+	case Debug:
+		return "debug"
+	case Info:
+		return "info"
+	case Warn:
+		return "warn"
+	case Error:
+		return "error"
+	default:
+		return ""
+	}
+}
+
+func (l LogLevel) MarshalJSON() ([]byte, error) {
+	return json.Marshal(l.String())
+}
+
+func (l *LogLevel) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return errors.New("unable to unmarshal LogLevel string")
+	}
+	if level, ok := ParseLogLevel(s); ok {
+		*l = level
+		return nil
+	}
+	return fmt.Errorf("unable to unmarshal LogLevel")
+}
+
 const (
 	// TimeKey aligns with the zerolog.TimeFieldName
 	TimeKey = "time"

management/events_test.go

@@ -11,14 +11,83 @@ import (
 	"github.com/cloudflare/cloudflared/internal/test"
 )
 
+var (
+	debugLevel *LogLevel
+	infoLevel  *LogLevel
+	warnLevel  *LogLevel
+	errorLevel *LogLevel
+)
+
+func init() {
+	// created here because we can't do a reference to a const enum, i.e. &Info
+	debugLevel := new(LogLevel)
+	*debugLevel = Debug
+	infoLevel := new(LogLevel)
+	*infoLevel = Info
+	warnLevel := new(LogLevel)
+	*warnLevel = Warn
+	errorLevel := new(LogLevel)
+	*errorLevel = Error
+}
+
 func TestIntoClientEvent_StartStreaming(t *testing.T) {
-	event := ClientEvent{
-		Type:  StartStreaming,
-		event: []byte(`{"type": "start_streaming"}`),
+	for _, test := range []struct {
+		name     string
+		expected EventStartStreaming
+	}{
+		{
+			name:     "no filters",
+			expected: EventStartStreaming{ClientEvent: ClientEvent{Type: StartStreaming}},
+		},
+		{
+			name: "level filter",
+			expected: EventStartStreaming{
+				ClientEvent: ClientEvent{Type: StartStreaming},
+				Filters: &StreamingFilters{
+					Level: infoLevel,
+				},
+			},
+		},
+		{
+			name: "events filter",
+			expected: EventStartStreaming{
+				ClientEvent: ClientEvent{Type: StartStreaming},
+				Filters: &StreamingFilters{
+					Events: []LogEventType{Cloudflared, HTTP},
+				},
+			},
+		},
+		{
+			name: "level and events filters",
+			expected: EventStartStreaming{
+				ClientEvent: ClientEvent{Type: StartStreaming},
+				Filters: &StreamingFilters{
+					Level:  infoLevel,
+					Events: []LogEventType{Cloudflared},
+				},
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			data, err := json.Marshal(test.expected)
+			require.NoError(t, err)
+			event := ClientEvent{}
+			err = json.Unmarshal(data, &event)
+			require.NoError(t, err)
+			event.event = data
+			ce, ok := IntoClientEvent[EventStartStreaming](&event, StartStreaming)
+			require.True(t, ok)
+			require.Equal(t, test.expected.ClientEvent, ce.ClientEvent)
+			if test.expected.Filters != nil {
+				f := ce.Filters
+				ef := test.expected.Filters
+				if ef.Level != nil {
+					require.Equal(t, *ef.Level, *f.Level)
+				}
+				require.ElementsMatch(t, ef.Events, f.Events)
+			}
+		})
 	}
-	ce, ok := IntoClientEvent[EventStartStreaming](&event, StartStreaming)
-	require.True(t, ok)
-	require.Equal(t, EventStartStreaming{ClientEvent: ClientEvent{Type: StartStreaming}}, *ce)
 }
 
 func TestIntoClientEvent_StopStreaming(t *testing.T) {