Building a Custom Execution Logic

Flyte uses a pluggable execution logic for its scheduled triggers. By implementing the Executor interface, you can customize how scheduled runs are fired, add validation logic, or integrate with external systems before a workflow is triggered.

This tutorial walks you through creating a custom Executor that implements a "kill-switch" mechanism to prevent scheduled runs from firing during maintenance windows.

Prerequisites

• A Go development environment.
• Familiarity with the Flyte runs service structure.
• The following imports from the Flyte codebase:
  • github.com/flyteorg/flyte/v2/runs/scheduler/core
  • github.com/flyteorg/flyte/v2/runs/repository/models
  • github.com/flyteorg/flyte/v2/flytestdlib/logger

Step 1: Understand the Executor Interface

The Executor interface is defined in runs/scheduler/core/job.go. It consists of a single method that is called every time a schedule fires or during a catch-up phase.

// From runs/scheduler/core/job.go
type Executor interface {
	Execute(ctx context.Context, t *models.Trigger, scheduledAt time.Time) error
}

Step 2: Implement a Custom Executor

We will create a MaintenanceExecutor that wraps the standard TriggerExecutor. It will check a boolean flag before delegating the execution. This pattern allows you to add logic without reimplementing the complex CreateRun RPC calls.

Create a new file or add this to your scheduler package:

package custom

import (
	"context"
	"fmt"
	"time"

	"github.com/flyteorg/flyte/v2/flytestdlib/logger"
	"github.com/flyteorg/flyte/v2/runs/repository/models"
	"github.com/flyteorg/flyte/v2/runs/scheduler/core"
)

// MaintenanceExecutor wraps another executor and can be disabled globally.
type MaintenanceExecutor struct {
	Inner    core.Executor
	Disabled bool
}

func (e *MaintenanceExecutor) Execute(ctx context.Context, t *models.Trigger, scheduledAt time.Time) error {
	if e.Disabled {
		logger.Infof(ctx, "Maintenance mode active: skipping trigger %s/%s/%s", 
            t.Project, t.Domain, t.Name)
		return nil
	}

	// Delegate to the standard execution logic
	return e.Inner.Execute(ctx, t, scheduledAt)
}

Step 3: Ensure Idempotency

When implementing custom execution logic, you must ensure that firing a run is idempotent. Flyte's default TriggerExecutor (found in runs/scheduler/executor/trigger_executor.go) achieves this by generating a deterministic run name based on the trigger identity and the scheduled time.

If you were to implement the Execute method from scratch using the RunServiceClient, you should use a similar naming strategy:

// Example of deterministic naming used in runs/scheduler/executor/trigger_executor.go
func runName(t *models.Trigger, scheduledAt time.Time) string {
	h := fnv.New64()
	_, _ = fmt.Fprintf(h, "%s:%s:%s:%s:%d:%d:%d:%d:%d:%d",
		t.Project, t.Domain, t.TaskName, t.Name,
		scheduledAt.Year(), scheduledAt.Month(), scheduledAt.Day(),
		scheduledAt.Hour(), scheduledAt.Minute(), scheduledAt.Second())
	return fmt.Sprintf("r%x", h.Sum64())
}

By using this name in the CreateRunRequest, the Flyte backend will return a CodeAlreadyExists error if the run was already fired (e.g., due to a scheduler restart), which you should handle as a success.

Step 4: Wire the Custom Executor into the Scheduler

The scheduler is initialized in runs/scheduler/start.go. To use your custom logic, you wrap the standard TriggerExecutor before passing it to the GoCronScheduler.

// Modified logic based on runs/scheduler/start.go
func StartWithMaintenance(
	ctx context.Context,
	triggerRepo interfaces.TriggerRepo,
	cfg config.TriggerSchedulerConfig,
	baseURL string,
	clientOpts ...connect.ClientOption,
) func(ctx context.Context) error {
    // 1. Initialize the standard executor
	standardExec := executor.NewTriggerExecutor(executor.TriggerExecutorConfig{
		BaseURL:    baseURL,
		QPS:        cfg.ExecutionQPS,
		Burst:      cfg.ExecutionBurst,
		ClientOpts: clientOpts,
	})

    // 2. Wrap it with your custom logic
    customExec := &MaintenanceExecutor{
        Inner:    standardExec,
        Disabled: false, // This could be loaded from a dynamic config
    }

    // 3. Pass the custom executor to the scheduler
	sched := core.NewGoCronScheduler(customExec)
	sched.Start()

	syncer := core.NewScheduleSyncer(triggerRepo, sched, cfg.ResyncInterval)

	return func(ctx context.Context) error {
        // ... rest of the start logic from runs/scheduler/start.go
        return syncer.Run(ctx)
    }
}

Summary of Results

By implementing the Executor interface, you have:

1. Created a middleware-style wrapper for Flyte's scheduling logic.
2. Gained the ability to intercept every scheduled fire event.
3. Maintained compatibility with Flyte's idempotency and rate-limiting features by wrapping the TriggerExecutor.

The GoCronScheduler will now use your MaintenanceExecutor for both steady-state cron fires and the CatchupAll logic that runs during service bootstrap.