knative-extensions · knative-prow · Aug 28, 2024 · Aug 22, 2024 · Aug 22, 2024 · Aug 26, 2024
diff --git a/control-plane/pkg/core/config/utils.go b/control-plane/pkg/core/config/utils.go
@@ -90,6 +90,10 @@ func EventPoliciesFromAppliedEventPoliciesStatus(status duck.AppliedEventPolicie
 			}
 		}
 
+		for _, filter := range policy.Spec.Filters {
+			contractPolicy.Filters = append(contractPolicy.Filters, contract.FromSubscriptionFilter(filter))
+		}
+
 		eventPolicies = append(eventPolicies, contractPolicy)
 	}
 

diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
@@ -23,6 +23,8 @@ import (
 	"testing"
 	"time"
 
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+
 	"google.golang.org/protobuf/encoding/protojson"
 	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing/pkg/apis/feature"
@@ -618,6 +620,80 @@ func TestEventPoliciesFromAppliedEventPoliciesStatus(t *testing.T) {
 					},
 				},
 			},
+		}, {
+			name: "Multiple policies with filters",
+			applyingPolicies: []string{
+				"policy-1",
+				"policy-2",
+			},
+			existingEventPolicies: []*eventingv1alpha1.EventPolicy{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "policy-1",
+						Namespace: "my-ns",
+					},
+					Spec: eventingv1alpha1.EventPolicySpec{
+						Filters: []eventingv1.SubscriptionsAPIFilter{
+							{
+								CESQL: "true",
+							},
+						},
+					},
+					Status: eventingv1alpha1.EventPolicyStatus{
+						From: []string{
+							"from-1",
+						},
+					},
+				}, {
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "policy-2",
+						Namespace: "my-ns",
+					},
+					Spec: eventingv1alpha1.EventPolicySpec{
+						Filters: []eventingv1.SubscriptionsAPIFilter{
+							{
+								CESQL: "false",
+							},
+						},
+					},
+					Status: eventingv1alpha1.EventPolicyStatus{
+						From: []string{
+							"from-2-*",
+						},
+					},
+				},
+			},
+			namespace:                "my-ns",
+			defaultAuthorizationMode: feature.AuthorizationDenyAll,
+			expected: []*contract.EventPolicy{
+				{
+					TokenMatchers: []*contract.TokenMatcher{
+						exactTokenMatcher("from-1"),
+					},
+					Filters: []*contract.DialectedFilter{
+						{
+							Filter: &contract.DialectedFilter_Cesql{
+								Cesql: &contract.CESQL{
+									Expression: "true",
+								},
+							},
+						},
+					},
+				}, {
+					TokenMatchers: []*contract.TokenMatcher{
+						prefixTokenMatcher("from-2-"),
+					},
+					Filters: []*contract.DialectedFilter{
+						{
+							Filter: &contract.DialectedFilter_Cesql{
+								Cesql: &contract.CESQL{
+									Expression: "false",
+								},
+							},
+						},
+					},
+				},
+			},
 		}, {
 			name:                     "No applying policies - allow-same-namespace default mode",
 			applyingPolicies:         []string{},

diff --git a/data-plane/benchmarks/pom.xml b/data-plane/benchmarks/pom.xml
@@ -54,6 +54,12 @@
       <artifactId>dispatcher</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>dev.knative.eventing.kafka.broker</groupId>
+      <artifactId>core</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
   </dependencies>
 
   <build>

diff --git a/...tcher/impl/filter/AllFilterBenchmark.java → ...roker/core/filter/AllFilterBenchmark.java b/...tcher/impl/filter/AllFilterBenchmark.java → ...roker/core/filter/AllFilterBenchmark.java
@@ -14,13 +14,12 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.AllFilter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.ExactFilter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.PrefixFilter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.SuffixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.AllFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.ExactFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.PrefixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.SuffixFilter;
 import io.cloudevents.CloudEvent;
 import java.util.List;
 import java.util.Map;

diff --git a/...tcher/impl/filter/AnyFilterBenchmark.java → ...roker/core/filter/AnyFilterBenchmark.java b/...tcher/impl/filter/AnyFilterBenchmark.java → ...roker/core/filter/AnyFilterBenchmark.java
@@ -14,10 +14,12 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.*;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.AnyFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.ExactFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.PrefixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.SuffixFilter;
 import io.cloudevents.CloudEvent;
 import java.util.List;
 import java.util.Map;

diff --git a/...her/impl/filter/ExactFilterBenchmark.java → ...ker/core/filter/ExactFilterBenchmark.java b/...her/impl/filter/ExactFilterBenchmark.java → ...ker/core/filter/ExactFilterBenchmark.java
@@ -14,10 +14,9 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.ExactFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.ExactFilter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
 import io.cloudevents.core.v1.CloudEventV1;

diff --git a/...spatcher/impl/filter/FilterBenchmark.java → ...a/broker/core/filter/FilterBenchmark.java b/...spatcher/impl/filter/FilterBenchmark.java → ...a/broker/core/filter/FilterBenchmark.java
@@ -14,9 +14,8 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import io.cloudevents.CloudEvent;
 import java.util.concurrent.TimeUnit;
 import org.openjdk.jmh.annotations.*;

diff --git a/...tcher/impl/filter/NotFilterBenchmark.java → ...roker/core/filter/NotFilterBenchmark.java b/...tcher/impl/filter/NotFilterBenchmark.java → ...roker/core/filter/NotFilterBenchmark.java
@@ -14,13 +14,12 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.ExactFilter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.NotFilter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.PrefixFilter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.SuffixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.ExactFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.NotFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.PrefixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.SuffixFilter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.v1.CloudEventV1;
 import java.util.Map;

diff --git a/...er/impl/filter/PrefixFilterBenchmark.java → ...er/core/filter/PrefixFilterBenchmark.java b/...er/impl/filter/PrefixFilterBenchmark.java → ...er/core/filter/PrefixFilterBenchmark.java
@@ -14,10 +14,9 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.PrefixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.PrefixFilter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.v1.CloudEventV1;
 import java.util.Map;

diff --git a/...r/dispatcher/impl/filter/SampleEvent.java → ...kafka/broker/core/filter/SampleEvent.java b/...r/dispatcher/impl/filter/SampleEvent.java → ...kafka/broker/core/filter/SampleEvent.java
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;

diff --git a/...er/impl/filter/SuffixFilterBenchmark.java → ...er/core/filter/SuffixFilterBenchmark.java b/...er/impl/filter/SuffixFilterBenchmark.java → ...er/core/filter/SuffixFilterBenchmark.java
@@ -14,10 +14,9 @@
  * limitations under the License.
  */
 
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi.SuffixFilter;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.SuffixFilter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.v1.CloudEventV1;
 import java.util.Map;

diff --git a/data-plane/core/pom.xml b/data-plane/core/pom.xml
@@ -146,6 +146,10 @@
       <artifactId>vertx-opentelemetry</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.cloudevents</groupId>
+      <artifactId>cloudevents-sql</artifactId>
+    </dependency>
     <dependency>
       <groupId>io.cloudevents</groupId>
       <artifactId>cloudevents-kafka</artifactId>

diff --git a/...patcher/impl/filter/AttributesFilter.java → .../broker/core/filter/AttributesFilter.java b/...patcher/impl/filter/AttributesFilter.java → .../broker/core/filter/AttributesFilter.java
@@ -13,11 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter;
+package dev.knative.eventing.kafka.broker.core.filter;
 
 import static java.time.format.DateTimeFormatter.ISO_INSTANT;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.v03.CloudEventV03;
 import io.cloudevents.core.v1.CloudEventV1;

diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/filter/Filter.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/filter/Filter.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.core.filter;
+
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi.*;
+import io.cloudevents.CloudEvent;
+import java.util.List;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+/**
+ * This interface provides an abstraction for filtering {@link CloudEvent} instances.
+ */
+@FunctionalInterface
+public interface Filter extends Predicate<CloudEvent> {
+
+    /**
+     * @return noop implementation that always returns true
+     */
+    static Filter noop() {
+        return ce -> true;
+    }
+
+    static Filter fromContract(DataPlaneContract.DialectedFilter filter) {
+        return switch (filter.getFilterCase()) {
+            case EXACT -> new ExactFilter(filter.getExact().getAttributesMap());
+            case PREFIX -> new PrefixFilter(filter.getPrefix().getAttributesMap());
+            case SUFFIX -> new SuffixFilter(filter.getSuffix().getAttributesMap());
+            case NOT -> new NotFilter(fromContract(filter.getNot().getFilter()));
+            case ANY -> new AnyFilter(filter.getAny().getFiltersList().stream()
+                    .map(Filter::fromContract)
+                    .collect(Collectors.toList()));
+            case ALL -> new AllFilter(filter.getAll().getFiltersList().stream()
+                    .map(Filter::fromContract)
+                    .collect(Collectors.toList()));
+            case CESQL -> new CeSqlFilter(filter.getCesql().getExpression());
+            default -> Filter.noop();
+        };
+    }
+
+    static Filter fromContract(List<DataPlaneContract.DialectedFilter> filters) {
+        return new AllFilter(filters.stream().map(Filter::fromContract).collect(Collectors.toList()));
+    }
+}
diff --git a/...pl/filter/subscriptionsapi/AllFilter.java → ...re/filter/subscriptionsapi/AllFilter.java b/...pl/filter/subscriptionsapi/AllFilter.java → ...re/filter/subscriptionsapi/AllFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
+import dev.knative.eventing.kafka.broker.core.filter.Filter;
 import io.cloudevents.CloudEvent;
 import java.util.List;
 import org.slf4j.Logger;

diff --git a/...pl/filter/subscriptionsapi/AnyFilter.java → ...re/filter/subscriptionsapi/AnyFilter.java b/...pl/filter/subscriptionsapi/AnyFilter.java → ...re/filter/subscriptionsapi/AnyFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
+import dev.knative.eventing.kafka.broker.core.filter.Filter;
 import io.cloudevents.CloudEvent;
 import java.util.List;
 import org.slf4j.Logger;

diff --git a/.../filter/subscriptionsapi/CeSqlFilter.java → .../filter/subscriptionsapi/CeSqlFilter.java b/.../filter/subscriptionsapi/CeSqlFilter.java → .../filter/subscriptionsapi/CeSqlFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
+import dev.knative.eventing.kafka.broker.core.filter.Filter;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.sql.EvaluationContext;
 import io.cloudevents.sql.EvaluationException;

diff --git a/.../filter/subscriptionsapi/ExactFilter.java → .../filter/subscriptionsapi/ExactFilter.java b/.../filter/subscriptionsapi/ExactFilter.java → .../filter/subscriptionsapi/ExactFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.AttributesFilter;
+import dev.knative.eventing.kafka.broker.core.filter.AttributesFilter;
 import java.util.Map;
 
 public class ExactFilter extends AttributesFilter {

diff --git a/...pl/filter/subscriptionsapi/NotFilter.java → ...re/filter/subscriptionsapi/NotFilter.java b/...pl/filter/subscriptionsapi/NotFilter.java → ...re/filter/subscriptionsapi/NotFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.Filter;
+import dev.knative.eventing.kafka.broker.core.filter.Filter;
 import io.cloudevents.CloudEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

diff --git a/...filter/subscriptionsapi/PrefixFilter.java → ...filter/subscriptionsapi/PrefixFilter.java b/...filter/subscriptionsapi/PrefixFilter.java → ...filter/subscriptionsapi/PrefixFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.AttributesFilter;
+import dev.knative.eventing.kafka.broker.core.filter.AttributesFilter;
 import java.util.Map;
 
 public class PrefixFilter extends AttributesFilter {

diff --git a/...filter/subscriptionsapi/SuffixFilter.java → ...filter/subscriptionsapi/SuffixFilter.java b/...filter/subscriptionsapi/SuffixFilter.java → ...filter/subscriptionsapi/SuffixFilter.java
@@ -13,9 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package dev.knative.eventing.kafka.broker.dispatcher.impl.filter.subscriptionsapi;
+package dev.knative.eventing.kafka.broker.core.filter.subscriptionsapi;
 
-import dev.knative.eventing.kafka.broker.dispatcher.impl.filter.AttributesFilter;
+import dev.knative.eventing.kafka.broker.core.filter.AttributesFilter;
 import java.util.Map;
 
 public class SuffixFilter extends AttributesFilter {
-Original file line number
+Diff line change
@@ Expand Up @@
     			}
     		}
+    		for _, filter := range policy.Spec.Filters {
+    			contractPolicy.Filters = append(contractPolicy.Filters, contract.FromSubscriptionFilter(filter))
+    		}
     		eventPolicies = append(eventPolicies, contractPolicy)
     	}
@@ Expand Down @@