Tangle Orchestration System

Tangle orchestrates containerized command-line programs using a graph-based model:

• Component: A description of a containerized command-line program. It includes a ComponentSpec/component.yaml defining its signature (inputs/outputs), metadata, and implementation (container image, program to start, and how to map inputs/outputs to CLI arguments). Components can be written in any language (Python, Shell, R, etc.).
• Task: An instance of a component with specific input arguments. Tasks are connected by linking the output of one task to the input of another.
• Pipeline: A graph of interconnected tasks. When a pipeline is submitted for execution, Tangle runs the tasks (in parallel where possible) and manages the transfer of output artifacts between them.

This architecture is compatible with Google Cloud Vertex AI Pipelines and Kubeflow Pipelines.

To run Tangle locally, you need Docker and uv installed. Follow these steps to clone the necessary repositories and start the application:

1. Download the app code (required once):

   git clone https://github.com/TangleML/tangle.git tangle/backend --branch stable
   git clone https://github.com/TangleML/tangle-ui.git tangle/ui_build --branch stable_local_build

2. Start the app:

   • Linux and Mac OS:

     cd tangle && backend/start_local.sh

   • Windows:

     cd tangle && backend\start_local.cmd

3. Access the UI: Once the terminal displays start_local: Starting the orchestrator, open http://localhost:8000 in your web browser.

git clone https://github.com/TangleML/tangle.git tangle/backend --branch stable
git clone https://github.com/TangleML/tangle-ui.git tangle/ui_build --branch stable_local_build

# For Linux/Mac
cd tangle && backend/start_local.sh

To enable tracing and metrics export from the Tangle API to the OTel collector, you must set the TANGLE_OTEL_EXPORTER_ENDPOINT environment variable. If this variable is unset, telemetry export is disabled.

Use the following environment variables:

You can set these in your shell or via a .env file in your project root.

export TANGLE_OTEL_EXPORTER_ENDPOINT=http://localhost:4317
export TANGLE_OTEL_EXPORTER_PROTOCOL=grpc
# then start your Tangle API

The Orchestrator manages the lifecycle of containerized tasks through a state machine that transitions between statuses like PENDING, RUNNING, SUCCEEDED, FAILED, and CANCELLED.

Key Lifecycle Transitions:

• Launching: When a task is launched, the Orchestrator creates a ContainerExecution record, links it to ExecutionNodes, and stores metadata via a launcher_interfaces.LaunchedContainer object.
• Polling/Refreshing: The Orchestrator periodically refreshes the container state using _launcher.get_refreshed_launched_container_from_dict. If the status changes (e.g., to SUCCEEDED), it triggers downstream updates.
• Success: Upon SUCCEEDED, the Orchestrator validates output artifacts, updates ExecutionNode statuses to SUCCEEDED, and wakes up direct downstream executions by moving them from WAITING_FOR_UPSTREAM to QUEUED.
• Failure: If the container fails or outputs are missing, the Orchestrator marks the ContainerExecution as FAILED and uses _mark_all_downstream_executions_as_skipped to recursively skip all dependent tasks.
• Cancellation: If an ExecutionNode has a desired_state of TERMINATED, the Orchestrator attempts to upload logs and then calls launched_container.terminate() to cancel the execution.

# Conceptual flow of status updates
if new_status == launcher_interfaces.ContainerStatus.RUNNING:
    container_execution.status = bts.ContainerExecutionStatus.RUNNING
    for execution_node in execution_nodes:
        execution_node.container_execution_status = bts.ContainerExecutionStatus.RUNNING

elif new_status == launcher_interfaces.ContainerStatus.SUCCEEDED:
    container_execution.status = bts.ContainerExecutionStatus.SUCCEEDED
    # ... update nodes and wake up downstream ...

elif new_status == launcher_interfaces.ContainerStatus.FAILED:
    container_execution.status = bts.ContainerExecutionStatus.FAILED
    # ... skip downstream ...

The QueuedExecutionInterceptor is a protocol used to take control of an execution before it is launched by the orchestrator.

An implementation of this protocol is called once the orchestrator determines an execution is ready to launch (inputs are present, not skipped, no cache hit, etc.). If the intercept method returns True, the interceptor 'owns' the execution: it is responsible for setting the desired status and committing the transaction. The orchestrator will then skip the standard launch path for that execution.

This is useful for implementing custom gating logic, manual approvals, or specialized handling of specific execution types.

class QueuedExecutionInterceptor(typing.Protocol):
    def intercept(self, *, session: orm.Session, execution: bts.ExecutionNode) -> bool:
        """True if this execution was taken over and must not launch; False to continue."""
        ...

# Example implementation
class MyCustomInterceptor:
    def intercept(self, *, session: orm.Session, execution: bts.ExecutionNode) -> bool:
        if some_custom_condition(execution):
            execution.container_execution_status = bts.ContainerExecutionStatus.SKIPPED
            session.commit()
            return True
        return False

Tangle uses a two-part system to manage data produced or consumed by tasks:

1. ArtifactNode: Represents the logical presence of an artifact in a pipeline graph. It tracks the type of data, its producer (an ExecutionNode), and its relationships to upstream and downstream tasks via InputArtifactLink and OutputArtifactLink.
2. ArtifactData: Represents the physical storage or value of the artifact. It contains the uri (for large files), the actual value (for small constants), the total_size, and a hash for integrity.

Key Relationships:

• ArtifactNode.artifact_data: Links the logical node to its physical data.
• ArtifactNode.producer_execution: The execution that created this artifact.
• ArtifactNode.upstream_execution_links / downstream_execution_links: Links to other nodes in the graph via artifact consumption/production.

class ArtifactData(_TableBase):
    __tablename__ = "artifact_data"
    id: orm.Mapped[IdType] = orm.mapped_column(primary_key=True, init=False, insert_default=generate_unique_id)
    total_size: orm.Mapped[int] = orm.mapped_column(sql.BigInteger())
    is_dir: orm.Mapped[bool]
    hash: orm.Mapped[str]
    uri: orm.Mapped[str | None] = orm.mapped_column(default=None)
    value: orm.Mapped[str | None] = orm.mapped_column(sql.Text(), default=None)

class ArtifactNode(_TableBase):
    __tablename__ = "artifact_node"
    id: orm.Mapped[IdType] = orm.mapped_column(primary_key=True, init=False, insert_default=generate_unique_id)
    type_name: orm.Mapped[str | None] = orm.mapped_column(default=None)
    type_properties: orm.Mapped[dict[str, Any] | None] = orm.mapped_column(default=None)
    producer_execution_id: orm.Mapped[IdType | None] = orm.mapped_column(sql.ForeignKey("execution_node.id"), init=False)
    artifact_data_id: orm.Mapped[IdType | None] = orm.mapped_column(sql.ForeignKey("artifact_data.id"), init=False)