Knative Eventing Working Group
Last Updated: Jan 27th, 2022
Main GitHub Issue: 1521
Contributor(s):
Knative Eventing flows constructs (Sequence and Parallel) are not composable, considerably limiting their usefulness. For instance, a Parallel branch cannot reference an existing Sequence and forward the result to the next step to, let's say, aggregate the results. This limitation cannot be easily solved at the application layer (for instance by calling a subflow) mostly due to the dependency on inner workings, such as reference resolution, to forward the result to an object somewhere in the middle of a flow. The only reasonable solution is for Knative Eventing to support this pattern out-of-the-box because 1. it is a very common pattern 2. it not something that can easily be done at the application level.
Knative Eventing provides two flows constructs, Sequence and Parallel. Both constructs allow references to external sinks with the implicit assumption that those sinks are callable (i.e. return 0 or 1 event) in order for the flow to keep going when an event is received and to be interrupted when no event is received.
For instance, consider this sequence:
apiVersion: flows.knative.dev/v1
kind: Sequence
metadata:
name: oneseq
spec:
steps:
- ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: identity
- ref:
apiVersion: serving.knative.dev/v1
kind: Service
name: last-step
The second step is only executed when the service in the first step synchronously returns a non-empty event. When the first step is a reference to a flow construct, never immediately returning an event, the second step is consequently never executed.
The proposal below leverages existing Knative Eventing capabilities, specifically error handling and delivery guarantees.
Under the cover, both Sequence and Parallel get realized as a set of Channel and Subscription objects by their respective reconciler.
Current flow constructs are not Callable, but they eventually produce zero or one event via reply
. We propose to call this type of objects composable:
status.address.url
field.spec.reply
field.In addition, sink objects can potentially invoked in two different ways:
Asynchronous invocation of both Callable and Composable objects is straightforward. The real challenge is synchronous invocation of composable objects, as discussed below.
A simple approach is to clone the referenced composable object and to inject the address of the next step into the spec.reply
field. While this solution could work (modulo some open questions like what happens when spec.reply
is already specified?), it leads to a lot of additional objects being created, all connected the same way (same topology) except for the last spec.reply
.
An alternative solution is to dynamic dispatch flow construct results to the next step. This can be achieved by maintaining a stack of callers (e.g. by using a CloudEvent extension attribute) and by adding stack manipulation functions, one for the flow entry (push) and one for the flow exit (pop). This is how most compilers handle function calls.
TODO: explain how errors are handled, maybe from a continuation passing style PoV.
The Composable duck-type is defined as follows:
type Composable struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec ComposableSpec `json:"spec,omitempty"`
Status ComposableStatus `json:"status,omitempty"`
}
// ComposableSpec contains Spec of the Composable object
type ComposableSpec struct {
// Reply is a Reference to where the Composable result
// is sent to
// +optional
Reply *duckv1.Destination `json:"reply,omitempty"`
}
// ComposableStatus contains the Status of a Composable object.
type ComposableStatus struct {
// AddressStatus is the part where the Composable fulfills the Addressable contract.
// +optional
duckv1.AddressStatus `json:",inline"`
}
The call stack is represented by a CloudEvent attribute. Let's call it knativeflowcallstack
. Its value is a space-separated list of resolved URLs.
Knative Eventing should enforce the limit on the maximum number of nested calls. A global configuration parameter will be added.
In this proposal the flow state is stored within the CloudEvent knativeflowcallstack
extension attribute, which is visible to applications, possibly leading to meta-data corruption. A solution to this problem is to remove and reinject this attribute before and after a non-composable function is invoked.
This solution relies on two managed stateless functions: flow-push
and flow-pop
. The first one manipulates the received knativeflowcallstack
by adding a new URL, passed as an HTTP query parameter, to the end of the list and directly forwards the modified event to the specified reference.
The second service removes the last URL from the received CloudEvent and directly forwards the modified event to the URL.
Side comment: these services could be removed if Knative Eventing Subscription would support ceOverrides
over list, or if it would support capabilities negociation.
When the referenced composable is asynchronously invoked, the flow reconciler configures the underlying Subscription to point to the built-in flow-push
function. Here an example of a generated Subscription:
apiVersion: messaging.knative.dev/v1
kind: Subscription
metadata:
name: <subscription-name> # Name of the Subscription.
spec:
subscriber:
uri: https://flow-push.knative-eventing?URL=<resolved URI>
On the egress, the flow operator checks the spec.reply
field is properly set to send event to the flow-pop
built-in function. If not, the reconciliation process fails.
TBD
TBD
TBD
TBD
TBD
As suggested in this comment, extend the delivery contract to include an optional Reply-Location
header that indicates a URL where reply events may be POSTed (instead of, or in addition to, being processed from the HTTP Response).
Reply-Location
would not be a CloudEvents attribute; it would be a hop-by-hop HTTP option on the delivery, similar to the Prefer: Reply
header (and possibly set in similar circumstances).
Pros:
Cons:
Reply-Location
would need to ensure that the event was persisted to stable storage before returning a 200.Reply-Location
information might need to be persisted across multiple Channel hops in a Sequence or Parallel, which is not supported in the current implementation.Reply-Location
set on a request and spec.reply
in Sequence or Parallel. (It seems to me that Broker and Channel would reasonably ignore this parameter.)