owned this note   changed 3 years ago  
Published Linked with GitHub

Composable flows

Knative Eventing Working Group

Last Updated: Jan 27th, 2022

Main GitHub Issue: 1521

Contributor(s):

Motivation / Abstract

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.

Background

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.

Proposal Design / Approach

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:

  • Composable objects are Addressable, i.e. an event can be delivered over HTTP to an address defined in their status.address.url field.
  • Composable objects can asynchronously forward events over HTTP to an address specified in their spec.reply field.

In addition, sink objects can potentially invoked in two different ways:

  • async mode: The current event is forwarded to both the sink and the next step. Any event returned by the sink ingress is dropped.
  • sync mode: The event(s) produced by the sink (synchronously or asynchronously) are forwarded to the next step.

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.

Implementation

Composable duck-type

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"`
}

Asynchronous Composable Invocation

Call Stack

The call stack is represented by a CloudEvent attribute. Let's call it knativeflowcallstack. Its value is a space-separated list of resolved URLs.

Maximum size

Knative Eventing should enforce the limit on the maximum number of nested calls. A global configuration parameter will be added.

Security and integrity

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.

Data-Plane

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.

Control-Plane

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.

Integration Checklist

Operations

TBD

Observability

TBD

Test Plan

TBD

Documentation

TBD

User Experience

TBD

Alternative Proposals

Delivery Contract Extension

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:

  • Possibly simpler implementation vs a call stack
  • Also applies to long-lived delivery to a Trigger or Subscription Target
  • Potentially provides an avenue for returning multiple response events to a single event.

Cons:

  • Recipients using 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.
  • This does not avoid the conflict between having 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.)
Select a repo