[KOGITO-7754] create Knative eventing resources

Makefile

@@ -186,7 +186,7 @@ docker-buildx: test ## Build and push docker image for the manager for cross-pla
 	sed -e '1 s/\(^FROM\)/FROM --platform=\$$\{BUILDPLATFORM\}/; t' -e ' 1,// s//FROM --platform=\$$\{BUILDPLATFORM\}/' Dockerfile > Dockerfile.cross
 	- docker buildx create --name project-v3-builder
 	docker buildx use project-v3-builder
-	- docker buildx build --push --platform=$(PLATFORMS) --tag ${IMG} -f Dockerfile.cross
+	- docker buildx build --platform=$(PLATFORMS) --push . -f Dockerfile.cross --tag ${IMG}
 	- docker buildx rm project-v3-builder
 	rm Dockerfile.cross
 

api/v1alpha08/sonataflow_types.go

@@ -657,6 +657,9 @@ type SonataFlowSpec struct {
 	// PodTemplate describes the deployment details of this SonataFlow instance.
 	//+operator-sdk:csv:customresourcedefinitions:type=spec,displayName="podTemplate"
 	PodTemplate PodTemplateSpec `json:"podTemplate,omitempty"`
+	// Sink describes the sinkBinding details of this SonataFlow instance.
+	//+operator-sdk:csv:customresourcedefinitions:type=spec,displayName="sink"
+	Sink *duckv1.Destination `json:"sink,omitempty"`
 }
 
 // SonataFlowStatus defines the observed state of SonataFlow

bundle/manifests/sonataflow-operator.clusterserviceversion.yaml

@@ -259,6 +259,9 @@ spec:
           definition. For example, a collection of OpenAPI specification files.
         displayName: resources
         path: resources
+      - description: Sink describes the sinkBinding details of this SonataFlow instance.
+        displayName: sink
+        path: sink
       statusDescriptors:
       - description: Address is used as a part of Addressable interface (status.address.url)
           for knative
@@ -347,6 +350,36 @@ spec:
           - patch
           - update
           - watch
+        - apiGroups:
+          - eventing.knative.dev
+          resources:
+          - triggers
+          - triggers/status
+          - triggers/finalizers
+          verbs:
+          - create
+          - delete
+          - deletecollection
+          - get
+          - list
+          - patch
+          - update
+          - watch
+        - apiGroups:
+          - sources.knative.dev
+          resources:
+          - sinkbindings
+          - sinkbindings/status
+          - sinkbindings/finalizers
+          verbs:
+          - create
+          - delete
+          - deletecollection
+          - get
+          - list
+          - patch
+          - update
+          - watch
         - apiGroups:
           - coordination.k8s.io
           resources:

bundle/manifests/sonataflow.org_sonataflows.yaml

@@ -9287,6 +9287,53 @@ spec:
                       type: object
                     type: array
                 type: object
+              sink:
+                description: Sink describes the sinkBinding details of this SonataFlow
+                  instance.
+                properties:
+                  CACerts:
+                    description: CACerts are Certification Authority (CA) certificates
+                      in PEM format according to https://www.rfc-editor.org/rfc/rfc7468.
+                      If set, these CAs are appended to the set of CAs provided by
+                      the Addressable target, if any.
+                    type: string
+                  ref:
+                    description: Ref points to an Addressable.
+                    properties:
+                      address:
+                        description: Address points to a specific Address Name.
+                        type: string
+                      apiVersion:
+                        description: API version of the referent.
+                        type: string
+                      group:
+                        description: 'Group of the API, without the version of the
+                          group. This can be used as an alternative to the APIVersion,
+                          and then resolved using ResolveGroup. Note: This API is
+                          EXPERIMENTAL and might break anytime. For more details:
+                          https://github.com/knative/eventing/issues/5086'
+                        type: string
+                      kind:
+                        description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                        type: string
+                      name:
+                        description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                        type: string
+                      namespace:
+                        description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/
+                          This is optional field, it gets defaulted to the object
+                          holding it if left out.'
+                        type: string
+                    required:
+                    - kind
+                    - name
+                    type: object
+                  uri:
+                    description: URI can be an absolute URL(non-empty scheme and non-empty
+                      host) pointing to the target or a relative URI. Relative URIs
+                      will be resolved using the base URI retrieved from Ref.
+                    type: string
+                type: object
             required:
             - flow
             type: object

config/crd/bases/sonataflow.org_sonataflows.yaml

@@ -9288,6 +9288,53 @@ spec:
                       type: object
                     type: array
                 type: object
+              sink:
+                description: Sink describes the sinkBinding details of this SonataFlow
+                  instance.
+                properties:
+                  CACerts:
+                    description: CACerts are Certification Authority (CA) certificates
+                      in PEM format according to https://www.rfc-editor.org/rfc/rfc7468.
+                      If set, these CAs are appended to the set of CAs provided by
+                      the Addressable target, if any.
+                    type: string
+                  ref:
+                    description: Ref points to an Addressable.
+                    properties:
+                      address:
+                        description: Address points to a specific Address Name.
+                        type: string
+                      apiVersion:
+                        description: API version of the referent.
+                        type: string
+                      group:
+                        description: 'Group of the API, without the version of the
+                          group. This can be used as an alternative to the APIVersion,
+                          and then resolved using ResolveGroup. Note: This API is
+                          EXPERIMENTAL and might break anytime. For more details:
+                          https://github.com/knative/eventing/issues/5086'
+                        type: string
+                      kind:
+                        description: 'Kind of the referent. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds'
+                        type: string
+                      name:
+                        description: 'Name of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/names/#names'
+                        type: string
+                      namespace:
+                        description: 'Namespace of the referent. More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/namespaces/
+                          This is optional field, it gets defaulted to the object
+                          holding it if left out.'
+                        type: string
+                    required:
+                    - kind
+                    - name
+                    type: object
+                  uri:
+                    description: URI can be an absolute URL(non-empty scheme and non-empty
+                      host) pointing to the target or a relative URI. Relative URIs
+                      will be resolved using the base URI retrieved from Ref.
+                    type: string
+                type: object
             required:
             - flow
             type: object

config/manifests/bases/sonataflow-operator.clusterserviceversion.yaml

@@ -156,6 +156,9 @@ spec:
           definition. For example, a collection of OpenAPI specification files.
         displayName: resources
         path: resources
+      - description: Sink describes the sinkBinding details of this SonataFlow instance.
+        displayName: sink
+        path: sink
       statusDescriptors:
       - description: Address is used as a part of Addressable interface (status.address.url)
           for knative

config/rbac/builder_role.yaml

@@ -57,6 +57,36 @@ rules:
   resources:
     - roles
     - rolebindings
+  verbs:
+    - create
+    - delete
+    - deletecollection
+    - get
+    - list
+    - patch
+    - update
+    - watch
+- apiGroups:
+    - eventing.knative.dev
+  resources:
+    - triggers
+    - triggers/status
+    - triggers/finalizers
+  verbs:
+    - create
+    - delete
+    - deletecollection
+    - get
+    - list
+    - patch
+    - update
+    - watch
+- apiGroups:
+    - sources.knative.dev
+  resources:
+    - sinkbindings
+    - sinkbindings/status
+    - sinkbindings/finalizers
   verbs:
     - create
     - delete

controllers/discovery/discovery.go

@@ -80,33 +80,33 @@ type ServiceCatalog interface {
 	Query(ctx context.Context, uri ResourceUri, outputFormat string) (string, error)
 }
 
-type sonataFlowServiceCatalog struct {
+type SonataFlowServiceCatalog struct {
 	kubernetesCatalog ServiceCatalog
-	knativeCatalog    ServiceCatalog
+	KnativeCatalog    ServiceCatalog
 	openshiftCatalog  ServiceCatalog
 }
 
 // NewServiceCatalog returns a new ServiceCatalog configured to resolve kubernetes, knative, and openshift resource addresses.
 func NewServiceCatalog(cli client.Client, knDiscoveryClient *KnDiscoveryClient) ServiceCatalog {
-	return &sonataFlowServiceCatalog{
+	return &SonataFlowServiceCatalog{
 		kubernetesCatalog: newK8SServiceCatalog(cli),
-		knativeCatalog:    newKnServiceCatalog(knDiscoveryClient),
+		KnativeCatalog:    newKnServiceCatalog(knDiscoveryClient),
 	}
 }
 
 func NewServiceCatalogForConfig(cli client.Client, cfg *rest.Config) ServiceCatalog {
-	return &sonataFlowServiceCatalog{
+	return &SonataFlowServiceCatalog{
 		kubernetesCatalog: newK8SServiceCatalog(cli),
-		knativeCatalog:    newKnServiceCatalogForConfig(cfg),
+		KnativeCatalog:    newKnServiceCatalogForConfig(cfg),
 	}
 }
 
-func (c *sonataFlowServiceCatalog) Query(ctx context.Context, uri ResourceUri, outputFormat string) (string, error) {
+func (c *SonataFlowServiceCatalog) Query(ctx context.Context, uri ResourceUri, outputFormat string) (string, error) {
 	switch uri.Scheme {
 	case KubernetesScheme:
 		return c.kubernetesCatalog.Query(ctx, uri, outputFormat)
 	case KnativeScheme:
-		return c.knativeCatalog.Query(ctx, uri, outputFormat)
+		return c.KnativeCatalog.Query(ctx, uri, outputFormat)
 	case OpenshiftScheme:
 		return "", fmt.Errorf("openshift service discovery is not yet implemented")
 	default:

controllers/profiles/common/constants/workflows.go

@@ -16,4 +16,10 @@ package constants
 
 const (
 	MicroprofileServiceCatalogPropertyPrefix = "org.kie.kogito.addons.discovery."
+	OutgoingEventsURL                        = "mp.messaging.outgoing.kogito_outgoing_stream.url"
+	OutgoingEventsConnector                  = "mp.messaging.outgoing.kogito_outgoing_stream.connector"
+	IncomingEventsConnector                  = "mp.messaging.incoming.kogito_incoming_stream.connector"
+	IncomingEventsPath                       = "mp.messaging.incoming.kogito_incoming_stream.path"
+	KnativeInjectedEnvVar                    = "${K_SINK}"
+	KnativeEventingBrokerDefault             = "default"
 )

controllers/profiles/common/ensurer.go

@@ -32,6 +32,7 @@ import (
 
 var _ ObjectEnsurer = &defaultObjectEnsurer{}
 var _ ObjectEnsurer = &noopObjectEnsurer{}
+var _ ObjectsEnsurer = &defaultObjectsEnsurer{}
 
 type ObjectEnsurer interface {
 	Ensure(ctx context.Context, workflow *operatorapi.SonataFlow, visitors ...MutateVisitor) (client.Object, controllerutil.OperationResult, error)
@@ -66,22 +67,10 @@ func (d *defaultObjectEnsurer) Ensure(ctx context.Context, workflow *operatorapi
 	result := controllerutil.OperationResultNone
 
 	object, err := d.creator(workflow)
-	if err != nil {
+	if err != nil || object == nil {
 		return nil, result, err
 	}
-	if result, err = controllerutil.CreateOrPatch(ctx, d.c, object,
-		func() error {
-			for _, v := range visitors {
-				if visitorErr := v(object)(); visitorErr != nil {
-					return visitorErr
-				}
-			}
-			return controllerutil.SetControllerReference(workflow, object, d.c.Scheme())
-		}); err != nil {
-		return nil, result, err
-	}
-	klog.V(log.I).InfoS("Object operation finalized", "result", result, "kind", object.GetObjectKind().GroupVersionKind().String(), "name", object.GetName(), "namespace", object.GetNamespace())
-	return object, result, nil
+	return ensureObject(ctx, workflow, visitors, result, err, d.c, object)
 }
 
 // NewNoopObjectEnsurer see noopObjectEnsurer
@@ -97,3 +86,61 @@ func (d *noopObjectEnsurer) Ensure(ctx context.Context, workflow *operatorapi.So
 	result := controllerutil.OperationResultNone
 	return nil, result, nil
 }
+
+// ObjectsEnsurer is an ensurer to apply multiple objects
+type ObjectsEnsurer interface {
+	Ensure(ctx context.Context, workflow *operatorapi.SonataFlow, visitors ...MutateVisitor) []ObjectEnsurerResult
+}
+
+type ObjectEnsurerResult struct {
+	client.Object
+	Result controllerutil.OperationResult
+	Error  error
+}
+
+func NewObjectsEnsurer(client client.Client, creator ObjectsCreator) ObjectsEnsurer {
+	return &defaultObjectsEnsurer{
+		c:       client,
+		creator: creator,
+	}
+}
+
+type defaultObjectsEnsurer struct {
+	ObjectsEnsurer
+	c       client.Client
+	creator ObjectsCreator
+}
+
+func (d *defaultObjectsEnsurer) Ensure(ctx context.Context, workflow *operatorapi.SonataFlow, visitors ...MutateVisitor) []ObjectEnsurerResult {
+	result := controllerutil.OperationResultNone
+
+	objects, err := d.creator(workflow)
+	if err != nil {
+		return []ObjectEnsurerResult{{nil, result, err}}
+	}
+	var ensureResult []ObjectEnsurerResult
+	for _, object := range objects {
+		ensureObject, c, err := ensureObject(ctx, workflow, visitors, result, err, d.c, object)
+		ensureResult = append(ensureResult, ObjectEnsurerResult{ensureObject, c, err})
+		if err != nil {
+			return ensureResult
+		}
+	}
+	return ensureResult
+}
+
+func ensureObject(ctx context.Context, workflow *operatorapi.SonataFlow, visitors []MutateVisitor, result controllerutil.OperationResult, err error, c client.Client, object client.Object) (client.Object, controllerutil.OperationResult, error) {
+	if result, err = controllerutil.CreateOrPatch(ctx, c, object,
+		func() error {
+			for _, v := range visitors {
+				if visitorErr := v(object)(); visitorErr != nil {
+					return visitorErr
+				}
+			}
+			return controllerutil.SetControllerReference(workflow, object, c.Scheme())
+		}); err != nil {
+		return nil, result, err
+	}
+	klog.V(log.I).InfoS("Object operation finalized", "result", result, "kind", object.GetObjectKind().GroupVersionKind().String(), "name", object.GetName(), "namespace", object.GetNamespace())
+	return object, result, nil
+}