Adding first Event Bridge Sink Control Plane work.

matzew · matzew · commit 89c308dfbeb8 · 2025-10-20T08:53:28.000+02:00
Signed-off-by: Matthias Wessendorf &lt;mwessend@redhat.com&gt;
diff --git a/config/core/deployments/controller.yaml b/config/core/deployments/controller.yaml
@@ -135,6 +135,13 @@ spec:
                 key: aws-sns-sink
                 name: eventing-integrations-images
 
+          - name: INTEGRATION_SINK_AWS_EVENTBRIDGE_IMAGE
+            valueFrom:
+              configMapKeyRef:
+                key: aws-eventbridge-sink
+                name: eventing-integrations-images
+
+
 ##         Adapter settings
 #          - name: K_LOGGING_CONFIG
 #            value: ''
diff --git a/config/core/resources/integrationsink.yaml b/config/core/resources/integrationsink.yaml
@@ -111,6 +111,30 @@ spec:
                 aws:
                   type: object
                   properties:
+                    eventbridge:
+                      type: object
+                      properties:
+                        arn:
+                          type: string
+                          title: Eventbus Name
+                          description: The Eventbridge Eventbus name or Amazon Resource Name (ARN).
+                        region:
+                          type: string
+                          title: AWS Region
+                          description: The AWS region to access.
+                        uriEndpointOverride:
+                          type: string
+                          title: Overwrite Endpoint URI
+                          description: The overriding endpoint URI. To use this option, you must
+                            also select the `overrideEndpoint` option.
+                        overrideEndpoint:
+                          type: boolean
+                          title: Endpoint Overwrite
+                          description: Select this option to override the endpoint URI. To use
+                            this option, you must also provide a URI for the `uriEndpointOverride`
+                            option.
+                          default: false
+
                     s3:
                       type: object
                       properties:
diff --git a/docs/eventing-api.md b/docs/eventing-api.md
@@ -6759,12 +6759,23 @@ JobSinkStatus
 <tbody>
 <tr>
 <td>
+<code>eventbridge</code><br/>
+<em>
+knative.dev/eventing/pkg/apis/common/integration/v1alpha1.AWSEventbridge
+</em>
+</td>
+<td>
+</td>
+</tr>
+<tr>
+<td>
 <code>s3</code><br/>
 <em>
 knative.dev/eventing/pkg/apis/common/integration/v1alpha1.AWSS3
 </em>
 </td>
 <td>
+<p>EB sink configuration</p>
 </td>
 </tr>
 <tr>
diff --git a/pkg/apis/common/integration/v1alpha1/aws.go b/pkg/apis/common/integration/v1alpha1/aws.go
@@ -74,3 +74,8 @@ type AWSSNS struct {
 	Arn             string           `json:"arn,omitempty" camel:"TOPIC_NAME_OR_ARN"` // SNS ARN
 	AutoCreateTopic bool             `json:"autoCreateTopic" default:"false"`         // Auto-create SNS topic
 }
+
+type AWSEventbridge struct {
+	AWSCommon `json:",inline"` // Embeds AWSCommon to inherit its fields in JSON
+	Arn       string           `json:"arn,omitempty" camel:"EVENTBUS_NAME_OR_ARN"` // EB ARN
+}
diff --git a/pkg/apis/common/integration/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/common/integration/v1alpha1/zz_generated.deepcopy.go
diff --git a/pkg/apis/sinks/v1alpha1/integration_sink_types.go b/pkg/apis/sinks/v1alpha1/integration_sink_types.go
@@ -74,10 +74,11 @@ type Log struct {
 }
 
 type Aws struct {
-	S3   *v1alpha1.AWSS3  `json:"s3,omitempty"`  // S3 sink configuration
-	SQS  *v1alpha1.AWSSQS `json:"sqs,omitempty"` // SQS sink configuration
-	SNS  *v1alpha1.AWSSNS `json:"sns,omitempty"` // SNS sink configuration
-	Auth *v1alpha1.Auth   `json:"auth,omitempty"`
+	EVENTBRIDGE *v1alpha1.AWSEventbridge `json:"eventbridge,omitempty"` // EB sink configuration
+	S3          *v1alpha1.AWSS3          `json:"s3,omitempty"`          // S3 sink configuration
+	SQS         *v1alpha1.AWSSQS         `json:"sqs,omitempty"`         // SQS sink configuration
+	SNS         *v1alpha1.AWSSNS         `json:"sns,omitempty"`         // SNS sink configuration
+	Auth        *v1alpha1.Auth           `json:"auth,omitempty"`
 }
 
 type IntegrationSinkStatus struct {
diff --git a/pkg/apis/sinks/v1alpha1/integration_sink_types_test.go b/pkg/apis/sinks/v1alpha1/integration_sink_types_test.go
@@ -98,6 +98,17 @@ func TestAWS(t *testing.T) {
 	if sns.Region != "eu-north-1" {
 		t.Errorf("AWSDDBStreams.Region = %v, want 'eu-north-1'", sns.Region)
 	}
+
+	eb := v1alpha1.AWSEventbridge{
+		AWSCommon: v1alpha1.AWSCommon{
+			Region: "eu-north-1",
+		},
+		Arn: "example-event-bus",
+	}
+
+	if eb.Region != "eu-north-1" {
+		t.Errorf("AWSEventbridge.Region = %v, want 'eu-north-1'", sns.Region)
+	}
 }
 
 // TestAuthFieldAccess tests the HasAuth method and field access in Auth struct
diff --git a/pkg/apis/sinks/v1alpha1/integration_sink_validation.go b/pkg/apis/sinks/v1alpha1/integration_sink_validation.go
@@ -45,6 +45,9 @@ func (spec *IntegrationSinkSpec) Validate(ctx context.Context) *apis.FieldError
 		if spec.Aws.SNS != nil {
 			sinkSetCount++
 		}
+		if spec.Aws.EVENTBRIDGE != nil {
+			sinkSetCount++
+		}
 	}
 
 	// Validate that only one sink field is set
@@ -56,7 +59,7 @@ func (spec *IntegrationSinkSpec) Validate(ctx context.Context) *apis.FieldError
 
 	// Only perform AWS-specific validation if exactly one AWS sink is configured
 	if sinkSetCount == 1 && spec.Aws != nil {
-		if spec.Aws.S3 != nil || spec.Aws.SQS != nil || spec.Aws.SNS != nil {
+		if spec.Aws.S3 != nil || spec.Aws.SQS != nil || spec.Aws.SNS != nil || spec.Aws.EVENTBRIDGE != nil {
 			// Check that AWS Auth is properly configured
 			if !spec.Aws.Auth.HasAuth() {
 				errs = errs.Also(apis.ErrMissingField("aws.auth.secret.ref.name"))
@@ -91,6 +94,15 @@ func (spec *IntegrationSinkSpec) Validate(ctx context.Context) *apis.FieldError
 				errs = errs.Also(apis.ErrMissingField("aws.sns.region"))
 			}
 		}
+		// Additional validation for AWS Eventbridge required fields
+		if spec.Aws.EVENTBRIDGE != nil {
+			if spec.Aws.EVENTBRIDGE.Arn == "" {
+				errs = errs.Also(apis.ErrMissingField("aws.eventbridge.arn"))
+			}
+			if spec.Aws.EVENTBRIDGE.Region == "" {
+				errs = errs.Also(apis.ErrMissingField("aws.eventbridge.region"))
+			}
+		}
 	}
 
 	return errs
diff --git a/pkg/apis/sinks/v1alpha1/integration_sink_validation_test.go b/pkg/apis/sinks/v1alpha1/integration_sink_validation_test.go
@@ -203,6 +203,26 @@ func TestIntegrationSinkSpecValidation(t *testing.T) {
 			},
 			want: apis.ErrMissingField("aws.sns.arn"),
 		},
+		{
+			name: "AWS Eventbridge sink without TopicNameOrArn (invalid)",
+			spec: IntegrationSinkSpec{
+				Aws: &Aws{
+					EVENTBRIDGE: &v1alpha1.AWSEventbridge{
+						AWSCommon: v1alpha1.AWSCommon{
+							Region: "us-east-1",
+						},
+					},
+					Auth: &v1alpha1.Auth{
+						Secret: &v1alpha1.Secret{
+							Ref: &v1alpha1.SecretReference{
+								Name: "aws-secret",
+							},
+						},
+					},
+				},
+			},
+			want: apis.ErrMissingField("aws.eventbridge.arn"),
+		},
 		{
 			name: "no sink type specified (invalid)",
 			spec: IntegrationSinkSpec{},
@@ -255,6 +275,24 @@ func TestIntegrationSinkSpecValidation(t *testing.T) {
 			},
 			want: apis.ErrMissingField("aws.s3.region"),
 		},
+		{
+			name: "AWS Eventbridge sink without region (invalid)",
+			spec: IntegrationSinkSpec{
+				Aws: &Aws{
+					EVENTBRIDGE: &v1alpha1.AWSEventbridge{
+						Arn: "example-bus",
+					},
+					Auth: &v1alpha1.Auth{
+						Secret: &v1alpha1.Secret{
+							Ref: &v1alpha1.SecretReference{
+								Name: "aws-secret",
+							},
+						},
+					},
+				},
+			},
+			want: apis.ErrMissingField("aws.eventbridge.region"),
+		},
 	}
 
 	for _, test := range tests {
diff --git a/pkg/apis/sinks/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/sinks/v1alpha1/zz_generated.deepcopy.go
diff --git a/pkg/reconciler/integration/sink/resources/container_image.go b/pkg/reconciler/integration/sink/resources/container_image.go
@@ -411,6 +411,18 @@ func makeEnv(sink *v1alpha1.IntegrationSink, featureFlags feature.Flags) []corev
 		return envVars
 	}
 
+	// AWS Eventbridge environment variables
+	if sink.Spec.Aws != nil && sink.Spec.Aws.EVENTBRIDGE != nil {
+		envVars = append(envVars, integration.GenerateEnvVarsFromStruct("CAMEL_KAMELET_AWS_EVENTBRIDGE_SINK", *sink.Spec.Aws.EVENTBRIDGE)...)
+		if secretName != "" {
+			envVars = append(envVars, []corev1.EnvVar{
+				integration.MakeSecretEnvVar("CAMEL_KAMELET_AWS_EVENTBRIDGE_SINK_ACCESSKEY", commonv1a1.AwsAccessKey, secretName),
+				integration.MakeSecretEnvVar("CAMEL_KAMELET_AWS_EVENTBRIDGE_SINK_SECRETKEY", commonv1a1.AwsSecretKey, secretName),
+			}...)
+		}
+		return envVars
+	}
+
 	// If no valid configuration is found, return empty envVars
 	return envVars
 }
@@ -433,6 +445,8 @@ func selectImage(sink *v1alpha1.IntegrationSink) string {
 		return os.Getenv("INTEGRATION_SINK_AWS_SQS_IMAGE")
 	case sink.Spec.Aws != nil && sink.Spec.Aws.SNS != nil:
 		return os.Getenv("INTEGRATION_SINK_AWS_SNS_IMAGE")
+	case sink.Spec.Aws != nil && sink.Spec.Aws.EVENTBRIDGE != nil:
+		return os.Getenv("INTEGRATION_SINK_AWS_EVENTBRIDGE_IMAGE")
 	default:
 		return ""
 	}

Original file line number	Diff line number	Diff line change
`@@ -74,3 +74,8 @@ type AWSSNS struct {`
`74`	`74`	Arn string `json:"arn,omitempty" camel:"TOPIC_NAME_OR_ARN"` // SNS ARN
`75`	`75`	AutoCreateTopic bool `json:"autoCreateTopic" default:"false"` // Auto-create SNS topic
`76`	`76`	`}`
	`77`	`+`
	`78`	`+type AWSEventbridge struct {`
	`79`	+ AWSCommon `json:",inline"` // Embeds AWSCommon to inherit its fields in JSON
	`80`	+ Arn string `json:"arn,omitempty" camel:"EVENTBUS_NAME_OR_ARN"` // EB ARN
	`81`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -74,10 +74,11 @@ type Log struct {`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`type Aws struct {`
`77`		- S3 *v1alpha1.AWSS3 `json:"s3,omitempty"` // S3 sink configuration
`78`		- SQS *v1alpha1.AWSSQS `json:"sqs,omitempty"` // SQS sink configuration
`79`		- SNS *v1alpha1.AWSSNS `json:"sns,omitempty"` // SNS sink configuration
`80`		- Auth *v1alpha1.Auth `json:"auth,omitempty"`
	`77`	+ EVENTBRIDGE *v1alpha1.AWSEventbridge `json:"eventbridge,omitempty"` // EB sink configuration
	`78`	+ S3 *v1alpha1.AWSS3 `json:"s3,omitempty"` // S3 sink configuration
	`79`	+ SQS *v1alpha1.AWSSQS `json:"sqs,omitempty"` // SQS sink configuration
	`80`	+ SNS *v1alpha1.AWSSNS `json:"sns,omitempty"` // SNS sink configuration
	`81`	+ Auth *v1alpha1.Auth `json:"auth,omitempty"`
`81`	`82`	`}`
`82`	`83`
`83`	`84`	`type IntegrationSinkStatus struct {`