Merge remote-tracking branch 'origin/main' into aws-default-creds

Author: qswinson

pkg/adapter/apiserver/adapter.go

@@ -157,14 +157,16 @@ func (a *apiServerAdapter) startFailFast(ctx context.Context, stopCh <-chan stru
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
-				if err := reflector.ListAndWatchWithContext(watchCtx); err != nil {
-					a.logger.Errorf("reflector failed: %v", err)
-					select {
-					case errorChan <- fmt.Errorf("failed to watch: %v", err):
-					default:
-					}
-					cancelWatchers()
+				// ListAndWatch could exit with `nil` under some circumstances, it shouldn't
+				// ever stop listening and watching, so we will treat any exit as a reason to restart
+				err := reflector.ListAndWatchWithContext(watchCtx)
+				a.logger.Error("reflector exited: %v", err)
+
+				select {
+				case errorChan <- fmt.Errorf("failed to watch: %v", err):
+				default:
 				}
+				cancelWatchers()
 			}()
 		}
 	}

pkg/channel/event_receiver.go

@@ -306,7 +306,7 @@ func (r *EventReceiver) ServeHTTP(response nethttp.ResponseWriter, request *neth
 		return
 	}
 
-	/// Here we do the OIDC audience verification
+	// Here we do the OIDC audience verification
 	features := feature.FromContext(ctx)
 	if features.IsOIDCAuthentication() {
 		r.logger.Debug("OIDC authentication is enabled")

pkg/channel/event_receiver_test.go

@@ -88,7 +88,7 @@ func TestEventReceiver_ServeHTTP(t *testing.T) {
 		},
 		"path based channel reference": {
 			path: "/new-namespace/new-channel",
-			host: "test-name.test-namespace.svc." + network.GetClusterDomainName(),
+			host: host(),
 			receiverFunc: func(ctx context.Context, r ChannelReference, m event.Event, additionalHeaders nethttp.Header) error {
 				if r.Namespace != "new-namespace" || r.Name != "new-channel" {
 					return fmt.Errorf("bad channel reference %v", r)
@@ -107,9 +107,9 @@ func TestEventReceiver_ServeHTTP(t *testing.T) {
 				"x-requEst-id":              {"1234"},
 				"knatIve-will-pass-through": {"true", "always"},
 			},
-			host: "test-name.test-namespace.svc." + network.GetClusterDomainName(),
+			host: host(),
 			receiverFunc: func(ctx context.Context, r ChannelReference, e event.Event, additionalHeaders nethttp.Header) error {
-				if r.Namespace != "test-namespace" || r.Name != "test-name" {
+				if r.Namespace != "test-namespace" || r.Name != "test-channel" {
 					return fmt.Errorf("test receiver func -- bad reference: %v", r)
 				}
 
@@ -138,7 +138,7 @@ func TestEventReceiver_ServeHTTP(t *testing.T) {
 		},
 		"OPTIONS okay": {
 			method:   nethttp.MethodOptions,
-			host:     "test-name.test-namespace.svc." + network.GetClusterDomainName(),
+			host:     host(),
 			expected: nethttp.StatusOK,
 			responseValidator: func(res httptest.ResponseRecorder) error {
 				expectedHeaders := nethttp.Header{
@@ -163,7 +163,7 @@ func TestEventReceiver_ServeHTTP(t *testing.T) {
 				tc.path = "/"
 			}
 			if tc.host == "" {
-				tc.host = "test-channel.test-namespace.svc." + network.GetClusterDomainName()
+				tc.host = host()
 			}
 
 			f := tc.receiverFunc
@@ -238,7 +238,7 @@ func TestEventReceiver_ServerStart_trace_propagation(t *testing.T) {
 	done := make(chan struct{}, 1)
 
 	receiverFunc := func(ctx context.Context, r ChannelReference, e event.Event, additionalHeaders nethttp.Header) error {
-		if r.Namespace != "test-namespace" || r.Name != "test-name" {
+		if r.Namespace != "test-namespace" || r.Name != "test-channel" {
 			return fmt.Errorf("test receiver func -- bad reference: %v", r)
 		}
 
@@ -253,8 +253,6 @@ func TestEventReceiver_ServerStart_trace_propagation(t *testing.T) {
 
 	// Default the common things.
 	method := nethttp.MethodPost
-	host := "test-name.test-namespace.svc." + network.GetClusterDomainName()
-
 	logger, _ := zap.NewDevelopment()
 
 	r, err := NewEventReceiver(receiverFunc, logger)
@@ -276,7 +274,7 @@ func TestEventReceiver_ServerStart_trace_propagation(t *testing.T) {
 			),
 		))
 	require.NoError(t, err)
-	p.RequestTemplate.Host = host
+	p.RequestTemplate.Host = host()
 
 	c, err := cloudevents.NewClient(p)
 	require.NoError(t, err)
@@ -291,8 +289,6 @@ func TestEventReceiver_ServerStart_trace_propagation(t *testing.T) {
 }
 
 func TestEventReceiver_WrongRequest(t *testing.T) {
-	host := "http://test-channel.test-namespace.svc." + network.GetClusterDomainName() + "/"
-
 	f := func(_ context.Context, _ ChannelReference, _ event.Event, _ nethttp.Header) error {
 		return errors.New("test induced receiver function error")
 	}
@@ -301,7 +297,7 @@ func TestEventReceiver_WrongRequest(t *testing.T) {
 		t.Fatalf("Error creating new event receiver. Error:%s", err)
 	}
 
-	req := httptest.NewRequest(nethttp.MethodPost, host, bytes.NewReader([]byte("{}")))
+	req := httptest.NewRequest(nethttp.MethodPost, "http://"+host()+"/", bytes.NewReader([]byte("{}")))
 	req.Header.Set("content-type", "application/json")
 
 	res := httptest.ResponseRecorder{}
@@ -313,8 +309,6 @@ func TestEventReceiver_WrongRequest(t *testing.T) {
 }
 
 func TestEventReceiver_UnknownHost(t *testing.T) {
-	host := "http://test-channel.test-namespace.svc." + network.GetClusterDomainName() + "/"
-
 	f := func(_ context.Context, _ ChannelReference, _ event.Event, _ nethttp.Header) error {
 		return errors.New("test induced receiver function error")
 	}
@@ -335,7 +329,7 @@ func TestEventReceiver_UnknownHost(t *testing.T) {
 	}
 
 	req := httptest.NewRequest("POST", "http://localhost:8080/", nil)
-	req.Host = host
+	req.Host = "http://" + host() + "/"
 
 	err = http.WriteRequest(context.TODO(), binding.ToMessage(&event), req)
 	if err != nil {
@@ -349,3 +343,7 @@ func TestEventReceiver_UnknownHost(t *testing.T) {
 		t.Fatal("Unexpected status code. Expected 404. Actual", res.Code)
 	}
 }
+
+func host() string {
+	return fmt.Sprintf("test-channel%s.test-namespace.svc.%s", K8ServiceNameSuffix, network.GetClusterDomainName())
+}

pkg/channel/references.go

@@ -21,6 +21,11 @@ import (
 	"strings"
 )
 
+const (
+	// K8ServiceNameSuffix is added to the k8 service name which is owned by the channel
+	K8ServiceNameSuffix = "-kn-channel"
+)
+
 // ChannelReference references a Channel within the cluster by name and
 // namespace.
 type ChannelReference struct {
@@ -38,6 +43,11 @@ func ParseChannelFromHost(host string) (ChannelReference, error) {
 	if len(chunks) < 2 {
 		return ChannelReference{}, BadRequestError(fmt.Sprintf("bad host format %q", host))
 	}
+
+	if channelName, found := strings.CutSuffix(chunks[0], K8ServiceNameSuffix); found {
+		chunks[0] = channelName
+	}
+
 	return ChannelReference{
 		Name:      chunks[0],
 		Namespace: chunks[1],

pkg/channel/references_test.go

@@ -46,16 +46,24 @@ func TestParseChannelFromHost(t *testing.T) {
 		wantErr            bool
 		expectedChannelRef ChannelReference
 	}{
+		"host based with channel k8 service suffix": {
+			host:    fmt.Sprintf("%s%s.%s.svc.cluster.local", referencesTestChannelName, K8ServiceNameSuffix, referencesTestNamespace),
+			wantErr: false,
+			expectedChannelRef: ChannelReference{
+				Namespace: referencesTestNamespace,
+				Name:      referencesTestChannelName,
+			},
+		},
 		"host based": {
-			host:    "test-channel.test-namespace.svc.cluster.local",
+			host:    fmt.Sprintf("%s.%s.svc.cluster.local", referencesTestChannelName, referencesTestNamespace),
 			wantErr: false,
 			expectedChannelRef: ChannelReference{
-				Namespace: "test-namespace",
-				Name:      "test-channel",
+				Namespace: referencesTestNamespace,
+				Name:      referencesTestChannelName,
 			},
 		},
 		"bad host format should return error": {
-			host:    "test-channel",
+			host:    referencesTestChannelName,
 			wantErr: true,
 		},
 	}
@@ -91,11 +99,11 @@ func TestParseChannelFromPath(t *testing.T) {
 		expectedChannelRef ChannelReference
 	}{
 		"path based": {
-			path:    "/new-namespace/new-channel/",
+			path:    fmt.Sprintf("/%s/%s/", referencesTestNamespace, referencesTestChannelName),
 			wantErr: false,
 			expectedChannelRef: ChannelReference{
-				Namespace: "new-namespace",
-				Name:      "new-channel",
+				Namespace: referencesTestNamespace,
+				Name:      referencesTestChannelName,
 			},
 		},
 

pkg/reconciler/inmemorychannel/controller/inmemorychannel_test.go

@@ -33,6 +33,7 @@ import (
 	"knative.dev/pkg/resolver"
 
 	"knative.dev/eventing/pkg/apis/feature"
+	"knative.dev/eventing/pkg/channel"
 	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
 	"knative.dev/eventing/pkg/eventingtls"
 	"knative.dev/eventing/pkg/eventingtls/eventingtlstesting"
@@ -70,6 +71,7 @@ const (
 	dlsName             = "test-dls"
 	testNS              = "test-namespace"
 	imcName             = "test-imc"
+	imcSvcName          = imcName + channel.K8ServiceNameSuffix
 	imageName           = "test-image"
 	maxIdleConns        = 2000
 	maxIdleConnsPerHost = 200
@@ -83,7 +85,7 @@ const (
 var (
 	channelServiceAddress = duckv1.Addressable{
 		Name: pointer.String("http"),
-		URL:  apis.HTTP("test-imc-kn-channel.test-namespace.svc.cluster.local"),
+		URL:  apis.HTTP(fmt.Sprintf("%s.test-namespace.svc.cluster.local", imcSvcName)),
 	}
 
 	channelAudience = fmt.Sprintf("messaging.knative.dev/inmemorychannel/%s/%s", testNS, imcName)
@@ -426,11 +428,11 @@ func TestAllCases(t *testing.T) {
 					WithInMemoryChannelDeploymentReady(),
 					WithInMemoryChannelServiceReady(),
 					WithInMemoryChannelEndpointsReady(),
-					WithInMemoryChannelChannelServiceNotReady("ChannelServiceFailed", `Channel Service failed: inmemorychannel: test-namespace/test-imc does not own Service: "test-imc-kn-channel"`),
+					WithInMemoryChannelChannelServiceNotReady("ChannelServiceFailed", fmt.Sprintf(`Channel Service failed: inmemorychannel: test-namespace/test-imc does not own Service: "%s"`, imcSvcName)),
 				),
 			}},
 			WantEvents: []string{
-				Eventf(corev1.EventTypeWarning, "InternalError", `inmemorychannel: test-namespace/test-imc does not own Service: "test-imc-kn-channel"`),
+				Eventf(corev1.EventTypeWarning, "InternalError", fmt.Sprintf(`inmemorychannel: test-namespace/test-imc does not own Service: "%s"`, imcSvcName)),
 			},
 		}, {
 			Name: "Works, channel exists with subscribers",
@@ -986,7 +988,7 @@ func makeChannelService(imc *v1.InMemoryChannel) *corev1.Service {
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: testNS,
-			Name:      fmt.Sprintf("%s-kn-channel", imcName),
+			Name:      imcSvcName,
 			Labels: map[string]string{
 				resources.MessagingRoleLabel: resources.MessagingRole,
 			},
@@ -1016,7 +1018,7 @@ func makeChannelServiceNotOwnedByUs(imc *v1.InMemoryChannel) *corev1.Service {
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: testNS,
-			Name:      fmt.Sprintf("%s-kn-channel", imcName),
+			Name:      imcSvcName,
 			Labels: map[string]string{
 				resources.MessagingRoleLabel: resources.MessagingRole,
 			},

pkg/reconciler/inmemorychannel/controller/resources/service.go

@@ -20,6 +20,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/eventing/pkg/channel"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/network"
 )
@@ -35,7 +36,7 @@ const (
 type K8sServiceOption func(*corev1.Service) error
 
 func CreateChannelServiceName(name string) string {
-	return kmeta.ChildName(name, "-kn-channel")
+	return kmeta.ChildName(name, channel.K8ServiceNameSuffix)
 }
 
 // ExternalService is a functional option for CreateK8sService to create a K8s service of type ExternalName

pkg/reconciler/inmemorychannel/controller/resources/service_test.go

@@ -18,20 +18,21 @@ package resources
 
 import (
 	"errors"
-	"fmt"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	v1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	"knative.dev/eventing/pkg/channel"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/network"
 )
 
 const (
 	serviceName    = "my-test-service"
 	imcName        = "my-test-imc"
+	imcSvcName     = imcName + channel.K8ServiceNameSuffix
 	testNS         = "my-test-ns"
 	dispatcherNS   = "dispatcher-namespace"
 	dispatcherName = "dispatcher-name"
@@ -44,7 +45,7 @@ func TestCreateExternalServiceAddress(t *testing.T) {
 }
 
 func TestCreateChannelServiceAddress(t *testing.T) {
-	if want, got := "my-test-imc-kn-channel", CreateChannelServiceName(imcName); want != got {
+	if want, got := imcSvcName, CreateChannelServiceName(imcName); want != got {
 		t.Errorf("Want: %q got %q", want, got)
 	}
 }
@@ -62,7 +63,7 @@ func TestNewK8sService(t *testing.T) {
 			Kind:       "Service",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%s-kn-channel", imcName),
+			Name:      CreateChannelServiceName(imcName),
 			Namespace: testNS,
 			Labels: map[string]string{
 				MessagingRoleLabel: MessagingRole,
@@ -105,7 +106,7 @@ func TestNewK8sServiceWithExternal(t *testing.T) {
 			Kind:       "Service",
 		},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("%s-kn-channel", imcName),
+			Name:      CreateChannelServiceName(imcName),
 			Namespace: testNS,
 			Labels: map[string]string{
 				MessagingRoleLabel: MessagingRole,

pkg/reconciler/inmemorychannel/dispatcher/inmemorychannel_test.go

@@ -18,6 +18,7 @@ package dispatcher
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"reflect"
 	"testing"
@@ -48,6 +49,7 @@ import (
 	"knative.dev/eventing/pkg/apis/feature"
 	v1 "knative.dev/eventing/pkg/apis/messaging/v1"
 	"knative.dev/eventing/pkg/auth"
+	"knative.dev/eventing/pkg/channel"
 	"knative.dev/eventing/pkg/channel/fanout"
 	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
 	"knative.dev/eventing/pkg/client/injection/reconciler/messaging/v1/inmemorychannel"
@@ -68,7 +70,7 @@ const (
 
 var (
 	channelServiceAddress = duckv1.Addressable{
-		URL: apis.HTTP("test-imc-kn-channel.test-namespace.svc.cluster.local"),
+		URL: apis.HTTP(fmt.Sprintf("%s%s.%s.svc.cluster.local", imcName, channel.K8ServiceNameSuffix, testNS)),
 	}
 
 	linear      = eventingduckv1.BackoffPolicyLinear

pkg/reconciler/integration/sink/resources/container_image.go

@@ -124,7 +124,7 @@ func MakeDeploymentSpec(sink *v1alpha1.IntegrationSink, authProxyImage string, f
 		deploy.Spec.Template.Spec.Containers = append(deploy.Spec.Template.Spec.Containers, corev1.Container{
 			Name:            "auth-proxy",
 			Image:           authProxyImage,
-			ImagePullPolicy: corev1.PullAlways,
+			ImagePullPolicy: corev1.PullIfNotPresent,
 			Ports: []corev1.ContainerPort{
 				{
 					ContainerPort: 3128,