minimum set of features, not tested

Author: odacremolbap

cmd/triggermesh-hook/start/start.go

@@ -7,7 +7,8 @@ import (
 	commoncmd "github.com/triggermesh/scoby-hook-triggermesh/pkg/common/cmd"
 	"github.com/triggermesh/scoby-hook-triggermesh/pkg/handler"
 	"github.com/triggermesh/scoby-hook-triggermesh/pkg/handler/kuards"
-	"github.com/triggermesh/scoby-hook-triggermesh/pkg/handler/sources/awss3"
+	"github.com/triggermesh/scoby-hook-triggermesh/pkg/sources/client/s3"
+	"github.com/triggermesh/scoby-hook-triggermesh/pkg/sources/reconciler/awss3source"
 
 	"github.com/triggermesh/scoby-hook-triggermesh/pkg/server"
 )
@@ -23,7 +24,7 @@ func (c *Cmd) Run(g *commoncmd.Globals) error {
 	r := handler.NewRegistry([]handler.Handler{
 		// Kuards is a temporary playground
 		kuards.New(),
-		awss3.New(),
+		awss3source.New(s3.NewClientGetter(g.KubeClient.CoreV1().Secrets), g.Logger),
 	})
 
 	s := server.New(c.Path, c.Address, r, g.DynClient, g.Logger)

pkg/handler/sources/awss3source/handler.go

@@ -23,108 +23,74 @@ import (
 	"net/http"
 	"sort"
 
-	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	"knative.dev/pkg/controller"
-	"knative.dev/pkg/reconciler"
-
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/aws/aws-sdk-go/service/sqs"
 
-	commonv1alpha1 "github.com/triggermesh/triggermesh/pkg/apis/common/v1alpha1"
-	"github.com/triggermesh/triggermesh/pkg/apis/sources/v1alpha1"
-	"github.com/triggermesh/triggermesh/pkg/reconciler/event"
+	"github.com/triggermesh/scoby-hook-triggermesh/pkg/apis/sources/v1alpha1"
 )
 
 // EnsureNotificationsEnabled ensures that event notifications are enabled in
 // the S3 bucket.
-func EnsureNotificationsEnabled(ctx context.Context, cli s3iface.S3API, queueARN string) error {
-	src := commonv1alpha1.ReconcilableFromContext(ctx)
-	typedSrc := src.(*v1alpha1.AWSS3Source)
-
-	status := &typedSrc.Status
-
-	bucketARN := typedSrc.Spec.ARN
+func EnsureNotificationsEnabled(ctx context.Context, src *v1alpha1.AWSS3Source, cli s3iface.S3API, queueARN string) error {
+	bucketARN := src.Spec.ARN
 
 	notifCfg, err := getNotificationsConfig(ctx, cli, bucketARN.Resource)
 	switch {
 	case isNotFound(err):
-		status.MarkNotSubscribed(v1alpha1.AWSS3ReasonNoBucket, "Bucket does not exist")
-		return controller.NewPermanentError(reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedSubscribe,
-			"The bucket does not exist: %s", toErrMsg(err)))
+		return fmt.Errorf("The bucket does not exist: %v", toErrMsg(err))
 	case isAWSError(err):
 		// All documented API errors require some user intervention and
 		// are not to be retried.
 		// https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
-		status.MarkNotSubscribed(v1alpha1.AWSS3ReasonAPIError, "Request to S3 API got rejected")
-		return controller.NewPermanentError(reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedSubscribe,
-			"Failed to synchronize bucket configuration: %s", toErrMsg(err)))
+		return fmt.Errorf("Failed to synchronize bucket configuration: %v", toErrMsg(err))
 	case err != nil:
-		status.MarkNotSubscribed(v1alpha1.AWSS3ReasonAPIError, "Cannot obtain current bucket configuration")
-		// wrap any other error to fail the reconciliation
-		return fmt.Errorf("%w", reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedSubscribe,
-			"Error reading current event notifications configuration: %s", toErrMsg(err)))
+		return fmt.Errorf("Cannot obtain current bucket configuration: %v", toErrMsg(err))
 	}
 
-	desiredQueueCfg := makeQueueConfiguration(typedSrc, queueARN)
+	desiredQueueCfg := makeQueueConfiguration(src, queueARN)
 
 	notifCfg, hasUpdates := setQueueConfiguration(notifCfg, desiredQueueCfg)
 
 	if hasUpdates {
 		if err := configureNotifications(ctx, cli, bucketARN.Resource, notifCfg); err != nil {
-			status.MarkNotSubscribed(v1alpha1.AWSS3ReasonAPIError, "Cannot configure event notifications")
-			return fmt.Errorf("%w", reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedSubscribe,
-				"Error configuring event notifications: %s", toErrMsg(err)))
+			return fmt.Errorf("Cannot configure event notifications: %v", toErrMsg(err))
 		}
 	}
 
-	if !status.GetCondition(v1alpha1.AWSS3ConditionSubscribed).IsTrue() {
-		event.Normal(ctx, ReasonSubscribed, "Configured event notifications for S3 bucket %q", bucketARN)
-	}
-	status.MarkSubscribed()
-
 	return nil
 }
 
 // EnsureNotificationsDisabled ensures that event notifications are disabled in
 // the S3 bucket.
-func EnsureNotificationsDisabled(ctx context.Context, cli s3iface.S3API) error {
-	src := commonv1alpha1.ReconcilableFromContext(ctx)
-	typedSrc := src.(*v1alpha1.AWSS3Source)
-
-	bucketARN := typedSrc.Spec.ARN
+func EnsureNotificationsDisabled(ctx context.Context, src *v1alpha1.AWSS3Source, cli s3iface.S3API) error {
+	bucketARN := src.Spec.ARN
 
 	notifCfg, err := getNotificationsConfig(ctx, cli, bucketARN.Resource)
 	switch {
 	case isNotFound(err):
-		return reconciler.NewEvent(corev1.EventTypeNormal, ReasonUnsubscribed,
-			"Bucket not found, skipping finalization")
+		return fmt.Errorf("Bucket not found: %v", toErrMsg(err))
 	case isDenied(err):
 		// it is unlikely that we recover from auth errors in the
 		// finalizer, so we simply record a warning event and return
-		event.Warn(ctx, ReasonFailedUnsubscribe,
-			"Authorization error getting bucket configuration. Ignoring: %s", toErrMsg(err))
-		return nil
+		return fmt.Errorf("Authorization error getting bucket configuration: %v", toErrMsg(err))
 	case err != nil:
-		return reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedUnsubscribe,
-			"Error reading current event notifications configuration: %s", toErrMsg(err))
+		return fmt.Errorf("Error reading current event notifications configuration: %v", toErrMsg(err))
 	}
 
 	notifCfg = removeQueueConfiguration(notifCfg, sourceID(src))
 
 	if err := configureNotifications(ctx, cli, bucketARN.Resource, notifCfg); err != nil {
-		return fmt.Errorf("%w", reconciler.NewEvent(corev1.EventTypeWarning, ReasonFailedUnsubscribe,
-			"Error configuring event notifications: %s", toErrMsg(err)))
+		return fmt.Errorf("Error configuring event notifications: %v", toErrMsg(err))
 	}
 
-	return reconciler.NewEvent(corev1.EventTypeNormal, ReasonUnsubscribed,
-		"Disabled event notifications for S3 bucket %q", bucketARN)
+	return nil
 }
 
 // getNotificationsConfig reads the current event notifications configuration