[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/condition_types.go

@@ -38,7 +38,8 @@ const (
 	// SucceedConditionType describes the readiness condition of a static resource, like a platform, a builder, a configuration, etc.
 	SucceedConditionType ConditionType = "Succeed"
 	// BuiltConditionType describes the condition of a resource that needs to be build.
-	BuiltConditionType ConditionType = "Built"
+	BuiltConditionType            ConditionType = "Built"
+	KnativeResourcesConditionType ConditionType = "KnativeResourcesReady"
 )
 
 const (
@@ -54,6 +55,9 @@ const (
 	BuildSkippedReason              = "BuildSkipped"
 	BuildSuccessfulReason           = "BuildSuccessful"
 	BuildMarkedToRestartReason      = "BuildMarkedToRestart"
+	WaitingForKnativeReason         = "WaitingForKnative"
+	KnativeSkippedReason            = "KnativeSkipped"
+	KnativeFailureReason            = "KnativeFailure"
 )
 
 // Condition describes the common structure for conditions in our types

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
@@ -690,7 +693,7 @@ func (s *SonataFlowStatus) GetTopLevelCondition() *api.Condition {
 }
 
 func (s *SonataFlowStatus) Manager() api.ConditionsManager {
-	return api.NewConditionManager(s, api.RunningConditionType, api.BuiltConditionType)
+	return api.NewConditionManager(s, api.RunningConditionType, api.BuiltConditionType, api.KnativeResourcesConditionType)
 }
 
 func (s *SonataFlowStatus) IsWaitingForPlatform() bool {
@@ -731,6 +734,26 @@ func (s *SonataFlowStatus) IsBuildFailed() bool {
 	return cond.IsFalse() && cond.Reason == api.BuildFailedReason
 }
 
+func (s *SonataFlowStatus) IsWaitingForKnative() bool {
+	cond := s.GetCondition(api.KnativeResourcesConditionType)
+	return cond.IsFalse() && cond.Reason == api.WaitingForKnativeReason
+}
+
+func (s *SonataFlowStatus) IsKnativeFailed() bool {
+	cond := s.GetCondition(api.KnativeResourcesConditionType)
+	return cond.IsFalse() && cond.Reason == api.KnativeFailureReason
+}
+
+func (s *SonataFlowStatus) IsKnativeReady() bool {
+	cond := s.GetCondition(api.KnativeResourcesConditionType)
+	return cond.IsTrue()
+}
+
+func (s *SonataFlowStatus) IsKnativeUnknown() bool {
+	cond := s.GetCondition(api.KnativeResourcesConditionType)
+	return cond.IsUnknown()
+}
+
 // SonataFlow is the descriptor representation for a workflow application based on the CNCF Serverless Workflow specification.
 // +kubebuilder:object:root=true
 // +kubebuilder:object:generate=true

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,9 @@ 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}"
 )