Subscription Control Proposal

[travis-minke-sap]

Knative Eventing - Subscription Control Proposal

NOTE - DISCUSSION "CLOSED" - SEE COMMENTS BELOW BEFORE READING!

The following proposal seeks to enhance the existing Knative Eventing Subscription API with optional configuration allowing Channel implementations to provide basic playback control such as Pause & Replay.

Due to their backing technologies, not all Channel implementations will be able to support these optional capabilities. The intent is that the API allow those that can, the ability to do so without requiring it of other implementations. While it is expected that all Subscriptions could support the "Pause" capability, most might not be able to support the "Replay" feature.

Problem Statement

The following use cases are driving the need for the proposed enhancements...

• Planned Subscriber Downtime: Subscribers often need to incur downtime for basic maintenance and upgrades. Subscribers would like to manage this downtime in a controlled pro-active manner without incurring message loss. Deleting and re-creating a Subscription is not adequate as the subscriber will lose its location in the event-stream and thus miss events. Currently, the way to manage this is to set a retry strategy where the retry time exceeds the expected downtime, or to send events to the deadLetterSink and manually process them via custom logic, neither of which is great.

• Unplanned Subscriber Downtime: Subscribers might also experience unplanned downtime due to bugs, hardware failures, event content incompatibility, etc. Upon correction of the failure's cause, the subscriber would like the ability to resend (replay) the missed events. They might want to do this in parallel with the processing of new/current events, or they might instead wish to back up and start again at the original failure point.

• Variable Starting Location: Independent of any downtime, a "new" subscriber might want to start receiving events starting from some prior point in time. Alternatively, it might want to just start receiving new events. Currently, there is no control over this, and the user is subject to the Channel implementations whims as to the point in time at which to start sending events. Usually this means receiving "new" events only. A recent example of this can be seen in Issue #420 and associated PR #428.

Proposal

The Subscription API changes being proposed are detailed in the following sections...

Features

The following new capabilities are being proposed...

• Pause: The ability to pro-actively "pause" (disable, inactivate, etc) and "resume" Subscriptions will allow administrators to manage planned-for subscriber downtime without suffering any event loss.

• Replay: The ability to "replay" old events provides a clean recovery for a subscriber which experienced an unexpected downtime. This replay might take several forms including the following...

  • The ability to revert a Subscription back to an earlier position in the history of events and start re-sending. The Subscription will eventually catch up and start processing new events again.
  • Leaving the current Subscription to process new events, while simultaneously creating another Subscription with delineated start / stop positions for which events to re-send in parallel.

• Initial Positioning: The "Replay" capability described above inherently includes the ability to create a new Subscription with a specific starting position.

Requirements

1. The proposed capabilities should be provided at the Subscription level to ensure fine-grained control.

2. Not all Channel implementations will be able to provide these capabilities, so all API changes should be marked as MAY or OPTIONAL leaving the decision up to Channel implementers.

3. Channel implementations would ideally document their ability to support these new features.

4. New configuration values should be open and generic enough to allow Channel implementations freedom as to their use. For example a KafkaChannel might expose the ability to specify timestamps or offsets for positional indicators, while another channel might require a different value altogether.

5. New configuration should be included in the Channel.spec.subscribers data, that is populated by the Subscription Controller, to facilitate implementation of these new features.

6. Subscription Status field should be expanded to indicate "enabled" state, thus impacting in the overall Subscription READY state.

Design

The following design is an initial rough-draft of how the above could be achieved. The exact field names or structures are open for improvement...

DeliverySpec Fields

The existing DeliverySpec duck type exists to control the actual "delivery" of events, and seems like the logical place to add new control fields. If, however, it is undesirable to include OPTIONAL fields in the DeliverySpec we could instead create a new ControlSpec structure to separate the two. The following new fields should allow the desired features to be controlled.

package v1

// DeliverySpec contains the delivery options for event senders,
// such as channelable and source.
type DeliverySpec struct {

	// ...existing content...

	// Enabled indicates whether or not delivery is currently allowed.
	// If not present it is assumed that event delivery is enabled.
	// Support for this flag is optional so check your specific event
	// sender (Channel, Source) to determine if it is supported.
	// +optional
	Enabled bool `json:"enabled,omitempty"`

	// StartPosition is an indicator of where in the event stream to start
	// sending from.  The value type is up to the specific event sender as
	// makes sense for the underlying technology.  For instance a
	// KafkaChannel might take a date, or an offset as the value, while
	// another Channel type might requires other information. Support
	// for this flag is optional so check your specific event sender
	// (Channel, Source) to determine if it is supported, and what
	// content is expected.
	//
	// This value is likely to be removed by the event sender once it
	// has started sending from the specified position.  This is to
	// prevent re-starting from the old position upon restart.  This
	// field allows for the ability to "replay" old events in Channels
	// capable of supporting that functionality.
	// +optional
	StartPosition string `json:"startPosition,omitempty"`

	// EndPosition is an indicator of where in the event stream to stop
	// sending.  The value type is up to the specific event sender as
	// makes sense for the underlying technology.  For instance a
	// KafkaChannel might take a date, or an offset as the value, while
	// another Channel type might requires other information. Support
	// for this flag is optional so check your specific event sender
	// (Channel, Source) to determine if it is supported, and what
	// content is expected.
	//
	// The presence of this value should prevent a Dispatcher from sending
	// any events beyond the specified position.  The Enabled flag might
	// still be true, and the Subscription status READY, but no events 
	// after the position will be sent.  This is to facilitate temporary
	// replay Subscriptions which process a range of events and then stop.
	// The user will be expected to remove such Subscriptions once their
	// purpose has been fulfilled.
	// +optional
	EndPosition string `json:"endPosition,omitempty"`
}

• Notes
  • The Enabled field could be renamed as "Paused" or "Active" or whatever term is preferred.
  • The StartPosition field is "temporal" in that it will be removed by the Channel Controller once it has "started" from that location. This is admittedly a bit odd, in that the user provides some config that is automatically removed, but is necessary to prevent successive restarts from that location upon restarts. Implementations would likely want to make use of the new control-protocol for coordinating the new positions between Controllers and Dispatchers.
  • Might make more sense for the StartPosition and EndPosition fields to be []byte instead of string?
  • Adding a StartPosition field to an existing Subscription should move its position in the event stream back to the specified location.
  • The EndPosition field is not temporal, and can block further event processing. If a user were to remove it then the Subscription would resume processing newer events.
  • While it might vary slightly per-Channel, the following control logic is expected: (Should this be part of the contract?)

    - None of the new fields are specified
        Everything works as it does today.
        Subscriptions start wherever Channel chooses (current / latest) and are always processing events.
    
    - Only StartPosition is specified
        Event stream is re-positioned and events processed forever as usual.
        StartPosition field will be removed to prevent accidental repositioning.
    
    - Only EndPosition is specified 
        If current position is less than the EndPosition 
        Then events will be processed until the EndPosition is reached.
        Otherwise event processing will cease immediately.
    
    - Both positions are specified
        If StartPosition < EndPosition
        Then resposition to StartPosition, remove StartPosition and play until reached EndPosition
        Else If StartPosition > EndPosition
        Then do not start processing events and sit idle and leave invalid config in place.        
    

Status Tracking

The proposed Enabled field above would be used (if present) to populate a new Subscription Status Condition which would impact the rolled-up value of the overall Ready status for the Subscription. The corresponding SubscriptionStatus lifecycle methods would be expanded to include a MarkSubscriptionEnabled() function which the Subscription Controller would make use of according to the current implementation.

Annotation History

In the case where a Subscription exists for a long time (months / years), and might have undergone several "pause" and "replay" cycles, it would be helpful for users of the system to be able to see a record of that history. A lightweight way to achieve this would be for Channels supporting these features to maintain this record as a custom annotation similar to the following...

annotations:
  messaging.knative.dev/control-history: |
  {
    { "timestamp": "2021-03-07T18:30:00Z", "type": "enabled", "value": "false" },
    { "timestamp": "2021-03-09T11:05:00Z", "type": "enabled", "value": "true" },
    { "timestamp": "2021-03-10T08:14:00Z", "type": "start", "position": "2021-03-06T23:59:59Z" }
    { "timestamp": "2021-03-10T08:14:00Z", "type": "stop", "position": "2021-03-07T23:59:59Z" }
  }

Alternatives

The following alternatives were considered but rejected due to inherent disadvantages. These ideas could be debated further or enhanced on if they warrant it. (Any other suggestions?)

• New Eventing CRD: Instead of modifying Subscriptions spec directly, a new Custom Resource (e.g. "SubscriptionControl") could be created to isolate this capability. This is very heavy-weight though, potentially requiring separate Controllers, and coordination with the actual Subscription custom resources out of band from the normal handling of the DeliverySpec configuration.

• Custom Annotations: If there is no Knative Eventing level API specified way to achieve the pausing and replay of events, then Channel implementations requiring this functionality would be left to create their own custom set of Kubernetes annotations. This would also perturb the lifecycle as it would likely require watching Subscriptions directly instead of relying on the Channel.spec.subscribers block.

• Non YAML API: Alternatively a Channel implementation could expose HTTP endpoints as an interface for these "control" operations. This API could be one-off per Channel or standardized in Knative, but this introduces a new control paradigm which might not be desirable?

Resources

• Knative Specifications
  • Knative Channel Specification
  • Knative Subscription Specification

[n3wscott]

See https://docs.google.com/document/d/1SyUGcVQpNZU3szALmHYNNEbTW37B-5sVEgES7ZLm9vA/edit?resourcekey=0-ZX15LAKLG_QD3q_jZF-5Gg

[n3wscott]

This is also why @slinkydeveloper wrote https://github.com/knative-sandbox/control-protocol

[travis-minke-sap]

Yeah, I mentioned in the write-up that using the control-protocol as a means to implement this inside a Channel/Dispatcher would be a good option. Or did you instead mean that the control-protocol would be exposed for outside use/interaction instead of exposing new configuration/control parameters?

[travis-minke-sap]

See https://docs.google.com/document/d/1SyUGcVQpNZU3szALmHYNNEbTW37B-5sVEgES7ZLm9vA/edit?resourcekey=0-ZX15LAKLG_QD3q_jZF-5Gg

Thanks for that - sorry I had missed it when writing this up. It appears that the document describes a richer, more rigorous version of the Enabled field proposed above which seems great! I did add a question/comment to the document asking whether Subscriptions could also be "paused"? We have multi-tenant Channels and pausing the entire Channel would not be feasible if only one Subscriber was experiencing downtime and needed to be paused.

Also... My most pressing need is to provide the ability to Replay events from the distributed KafkaChannel. Either of the replay "modes" described above will suffice for now. Essentially we need the ability to reposition the ConsumerGroup offsets in the Dispatchers. Since this requires some persistence in the Channel's backing technology not all Channels will be able to support it... Do we think this is something that should even be considered in the API or just something KafkaChannels can choose to expose/support? (Point me in the right direction please ; )

[travis-minke-sap]

OK... after some side conversations in Slack it seems that...

• The "Pause" feature will be designed in @n3wscott 's document which is further along and provides more detailed consideration of other Eventing components (Sources, Brokers, etc).
• The "Replay" feature (re-positioning in the event stream) is not expected to be widely enough supported by various implementations to warrant inclusion in the larger Eventing API, and will instead be handled as a custom feature of the KafkaChannel(s). This does restrict options somewhat, but also opens new possibilities. I will pursue a different design specific to the eventing-kafka repository instead.

...and so we can consider this discussion... "closed" ; )

[travis-minke-sap]

Next attempt is to only to replay in eventing-kafka, but trying to find agreeable solution to larger issue of how to expose dynamic / ephemeral control apis in knative / kubernetes...

eventing-kafka replay issue
eventing control api discussion