Implement Advanced Filtering proposal

Gopkg.toml

@@ -105,7 +105,11 @@ required = [
   name = "github.com/cloudevents/sdk-go"
   version = "=0.6.0"
 
+[[constraint]]
+  name = "github.com/google/cel-go"
+  version = "=0.2.0"
+
 # needed because pkg upgraded
 [[override]]
   name = "go.uber.org/zap"
-  revision = "67bc79d13d155c02fd008f721863ff8cc5f30659"
+  revision = "67bc79d13d155c02fd008f721863ff8cc5f30659"

cmd/broker/filter/main.go

@@ -18,21 +18,38 @@ package main
 
 import (
 	"flag"
+	"fmt"
 	"log"
+	"net/http"
 	"os"
+	"sync"
+	"time"
 
 	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
 	"github.com/knative/eventing/pkg/broker"
 	"github.com/knative/eventing/pkg/provisioners"
+	"github.com/knative/eventing/pkg/utils"
 	"github.com/knative/pkg/signals"
+	"go.opencensus.io/exporter/prometheus"
+	"go.opencensus.io/stats/view"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 )
 
 const (
-	NAMESPACE = "NAMESPACE"
+	NAMESPACE        = "NAMESPACE"
+	metricsNamespace = "broker_filter"
+)
+
+var (
+	metricsPort = 9090
+
+	writeTimeout    = 1 * time.Minute
+	shutdownTimeout = 1 * time.Minute
+
+	wg sync.WaitGroup
 )
 
 func main() {
@@ -67,6 +84,30 @@ func main() {
 		logger.Fatal("Unable to start the receiver", zap.Error(err), zap.Any("receiver", receiver))
 	}
 
+	// Metrics
+	e, err := prometheus.NewExporter(prometheus.Options{Namespace: metricsNamespace})
+	if err != nil {
+		logger.Fatal("Unable to create Prometheus exporter", zap.Error(err))
+	}
+	view.RegisterExporter(e)
+	sm := http.NewServeMux()
+	sm.Handle("/metrics", e)
+	metricsSrv := &http.Server{
+		Addr:         fmt.Sprintf(":%d", metricsPort),
+		Handler:      e,
+		ErrorLog:     zap.NewStdLog(logger),
+		WriteTimeout: writeTimeout,
+	}
+
+	err = mgr.Add(&utils.RunnableServer{
+		Server:          metricsSrv,
+		ShutdownTimeout: shutdownTimeout,
+		WaitGroup:       &wg,
+	})
+	if err != nil {
+		logger.Fatal("Unable to add metrics runnableServer", zap.Error(err))
+	}
+
 	// Set up signals so we handle the first shutdown signal gracefully.
 	stopCh := signals.SetupSignalHandler()
 
@@ -77,6 +118,18 @@ func main() {
 		logger.Fatal("Manager.Start() returned an error", zap.Error(err))
 	}
 	logger.Info("Exiting...")
+
+	// TODO Gracefully shutdown the filter server. CloudEvents SDK doesn't seem
+	// to let us do that today.
+	go func() {
+		<-time.After(shutdownTimeout)
+		log.Fatalf("Shutdown took longer than %v", shutdownTimeout)
+	}()
+
+	// Wait for runnables to stop. This blocks indefinitely, but the above
+	// goroutine will exit the process if it takes longer than shutdownTimeout.
+	wg.Wait()
+	logger.Info("Done.")
 }
 
 func getRequiredEnv(envKey string) string {

pkg/apis/eventing/v1alpha1/trigger_defaults.go

@@ -30,16 +30,4 @@ func (ts *TriggerSpec) SetDefaults(ctx context.Context) {
 	if ts.Filter == nil {
 		ts.Filter = &TriggerFilter{}
 	}
-
-	// Note that this logic will need to change once there are other filtering options, as it should
-	// only apply if no other filter is applied.
-	if ts.Filter.SourceAndType == nil {
-		ts.Filter.SourceAndType = &TriggerFilterSourceAndType{}
-	}
-	if ts.Filter.SourceAndType.Type == "" {
-		ts.Filter.SourceAndType.Type = TriggerAnyFilter
-	}
-	if ts.Filter.SourceAndType.Source == "" {
-		ts.Filter.SourceAndType.Source = TriggerAnyFilter
-	}
 }

pkg/apis/eventing/v1alpha1/trigger_defaults_test.go

@@ -26,16 +26,13 @@ import (
 var (
 	defaultBroker        = "default"
 	otherBroker          = "other_broker"
-	defaultTriggerFilter = &TriggerFilter{
-		SourceAndType: &TriggerFilterSourceAndType{
-			Type:   TriggerAnyFilter,
-			Source: TriggerAnyFilter},
-	}
-	otherTriggerFilter = &TriggerFilter{
-		SourceAndType: &TriggerFilterSourceAndType{
+	defaultTriggerFilter = &TriggerFilter{}
+	otherTriggerFilter   = &TriggerFilter{
+		DeprecatedSourceAndType: &TriggerFilterSourceAndType{
 			Type:   "other_type",
 			Source: "other_source"},
 	}
+
 	defaultTrigger = Trigger{
 		Spec: TriggerSpec{
 			Broker: defaultBroker,

pkg/apis/eventing/v1alpha1/trigger_types.go

@@ -64,8 +64,31 @@ type TriggerSpec struct {
 	Subscriber *SubscriberSpec `json:"subscriber,omitempty"`
 }
 
+// TriggerFilter specifies the event filtering strategy for the Trigger. Only
+// one field may be set.
 type TriggerFilter struct {
-	SourceAndType *TriggerFilterSourceAndType `json:"sourceAndType,omitempty"`
+	// DeprecatedSourceAndType filters events based on exact matches on the
+	// CloudEvents type and source attributes. This field has been replaced by the
+	// Attributes field.
+	//
+	// +optional
+	DeprecatedSourceAndType *TriggerFilterSourceAndType `json:"sourceAndType,omitempty"`
+
+	// Attributes filters events by exact match on event context attributes.
+	// Each key in the map is compared with the equivalent key in the event
+	// context. An event passes the filter if all values are equal to the
+	// specified values.
+	//
+	// Nested context attributes are not supported as keys. Numeric values are
+	// not supported.
+	Attributes *TriggerFilterAttributes `json:"attributes,omitempty"`
+
+	// Expression filters events by evaluating the expression with the Common
+	// Expression Language runtime. An event passes the filter if the expression
+	// evaluates to true.
+	//
+	// +optional
+	Expression *TriggerFilterExpression `json:"expression,omitempty"`
 }
 
 // TriggerFilterSourceAndType filters events based on exact matches on the cloud event's type and
@@ -76,6 +99,14 @@ type TriggerFilterSourceAndType struct {
 	Source string `json:"source,omitempty"`
 }
 
+// TriggerFilterAttributes is a map of context attribute names to values for
+// filtering by equality.
+type TriggerFilterAttributes map[string]string
+
+// TriggerFilterExpression is a string containing the filter expression to
+// evaluate.
+type TriggerFilterExpression string
+
 // TriggerStatus represents the current state of a Trigger.
 type TriggerStatus struct {
 	// inherits duck/v1alpha1 Status, which currently provides:

pkg/apis/eventing/v1alpha1/trigger_validation.go

@@ -23,10 +23,12 @@ import (
 	"github.com/knative/pkg/kmp"
 )
 
+// Validate the Trigger.
 func (t *Trigger) Validate(ctx context.Context) *apis.FieldError {
 	return t.Spec.Validate(ctx).ViaField("spec")
 }
 
+// Validate the TriggerSpec.
 func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 	var errs *apis.FieldError
 	if ts.Broker == "" {
@@ -39,9 +41,33 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe)
 	}
 
-	if ts.Filter != nil && ts.Filter.SourceAndType == nil {
-		fe := apis.ErrMissingField("filter.sourceAndType")
-		errs = errs.Also(fe)
+	if ts.Filter != nil {
+		filtersSpecified := []string{}
+
+		if ts.Filter.DeprecatedSourceAndType != nil {
+			filtersSpecified = append(filtersSpecified, "filter.sourceAndType")
+		}
+
+		if ts.Filter.Attributes != nil {
+			filtersSpecified = append(filtersSpecified, "filter.attributes")
+			if len(*ts.Filter.Attributes) == 0 {
+				fe := &apis.FieldError{
+					Message: "At least one filtered attribute must be specified",
+					Paths:   []string{"filter.attributes"},
+				}
+				errs = errs.Also(fe)
+			}
+		}
+
+		if ts.Filter.Expression != nil {
+			filtersSpecified = append(filtersSpecified, "filter.expression")
+			// TODO validate expression here
+		}
+
+		if len(filtersSpecified) > 1 {
+			fe := apis.ErrMultipleOneOf(filtersSpecified...)
+			errs = errs.Also(fe)
+		}
 	}
 
 	if isSubscriberSpecNilOrEmpty(ts.Subscriber) {
@@ -54,6 +80,7 @@ func (ts *TriggerSpec) Validate(ctx context.Context) *apis.FieldError {
 	return errs
 }
 
+// CheckImmutableFields checks that any immutable fields were not changed.
 func (t *Trigger) CheckImmutableFields(ctx context.Context, og apis.Immutable) *apis.FieldError {
 	if og == nil {
 		return nil