Merge remote-tracking branch 'upstream/main'

merged with forked repo

Author: vichekaoeun

.github/workflows/kind-e2e.yaml

@@ -17,8 +17,8 @@ jobs:
       fail-fast: false # Keep running if one leg fails.
       matrix:
         k8s-version:
-        - v1.28.7
-        - v1.29.0
+          - v1.30.8
+          - v1.31.4
 
         test-suite:
         - ./test/e2e
@@ -29,12 +29,12 @@ jobs:
         # This is attempting to make it a bit clearer what's being tested.
         # See: https://github.com/kubernetes-sigs/kind/releases/tag/v0.20.0
         include:
-        - k8s-version: v1.28.7
-          kind-version: v0.21.0
-          kind-image-sha: sha256:9bc6c451a289cf96ad0bbaf33d416901de6fd632415b076ab05f5fa7e4f65c58
-        - k8s-version: v1.29.0
-          kind-version: v0.21.0
-          kind-image-sha: sha256:eaa1450915475849a73a9227b8f201df25e55e268e5d619312131292e324d570
+        - k8s-version: v1.30.8
+          kind-version: v0.26.0
+          kind-image-sha: sha256:17cd608b3971338d9180b00776cb766c50d0a0b6b904ab4ff52fd3fc5c6369bf
+        - k8s-version: v1.31.4
+          kind-version: v0.26.0
+          kind-image-sha: sha256:2cb39f7295fe7eafee0842b1052a599a4fb0f8bcf3f83d96c7f4864c357c6c30
 
         # Add the flags we use for each of these test suites.
         - test-suite: ./test/e2e

OWNERS_ALIASES

@@ -13,11 +13,9 @@ aliases:
   - vyasgun
   docs-reviewers:
   - nainaz
-  - retocode
   - skonto
   docs-writers:
   - csantanapr
-  - retocode
   - skonto
   eventing-reviewers:
   - Leo6Leo
@@ -51,7 +49,6 @@ aliases:
   knative-admin:
   - aliok
   - cardil
-  - davidhadas
   - dprotaso
   - dsimansk
   - evankanderson
@@ -60,15 +57,13 @@ aliases:
   - knative-prow-robot
   - knative-prow-updater-robot
   - knative-test-reporter-robot
-  - nainaz
-  - psschwei
-  - retocode
-  - salaboy
+  - matzew
+  - nrrso
   - skonto
   - upodroid
   knative-release-leads:
+  - dprotaso
   - dsimansk
-  - retocode
   - skonto
   knative-robots:
   - knative-automation
@@ -105,32 +100,22 @@ aliases:
   - davidhadas
   - evankanderson
   serving-approvers:
-  - ReToCode
   - skonto
   serving-reviewers:
-  - izabelacg
-  - retocode
   - skonto
   serving-triage:
-  - izabelacg
-  - retocode
   - skonto
   serving-wg-leads:
   - dprotaso
   serving-writers:
-  - ReToCode
   - dprotaso
   - skonto
   steering-committee:
   - aliok
-  - davidhadas
   - dprotaso
-  - dsimansk
   - evankanderson
-  - nainaz
-  - psschwei
-  - salaboy
-  technical-oversight-committee: []
+  - matzew
+  - nrrso
   ux-wg-leads:
   - cali0707
   - leo6leo

cmd/controller/main.go

@@ -20,12 +20,6 @@ import (
 	// Uncomment the following line to load the gcp plugin (only required to authenticate against GKE clusters).
 	// _ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 
-	"errors"
-	"log"
-	"net/http"
-	"os"
-	"time"
-
 	"knative.dev/pkg/injection/sharedmain"
 
 	filteredFactory "knative.dev/pkg/client/injection/kube/informers/factory/filtered"
@@ -41,7 +35,8 @@ import (
 	"knative.dev/eventing/pkg/reconciler/channel"
 	"knative.dev/eventing/pkg/reconciler/containersource"
 	"knative.dev/eventing/pkg/reconciler/eventtype"
-	"knative.dev/eventing/pkg/reconciler/integrationsource"
+	integrationsink "knative.dev/eventing/pkg/reconciler/integration/sink"
+	integrationsource "knative.dev/eventing/pkg/reconciler/integration/source"
 	"knative.dev/eventing/pkg/reconciler/parallel"
 	"knative.dev/eventing/pkg/reconciler/pingsource"
 	"knative.dev/eventing/pkg/reconciler/sequence"
@@ -52,36 +47,8 @@ import (
 )
 
 func main() {
-
 	ctx := signals.NewContext()
 
-	port := os.Getenv("PROBES_PORT")
-	if port == "" {
-		port = "8080"
-	}
-
-	// sets up liveness and readiness probes.
-	server := http.Server{
-		ReadTimeout: 5 * time.Second,
-		Handler:     http.HandlerFunc(handler),
-		Addr:        ":" + port,
-	}
-
-	go func() {
-
-		go func() {
-			<-ctx.Done()
-			_ = server.Shutdown(ctx)
-		}()
-
-		// start the web server on port and accept requests
-		log.Printf("Readiness and health check server listening on port %s", port)
-
-		if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			log.Fatal(err)
-		}
-	}()
-
 	ctx = filteredFactory.WithSelectors(ctx,
 		auth.OIDCLabelSelector,
 		eventingtls.TrustBundleLabelSelector,
@@ -112,13 +79,10 @@ func main() {
 
 		// Sinks
 		jobsink.NewController,
+		integrationsink.NewController,
 
 		// Sugar
 		sugarnamespace.NewController,
 		sugartrigger.NewController,
 	)
 }
-
-func handler(w http.ResponseWriter, r *http.Request) {
-	w.WriteHeader(http.StatusOK)
-}

cmd/jobsink/main.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/md5" //nolint:gosec
 	"crypto/tls"
+	"encoding/hex"
 	"fmt"
 	"log"
 	"net/http"
@@ -65,7 +66,7 @@ import (
 	"knative.dev/eventing/pkg/utils"
 )
 
-const component = "job-sink"
+const component = "job_sink"
 
 func main() {
 
@@ -231,11 +232,11 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	id := toIdHashLabelValue(event.Source(), event.ID())
-	logger.Debug("Getting job for event", zap.String("URI", r.RequestURI), zap.String("id", id))
+	jobName := toJobName(ref.Name, event.Source(), event.ID())
+	logger.Debug("Getting job for event", zap.String("URI", r.RequestURI), zap.String("jobName", jobName))
 
 	jobs, err := h.k8s.BatchV1().Jobs(js.GetNamespace()).List(r.Context(), metav1.ListOptions{
-		LabelSelector: jobLabelSelector(ref, id),
+		LabelSelector: jobLabelSelector(ref, jobName),
 		Limit:         1,
 	})
 	if err != nil {
@@ -256,56 +257,24 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	jobName := kmeta.ChildName(ref.Name, id)
-
-	logger.Debug("Creating secret for event", zap.String("URI", r.RequestURI), zap.String("jobName", jobName))
-
-	jobSinkUID := js.GetUID()
-
-	or := metav1.OwnerReference{
-		APIVersion:         sinksv.SchemeGroupVersion.String(),
-		Kind:               sinks.JobSinkResource.Resource,
-		Name:               js.GetName(),
-		UID:                jobSinkUID,
-		Controller:         ptr.Bool(true),
-		BlockOwnerDeletion: ptr.Bool(false),
-	}
-
-	secret := &corev1.Secret{
-		TypeMeta: metav1.TypeMeta{},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      jobName,
-			Namespace: ref.Namespace,
-			Labels: map[string]string{
-				sinks.JobSinkIDLabel:   id,
-				sinks.JobSinkNameLabel: ref.Name,
-			},
-			OwnerReferences: []metav1.OwnerReference{or},
-		},
-		Immutable: ptr.Bool(true),
-		Data:      map[string][]byte{"event": eventBytes},
-		Type:      corev1.SecretTypeOpaque,
-	}
-
-	_, err = h.k8s.CoreV1().Secrets(ref.Namespace).Create(r.Context(), secret, metav1.CreateOptions{})
-	if err != nil && !apierrors.IsAlreadyExists(err) {
-		logger.Warn("Failed to create secret", zap.Error(err))
-
-		w.Header().Add("Reason", err.Error())
-		w.WriteHeader(http.StatusInternalServerError)
-		return
-	}
-
-	logger.Debug("Creating job for event", zap.String("URI", r.RequestURI), zap.String("jobName", jobName))
+	js = js.DeepCopy() // Do not modify informer copy.
+	js.SetDefaults(ctx)
 
 	job := js.Spec.Job.DeepCopy()
 	job.Name = jobName
 	if job.Labels == nil {
 		job.Labels = make(map[string]string, 4)
 	}
-	job.Labels[sinks.JobSinkIDLabel] = id
+	job.Labels[sinks.JobSinkIDLabel] = jobName
 	job.Labels[sinks.JobSinkNameLabel] = ref.Name
-	job.OwnerReferences = append(job.OwnerReferences, or)
+	job.OwnerReferences = append(job.OwnerReferences, metav1.OwnerReference{
+		APIVersion:         sinksv.SchemeGroupVersion.String(),
+		Kind:               sinks.JobSinkResource.Resource,
+		Name:               js.GetName(),
+		UID:                js.GetUID(),
+		Controller:         ptr.Bool(true),
+		BlockOwnerDeletion: ptr.Bool(false),
+	})
 	var mountPathName string
 	for i := range job.Spec.Template.Spec.Containers {
 		found := false
@@ -346,14 +315,66 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		})
 	}
 
-	_, err = h.k8s.BatchV1().Jobs(ref.Namespace).Create(r.Context(), job, metav1.CreateOptions{})
-	if err != nil {
+	logger.Debug("Creating job for event",
+		zap.String("URI", r.RequestURI),
+		zap.String("jobName", jobName),
+		zap.Any("job", job),
+	)
+
+	createdJob, err := h.k8s.BatchV1().Jobs(ref.Namespace).Create(r.Context(), job, metav1.CreateOptions{})
+	if err != nil && !apierrors.IsAlreadyExists(err) {
 		logger.Warn("Failed to create job", zap.Error(err))
 
 		w.Header().Add("Reason", err.Error())
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
+	if apierrors.IsAlreadyExists(err) {
+		logger.Debug("Job already exists", zap.String("URI", r.RequestURI), zap.String("jobName", jobName))
+	}
+
+	secret := &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      jobName,
+			Namespace: ref.Namespace,
+			Labels: map[string]string{
+				sinks.JobSinkIDLabel:   jobName,
+				sinks.JobSinkNameLabel: ref.Name,
+			},
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion:         "batch/v1",
+					Kind:               "Job",
+					Name:               createdJob.Name,
+					UID:                createdJob.UID,
+					Controller:         ptr.Bool(true),
+					BlockOwnerDeletion: ptr.Bool(false),
+				},
+			},
+		},
+		Immutable: ptr.Bool(true),
+		Data:      map[string][]byte{"event": eventBytes},
+		Type:      corev1.SecretTypeOpaque,
+	}
+
+	logger.Debug("Creating secret for event",
+		zap.String("URI", r.RequestURI),
+		zap.String("jobName", jobName),
+		zap.Any("secret.metadata", secret.ObjectMeta),
+	)
+
+	_, err = h.k8s.CoreV1().Secrets(ref.Namespace).Create(r.Context(), secret, metav1.CreateOptions{})
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+		logger.Warn("Failed to create secret", zap.Error(err))
+
+		w.Header().Add("Reason", err.Error())
+		w.WriteHeader(http.StatusInternalServerError)
+		return
+	}
+	if apierrors.IsAlreadyExists(err) {
+		logger.Debug("Secret already exists", zap.String("URI", r.RequestURI), zap.String("jobName", jobName))
+	}
 
 	w.Header().Add("Location", locationHeader(ref, event.Source(), event.ID()))
 	w.WriteHeader(http.StatusAccepted)
@@ -391,8 +412,7 @@ func (h *Handler) handleGet(ctx context.Context, w http.ResponseWriter, r *http.
 	eventSource := parts[6]
 	eventID := parts[8]
 
-	id := toIdHashLabelValue(eventSource, eventID)
-	jobName := kmeta.ChildName(ref.Name, id)
+	jobName := toJobName(ref.Name, eventSource, eventID)
 
 	job, err := h.k8s.BatchV1().Jobs(ref.Namespace).Get(r.Context(), jobName, metav1.GetOptions{})
 	if err != nil {
@@ -445,6 +465,7 @@ func jobLabelSelector(ref types.NamespacedName, id string) string {
 	return fmt.Sprintf("%s=%s,%s=%s", sinks.JobSinkIDLabel, id, sinks.JobSinkNameLabel, ref.Name)
 }
 
-func toIdHashLabelValue(source, id string) string {
-	return utils.ToDNS1123Subdomain(fmt.Sprintf("%s", md5.Sum([]byte(fmt.Sprintf("%s-%s", source, id))))) //nolint:gosec
+func toJobName(js string, source, id string) string {
+	h := md5.Sum([]byte(source + id)) //nolint:gosec
+	return kmeta.ChildName(js+"-", utils.ToDNS1123Subdomain(hex.EncodeToString(h[:])))
 }