Implement new trigger filters experimental feature (SubscriptionsAPI Filters) #1922

devguyio · 2022-02-17T02:56:42Z

Implementation of the new filters experimental feature

Based on #1921 , #1977 & #1978

Fixes #1763

Proposed Changes

Implement SubscriptionsAPI Filters / aka new trigger filters experimental feature.

TODO

Contract tests

Release Note

- 🎁 Add new `new-trigger-filters` experimental feature. When enabled, Triggers support a new `filters` field that conforms to the filters API field defined in the [`CloudEvents Subscriptions API`](https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#324-filters). It allows you to specify a set of powerful filter expressions, where each expression evaluates to either true or false for each event.

devguyio · 2022-02-17T02:57:42Z

@matzew @aliok @pierDipi you can follow the commits flow for reviewing.

codecov · 2022-02-17T03:01:17Z

Codecov Report

Merging #1922 (d78f278) into main (4d8f68b) will increase coverage by 0.24%.
The diff coverage is 78.23%.

@@             Coverage Diff              @@
##               main    #1922      +/-   ##
============================================
+ Coverage     66.98%   67.23%   +0.24%     
- Complexity      630      632       +2     
============================================
  Files           140      141       +1     
  Lines          6567     6687     +120     
  Branches        179      180       +1     
============================================
+ Hits           4399     4496      +97     
- Misses         1800     1823      +23     
  Partials        368      368

Flag	Coverage Δ
java-unittests	`81.97% <78.12%> (-0.26%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
control-plane/pkg/contract/contract.pb.go	`8.66% <20.00%> (+0.06%)`	⬆️
.../core/reconciler/impl/ResourcesReconcilerImpl.java	`90.20% <50.00%> (+0.06%)`	⬆️
...r/dispatcher/main/ConsumerVerticleFactoryImpl.java	`77.27% <60.00%> (-10.66%)`	⬇️
control-plane/pkg/contract/subscriptionsapi.go	`77.94% <77.94%> (ø)`
control-plane/pkg/reconciler/trigger/controller.go	`85.00% <83.33%> (-0.30%)`	⬇️
control-plane/pkg/reconciler/trigger/trigger.go	`73.41% <100.00%> (+1.60%)`	⬆️
...roker/dispatcher/impl/filter/AttributesFilter.java	`90.00% <100.00%> (+4.81%)`	⬆️
...atcher/impl/filter/subscriptionsapi/AllFilter.java	`100.00% <100.00%> (ø)`
...atcher/impl/filter/subscriptionsapi/AnyFilter.java	`100.00% <100.00%> (ø)`
...cher/impl/filter/subscriptionsapi/CeSqlFilter.java	`100.00% <100.00%> (ø)`
... and 5 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 4d8f68b...d78f278. Read the comment docs.

aliok · 2022-02-17T06:42:50Z

/rebase

matzew · 2022-02-17T07:13:19Z

/rebase

does that really work ? 😂

...ava/dev/knative/eventing/kafka/broker/dispatcher/impl/filter/subscriptionsapi/AllFilter.java

...ain/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java

matzew · 2022-02-17T08:01:07Z

I think, as said on the other PR this looks good

I had only a nit, do to import/format changes.

control-plane/pkg/reconciler/trigger/trigger.go

...main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java

.github/workflows/kind-e2e.yaml

control-plane/pkg/reconciler/trigger/trigger_test.go

test/experimental/main_test.go

test/experimental/features_config/install.go

control-plane/pkg/contract/subscriptionsapi.go

matzew · 2022-02-22T09:55:47Z

LGTM, as already stated on the other PR.

pierDipi · 2022-02-17T17:24:52Z

control-plane/pkg/contract/subscriptionsapi.go

+// https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#any-filter-dialect
+func NewAnyFilter(filters []v1.SubscriptionsAPIFilter) *DialectedFilter {
+	any := Any{
+		Filters: []*DialectedFilter{},


Suggested change

Filters: []*DialectedFilter{},

Filters: make(DialectedFilter, 0, len(filters)),

pierDipi · 2022-02-17T17:25:12Z

control-plane/pkg/contract/subscriptionsapi.go

+// https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#all-filter-dialect
+func NewAllFilter(filters []v1.SubscriptionsAPIFilter) *DialectedFilter {
+	all := All{
+		Filters: []*DialectedFilter{},


We can pre-allocate this array.

pierDipi · 2022-02-22T10:06:10Z

control-plane/pkg/contract/subscriptionsapi.go

+// NewExactFilter converts the SubscriptionsAPIFilter into the exact dialect of
+// the DialectedFilter as defined in CloudEvents Subscriptions API.
+//
+// Exact contains exactly one attribute where the key is the name of the CloudEvent
+// attribute and its value is the string value which must exactly match the value
+// of the CloudEvent attribute.
+//
+// See "CNCF CloudEvents Subscriptions API" > "3.2.4.1 Filters Dialects"
+// https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#all-filter-dialect
+func NewExactFilter(f v1.SubscriptionsAPIFilter) *DialectedFilter {
+	return &DialectedFilter{
+		Filter: &DialectedFilter_Exact{
+			Exact: &Exact{
+				Attributes: f.Exact,
+			},
+		},
+	}
+}
+
+// NewPrefixFilter converts the SubscriptionsAPIFilter into the suffix dialect of
+// the DialectedFilter as defined in CloudEvents Subscriptions API.
+//
+// Prefix contains exactly one attribute where the key is the name of the CloudEvent
+// attribute which value must start with the value specified.
+//
+// See "CNCF CloudEvents Subscriptions API" > "3.2.4.1 Filters Dialects"
+// https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#prefix-filter-dialect
+func NewPrefixFilter(f v1.SubscriptionsAPIFilter) *DialectedFilter {
+
+	p := &Prefix{}
+	// Eventing Trigger API is conforming to the Subscriptions API, so this map has only a single
+	// key-value pair
+	for k, v := range f.Prefix {
+		p.Attribute = k
+		p.Prefix = v
+		// We only expect a single element, but let's break anyway
+		break
+	}
+
+	return &DialectedFilter{
+		Filter: &DialectedFilter_Prefix{
+			Prefix: p,
+		},
+	}
+}
+
+// NewSuffixFilter converts the SubscriptionsAPIFilter into the suffix dialect of
+// the DialectedFilter as defined in CloudEvents Subscriptions API.
+//
+// Suffix contains exactly one attribute where the key is the name of the CloudEvent
+// attribute which value must end with the value specified.
+//
+// See "CNCF CloudEvents Subscriptions API" > "3.2.4.1 Filters Dialects"
+// https://github.com/cloudevents/spec/blob/main/subscriptions/spec.md#suffix-filter-dialect
+func NewSuffixFilter(f v1.SubscriptionsAPIFilter) *DialectedFilter {
+
+	p := &Suffix{}
+	// Eventing Trigger API is conforming to the Subscriptions API, so this map has only a single
+	// key-value pair
+	for k, v := range f.Suffix {
+		p.Attribute = k
+		p.Suffix = v
+		// We only expect a single element, but let's break anyway
+		break
+	}
+
+	return &DialectedFilter{
+		Filter: &DialectedFilter_Suffix{
+			Suffix: p,
+		},
+	}
+}


As the other comment on the PR, there shouldn't be a difference between these 3 filters.

pierDipi · 2022-02-22T10:07:49Z

control-plane/pkg/contract/subscriptionsapi.go

+func FromSubscriptionFilter(f v1.SubscriptionsAPIFilter) *DialectedFilter {
+	switch {
+	case len(f.All) > 0:
+		return NewAllFilter(f.All)
+	case len(f.Any) > 0:
+		return NewAnyFilter(f.Any)
+	case f.Not != nil:
+		return NewNotFilter(*f.Not)
+	case f.Exact != nil:
+		return NewExactFilter(f)
+	case f.Prefix != nil:
+		return NewPrefixFilter(f)
+	case f.Suffix != nil:
+		return NewSuffixFilter(f)
+	case f.SQL != "":
+		return NewCESQLFilter(f)
+	}
+	return &DialectedFilter{}
+}


Can we decouple the one filter constraint from the contract?
I remember making this comment in another PR 🤔

In addition, the CE subscription spec doesn't have this constraint, so I would really avoid designing a contract in this way

Can we decouple the one filter constraint from the contract? I remember making this comment in another PR thinking

Yes I'm working on it.

In addition, the CE subscription spec doesn't have this constraint

Yeah while it's not in the spec itself, it's in the reference openapi file and I remember that it was brought up in one of the PRs introducing the spec. But I agree that we shouldn't have it tightly coupled here.

...dev/knative/eventing/kafka/broker/dispatcher/impl/filter/subscriptionsapi/SqlFilterTest.java

.../knative/eventing/kafka/broker/dispatcher/impl/filter/subscriptionsapi/SuffixFilterTest.java

.../knative/eventing/kafka/broker/dispatcher/impl/filter/subscriptionsapi/PrefixFilterTest.java

...v/knative/eventing/kafka/broker/dispatcher/impl/filter/subscriptionsapi/ExactFilterTest.java

...ain/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java

devguyio · 2022-03-09T19:07:42Z

/unhold

Signed-off-by: Ahmed Abdalla <aabdelre@redhat.com>

knative-metrics-robot · 2022-03-09T19:57:11Z

The following is the coverage report on the affected files.
Say /test pull-knative-sandbox-eventing-kafka-broker-go-coverage to re-run this coverage report

File	Old Coverage	New Coverage	Delta
control-plane/pkg/contract/subscriptionsapi.go	Do not exist	81.2%
control-plane/pkg/reconciler/trigger/controller.go	76.9%	79.6%	2.7
control-plane/pkg/reconciler/trigger/trigger.go	84.7%	85.8%	1.1

devguyio · 2022-03-09T20:46:44Z

Seems like the webhook was unavailable

    broker_trigger_sink_test.go:77: 
        	Error Trace:	broker_trigger_sink_test.go:77
        	Error:      	Expected nil, but got: &errors.StatusError{ErrStatus:v1.Status{TypeMeta:v1.TypeMeta{Kind:"", APIVersion:""}, ListMeta:v1.ListMeta{SelfLink:"", ResourceVersion:"", Continue:"", RemainingItemCount:(*int64)(nil)}, Status:"Failure", Message:"Internal error occurred: failed calling webhook \"validation.webhook.kafka.eventing.knative.dev\": failed to call webhook: Post \"https://kafka-webhook-eventing.knative-eventing.svc:443/resource-validation?timeout=2s\": EOF", Reason:"InternalError", Details:(*v1.StatusDetails)(0xc007792c00), Code:500}}
        	Test:       	TestBrokerV1TriggersV1SinkV1Alpha1/5

/retest

aliok

/lgtm
/approve

knative-prow-robot · 2022-03-10T06:47:06Z

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: aliok, devguyio, pierDipi

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Needs approval from an approver in each of these files:

~~OWNERS~~ [aliok,devguyio,pierDipi]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

devguyio · 2022-03-16T09:47:13Z

/cherry-pick release-1.3

knative-prow-robot · 2022-03-16T09:47:54Z

@devguyio: new pull request created: #2008

In response to this:

/cherry-pick release-1.3

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

knative-prow-robot added the area/control-plane label Feb 17, 2022

devguyio requested a review from matzew February 17, 2022 02:56

knative-prow-robot added area/data-plane area/test approved Indicates a PR has been approved by an approver from all required OWNERS files. size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. labels Feb 17, 2022

devguyio requested review from pierDipi and aliok February 17, 2022 02:56

matzew reviewed Feb 17, 2022

View reviewed changes

...ava/dev/knative/eventing/kafka/broker/dispatcher/impl/filter/subscriptionsapi/AllFilter.java Show resolved Hide resolved

matzew reviewed Feb 17, 2022

View reviewed changes

...ain/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java Outdated Show resolved Hide resolved

devguyio force-pushed the new-filters branch from 9fa00ee to d730363 Compare February 17, 2022 10:03

knative-prow-robot added size/XL Denotes a PR that changes 500-999 lines, ignoring generated files. and removed size/XXL Denotes a PR that changes 1000+ lines, ignoring generated files. labels Feb 17, 2022