Add log type include filter

src/pkg/egress/syslog/filtering_drain_writer.go

@@ -50,7 +50,8 @@ func (w *FilteringDrainWriter) Write(env *loggregator_v2.Envelope) error {
 		}
 	}
 	if env.GetLog() != nil {
-		if sendsLogs(w.binding.DrainData) {
+		sourceType := env.GetTags()["source_type"]
+		if sendsLogs(w.binding.DrainData, w.binding.LogFilter, sourceType) {
 			return w.writer.Write(env)
 		}
 	}
@@ -63,17 +64,12 @@ func (w *FilteringDrainWriter) Write(env *loggregator_v2.Envelope) error {
 	return nil
 }
 
-func sendsLogs(drainData DrainData) bool {
-	switch drainData {
-	case LOGS:
-		return true
-	case LOGS_AND_METRICS:
-		return true
-	case LOGS_NO_EVENTS:
-		return true
-	default:
+func sendsLogs(drainData DrainData, logFilter *LogFilter, sourceTypeTag string) bool {
+	if drainData != LOGS && drainData != LOGS_AND_METRICS && drainData != LOGS_NO_EVENTS {
 		return false
 	}
+
+	return logFilter.ShouldInclude(sourceTypeTag)
 }
 
 func sendsMetrics(drainData DrainData) bool {

src/pkg/egress/syslog/filtering_drain_writer_test.go

@@ -53,6 +53,178 @@ var _ = Describe("Filtering Drain Writer", func() {
 		_, err := syslog.NewFilteringDrainWriter(binding, &fakeWriter{})
 		Expect(err).To(HaveOccurred())
 	})
+
+	Context("when source_type tag is missing", func() {
+		var envelope *loggregator_v2.Envelope
+
+		BeforeEach(func() {
+			envelope = &loggregator_v2.Envelope{
+				Message: &loggregator_v2.Envelope_Log{
+					Log: &loggregator_v2.Log{
+						Payload: []byte("test log"),
+					},
+				},
+				Tags: map[string]string{
+					// source_type tag is intentionally missing
+				},
+			}
+		})
+
+		It("filters out the log with missing source_type when include filter is configured", func() {
+			binding := syslog.Binding{
+				DrainData: syslog.LOGS,
+				LogFilter: syslog.NewLogFilter(syslog.LogSourceTypeSet{syslog.LOG_SOURCE_APP: struct{}{}}, syslog.LogFilterModeInclude),
+			}
+			fakeWriter := &fakeWriter{}
+			drainWriter, err := syslog.NewFilteringDrainWriter(binding, fakeWriter)
+			Expect(err).NotTo(HaveOccurred())
+
+			err = drainWriter.Write(envelope)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeWriter.received).To(Equal(0))
+
+			appEnvelope := &loggregator_v2.Envelope{
+				Message: &loggregator_v2.Envelope_Log{
+					Log: &loggregator_v2.Log{Payload: []byte("app log")},
+				},
+				Tags: map[string]string{"source_type": "APP/PROC/WEB/0"},
+			}
+			err = drainWriter.Write(appEnvelope)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeWriter.received).To(Equal(1))
+		})
+
+		It("sends the log with missing source_type when exclude filter is configured", func() {
+			binding := syslog.Binding{
+				DrainData: syslog.LOGS,
+				LogFilter: syslog.NewLogFilter(syslog.LogSourceTypeSet{syslog.LOG_SOURCE_RTR: struct{}{}}, syslog.LogFilterModeExclude),
+			}
+			fakeWriter := &fakeWriter{}
+			drainWriter, err := syslog.NewFilteringDrainWriter(binding, fakeWriter)
+			Expect(err).NotTo(HaveOccurred())
+
+			err = drainWriter.Write(envelope)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeWriter.received).To(Equal(1))
+
+			rtrEnvelope := &loggregator_v2.Envelope{
+				Message: &loggregator_v2.Envelope_Log{
+					Log: &loggregator_v2.Log{Payload: []byte("rtr log")},
+				},
+				Tags: map[string]string{"source_type": "RTR/1"},
+			}
+			err = drainWriter.Write(rtrEnvelope)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeWriter.received).To(Equal(1))
+		})
+	})
+
+	Context("when source_type tag is present", func() {
+		It("filters logs based on include filter - includes only APP logs", func() {
+			binding := syslog.Binding{
+				DrainData: syslog.LOGS,
+				LogFilter: syslog.NewLogFilter(syslog.LogSourceTypeSet{syslog.LOG_SOURCE_APP: struct{}{}}, syslog.LogFilterModeInclude),
+			}
+			fakeWriter := &fakeWriter{}
+			drainWriter, err := syslog.NewFilteringDrainWriter(binding, fakeWriter)
+			Expect(err).NotTo(HaveOccurred())
+
+			envelopes := []*loggregator_v2.Envelope{
+				{
+					Message: &loggregator_v2.Envelope_Log{
+						Log: &loggregator_v2.Log{Payload: []byte("app log")},
+					},
+					Tags: map[string]string{"source_type": "APP/PROC/WEB/0"},
+				},
+				{
+					Message: &loggregator_v2.Envelope_Log{
+						Log: &loggregator_v2.Log{Payload: []byte("rtr log")},
+					},
+					Tags: map[string]string{"source_type": "RTR/1"},
+				},
+				{
+					Message: &loggregator_v2.Envelope_Log{
+						Log: &loggregator_v2.Log{Payload: []byte("stg log")},
+					},
+					Tags: map[string]string{"source_type": "STG/0"},
+				},
+			}
+
+			for _, envelope := range envelopes {
+				err = drainWriter.Write(envelope)
+				Expect(err).NotTo(HaveOccurred())
+			}
+
+			// Only APP log should be sent
+			Expect(fakeWriter.received).To(Equal(1))
+		})
+
+		It("filters logs based on exclude filter - excludes RTR logs", func() {
+			binding := syslog.Binding{
+				DrainData: syslog.LOGS,
+				LogFilter: syslog.NewLogFilter(syslog.LogSourceTypeSet{syslog.LOG_SOURCE_RTR: struct{}{}}, syslog.LogFilterModeExclude),
+			}
+			fakeWriter := &fakeWriter{}
+			drainWriter, err := syslog.NewFilteringDrainWriter(binding, fakeWriter)
+			Expect(err).NotTo(HaveOccurred())
+
+			envelopes := []*loggregator_v2.Envelope{
+				{
+					Message: &loggregator_v2.Envelope_Log{
+						Log: &loggregator_v2.Log{Payload: []byte("app log")},
+					},
+					Tags: map[string]string{"source_type": "APP/PROC/WEB/0"},
+				},
+				{
+					Message: &loggregator_v2.Envelope_Log{
+						Log: &loggregator_v2.Log{Payload: []byte("rtr log")},
+					},
+					Tags: map[string]string{"source_type": "RTR/1"},
+				},
+				{
+					Message: &loggregator_v2.Envelope_Log{
+						Log: &loggregator_v2.Log{Payload: []byte("stg log")},
+					},
+					Tags: map[string]string{"source_type": "STG/0"},
+				},
+			}
+
+			for _, envelope := range envelopes {
+				err = drainWriter.Write(envelope)
+				Expect(err).NotTo(HaveOccurred())
+			}
+
+			// APP and STG logs should be sent, RTR should be filtered out
+			Expect(fakeWriter.received).To(Equal(2))
+		})
+
+		It("sends logs with unknown source_type prefix when filter is set", func() {
+			binding := syslog.Binding{
+				DrainData: syslog.LOGS,
+				LogFilter: syslog.NewLogFilter(syslog.LogSourceTypeSet{syslog.LOG_SOURCE_APP: struct{}{}}, syslog.LogFilterModeExclude),
+			}
+			fakeWriter := &fakeWriter{}
+			drainWriter, err := syslog.NewFilteringDrainWriter(binding, fakeWriter)
+			Expect(err).NotTo(HaveOccurred())
+
+			envelope := &loggregator_v2.Envelope{
+				Message: &loggregator_v2.Envelope_Log{
+					Log: &loggregator_v2.Log{
+						Payload: []byte("test log"),
+					},
+				},
+				Tags: map[string]string{
+					"source_type": "UNKNOWN/some/path",
+				},
+			}
+
+			err = drainWriter.Write(envelope)
+
+			// Should send the log because unknown types default to being included for exclude filter
+			Expect(err).NotTo(HaveOccurred())
+			Expect(fakeWriter.received).To(Equal(1))
+		})
+	})
 })
 
 type fakeWriter struct {

src/pkg/egress/syslog/log_source.go

@@ -0,0 +1,133 @@
+package syslog
+
+import "strings"
+
+// LogSourceType defines the source types used within Cloud Foundry
+// Their order in the code is as documented in https://docs.cloudfoundry.org/devguide/deploy-apps/streaming-logs.html#format
+type LogSourceType string
+
+const (
+	LOG_SOURCE_API    LogSourceType = "API"
+	LOG_SOURCE_STG    LogSourceType = "STG"
+	LOG_SOURCE_RTR    LogSourceType = "RTR"
+	LOG_SOURCE_LGR    LogSourceType = "LGR"
+	LOG_SOURCE_APP    LogSourceType = "APP"
+	LOG_SOURCE_SSH    LogSourceType = "SSH"
+	LOG_SOURCE_CELL   LogSourceType = "CELL"
+	LOG_SOURCE_PROXY  LogSourceType = "PROXY"
+	LOG_SOURCE_HEALTH LogSourceType = "HEALTH"
+	LOG_SOURCE_SYS    LogSourceType = "SYS"
+	LOG_SOURCE_STATS  LogSourceType = "STATS"
+)
+
+// validSourceTypes contains SourceType prefixes for efficient lookup
+var validSourceTypes = map[LogSourceType]struct{}{
+	LOG_SOURCE_API:  {},
+	LOG_SOURCE_STG:  {},
+	LOG_SOURCE_RTR:  {},
+	LOG_SOURCE_LGR:  {},
+	LOG_SOURCE_APP:  {},
+	LOG_SOURCE_SSH:  {},
+	LOG_SOURCE_CELL: {},
+}
+
+// IsValid checks if the provided SourceType is valid
+func (lt LogSourceType) IsValid() bool {
+	_, ok := validSourceTypes[lt]
+	return ok
+}
+
+// ParseSourceType parses a string into a SourceType value
+func ParseSourceType(s string) (LogSourceType, bool) {
+	lt := LogSourceType(strings.ToUpper(s))
+	return lt, lt.IsValid()
+}
+
+// ParseSourceTypeList parses a comma-separated list of source types and returns
+// the valid types as a set and any unknown types as a slice.
+func ParseSourceTypeList(sourceTypeList string) (LogSourceTypeSet, []string) {
+	sourceTypes := strings.Split(sourceTypeList, ",")
+	set := make(LogSourceTypeSet, len(sourceTypes))
+	var unknownTypes []string
+
+	for _, sourceType := range sourceTypes {
+		t, ok := ParseSourceType(sourceType)
+		if !ok {
+			unknownTypes = append(unknownTypes, sourceType)
+			continue
+		}
+		set.Add(t)
+	}
+
+	return set, unknownTypes
+}
+
+// ExtractPrefix extracts the prefix from a source_type tag (e.g., "APP/PROC/WEB/0" -> "APP")
+func ExtractPrefix(sourceTypeTag string) string {
+	if idx := strings.IndexByte(sourceTypeTag, '/'); idx != -1 {
+		return sourceTypeTag[:idx]
+	}
+	return sourceTypeTag
+}
+
+// LogSourceTypeSet is a set of SourceTypes for efficient membership checking
+type LogSourceTypeSet map[LogSourceType]struct{}
+
+// Add adds a SourceType to the set
+func (s LogSourceTypeSet) Add(lt LogSourceType) {
+	s[lt] = struct{}{}
+}
+
+// Contains checks if the set contains a SourceType
+func (s LogSourceTypeSet) Contains(lt LogSourceType) bool {
+	_, exists := s[lt]
+	return exists
+}
+
+// LogFilterMode determines how the log filter should be applied
+type LogFilterMode int
+
+const (
+	// LogFilterModeInclude only includes logs matching the specified types (strict)
+	LogFilterModeInclude LogFilterMode = iota
+	// LogFilterModeExclude excludes logs matching the specified types (permissive)
+	LogFilterModeExclude
+)
+
+// LogFilter encapsulates source type filtering configuration
+type LogFilter struct {
+	Types LogSourceTypeSet
+	Mode  LogFilterMode
+}
+
+// NewLogFilter creates a new LogFilter with the given types and mode
+func NewLogFilter(types LogSourceTypeSet, mode LogFilterMode) *LogFilter {
+	return &LogFilter{
+		Types: types,
+		Mode:  mode,
+	}
+}
+
+// ShouldInclude determines if a log with the given sourceTypeTag should be forwarded
+// Include mode omits missing/unknown source types, exclude mode forwards them
+func (f *LogFilter) ShouldInclude(sourceTypeTag string) bool {
+	if f == nil {
+		return true
+	}
+
+	if sourceTypeTag == "" {
+		return f.Mode == LogFilterModeExclude
+	}
+
+	prefix := ExtractPrefix(sourceTypeTag)
+	sourceType := LogSourceType(prefix)
+	if !sourceType.IsValid() {
+		return f.Mode == LogFilterModeExclude
+	}
+
+	inSet := f.Types.Contains(sourceType)
+	if f.Mode == LogFilterModeInclude {
+		return inSet
+	}
+	return !inSet
+}

src/pkg/egress/syslog/syslog_connector.go

@@ -20,6 +20,7 @@ type Binding struct {
 	DrainData    DrainData `json:"type,omitempty"`
 	OmitMetadata bool
 	InternalTls  bool
+	LogFilter    *LogFilter
 }
 
 type Drain struct {

src/pkg/ingress/bindings/binding_config.go

@@ -35,6 +35,7 @@ func (d *DrainParamParser) FetchBindings() ([]syslog.Binding, error) {
 		b.OmitMetadata = getOmitMetadata(urlParsed, d.defaultDrainMetadata)
 		b.InternalTls = getInternalTLS(urlParsed)
 		b.DrainData = getBindingType(urlParsed)
+		b.LogFilter = d.getLogFilter(urlParsed)
 
 		processed = append(processed, b)
 	}
@@ -84,6 +85,29 @@ func getBindingType(u *url.URL) syslog.DrainData {
 	return drainData
 }
 
+func (d *DrainParamParser) getLogFilter(u *url.URL) *syslog.LogFilter {
+	includeSourceTypes := u.Query().Get("include-log-source-types")
+	excludeSourceTypes := u.Query().Get("exclude-log-source-types")
+
+	if excludeSourceTypes != "" {
+		return d.newLogFilter(excludeSourceTypes, syslog.LogFilterModeExclude)
+	} else if includeSourceTypes != "" {
+		return d.newLogFilter(includeSourceTypes, syslog.LogFilterModeInclude)
+	}
+	return nil
+}
+
+// newLogFilter parses a URL query parameter into a LogFilter.
+// sourceTypeList is assumed to be a comma-separated list of valid source types.
+func (d *DrainParamParser) newLogFilter(sourceTypeList string, mode syslog.LogFilterMode) *syslog.LogFilter {
+	if sourceTypeList == "" {
+		return nil
+	}
+
+	set, _ := syslog.ParseSourceTypeList(sourceTypeList)
+	return syslog.NewLogFilter(set, mode)
+}
+
 func getRemoveMetadataQuery(u *url.URL) string {
 	q := u.Query().Get("disable-metadata")
 	if q == "" {