Add log type include filter

src/pkg/binding/poller.go

@@ -9,9 +9,11 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 
 	metrics "code.cloudfoundry.org/go-metric-registry"
+	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog"
 	v2 "code.cloudfoundry.org/loggregator-agent-release/src/pkg/ingress/v2"
 	"code.cloudfoundry.org/loggregator-agent-release/src/pkg/simplecache"
 )
@@ -236,6 +238,17 @@ func (bc *bindingChecker) checkBindings(bindings []Binding) []Binding {
 			continue
 		}
 
+		if invalidLogFilter(u) {
+			bc.rejectBinding(b.Credentials, fmt.Sprintf("include-log-source-types and exclude-log-source-types cannot be used at the same time in syslog drain url %s", anonymousUrl.String()), true)
+			continue
+		}
+
+		sourceTypes := getUnknownSourceTypes(u.Query())
+		if sourceTypes != nil {
+			bc.rejectBinding(b.Credentials, fmt.Sprintf("Unknown source types '%s' in source type filter in syslog drain url %s", strings.Join(sourceTypes, ", "), anonymousUrl.String()), true)
+			continue
+		}
+
 		_, exists := bc.failedHostsCache.Get(u.Host)
 		if exists {
 			bc.rejectBinding(b.Credentials, fmt.Sprintf("Skipped resolve ip address for syslog drain with url %s due to prior failure", anonymousUrl.String()), false)
@@ -308,6 +321,34 @@ func invalidScheme(scheme string) bool {
 	return true
 }
 
+// invalidLogFilter checks if both include-log-source-types and exclude-log-source-types
+func invalidLogFilter(u *url.URL) bool {
+	includeSourceTypes := u.Query().Get("include-log-source-types")
+	excludeSourceTypes := u.Query().Get("exclude-log-source-types")
+	if excludeSourceTypes != "" && includeSourceTypes != "" {
+		return true
+	}
+	return false
+}
+
+// assumes only one of include-log-source-types or exclude-log-source-types is set
+func getUnknownSourceTypes(u url.Values) []string {
+	var sourceTypeList string
+	includeSourceTypes := u.Get("include-log-source-types")
+	excludeSourceTypes := u.Get("exclude-log-source-types")
+
+	if includeSourceTypes != "" {
+		sourceTypeList = includeSourceTypes
+	} else if excludeSourceTypes != "" {
+		sourceTypeList = excludeSourceTypes
+	} else {
+		return nil
+	}
+
+	_, unknownTypes := syslog.ParseSourceTypeList(sourceTypeList)
+	return unknownTypes
+}
+
 func CalculateBindingCount(bindings []Binding) int {
 	apps := make(map[string]bool)
 	for _, b := range bindings {

src/pkg/binding/poller_test.go

@@ -563,6 +563,125 @@ var _ = Describe("Poller", func() {
 			Expect(bndChecker.invalidDrains).To(Equal(float64(0)))
 			Expect(bndChecker.blacklistedDrains).To(Equal(float64(0)))
 		})
+
+		Context("when both include-log-source-types and exclude-log-source-types are specified", func() {
+			It("ignores the drain and counts as invalid", func() {
+				bindings := []Binding{
+					{
+						Url: "https://test.org/drain?include-log-source-types=app&exclude-log-source-types=rtr",
+						Credentials: []Credentials{
+							{
+								Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}},
+							},
+						},
+					},
+				}
+
+				filteredBindings := bndChecker.checkBindings(bindings)
+
+				Expect(filteredBindings).To(BeEmpty())
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("include-log-source-types and exclude-log-source-types cannot be used at the same time")))
+				Expect(bndChecker.invalidDrains).To(BeNumerically("==", 1))
+				Expect(bndChecker.blacklistedDrains).To(BeNumerically("==", 0))
+			})
+
+			It("doesn't log the conflicting filters warning when warn is false", func() {
+				bndChecker.warn = false
+				bindings := []Binding{
+					{
+						Url: "https://test.org/drain?include-log-source-types=app&exclude-log-source-types=rtr",
+						Credentials: []Credentials{
+							{
+								Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}},
+							},
+						},
+					},
+				}
+				bndChecker.checkBindings(bindings)
+
+				for _, msg := range logClient.Message() {
+					Expect(msg).ToNot(MatchRegexp("include-log-source-types and exclude-log-source-types cannot be used at the same time"))
+				}
+			})
+		})
+
+		Context("when unknown source types are provided", func() {
+			It("logs a warning and ignores the drain in include mode", func() {
+				bindings := []Binding{
+					{
+						Url: "https://test.org/drain?include-log-source-types=app,unknown,invalid,rtr",
+						Credentials: []Credentials{
+							{
+								Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}},
+							},
+						},
+					},
+				}
+				filteredBindings := bndChecker.checkBindings(bindings)
+
+				Expect(filteredBindings).To(BeEmpty())
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("Unknown source types")))
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("unknown")))
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("invalid")))
+				Expect(bndChecker.invalidDrains).To(BeNumerically("==", 1))
+			})
+
+			It("logs a warning and ignores the drain in exclude mode", func() {
+				bindings := []Binding{
+					{
+						Url: "https://test.org/drain?exclude-log-source-types=rtr,unknown",
+						Credentials: []Credentials{
+							{
+								Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}},
+							},
+						},
+					},
+				}
+				filteredBindings := bndChecker.checkBindings(bindings)
+
+				Expect(filteredBindings).To(BeEmpty())
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("Unknown source types")))
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("unknown")))
+				Expect(bndChecker.invalidDrains).To(BeNumerically("==", 1))
+			})
+
+			It("logs a warning and ignores the drain when source types have spaces", func() {
+				bindings := []Binding{
+					{
+						Url: "https://test.org/drain?include-log-source-types=app, rtr",
+						Credentials: []Credentials{
+							{
+								Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}},
+							},
+						},
+					},
+				}
+				filteredBindings := bndChecker.checkBindings(bindings)
+
+				Expect(filteredBindings).To(BeEmpty())
+				Expect(logClient.Message()).To(ContainElement(MatchRegexp("Unknown source types")))
+				Expect(bndChecker.invalidDrains).To(BeNumerically("==", 1))
+			})
+
+			It("doesn't log the warning when warn is false", func() {
+				bndChecker.warn = false
+				bindings := []Binding{
+					{
+						Url: "https://test.org/drain?include-log-source-types=app,unknown,rtr",
+						Credentials: []Credentials{
+							{
+								Apps: []App{{Hostname: "app-hostname0", AppID: "app-id-0"}},
+							},
+						},
+					},
+				}
+				bndChecker.checkBindings(bindings)
+
+				for _, msg := range logClient.Message() {
+					Expect(msg).ToNot(MatchRegexp("Unknown source types"))
+				}
+			})
+		})
 	})
 })
 

src/pkg/egress/syslog/filtering_drain_writer.go

@@ -35,22 +35,18 @@ func NewFilteringDrainWriter(binding Binding, writer egress.Writer) (*FilteringD
 }
 
 func (w *FilteringDrainWriter) Write(env *loggregator_v2.Envelope) error {
-	if w.binding.DrainData == ALL {
-		return w.writer.Write(env)
-	}
-
 	if env.GetTimer() != nil {
-		if w.binding.DrainData == TRACES {
+		if w.binding.DrainData == TRACES || w.binding.DrainData == ALL {
 			return w.writer.Write(env)
 		}
 	}
 	if env.GetEvent() != nil {
-		if w.binding.DrainData == LOGS {
+		if sendsEvents(w.binding.DrainData, w.binding.LogFilter, env.GetTags()["source_type"]) {
 			return w.writer.Write(env)
 		}
 	}
 	if env.GetLog() != nil {
-		if sendsLogs(w.binding.DrainData) {
+		if sendsLogs(w.binding.DrainData, w.binding.LogFilter, env.GetTags()["source_type"]) {
 			return w.writer.Write(env)
 		}
 	}
@@ -63,26 +59,25 @@ 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 sendsEvents(drainData DrainData, logFilter *LogFilter, sourceTypeTag string) bool {
+	if drainData != LOGS && drainData != ALL {
+		return false
+	}
+
+	return logFilter.ShouldInclude(sourceTypeTag)
+}
+
+func sendsLogs(drainData DrainData, logFilter *LogFilter, sourceTypeTag string) bool {
+	if drainData == TRACES || drainData == METRICS {
 		return false
 	}
+
+	return logFilter.ShouldInclude(sourceTypeTag)
 }
 
 func sendsMetrics(drainData DrainData) bool {
-	switch drainData {
-	case LOGS_AND_METRICS:
-		return true
-	case METRICS:
-		return true
-	default:
+	if drainData != LOGS_AND_METRICS && drainData != METRICS && drainData != ALL {
 		return false
 	}
+	return true
 }