Skip to content

Commit

Permalink
JobSink: Delete secrets associated with jobs when jobs are deleted (#…
Browse files Browse the repository at this point in the history
…8331)

* JobSink: Delete secrets associated with jobs when jobs are deleted

As reported in #8323 old
JobSink secrets lead to processing old events again while new events
are lost.

Using OwnerReference and k8s garbage collection, now a secret created
for a given event is bound to a given Job lifecycle, so that when a job
is deleted, the associated secret will be deleted.

Signed-off-by: Pierangelo Di Pilato <[email protected]>

* Fix jobsink name generator + add unit and fuzz tests

Signed-off-by: Pierangelo Di Pilato <[email protected]>

* Fix e2e test

Signed-off-by: Pierangelo Di Pilato <[email protected]>

* Lint

Signed-off-by: Pierangelo Di Pilato <[email protected]>

---------

Signed-off-by: Pierangelo Di Pilato <[email protected]>
  • Loading branch information
pierDipi authored Nov 19, 2024
1 parent 8fed0be commit 7abb04d
Show file tree
Hide file tree
Showing 5 changed files with 250 additions and 56 deletions.
124 changes: 71 additions & 53 deletions cmd/jobsink/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"context"
"crypto/md5" //nolint:gosec
"crypto/tls"
"encoding/hex"
"fmt"
"log"
"net/http"
Expand Down Expand Up @@ -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 {
Expand All @@ -256,56 +257,21 @@ 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))

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
Expand Down Expand Up @@ -346,14 +312,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)
Expand Down Expand Up @@ -391,8 +409,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 {
Expand Down Expand Up @@ -445,6 +462,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[:])))
}
106 changes: 106 additions & 0 deletions cmd/jobsink/main_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
/*
Copyright 2024 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 main

import (
"testing"

"k8s.io/apimachinery/pkg/api/validation"

"knative.dev/eventing/pkg/utils"
)

type testCase struct {
JobSinkName string
Source string
Id string
}

func TestToJobName(t *testing.T) {
testcases := []testCase{
{
JobSinkName: "job-sink-success",
Source: "mysource3/myservice",
Id: "2234-5678",
},
{
JobSinkName: "a",
Source: "0",
Id: "0",
},
}

for _, tc := range testcases {
t.Run(tc.JobSinkName+"_"+tc.Source+"_"+tc.Id, func(t *testing.T) {
if errs := validation.NameIsDNS1035Label(tc.JobSinkName, false); len(errs) != 0 {
t.Errorf("Invalid JobSinkName: %v", errs)
}

name := toJobName(tc.JobSinkName, tc.Source, tc.Id)
doubleName := toJobName(tc.JobSinkName, tc.Source, tc.Id)
if name != doubleName {
t.Errorf("Before: %q, after: %q", name, doubleName)
}

if got := utils.ToDNS1123Subdomain(name); got != name {
t.Errorf("ToDNS1123Subdomain(Want) returns a different result, Want: %q, Got: %q", name, got)
}

if errs := validation.NameIsDNS1035Label(name, false); len(errs) != 0 {
t.Errorf("toJobName produced invalid name %q given %q, %q, %q: errors: %#v", name, tc.JobSinkName, tc.Source, tc.Id, errs)
}
})
}
}

func FuzzToJobName(f *testing.F) {
testcases := []testCase{
{
JobSinkName: "job-sink-success",
Source: "mysource3/myservice",
Id: "2234-5678",
},
{
JobSinkName: "a",
Source: "0",
Id: "0",
},
}

for _, tc := range testcases {
f.Add(tc.JobSinkName, tc.Source, tc.Id) // Use f.Add to provide a seed corpus
}
f.Fuzz(func(t *testing.T, js, source, id string) {
if errs := validation.NameIsDNSLabel(js, false); len(errs) != 0 {
t.Skip("Prerequisite: invalid jobsink name")
}

name := toJobName(js, source, id)
doubleName := toJobName(js, source, id)
if name != doubleName {
t.Errorf("Before: %q, after: %q", name, doubleName)
}

if got := utils.ToDNS1123Subdomain(name); got != name {
t.Errorf("ToDNS1123Subdomain(Want) returns a different result, Want: %q, Got: %q", name, got)
}

if errs := validation.NameIsDNSLabel(name, false); len(errs) != 0 {
t.Errorf("toJobName produced invalid name %q given %q, %q, %q: errors: %#v", name, js, source, id, errs)
}
})
}
2 changes: 1 addition & 1 deletion hack/e2e-debug.sh
Original file line number Diff line number Diff line change
Expand Up @@ -35,4 +35,4 @@ wait_until_pods_running knative-eventing || fail_test "Pods in knative-eventing

header "Running tests"

go test -tags=e2e -v -timeout=30m -run="${test_name}" "${test_dir}" || fail_test "Test(s) failed"
go test -tags=e2e -v -timeout=30m -parallel=12 -run="${test_name}" "${test_dir}" || fail_test "Test(s) failed"
56 changes: 55 additions & 1 deletion test/rekt/features/jobsink/jobsink.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import (
cetest "github.com/cloudevents/sdk-go/v2/test"
"github.com/google/uuid"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/wait"
Expand All @@ -43,11 +44,14 @@ import (
"knative.dev/eventing/test/rekt/resources/jobsink"
)

func Success() *feature.Feature {
func Success(jobSinkName string) *feature.Feature {
f := feature.NewFeature()

sink := feature.MakeRandomK8sName("sink")
jobSink := feature.MakeRandomK8sName("jobsink")
if jobSinkName != "" {
jobSink = jobSinkName
}
source := feature.MakeRandomK8sName("source")

event := cetest.FullEvent()
Expand Down Expand Up @@ -83,6 +87,32 @@ func Success() *feature.Feature {
return f
}

func DeleteJobsCascadeSecretsDeletion(jobSink string) *feature.Feature {
f := feature.NewFeature()

f.Setup("Prerequisite: At least one secret for jobsink present", verifySecretsForJobSink(jobSink, func(secrets *corev1.SecretList) bool {
return len(secrets.Items) > 0
}))

f.Requirement("delete jobs for jobsink", func(ctx context.Context, t feature.T) {
policy := metav1.DeletePropagationBackground
err := kubeclient.Get(ctx).BatchV1().
Jobs(environment.FromContext(ctx).Namespace()).
DeleteCollection(ctx, metav1.DeleteOptions{PropagationPolicy: &policy}, metav1.ListOptions{
LabelSelector: fmt.Sprintf("%s=%s", sinks.JobSinkNameLabel, jobSink),
})
if err != nil {
t.Error(err)
}
})

f.Assert("No secrets for jobsink are present", verifySecretsForJobSink(jobSink, func(secrets *corev1.SecretList) bool {
return len(secrets.Items) == 0
}))

return f
}

func SuccessTLS() *feature.Feature {
f := feature.NewFeature()

Expand Down Expand Up @@ -239,3 +269,27 @@ func AtLeastOneJobIsComplete(jobSinkName string) feature.StepFn {
t.Errorf("No job is complete:\n%v", string(bytes))
}
}

func verifySecretsForJobSink(jobSink string, verify func(secrets *corev1.SecretList) bool) feature.StepFn {
return func(ctx context.Context, t feature.T) {

interval, timeout := environment.PollTimingsFromContext(ctx)
lastSecretList := &corev1.SecretList{}
err := wait.PollUntilContextTimeout(ctx, interval, timeout, true, func(ctx context.Context) (bool, error) {
var err error
lastSecretList, err = kubeclient.Get(ctx).CoreV1().
Secrets(environment.FromContext(ctx).Namespace()).
List(ctx, metav1.ListOptions{
LabelSelector: fmt.Sprintf("%s=%s", sinks.JobSinkNameLabel, jobSink),
})
if err != nil {
return false, fmt.Errorf("failed to list secrets: %w", err)
}
return verify(lastSecretList), nil
})
if err != nil {
bytes, _ := json.Marshal(lastSecretList)
t.Errorf("failed to wait for no secrets: %v\nSecret list:\n%s", err, string(bytes))
}
}
}
Loading

0 comments on commit 7abb04d

Please sign in to comment.