Send namespace header in MT components

When running MT components in mesh mode with Istio, we lose
the ability to define fine grained policies since we don't
know the resource namespace that originated such request,
therefore, by having a `Kn-Namespace` header, in mesh mode,
users case define fine-grained policies and isolate namespaces.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

pkg/adapter/mtping/runner.go

@@ -192,7 +192,7 @@ func (a *cronJobsRunner) newPingSourceClient(source *sourcesv1.PingSource) (adap
 	var env adapter.EnvConfig
 	if a.clientConfig.Env != nil {
 		env = adapter.EnvConfig{
-			Namespace:      a.clientConfig.Env.GetNamespace(),
+			Namespace:      source.GetNamespace(),
 			Name:           a.clientConfig.Env.GetName(),
 			EnvSinkTimeout: fmt.Sprintf("%d", a.clientConfig.Env.GetSinktimeout()),
 		}

pkg/adapter/v2/cloudevents.go

@@ -35,6 +35,7 @@ import (
 	"knative.dev/pkg/tracing/propagation/tracecontextb3"
 
 	"knative.dev/eventing/pkg/adapter/v2/util/crstatusevent"
+	"knative.dev/eventing/pkg/apis"
 	"knative.dev/eventing/pkg/eventingtls"
 	"knative.dev/eventing/pkg/metrics/source"
 	obsclient "knative.dev/eventing/pkg/observability/client"
@@ -171,6 +172,8 @@ func NewClient(cfg ClientConfig) (Client, error) {
 				return nil, err
 			}
 		}
+
+		pOpts = append(pOpts, http.WithHeader(apis.KnNamespaceHeader, cfg.Env.GetNamespace()))
 	}
 
 	pOpts = append(pOpts, http.WithRoundTripper(transport))

pkg/apis/http.go

@@ -0,0 +1,21 @@
+/*
+Copyright 2023 The Knative Authors
+
+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 apis
+
+const (
+	KnNamespaceHeader = "Kn-Namespace"
+)

pkg/broker/filter/filter_handler.go

@@ -34,13 +34,15 @@ import (
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	"k8s.io/client-go/tools/cache"
-	channelAttributes "knative.dev/eventing/pkg/channel/attributes"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
 
+	"knative.dev/eventing/pkg/apis"
+	channelAttributes "knative.dev/eventing/pkg/channel/attributes"
+
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/feature"
-	broker "knative.dev/eventing/pkg/broker"
+	"knative.dev/eventing/pkg/broker"
 	v1 "knative.dev/eventing/pkg/client/informers/externalversions/eventing/v1"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 	"knative.dev/eventing/pkg/eventfilter"
@@ -240,12 +242,12 @@ func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 		URL:     t.Status.SubscriberURI,
 		CACerts: t.Status.SubscriberCACerts,
 	}
-	h.send(ctx, writer, request.Header, target, reportArgs, event, ttl)
+	h.send(ctx, writer, request.Header, target, reportArgs, event, t, ttl)
 }
 
-func (h *Handler) send(ctx context.Context, writer http.ResponseWriter, headers http.Header, target duckv1.Addressable, reportArgs *ReportArgs, event *cloudevents.Event, ttl int32) {
+func (h *Handler) send(ctx context.Context, writer http.ResponseWriter, headers http.Header, target duckv1.Addressable, reportArgs *ReportArgs, event *cloudevents.Event, t *eventingv1.Trigger, ttl int32) {
 	// send the event to trigger's subscriber
-	response, responseErr := h.sendEvent(ctx, headers, target, event, reportArgs)
+	response, responseErr := h.sendEvent(ctx, headers, target, event, t, reportArgs)
 
 	if responseErr.err != nil {
 		h.logger.Error("failed to send event", zap.Error(responseErr.err))
@@ -288,7 +290,7 @@ func (h *Handler) send(ctx context.Context, writer http.ResponseWriter, headers
 	_ = h.reporter.ReportEventCount(reportArgs, statusCode)
 }
 
-func (h *Handler) sendEvent(ctx context.Context, headers http.Header, target duckv1.Addressable, event *cloudevents.Event, reporterArgs *ReportArgs) (*http.Response, ErrHandler) {
+func (h *Handler) sendEvent(ctx context.Context, headers http.Header, target duckv1.Addressable, event *cloudevents.Event, t *eventingv1.Trigger, reporterArgs *ReportArgs) (*http.Response, ErrHandler) {
 	responseErr := ErrHandler{
 		ResponseCode: NoResponse,
 	}
@@ -304,6 +306,7 @@ func (h *Handler) sendEvent(ctx context.Context, headers http.Header, target duc
 	defer message.Finish(nil)
 
 	additionalHeaders := utils.PassThroughHeaders(headers)
+	additionalHeaders.Set(apis.KnNamespaceHeader, t.GetNamespace())
 
 	// Following the spec https://github.com/knative/specs/blob/main/specs/eventing/data-plane.md#derived-reply-events
 	additionalHeaders.Set("prefer", "reply")

pkg/channel/fanout/fanout_message_handler.go

@@ -32,10 +32,12 @@ import (
 	"github.com/cloudevents/sdk-go/v2/binding/buffering"
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+
+	"knative.dev/eventing/pkg/apis"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/eventing/pkg/channel"
 	"knative.dev/eventing/pkg/kncloudevents"
-	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
 const (
@@ -191,6 +193,7 @@ func createMessageReceiverFunction(f *FanoutMessageHandler) func(context.Context
 			go func(m binding.Message, h nethttp.Header, s *trace.Span, r *channel.StatsReporter, args *channel.ReportArgs) {
 				// Run async dispatch with background context.
 				ctx = trace.NewContext(context.Background(), s)
+				h.Set(apis.KnNamespaceHeader, ref.Namespace)
 				// Any returned error is already logged in f.dispatch().
 				dispatchResultForFanout := f.dispatch(ctx, subs, m, h)
 				_ = ParseDispatchResultAndReportMetrics(dispatchResultForFanout, *r, *args)

pkg/channel/message_dispatcher.go

@@ -38,6 +38,8 @@ import (
 	"knative.dev/pkg/network"
 	"knative.dev/pkg/system"
 
+	eventingapis "knative.dev/eventing/pkg/apis"
+
 	"knative.dev/eventing/pkg/broker"
 	"knative.dev/eventing/pkg/channel/attributes"
 	"knative.dev/eventing/pkg/kncloudevents"
@@ -148,6 +150,8 @@ func (d *MessageDispatcherImpl) DispatchMessageWithRetries(ctx context.Context,
 		responseAdditionalHeaders = additionalHeaders
 	}
 
+	responseAdditionalHeaders.Set(eventingapis.KnNamespaceHeader, additionalHeaders.Get(eventingapis.KnNamespaceHeader))
+
 	// No response, dispatch completed
 	if responseMessage == nil {
 		return dispatchExecutionInfo, nil

test/rekt/features/broker/feature.go

@@ -26,9 +26,11 @@ import (
 	"github.com/cloudevents/sdk-go/v2/binding/spec"
 	"github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
+	"knative.dev/reconciler-test/pkg/environment"
 
 	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/eventing/test/rekt/features"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/channel"
 	"knative.dev/eventing/test/rekt/resources/subscription"
@@ -186,10 +188,12 @@ func BrokerWithManyTriggers() *feature.Feature {
 			for _, matcher := range matchers {
 				// One match per event is enough
 				f.Stable("test message without explicit prefer header should have the header").
-					Must("delivers events",
-						eventasssert.OnStore(sink).Match(
-							matcher,
-						).AtLeast(1))
+					Must("delivers events", func(ctx context.Context, t feature.T) {
+						eventasssert.OnStore(sink).
+							Match(features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace())).
+							Match(matcher).
+							AtLeast(1)(ctx, t)
+					})
 			}
 		}
 	}
@@ -628,14 +632,19 @@ func brokerRedeliveryDropN(retryNum int32, dropNum uint) *feature.Feature {
 
 	f.Stable("Broker Redelivery failed the first n events").
 		Must("delivers events",
-			eventasssert.OnStore(sink).Match(
-				eventasssert.MatchKind(eventasssert.EventReceived),
-				eventasssert.MatchEvent(
-					test.HasSource(eventSource),
-					test.HasType(eventType),
-					test.HasData([]byte(eventBody)),
-				),
-			).AtLeast(1))
+			func(ctx context.Context, t feature.T) {
+				eventasssert.OnStore(sink).
+					Match(features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace())).
+					Match(
+						eventasssert.MatchKind(eventasssert.EventReceived),
+						eventasssert.MatchEvent(
+							test.HasSource(eventSource),
+							test.HasType(eventType),
+							test.HasData([]byte(eventBody)),
+						),
+					).
+					AtLeast(1)(ctx, t)
+			})
 
 	return f
 }
@@ -693,11 +702,14 @@ func brokerSubscriberUnreachable() *feature.Feature {
 	))
 
 	f.Assert("Receives dls extensions when subscriber is unreachable",
-		eventasssert.OnStore(sink).
-			MatchEvent(
-				test.HasExtension("knativeerrordest", "http://fake.svc.cluster.local"),
-			).
-			AtLeast(1),
+		func(ctx context.Context, t feature.T) {
+			eventasssert.OnStore(sink).
+				Match(features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace())).
+				MatchEvent(
+					test.HasExtension("knativeerrordest", "http://fake.svc.cluster.local"),
+				).
+				AtLeast(1)(ctx, t)
+		},
 	)
 	return f
 }

test/rekt/features/channel/features.go

@@ -26,6 +26,7 @@ import (
 	"github.com/cloudevents/sdk-go/v2/test"
 	"github.com/google/uuid"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
@@ -35,6 +36,7 @@ import (
 
 	eventasssert "knative.dev/reconciler-test/pkg/eventshub/assert"
 
+	"knative.dev/eventing/test/rekt/features"
 	"knative.dev/eventing/test/rekt/resources/channel"
 	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/containersource"
@@ -107,10 +109,12 @@ func DeadLetterSink(createSubscriberFn func(ref *duckv1.KReference, uri string)
 	f.Requirement("containersource is ready", containersource.IsReady(cs))
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(name, channel_impl.GVR()))
 
-	f.Assert("dls receives events", assert.OnStore(sink).
-		MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).
-		AtLeast(1),
-	)
+	f.Assert("dls receives events", func(ctx context.Context, t feature.T) {
+		assert.OnStore(sink).
+			Match(features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace())).
+			MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).
+			AtLeast(1)(ctx, t)
+	})
 
 	return f
 }
@@ -140,10 +144,12 @@ func DeadLetterSinkGenericChannel(createSubscriberFn func(ref *duckv1.KReference
 	f.Requirement("containersource is ready", containersource.IsReady(cs))
 	f.Requirement("Channel has dead letter sink uri", channel_impl.HasDeadLetterSinkURI(name, channel.GVR()))
 
-	f.Assert("dls receives events", assert.OnStore(sink).
-		MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).
-		AtLeast(1),
-	)
+	f.Assert("dls receives events", func(ctx context.Context, t feature.T) {
+		assert.OnStore(sink).
+			Match(features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace())).
+			MatchEvent(test.HasType("dev.knative.eventing.samples.heartbeat")).
+			AtLeast(1)(ctx, t)
+	})
 
 	return f
 }
@@ -306,10 +312,12 @@ func ChannelPreferHeaderCheck(createSubscriberFn func(ref *duckv1.KReference, ur
 	))
 
 	f.Stable("test message without explicit prefer header should have the header").
-		Must("delivers events",
+		Must("delivers events", func(ctx context.Context, t feature.T) {
 			eventasssert.OnStore(sink).Match(
+				features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace()),
 				eventasssert.HasAdditionalHeader("Prefer", "reply"),
-			).AtLeast(1))
+			).AtLeast(1)(ctx, t)
+		})
 
 	return f
 }

test/rekt/features/matchers.go

@@ -0,0 +1,40 @@
+/*
+Copyright 2023 The Knative Authors
+
+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 features
+
+import (
+	"fmt"
+
+	"knative.dev/reconciler-test/pkg/eventshub"
+
+	"knative.dev/eventing/pkg/apis"
+)
+
+func HasKnNamespaceHeader(ns string) eventshub.EventInfoMatcher {
+	return func(info eventshub.EventInfo) error {
+		values, ok := info.HTTPHeaders[apis.KnNamespaceHeader]
+		if !ok {
+			return fmt.Errorf("%s header not found", apis.KnNamespaceHeader)
+		}
+		for _, v := range values {
+			if v == ns {
+				return nil
+			}
+		}
+		return fmt.Errorf("wanted %s header to have value %s, got %+v", apis.KnNamespaceHeader, ns, values)
+	}
+}

test/rekt/features/pingsource/features.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cloudevents/sdk-go/v2/test"
 	"k8s.io/apimachinery/pkg/util/sets"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/reconciler-test/pkg/environment"
 	"knative.dev/reconciler-test/pkg/eventshub"
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/manifest"
@@ -31,6 +32,7 @@ import (
 	eventassert "knative.dev/reconciler-test/pkg/eventshub/assert"
 
 	sourcesv1 "knative.dev/eventing/pkg/apis/sources/v1"
+	"knative.dev/eventing/test/rekt/features"
 	"knative.dev/eventing/test/rekt/features/featureflags"
 	"knative.dev/eventing/test/rekt/features/source"
 	"knative.dev/eventing/test/rekt/resources/broker"
@@ -51,7 +53,13 @@ func SendsEventsWithSinkRef() *feature.Feature {
 
 	f.Stable("pingsource as event source").
 		Must("delivers events",
-			assert.OnStore(sink).MatchEvent(test.HasType("dev.knative.sources.ping")).AtLeast(1))
+			func(ctx context.Context, t feature.T) {
+				assert.OnStore(sink).
+					Match(features.HasKnNamespaceHeader(environment.FromContext(ctx).Namespace())).
+					MatchEvent(test.HasType("dev.knative.sources.ping")).
+					AtLeast(1)(ctx, t)
+			},
+		)
 
 	return f
 }