flyte

GitHub

Dynamic, resilient AI orchestration. Coordinate data, models, and compute as you build AI workflows.

7,181 stars Go #agentic#ai-agents#ai-development-tools#data-analysis#data-science#declarative#fine-tuning#flyte
RAW Doc

README

Building the Flyte docs locally

Prerequisites

* conda (We recommend Miniconda installed with an official installer)

* conda-lock


Set up the build environment

In the flyteorg/flyte root directory do:

bash
$ conda-lock install --name monodocs-env monodocs-environment.lock.yaml
$ conda activate monodocs-env
$ pip install ./flyteidl

This creates a new environment called monodocs-env with all the dependencies needed to build the docs. You can choose a different environment name if you like.


Building the docs

In the flyteorg/flyte root directory make sure you have activated the monodocs-env (or whatever you called it) environment and do:

bash
need to set this to a fake value to build the docs locally

$ export DOCSEARCH_API_KEY=fake-api-key

bash
$ make docs

The resulting html files will be in docs/_build/html.

---

BACKEND README

Flyte 2 Backend

This repository contains the backend infrastructure for deploying a distributed, multi-node version of Flyte 2. The backend is Kubernetes-native — it orchestrates workflow execution using Kubernetes primitives, scheduling tasks as pods across clusters with built-in support for multi-cluster routing, service account-based identity, and pod-level log tracking. The core architecture consists of gRPC services (QueueService, RunService, StateService) backed by PostgreSQL, using async processing and real-time streaming via PostgreSQL LISTEN/NOTIFY. See the full Implementation Spec for details.

This repo also defines the protocol buffer schemas for Flyte's APIs and generates client libraries for Go, TypeScript, Python, and Rust. Deploy this when you need Flyte running as a scalable, distributed service across your organization.

Want to contribute? Join us on slack.flyte.org to get involved.

Repository Structure

text
flyte/
├── flyteidl2/           # Protocol buffer definitions
│   ├── common/          # Common types and utilities
│   ├── core/            # Core Flyte types (tasks, workflows, literals)
│   ├── imagebuilder/    # Image builder service definitions
│   ├── logs/            # Logging types
│   ├── secret/          # Secret management types
│   ├── task/            # Task execution types
│   ├── trigger/         # Trigger service definitions
│   ├── workflow/        # Workflow types
│   └── gen_utils/       # Language-specific generation utilities
├── gen/                 # Generated code (not checked into version control)
│   ├── go/              # Generated Go code
│   ├── ts/              # Generated TypeScript code
│   ├── python/          # Generated Python code
│   └── rust/            # Generated Rust code
├── buf.yaml             # Buf configuration
├── buf.gen.*.yaml       # Language-specific generation configs
└── Makefile             # Build automation

Prerequisites

- Buf CLI - Protocol buffer tooling
- Go 1.26.5 or later
- Node.js/npm (for TypeScript generation)
- Python 3.9+ with uv package manager (for Python generation)
- Rust toolchain (for Rust generation)

Quick Start

Generate All Code

To generate code for all supported languages:

bash
make gen

This will:
1. Update buf dependencies
2. Format and lint proto files
3. Generate code for Go, TypeScript, Python, and Rust
4. Generate mocks for Go
5. Run go mod tidy

Generate for Specific Languages Locally

bash
make buf-go      # Generate Go code only
make buf-ts      # Generate TypeScript code only
make buf-python  # Generate Python code only
make buf-rust    # Generate Rust code only

Making Changes

1. Modify Protocol Buffers

Edit .proto files in the flyteidl2/ directory following these guidelines:
- Follow the existing naming conventions
- Use proper protobuf style (snake_case for fields, PascalCase for messages)
- Add appropriate comments and documentation
- Ensure backward compatibility when modifying existing messages

2. Generate Code

After modifying proto files:

bash
make docker-pull   # Pull the docker image for generation
make gen

3. Verify Your Changes

Run the following to ensure everything builds correctly:

bash
For Go

make go-tidy
go build ./...
For Rust

make build-crate
For Python

cd gen/python && uv lock
For TypeScript

cd gen/ts && npm install

4. Generate Mocks (Go only)

If you've added or modified Go interfaces:

bash
make gen

Development Workflow

1. Format proto files: make buf-format
2. Lint proto files: make buf-lint
3. Generate code: make buf or make gen
4. Verify builds: Build generated code in your target language
5. Commit changes: Commit both proto files and generated code

Common Tasks

Update Buf Dependencies

bash
make gen

View Available Commands

bash
make help

Versioning and Releases

See CONTRIBUTING.md for detailed release instructions.

Generated Code

The gen/ directory contains auto-generated code and should not be manually edited. Changes to generated code should be made by:
1. Modifying the source .proto files in flyteidl2/
2. Updating generation utilities in flyteidl2/gen_utils/ if needed
3. Running make gen to regenerate all code

Troubleshooting

Buf Errors


- Ensure you have the latest version of Buf: buf --version
- Update dependencies: make buf-dep
- Check buf.lock for dependency conflicts

Go Module Issues


- Run make go-tidy to clean up dependencies
- Ensure you're using Go 1.26.5 or later

Python Generation Issues


- Ensure uv is installed: pip install uv
- Set the environment variable: export SETUPTOOLS_SCM_PRETEND_VERSION=0.0.0

Rust Build Issues


- Update Rust toolchain: rustup update
- Navigate to gen/rust and run cargo update

Contributing

We welcome contributions to Flyte 2! Please follow the guide here.

---

DOCKER QUICK START

Docker Development - Quick Start

This guide gets you started with Docker-based development in under 5 minutes.

Why Docker?

Using Docker ensures your local environment matches CI exactly, eliminating "works on my machine" issues.

Quick Start

1. Pull the Image

bash
make docker-pull

or

bash
docker pull ghcr.io/flyteorg/flyte/ci:v2

2. Run Common Commands

#### Generate Protocol Buffers

bash
make gen

#### Interactive Shell

bash
make docker-shell

3. Manual Docker Commands

If you prefer not to use Make:

bash
Generate files

docker run --rm -v $(pwd):/workspace -w /workspace \
  ghcr.io/flyteorg/flyte/ci:v2 make gen
Interactive shell

docker run --rm -it -v $(pwd):/workspace -w /workspace \
  ghcr.io/flyteorg/flyte/ci:v2 bash

Available Make Targets

Run make help to see all available targets including Docker-based ones:

bash
make help

Docker-specific targets:
- make docker-pull - Pull the latest CI image
- make docker-shell - Start interactive shell
- make gen - Run code generation
- make build-crate - Build Rust crate

Troubleshooting

Permission Issues

If generated files have wrong ownership:

bash
docker run --rm -v $(pwd):/workspace -w /workspace \
  --user $(id -u):$(id -g) \
  ghcr.io/flyteorg/flyte/ci:v2 make gen

Authentication Issues

Login to GitHub Container Registry:

bash
gh auth token | docker login ghcr.io -u YOUR_USERNAME --password-stdin

Updating the Docker Image

If you're modifying gen.Dockerfile, build and test locally first:

bash
One command to build and generate

make docker-dev
Or step-by-step

make docker-build          # Build image
make docker-shell    # Test interactively
make gen      # Run generation

This is much faster than waiting for PR builds!

PR Testing (After Local Testing)

Once your local changes work:

1. Create a PR with your changes
2. Wait for build - A bot will comment with the PR-specific image tag
3. Test with PR image to verify CI works:

bash
docker pull ghcr.io/flyteorg/flyte/ci:pr-123  # Use your PR number
   docker run --rm -it -v $(pwd):/workspace -w /workspace \
     ghcr.io/flyteorg/flyte/ci:pr-123 bash

4. CI automatically uses your new image in the PR

Workflow Comparison

Local iteration (seconds to minutes):

bash
vim gen.Dockerfile
make docker-dev           # Fast!
Repeat until it works

PR iteration (5-10 minutes per build):

bash
git push
Wait for build...

docker pull ghcr.io/flyteorg/flyte/ci:pr-123
Test

Use local iteration first, then validate with PR!

More Information

See docs/docker-image-workflow.md for comprehensive documentation.

---

Docker Image Workflow

Docker Image Build Workflow

This document explains how the Docker CI image is built and used across different scenarios.

Workflow Diagrams

Scenario 1: Regular PR (No Dockerfile Changes)

text
┌─────────────────────────────────────────────────────────────┐
│ Developer creates PR (no gen.Dockerfile changes)             │
└─────────────────┬───────────────────────────────────────────┘
                  │
                  ▼
┌─────────────────────────────────────────────────────────────┐
│ check-generate workflow triggers                            │
│ • Checks if gen.Dockerfile modified: NO                      │
│ • Uses image: ghcr.io/flyteorg/flyte/ci:v2                 │
└─────────────────┬───────────────────────────────────────────┘
                  │
                  ▼
┌─────────────────────────────────────────────────────────────┐
│ Pulls existing v2 image and runs checks                     │
│ ✓ Fast: No image build needed                              │
└─────────────────────────────────────────────────────────────┘

Scenario 2: PR with Dockerfile Changes

text
/ Detailed source-code truncated for AI context efficiency. /

Scenario 3: Merged to v2 Branch

text
┌─────────────────────────────────────────────────────────────┐
│ PR merged to v2 branch                                      │
└─────────────────┬───────────────────────────────────────────┘
                  │
                  ▼
┌─────────────────────────────────────────────────────────────┐
│ build-ci-image workflow triggers on push                    │
└─────────────────┬───────────────────────────────────────────┘
                  │
                  ▼
┌─────────────────────────────────────────────────────────────┐
│ Builds and pushes with tags:                                │
│ • ghcr.io/flyteorg/flyte/ci:v2                             │
│ • ghcr.io/flyteorg/flyte/ci:v2-sha-abc123                  │
└─────────────────┬───────────────────────────────────────────┘
                  │
                  ▼
┌─────────────────────────────────────────────────────────────┐
│ Future PRs use updated v2 image                             │
└─────────────────────────────────────────────────────────────┘

Image Tag Strategy

| Context | Image Tag | When Created |
|--------------------|----------------------------------------|---------------------------------|
| Regular PR | v2 or latest | Uses existing branch image |
| PR with Dockerfile | pr-123 | Built when PR is created/updated|
| v2 branch push | v2 | Built on every push to v2 |
| main branch push | latest | Built on every push to master |
| Any branch push | {branch}-sha-{commit} | Built on push (with commit SHA) |

Workflow Files

Primary Workflows

1. .github/workflows/build-ci-image.yml
- Builds Docker image
- Runs on: PR with Dockerfile changes, push to main/v2
- Publishes to GHCR with appropriate tags
- Comments on PR with image information

2. .github/workflows/check-generate.yml
- Validates generated files
- Automatically detects if Dockerfile was modified
- Uses PR-specific image if available, otherwise uses v2

3. .github/workflows/regenerate-on-comment.yml
- Regenerates files via /regen comment
- Uses PR-specific image if available

Developer Experience

For Regular Development (No Docker Changes)

bash
Just use the standard v2 image

make docker-pull
make gen

For Docker Image Updates

#### Option 1: Local Development (Fastest - Recommended)

bash
Modify gen.Dockerfile

vim gen.Dockerfile
Build and test locally

make docker-dev
Iterate quickly

vim gen.Dockerfile
make docker-build  # Uses cache, faster rebuilds
make gen
When it works, push to PR

git commit -am "Update Python to 3.13"
git push

Benefits:
- ⚡ Fast iteration (seconds to minutes)
- 🔄 No network dependency
- 🧪 Test before pushing
- 💰 No CI minutes wasted

#### Option 2: PR-Based Testing

bash
1. Modify gen.Dockerfile

vim gen.Dockerfile
2. Create PR

git checkout -b update-python
git commit -am "Update Python to 3.13"
git push origin update-python
Create PR on GitHub
3. Wait for build (~5-10 minutes)

Bot will comment: "🐳 Docker CI Image Built"

Image: ghcr.io/flyteorg/flyte/ci:pr-456
4. Test locally with YOUR image

make docker-pull gen DOCKER_CI_IMAGE=ghcr.io/flyteorg/flyte/ci:pr-456
5. CI automatically uses the same image!

No need to merge before testing

Benefits of This Approach

1. Test Before Merge: You can fully test Docker image changes before merging
2. True Parity: Local testing uses the exact same image as CI
3. Fast Iteration: No need to merge to test, iterate in the PR
4. Automatic Detection: Workflows automatically detect which image to use
5. Clean Fallback: If Docker isn't modified, uses the stable v2 image
6. Transparent: Bot comments show exactly what image is being used

Implementation Details

Image Detection Logic

Workflows check if gen.Dockerfile or .github/workflows/build-ci-image.yml were modified:

bash
git diff --name-only origin/$BASE_BRANCH...HEAD | \
  grep -E '^(Dockerfile\.ci|\.github/workflows/build-ci-image\.yml)$'

If modified:
- Use ghcr.io/flyteorg/flyte/ci:pr-{NUMBER}
- Wait for image to be available (with timeout)

If not modified:
- Use ghcr.io/flyteorg/flyte/ci:v2
- Proceed immediately

Wait Strategy

For PRs with Dockerfile changes, workflows intelligently wait for the build:

1. Check for build workflow: Queries GitHub API for build-ci-image.yml runs
2. Find matching run: Looks for run with same commit SHA
3. Wait for completion: Polls every 20 seconds for up to 20 minutes
4. Verify success: Ensures build succeeded before proceeding
5. Pull fresh image: Downloads the newly built image

Benefits:
- Always waits for the actual build to complete (not just image existence)
- Works correctly on subsequent pushes to the same PR
- Provides clear feedback on build status
- Fails fast if build fails

This ensures tests always run with the freshly built image, even when pushing multiple commits to a PR.

Build Performance

The Docker image is optimized for fast builds using several techniques:

Multi-Stage Build Strategy

The Dockerfile uses parallel multi-stage builds to download tools simultaneously:

text
┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│ Go Stage    │  │ Node Stage  │  │Python Stage │  │ Buf Stage   │
│ (parallel)  │  │ (parallel)  │  │ (parallel)  │  │ (parallel)  │
└──────┬──────┘  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘
       │                │                │                │
       └────────────────┴────────────────┴────────────────┘
                              │
                       ┌──────▼──────┐
                       │Final Image  │
                       │ (assembly)  │
                       └─────────────┘

Benefits:
- 4x parallelization of downloads
- Leverages official Docker images (pre-built, cached)
- Faster than sequential downloads

Caching Strategy

The build uses a multi-layer caching approach:

1. Registry cache (primary): Stored in GHCR, fastest to pull
2. GitHub Actions cache (secondary): Fallback for layers
3. BuildKit inline cache: Metadata in image layers
4. Cache mounts: For package managers (apt, go mod, cargo)

Cache hierarchy:

text
1. Try buildcache tag (dedicated cache image)
2. Try current PR tag (if exists)
3. Try v2 tag (stable baseline)
4. Try GHA cache
5. Build from scratch

Performance Improvements

| Optimization | Time Saved |
|--------------|------------|
| Multi-stage parallel builds | ~5-8 min |
| Official image copying vs downloads | ~2-3 min |
| Registry cache (vs no cache) | ~10-12 min |
| Cache mounts for packages | ~1-2 min |
| Total potential savings | 15-20 min |

Build times:
- Cold build (no cache): ~15 min
- Warm build (full cache): ~2-3 min
- Incremental build (partial cache): ~5-8 min

Cache Mounts

The Dockerfile uses BuildKit cache mounts for package managers:

dockerfile
APT packages cached

RUN --mount=type=cache,target=/var/cache/apt
Go modules cached

RUN --mount=type=cache,target=/root/go/pkg/mod
Cargo packages cached

RUN --mount=type=cache,target=/root/.cargo/registry

These persist across builds, dramatically speeding up package installation.

---

IMPLEMENTATION SPEC

Implementation Specification: Queue, Runs, and State Services

Overview

This document specifies the implementation of three gRPC services using buf connect:
- QueueService - Manages execution queue for actions
- RunService - Manages workflow runs and their lifecycle
- StateService - Manages state persistence for actions

Each service will have a simple implementation backed by PostgreSQL, using pgx for database operations, and leveraging existing flytestdlib packages for database connectivity and configuration management.

Architecture

text
/ Detailed source-code truncated for AI context efficiency. /

Technology Stack

Core Dependencies


- Protocol: buf connect (using generated code from gen/go/flyteidl2/workflow/workflowconnect/)
- Database: PostgreSQL
- Database Driver: pgx/v5 (for raw queries)
- ORM: gorm (for migrations and basic operations)
- Config Management: github.com/flyteorg/flyte/v2/flytestdlib/config
- Database Utils: github.com/flyteorg/flyte/v2/flytestdlib/database
- Logging: github.com/flyteorg/flyte/v2/flytestdlib/logger
- CLI: github.com/spf13/cobra

Service Framework


- HTTP server using net/http
- buf connect handlers mounted on HTTP server
- Graceful shutdown support
- Health check endpoints

Service Specifications

1. QueueService

Location: queue/

Proto Definition: flyteidl2/workflow/queue_service.proto

Connect Interface: workflowconnect.QueueServiceHandler

#### RPCs to Implement:
1. EnqueueAction(EnqueueActionRequest) -> EnqueueActionResponse
- Validates request
- Persists action to queue table
- Returns immediately (async processing)

2. AbortQueuedRun(AbortQueuedRunRequest) -> AbortQueuedRunResponse
- Marks all actions in a run as aborted
- Updates abort metadata

3. AbortQueuedAction(AbortQueuedActionRequest) -> AbortQueuedActionResponse
- Marks specific action as aborted
- Updates abort metadata

#### Database Schema:

text
/ Detailed source-code truncated for AI context efficiency. /

#### Configuration:

go
// queue/config/config.go
type Config struct {
    // HTTP server configuration
    Server ServerConfig json:"server"
    // Database configuration (reuses flytestdlib)
    Database database.DbConfig json:"database"
    // Queue specific settings
    MaxQueueSize int json:"maxQueueSize" pflag:",Maximum number of queued actions"
    WorkerCount  int json:"workerCount" pflag:",Number of worker goroutines for processing queue"
}
type ServerConfig struct {
    Port int    json:"port" pflag:",Port to bind the HTTP server"
    Host string json:"host" pflag:",Host to bind the HTTP server"
}

---

2. RunService

Location: runs/

Proto Definition: flyteidl2/workflow/run_service.proto

Connect Interface: workflowconnect.RunServiceHandler

#### RPCs to Implement:
1. CreateRun(CreateRunRequest) -> CreateRunResponse
- Creates a new run record
- Initializes root action
- Returns run metadata

2. AbortRun(AbortRunRequest) -> AbortRunResponse
- Marks run as aborted
- Cascades to all actions

3. GetRunDetails(GetRunDetailsRequest) -> GetRunDetailsResponse
- Fetches complete run information
- Includes root action details

4. WatchRunDetails(WatchRunDetailsRequest) -> stream WatchRunDetailsResponse
- Streams run updates
- Uses PostgreSQL LISTEN/NOTIFY for real-time updates

5. GetActionDetails(GetActionDetailsRequest) -> GetActionDetailsResponse
- Fetches detailed action information
- Includes all attempts

6. WatchActionDetails(WatchActionDetailsRequest) -> stream WatchActionDetailsResponse
- Streams action updates

7. GetActionData(GetActionDataRequest) -> GetActionDataResponse
- Returns input/output references
- Does NOT load actual data (just URIs)

8. ListRuns(ListRunsRequest) -> ListRunsResponse
- Paginated run listing
- Supports filtering by org/project/trigger

9. WatchRuns(WatchRunsRequest) -> stream WatchRunsResponse
- Streams run updates matching filter

10. ListActions(ListActionsRequest) -> ListActionsResponse
- Lists actions for a run
- Paginated

11. WatchActions(WatchActionsRequest) -> stream WatchActionsResponse
- Streams action updates for a run
- Supports filtering

12. WatchClusterEvents(WatchClusterEventsRequest) -> stream WatchClusterEventsResponse
- Streams cluster events for an action attempt

13. AbortAction(AbortActionRequest) -> AbortActionResponse
- Aborts a specific action

#### Database Schema:

text
/ Detailed source-code truncated for AI context efficiency. /

#### Configuration:

go
// runs/config/config.go
type Config struct {
    // HTTP server configuration
    Server ServerConfig json:"server"
    // Database configuration
    Database database.DbConfig json:"database"
    // Watch/streaming settings
    WatchBufferSize int json:"watchBufferSize" pflag:",Buffer size for watch streams"
}
type ServerConfig struct {
    Port int    json:"port" pflag:",Port to bind the HTTP server"
    Host string json:"host" pflag:",Host to bind the HTTP server"
}

---

3. StateService

Location: state/

Proto Definition: flyteidl2/workflow/state_service.proto

Connect Interface: workflowconnect.StateServiceHandler

#### RPCs to Implement:
1. Put(stream PutRequest) -> stream PutResponse
- Bidirectional streaming
- Persists action state (NodeStatus JSON)
- Returns status for each request

2. Get(stream GetRequest) -> stream GetResponse
- Bidirectional streaming
- Retrieves action state

3. Watch(WatchRequest) -> stream WatchResponse
- Server streaming
- Watches state changes for child actions
- Uses PostgreSQL LISTEN/NOTIFY

#### Database Schema:

text
/ Detailed source-code truncated for AI context efficiency. /

#### Configuration:

go
// state/config/config.go
type Config struct {
    // HTTP server configuration
    Server ServerConfig json:"server"
    // Database configuration
    Database database.DbConfig json:"database"
    // State specific settings
    MaxStateSizeBytes int json:"maxStateSizeBytes" pflag:",Maximum size of state JSON in bytes"
}
type ServerConfig struct {
    Port int    json:"port" pflag:",Port to bind the HTTP server"
    Host string json:"host" pflag:",Host to bind the HTTP server"
}

---

Unified Binary

Location: cmd/flyte-services/main.go

The unified binary provides a single entrypoint that can run:
1. queue - QueueService only
2. runs - RunService only
3. state - StateService only
4. executor - Kubernetes controller only
5. all - All services together on different ports

Command Structure:

bash
Run queue service only

flyte-services queue --config config.yaml
Run runs service only

flyte-services runs --config config.yaml
Run state service only

flyte-services state --config config.yaml
Run executor only

flyte-services executor --config config.yaml
Run all services

flyte-services all --config config.yaml

Implementation:

text
/ Detailed source-code truncated for AI context efficiency. /

---

Database Connection Management

All services use flytestdlib for database management:

go
// Example from queue/service/queue_service.go
import (
    "github.com/flyteorg/flyte/v2/flytestdlib/database"
    "gorm.io/gorm"
)
func initDB(ctx context.Context, cfg database.DbConfig) (gorm.DB, error) {
    gormConfig := &gorm.Config{
        // Configuration options
    }
    // Create database if it doesn't exist
    db, err := database.CreatePostgresDbIfNotExists(ctx, gormConfig, cfg.Postgres)
    if err != nil {
        return nil, err
    }
    // Apply connection pool settings
    sqlDB, err := db.DB()
    if err != nil {
        return nil, err
    }
    sqlDB.SetMaxIdleConns(cfg.MaxIdleConnections)
    sqlDB.SetMaxOpenConns(cfg.MaxOpenConnections)
    sqlDB.SetConnMaxLifetime(cfg.ConnMaxLifeTime.Duration)
    return db, nil
}

For pgx-specific operations (like LISTEN/NOTIFY for streaming):

go
import (
    "github.com/jackc/pgx/v5/pgxpool"
)
func initPgxPool(ctx context.Context, cfg database.PostgresConfig) (pgxpool.Pool, error) {
    connString := fmt.Sprintf(
        "postgres://%s:%s@%s:%d/%s?%s",
        cfg.User,
        resolvePassword(ctx, cfg.Password, cfg.PasswordPath),
        cfg.Host,
        cfg.Port,
        cfg.DbName,
        cfg.ExtraOptions,
    )
    return pgxpool.New(ctx, connString)
}

---

Configuration Management

All services use flytestdlib config:

go
// Example from queue/cmd/main.go
import (
    "github.com/flyteorg/flyte/v2/flytestdlib/config"
    queueconfig "github.com/flyteorg/flyte/v2/queue/config"
)
var (
    configSection = config.MustRegisterSection("queue", &queueconfig.Config{})
)
func main() {
    // Parse config from file and flags
    if err := config.LoadConfig(...); err != nil {
        panic(err)
    }
    cfg := configSection.GetConfig().(*queueconfig.Config)
    // Use cfg...
}

---

Service Implementation Pattern

Each service follows this pattern:

go
// queue/service/queue_service.go
package service
import (
    "context"
    "connectrpc.com/connect"
    "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow"
    "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow/workflowconnect"
    "github.com/flyteorg/flyte/v2/queue/repository"
)
type QueueService struct {
    repo repository.Repository
}
func NewQueueService(repo repository.Repository) *QueueService {
    return &QueueService{repo: repo}
}
// Ensure we implement the interface
var _ workflowconnect.QueueServiceHandler = (*QueueService)(nil)
func (s *QueueService) EnqueueAction(
    ctx context.Context,
    req *connect.Request[workflow.EnqueueActionRequest],
) (*connect.Response[workflow.EnqueueActionResponse], error) {
    // Validate request
    if err := req.Msg.Validate(); err != nil {
        return nil, connect.NewError(connect.CodeInvalidArgument, err)
    }
    // Persist to database
    if err := s.repo.EnqueueAction(ctx, req.Msg); err != nil {
        return nil, connect.NewError(connect.CodeInternal, err)
    }
    return connect.NewResponse(&workflow.EnqueueActionResponse{}), nil
}
// ... other methods

---

HTTP Server Setup

Each service's main.go sets up an HTTP server:

go
// queue/cmd/main.go
package cmd
import (
    "context"
    "fmt"
    "net/http"
    "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/workflow/workflowconnect"
    "github.com/flyteorg/flyte/v2/queue/service"
    "github.com/flyteorg/flyte/v2/queue/repository"
)
func RunQueue(ctx context.Context) error {
    // Initialize database
    db, err := initDB(ctx)
    if err != nil {
        return err
    }
    // Run migrations
    if err := runMigrations(db); err != nil {
        return err
    }
    // Create repository
    repo := repository.NewPostgresRepository(db)
    // Create service
    svc := service.NewQueueService(repo)
    // Create HTTP handler
    mux := http.NewServeMux()
    path, handler := workflowconnect.NewQueueServiceHandler(svc)
    mux.Handle(path, handler)
    // Add health check
    mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
        w.WriteHeader(http.StatusOK)
        w.Write([]byte("OK"))
    })
    // Setup HTTP/2 support
    addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.Port)
    protocols := &http.Protocols{}
    protocols.SetHTTP1(true)
    protocols.SetUnencryptedHTTP2(true)
    server := &http.Server{
        Addr:      addr,
        Handler:   mux,
        Protocols: protocols,
    }
    logger.Infof(ctx, "Starting Queue Service on %s", addr)
    return server.ListenAndServe()
}

---

Migration Management

Each service uses golang-migrate or similar:

go
// queue/repository/migrations.go
func RunMigrations(db *gorm.DB) error {
    return db.AutoMigrate(
        &models.QueuedAction{},
    )
}

Or use raw SQL migrations with a migration tool.

---

Repository Pattern

Each service implements a repository interface:

go
// queue/repository/interfaces.go
type Repository interface {
    EnqueueAction(ctx context.Context, req *workflow.EnqueueActionRequest) error
    AbortQueuedRun(ctx context.Context, runID *common.RunIdentifier, reason string) error
    AbortQueuedAction(ctx context.Context, actionID *common.ActionIdentifier, reason string) error
    GetQueuedActions(ctx context.Context, limit int) ([]*models.QueuedAction, error)
}

go
// queue/repository/postgres.go
type PostgresRepository struct {
    db *gorm.DB
}
func NewPostgresRepository(db *gorm.DB) Repository {
    return &PostgresRepository{db: db}
}
func (r PostgresRepository) EnqueueAction(ctx context.Context, req workflow.EnqueueActionRequest) error {
    action := &models.QueuedAction{
        Org:            req.ActionId.RunId.Org,
        Project:        req.ActionId.RunId.Project,
        Domain:         req.ActionId.RunId.Domain,
        RunName:        req.ActionId.RunId.Name,
        ActionName:     req.ActionId.Name,
        ParentActionName: req.ParentActionName,
        ActionGroup:    req.Group,
        Subject:        req.Subject,
        ActionSpec:     req, // Will be marshaled to JSONB
        InputUri:       req.InputUri,
        RunOutputBase:  req.RunOutputBase,
        Status:         "queued",
    }
    return r.db.WithContext(ctx).Create(action).Error
}

---

Streaming Implementation (Watch/Listen)

For streaming RPCs, use PostgreSQL LISTEN/NOTIFY:

go
// runs/service/run_service.go
func (s *RunService) WatchRunDetails(
    ctx context.Context,
    req *connect.Request[workflow.WatchRunDetailsRequest],
    stream *connect.ServerStream[workflow.WatchRunDetailsResponse],
) error {
    // Get initial state
    details, err := s.repo.GetRunDetails(ctx, req.Msg.RunId)
    if err != nil {
        return err
    }
    if err := stream.Send(&workflow.WatchRunDetailsResponse{Details: details}); err != nil {
        return err
    }
    // Subscribe to updates via PostgreSQL LISTEN
    updates := make(chan *workflow.RunDetails)
    errs := make(chan error)
    go s.repo.WatchRunDetails(ctx, req.Msg.RunId, updates, errs)
    for {
        select {
        case <-ctx.Done():
            return nil
        case err := <-errs:
            return err
        case details := <-updates:
            if err := stream.Send(&workflow.WatchRunDetailsResponse{Details: details}); err != nil {
                return err
            }
        }
    }
}

go
// runs/repository/postgres.go
func (r *PostgresRepository) WatchRunDetails(
    ctx context.Context,
    runID *common.RunIdentifier,
    updates chan<- *workflow.RunDetails,
    errs chan<- error,
) {
    conn, err := r.pgxPool.Acquire(ctx)
    if err != nil {
        errs <- err
        return
    }
    defer conn.Release()
    // Listen for notifications
    _, err = conn.Exec(ctx, "LISTEN run_updates")
    if err != nil {
        errs <- err
        return
    }
    for {
        notification, err := conn.Conn().WaitForNotification(ctx)
        if err != nil {
            errs <- err
            return
        }
        // Fetch updated run details
        details, err := r.GetRunDetails(ctx, runID)
        if err != nil {
            errs <- err
            return
        }
        select {
        case updates <- details:
        case <-ctx.Done():
            return
        }
    }
}

---

Testing Strategy

Unit Tests


- Repository layer: Mock database using testcontainers with PostgreSQL
- Service layer: Mock repository interface
- Use table-driven tests

Integration Tests


- End-to-end tests with real PostgreSQL
- Use docker-compose for local testing
- Test streaming with multiple concurrent clients

Example:

go
// queue/service/queue_service_test.go
func TestEnqueueAction(t *testing.T) {
    mockRepo := &mocks.Repository{}
    svc := service.NewQueueService(mockRepo)
    req := connect.NewRequest(&workflow.EnqueueActionRequest{
        // ... populate request
    })
    mockRepo.On("EnqueueAction", mock.Anything, req.Msg).Return(nil)
    resp, err := svc.EnqueueAction(context.Background(), req)
    assert.NoError(t, err)
    assert.NotNil(t, resp)
}

---

Deployment Considerations

Configuration Files

Example config.yaml:

yaml
database:
  postgres:
    host: postgres.flyte.svc.cluster.local
    port: 5432
    dbname: flyte_queue
    username: flyte
    passwordPath: /etc/secrets/db-password
    extraOptions: "sslmode=require"
  maxIdleConnections: 10
  maxOpenConnections: 100
  connMaxLifeTime: 1h
queue:
  server:
    host: 0.0.0.0
    port: 8089
  maxQueueSize: 10000
  workerCount: 10
runs:
  server:
    host: 0.0.0.0
    port: 8090
  watchBufferSize: 100
state:
  server:
    host: 0.0.0.0
    port: 8091
  maxStateSizeBytes: 1048576  # 1MB

Docker Compose (for local development)

yaml
version: '3.8'
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: flyte
      POSTGRES_PASSWORD: flyte
      POSTGRES_DB: flyte
    ports:
      - "5432:5432"
  queue:
    build: .
    command: queue --config /etc/flyte/config.yaml
    ports:
      - "8089:8089"
    depends_on:
      - postgres
  runs:
    build: .
    command: runs --config /etc/flyte/config.yaml
    ports:
      - "8090:8090"
    depends_on:
      - postgres
  state:
    build: .
    command: state --config /etc/flyte/config.yaml
    ports:
      - "8091:8091"
    depends_on:
      - postgres

---

Implementation Phases

Phase 1: Core Infrastructure


1. Setup project structure
2. Implement database schemas and migrations
3. Implement repository interfaces and PostgreSQL implementations
4. Setup configuration management using flytestdlib

Phase 2: Service Implementation


1. Implement QueueService
2. Implement RunService (non-streaming RPCs first)
3. Implement StateService (non-streaming RPCs first)

Phase 3: Streaming Support


1. Add PostgreSQL LISTEN/NOTIFY support
2. Implement streaming RPCs (Watch*)
3. Test concurrent streaming clients

Phase 4: Integration


1. Implement unified binary command structure
2. Add health checks and metrics
3. Integration testing
4. Documentation

Phase 5: Production Readiness


1. Add observability (metrics, tracing, logging)
2. Performance testing and optimization
3. Security audit
4. Deployment documentation

---

Open Questions

1. Migration Strategy: Should we use golang-migrate, gorm AutoMigrate, or custom SQL scripts?
2. Protobuf Serialization: Store protobuf as JSONB or use binary serialization?
3. Queue Processing: Should QueueService also include worker implementation for processing queued actions?
4. Multi-tenancy: How to handle org/project isolation at the database level?
5. Metrics: What metrics should each service expose?
6. Rate Limiting: Should services implement rate limiting per org/project?

---

References

- Protocol Buffers: queue_service.proto, run_service.proto, state_service.proto
- Generated Code: gen/go/flyteidl2/workflow/workflowconnect/
- Database Utils: flytestdlib/database/
- Config Management: flytestdlib/config/
- Buf Connect: https://connectrpc.com/docs/go/getting-started
- PostgreSQL LISTEN/NOTIFY: https://www.postgresql.org/docs/current/sql-notify.html

---

CONTRIBUTING

Contributing to Flyte

First off, thanks for taking the time to contribute! ❤️

All types of contributions are encouraged and valued. See the Table of Contents for different ways to help and details about how this project handles them. Please make sure to read the relevant section before making your contribution. It will make it a lot easier for the team and smooth out the experience for all involved. The community looks forward to your contributions. 🎉

If you don't have time to contribute, that's fine. There are other easy ways to support the project and show your appreciation, which we would also be very happy about:

- Star the project
- Post on X or Linkedin about Flyte #flyte
- Mention the project at local meetups and tell your friends/colleagues

Table of Contents

- Code of Conduct
- I Have a Question
- I Want To Contribute
- Code Contributors
- Recommendation Order (For Beginners)
- Reporting Bugs
- Feature Requests
- Improving The Documentation
- Improving The Design
- Publish a Blog Post or Case Study
- Commit Messages
- Pull requests
- Contributor Ladder

Code of Conduct

This project and everyone participating in it is governed by the Code of Conduct. By participating, you are expected to uphold this code. Please report unacceptable behavior
to [email protected].

I Have a Question

If you need clarification after reading this document, we encourage you to join our slack workspace and join channels #flyte-support and #contribute channel.

I Want To Contribute

Code Contributors

We’re excited that you’re interested in contributing code to Flyte! Before you start, please take a look at our Getting started docs, it includes setup instructions, build steps, and details on running your first workflow locally.

#### Recommendation Order (For Beginners)

- Finish reading Core Concepts
- Finish reading Connecting a Cluster
- Finish reading Projects and Domains

Reporting Bugs

Please use our issues templates that provide hints on what information we need to help you.

You must never report security related issues, vulnerabilities or bugs including sensitive information to the issue tracker, or elsewhere in public. Instead sensitive bugs must be sent by email to [email protected] or use the Report a security vulnerability issue template.

Feature Requests

Suggest an idea for the project by using the Issues template and choose your desired option and kindly provide as much context as you can about what you're running into. Do not open issues for questions or support, instead join our slack workspace and ask there.

Unsure if your contribution is “small”, “large” or whether it fits into the project's goal? Kindly start a quick discussion on github.

Improving The Documentation

If you notice outdated information or areas that could be clarified, kindly start a discussion in the contribute channel on slack. For more information, please checkout Contributing to documentation.

Improving The Design

Design contributions are welcome! To ensure smooth collaboration, please use the UI Feature Request Template when opening a design-related issue. This helps us gather the right context (such as wireframes, mockups, or visual references) and maintain a consistent design language across the project. Feedback and iterations are highly encouraged, design is always a shared process.

Publish a Blog Post or Case Study

We love hearing how people use or extend Flyte in their own projects. If you’ve written about your experience, we’re happy to review it! To share your work, please start a discussion in the #contribute channel on slack, summarizing your post or case study, with a link to the full content.

Commit Messages

Writing clear and consistent commit messages helps maintainers understand the purpose of your changes. A good commit message should:

- Be written in the present tense (e.g., Add new feature, not Added new feature)

- Be short (50 characters or less for the summary line)

- Include additional context in the body if needed

- Reference related issue numbers (e.g., Fixes #123)

- Keep each commit focused on one logical change

Pull Requests

When you’re ready to contribute your changes, follow these steps to create a clear and reviewable pull request:

- Push your changes to your fork:

bash
git push origin your-branch-name

- Open a Pull Request against the main Flyte Code repository.

- Select "Compare across forks" and choose your fork and branch.

- Fill out the PR template with:

- A clear description of your changes

- Any related issues (e.g., “changed `#123”).

- Testing steps or screenshots (if applicable)

- Notes for reviewers, if special attention is needed.

For more context, kindly read the official Before Submitting Your PR docs.

Contributor Ladder

The Flyte Contributor Ladder is a key resource for understanding how to grow within Flyte, outlining expectations, responsibilities, and progression across roles. It helps create transparency, recognize contributions, and ensure a sustainable and inclusive path for community members to deepen their impact. Read more.


Your contributions, big or small help make Flyte better for everyone!🫶

References

This document was adapted from https://contributing.md!

---

README

IMPORTANT

## Flyte 2 is now generally available!


> Read more in the announcement here.

> Want to try Flyte 2 locally? Run the Devbox.

> Looking for Flyte 1? Go to the master branch, where Flyte 1 is now maintained.

---

Flyte 2

Reliably orchestrate ML pipelines, models, and agents at scale — in pure Python.

[](https://pypi.org/project/flyte/)
[](https://pypi.org/project/flyte/)
[](LICENSE)
[](https://flyte2intro.apps.demo.hosted.unionai.cloud/)
[](https://www.union.ai/docs/v2/flyte/user-guide/running-locally/)
[](https://www.union.ai/docs/v2/byoc/api-reference/flyte-sdk/)
[](https://www.union.ai/docs/v2/byoc/api-reference/flyte-cli/)

Flyte is a Graduated project of the LF AI & Data Foundation.

<a href="https://lfaidata.foundation/projects/flyte/">
<img src="https://raw.githubusercontent.com/flyteorg/static-resources/main/flyte/readme/flyte_and_lf.png" alt="Flyte and LF AI & Data Logo" width="250">
</a>

Install

bash
uv pip install flyte

For the full SDK and development tools, see the flyte-sdk repository.

Example

python
import asyncio
import flyte
env = flyte.TaskEnvironment(
    name="hello_world",
    image=flyte.Image.from_debian_base(python_version=(3, 12)),
)
@env.task
def calculate(x: int) -> int:
    return x * 2 + 5
@env.task
async def main(numbers: list[int]) -> float:
    results = await asyncio.gather(*[
        calculate.aio(num) for num in numbers
    ])
    return sum(results) / len(results)
if __name__ == "__main__":
    flyte.init()
    run = flyte.run(main, numbers=list(range(10)))
    print(f"Result: {run.result}")

<table>
<tr><td><b>Python</b></td><td><b>Flyte CLI</b></td></tr>
<tr>
<td>

bash
python hello.py

</td>
<td>

bash
flyte run hello.py main --numbers '[1,2,3]'

</td>
</tr>
</table>

Serve a Model

python
serving.py

from fastapi import FastAPI
import flyte
from flyte.app.extras import FastAPIAppEnvironment
app = FastAPI()
env = FastAPIAppEnvironment(
    name="my-model",
    app=app,
    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages(
        "fastapi", "uvicorn"
    ),
)
@app.get("/predict")
async def predict(x: float) -> dict:
    return {"result": x * 2 + 5}
if __name__ == "__main__":
    flyte.init_from_config()
    flyte.serve(env)

<table>
<tr><td><b>Python</b></td><td><b>Flyte CLI</b></td></tr>
<tr>
<td>

bash
python serving.py

</td>
<td>

bash
flyte serve serving.py env

</td>
</tr>
</table>

Local Development Experience

Install the TUI for a rich local development experience:

bash
uv pip install flyte[tui]

[](https://www.youtube.com/watch?v=lsfy-7DbbRM)

Try the hosted demo in your browser — no installation required.

Open Source Backend

The open source backend for Flyte 2 is coming soon. This repository will contain the Kubernetes-native backend infrastructure for deploying Flyte 2 as a distributed, multi-node service. See the Backend README for the current state of the backend, protocol buffer definitions, and contribution guide.

If you need an enterprise-ready, production-grade backend for Flyte 2 today, it is available on Union.ai.

Learn More

- Try DevBox - Get started
- SDK Reference — API reference docs
- CLI Reference — CLI docs
- flyte-sdk — The Flyte 2 Python SDK repository
- Slack | GitHub Discussions | Issues

Contributing

We welcome contributions! See the Backend README for backend development, or join us on slack.flyte.org.

Sponsors

CI container image builds for this repository are sponsored by Depot — fast, native multi-arch Docker builds with persistent layer caching.

[](https://depot.dev)

License

Apache 2.0 — see LICENSE.

---

Explore Related Technologies & Topics

View All 9,111+ Repositories →

Discover similar high-velocity repositories, agent skills, and OpenAPI specifications across the ecosystem.

#agentic Browse agentic catalog → #ai-agents Browse ai-agents catalog → #ai-development-tools Browse ai-development-tools catalog → #data-analysis Browse data-analysis catalog → #data-science Browse data-science catalog → #declarative Browse declarative catalog → ⚡ MCP Servers Model Context Protocol → 🧠 Agent Skills Prompt & rule files → 🔌 OpenAPI Specs REST & Swagger schemas →