Architecture Proposal: Output Topic

[coltmcnealy-lh]

Motivation

The Output Topic will allow users of LittleHorse to export data in real-time from their LittleHorse Workflows into external systems.

On a personal note, when I started the LittleHorse Server project over three years ago, I did it with the intention of bridging the gap between Workflows, Streams, and Tables.

Workflows as Tables

I believe that Workflows are Data. For example, consider the following orders workflow:

var userId = wf.declareStr("user-id");
var itemId = wf.declareStr("item-id");
var orderStatus = wf.declareStr("order-status").withDefault("PENDING");

wf.execute("charge-credit-card", userId, wf.execute("fetch-price", itemId));
wf.execute("ship-item", userId, itemId);
orderStatus.assign("SHIPPING");

wf.waitForEvent("item-delivered");
orderStatus.assign("COMPLETED");

If you wanted to "export this workflow" into a database such as Postgres or Snowflake, you might create a database table that looks like the following:

CREATE TABLE orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id TEXT NOT NULL,
    item_id TEXT NOT NULL,
    order_status TEXT NOT NULL DEFAULT 'PENDING',
    created_at TIMESTAMPTZ DEFAULT now(),
    updated_at TIMESTAMPTZ DEFAULT now()
);

And then insert a new row for every single WfRun. This would allow you to do analytics based on your orders.

This can be accomplished with an Output Topic that publishes updates to WfRun data in real time to Apache Kafka.

Workflows as Streams

Another motivation for the Output Topic is that updates to your WfRuns can be streams of events. For example, the following use-cases have come up:

• Notifications when a TaskDef fails five times in a minute.
• Trigger an "incident response" WfSpec when 10 WfRuns of a specific type reach a certain failure scenario in a given time window.
• Send an alert to the business team when there is an excessively long backlog of UserTaskRuns assigned to the same group.
• Create "cascading workflows," wherein the completion of one WfRun triggers another WfRun in a loosely-coupled manner.

The above can also be accomplished as well thorugh Kafka.

Topic Structure

Metadata and Execution Data

There are two types of data in LittleHorse:

1. Metadata, such as WfSpec, TaskDef, etc.
2. Execution data, such as WfRun, TaskRun, etc.

Metadata is small, relatively static, and global to a cluster. Execution data is large, partitioned, and constantly changing. Consumers doing stream processing on Execution data will often need access to Metadata in order to properly make sense of the Execution Data.

Therefore, we will separate metadata and execution data into two topics:

1. The metatadata-output topic, which is a single-partition, compacted topic containing metadata updates.
2. The execution-output topic, which is a multi-partition, non-compacted topic containing execution data updates.

This will allow stream processors to load the current metadata snapshot through the compacted topic (think of a Kafka Streams Global Store), and then join the Execution Data against that snapshot in real time.

Note that most metadata in LittleHorse is immutable—when you want to change it, you end up creating a new version, which is a separate LittleHorse API Object with its own ID—so historical version mismatching shouldn't be a problem if the consumer is up-to-date on metadata but way behind on execution data.

Multi-Tenancy

There are a few considerations regarding topic structure, ownership, and multi-tenancy:

• LittleHorse is multi-tenant: a certain Principal might be able to do something in Tenant A but not in B.
• Kafka topics and partitions are not "free." There is a metadata overhead to each topic in Kafka.
• In Kafka, it is not possible to give a client permission to read certain messages in a topic but not others. However, you can allow a certain client to read one topic but not another.

Due to the above reasons, I propose that:

1. Each LittleHorse Tenant gets its own Output Topics (one for metadata and execution data).
2. We utilize protobuf oneofs to allow putting all data into the two topics above, and clients can filter it out as needed.

This prevents an expensive proliferation of Kafka topics and partitions as much as possible while still allowing different LittleHorse Tenants to have isolated data.

Proto Schemas

Naturally, LittleHorse is a protobuf-first system. The output topic will inherit this characteristic.

Tenant

Users should be able to enable or disable the Output Topic on a per-tenant basis.

// Configurations for the Output Topic of a certain Tenant.
message OutputTopicConfig {
    // Set to true if the Output Topic is enabled.
    bool enabled = 1;
}

message Tenant {
    // ...

    // Configurations related to the Output Topic.
    optional OutputTopicConfig output_topic = 3;
}

message PutTenantRequest {
    // ...

    // Configures the behavior of the Output Topic for this Tenant.
    optional OutputTopicConfig output_topic_config = 2;
}

Output Topic Schemas

Every message in the execution-output topic will be an OutputTopicRecord, and we will make heavy use of oneof to allow multiple data types.

The initial implementation will allow five types of records to be pushed into the Output Topic. However, we can always extend with more.

1. Notifications when a taskrun is executed (failed, completed).
2. Any WorkflowEvents thrown by a WfRun.
3. Notifications about changes to a WfRun which is treated as a data entity.
4. Updates for a UserTaskRun.
5. Updates about a specific Variable.

// An OutputTopicRecord is a single record in the output topic, which can
// denote one of several different types of events.
message OutputTopicRecord {
    // The ID of the WfRun that produced this record.
    WfRunId id = 1;

    // The time at which the event occurred.
    google.protobuf.Timestamp timestamp = 2;

    oneof payload {
        // Records the results of a TaskRun in the Output Topic.
        TaskRunExecutedRecord task_run_executed = 3;

        // Records a WorkflowEvent that was thrown into the Output Topic.
        WorkflowEventRecord workflow_event = 4;

        // Records an update to a WfRun which is treated as an "Entity", with
        // the public variables of the WfRun treated as fields in the Entity.
        WfRunEntityRecord wf_run_entity = 5;

        // Updates about a user task run.
        UserTaskRunUpdateRecord user_task_run = 6;

        // Updates about a specific Variable changing.
        VariableUpdateRecord variable_update = 7;
    }
}

// Record to state that a TaskRun was started.
message TaskRunExecutedRecord {
    // The TaskRun that was executed. All information about TaskAttempts,
    // input variables, start times, failures, etc is included in the
    // TaskRun itself.
    TaskRun task_run = 1;
}

// Record in the Output Topic to denote that a WorkflowEvent was thrown
// by a WfRun.
message WorkflowEventRecord {
    // The WorkflowEvent that was thrown.
    WorkflowEvent workflow_event = 1;

    // The WfSpecId for the WfRun that threw the WorkflowEvent.
    WfSpecId wf_spec_id = 2;
}

// Represents a snapshot of a WfRun as an entity. Used in the Output Topic
// to allow exporting a WfRun's public variables into external systems. This
// only includes Variables that are of type `PUBLIC_VAR` and in the entrypoint
// ThreadRun.
message WfRunEntityRecord {
    // The current status of the WfRun.
    LHStatus status = 1;

    // Denotes the reason that this `WfRunEntityRecord` was recorded to the
    // Output Topic.
    enum UpdateReason {
        // Recorded because the `WfRun` was started.
        WFRUN_STARTED = 0;

        // Recorded because a public variable was changed.
        VARIABLE_CHANGED = 1;

        // Recorded because the status of the `WfRun` was changed.
        WFRUN_STATUS_CHANGED = 2;
    }

    // The public Variables in the entrypoint ThreadRun
    map<String, Variable> public_variables = 2;

    // The reason(s) for this update.
    repeated UpdateReason update_reasons = 3;
}

// Represents a snapshot of a UserTaskRun being updated. Used in the Output Topic
// to allow exporting information about User Tasks into external systems.
message UserTaskRunUpdateRecord {
    // The current snapshot of the UserTaskRun.
    UserTaskRun user_task_run = 1;

    // Denotes the reason for a UserTaskRunUpdateReason to be recorded to the Output
    // Topic.
    enum UserTaskRunUpdateReason {
        // UserTaskRun was created.
        USER_TASK_CREATED = 0;

        // UserTaskRun was assigned.
        USER_TASK_ASSIGNED = 1;

        // UserTaskRun was saved.
        USER_TASK_SAVED = 2;

        // UserTaskRun was canceled.
        USER_TASK_CANCELED = 3;

        // UserTaskRun was completed.
        USER_TASK_COMPLETED = 4;
    }

    // The reason(s) for the UserTaskRunUpdateRecord to be recorded into the Output
    // Topic.
    repeated UserTaskRunUpdateReason reasons = 2;
}

// Represents a snapshot of an individual Variable being updated. Used in the Output
// Topic to allow exporting information for specific Variables to external systems.
message VariableUpdateRecord {
    // The cuurrent snapshot of the Variable in question.
    Variable current_variable = 1;
}

Metadata Output Topic

// TODO

Configuring What's Sent

WfRun Entities

Background: Public Variables

Configuring Recording Levels

WorkflowEvent records

TaskRun records

UserTaskRun records

Implementation

Future Work