Skip to main content

Configuring the Trigger Executor

The TriggerExecutor is responsible for firing CreateRun calls to the RunService when a scheduled trigger fires or when the scheduler catches up on missed runs. It manages rate limiting to protect the RunService and ensures that duplicate execution attempts are idempotent.

Initializing the Trigger Executor​

To set up the TriggerExecutor, you must provide a TriggerExecutorConfig which defines the target service URL, rate limits, and client options.

import (
    "connectrpc.com/connect"
    "github.com/flyteorg/flyte/v2/runs/scheduler/executor"
)

// Define configuration
cfg := executor.TriggerExecutorConfig{
    BaseURL:    "http://localhost:8090",
    QPS:        10.0,
    Burst:      20,
    ClientOpts: []connect.ClientOption{
        // Add custom connect client options here
    },
}

// Create the executor
exec := executor.NewTriggerExecutor(cfg)

In the Flyte runs service, this initialization typically happens in runs/scheduler/start.go, where it pulls values from the TriggerSchedulerConfig defined in the runs configuration section.

Configuring Rate Limiting​

The TriggerExecutor uses a token-bucket rate limiter (golang.org/x/time/rate) to control the frequency of outbound CreateRun requests. This prevents the scheduler from overwhelming the RunService during bursts of scheduled activity or during a catchup phase.

  • QPS (Queries Per Second): Set via ExecutionQPS (default: 10.0). This defines the steady-state rate at which tokens are added to the bucket.
  • Burst: Set via ExecutionBurst (default: 20). This defines the maximum number of tokens the bucket can hold, allowing for brief spikes in execution requests.

If the rate limit is reached, the Execute method will block until a token becomes available or the provided context.Context is cancelled.

Connection and Client Options​

The executor communicates with the RunService using a Connect-based gRPC client.

  • BaseURL: The endpoint of the RunService (e.g., http://localhost:8090).
  • ClientOpts: A slice of connect.ClientOption used when constructing the RunServiceClient. This allows you to inject interceptors, custom headers, or timeout configurations.

The client is instantiated inside NewTriggerExecutor using the default HTTP client:

// From runs/scheduler/executor/trigger_executor.go
func NewTriggerExecutor(cfg TriggerExecutorConfig) *TriggerExecutor {
	return &TriggerExecutor{
		runClient: workflowconnect.NewRunServiceClient(http.DefaultClient, cfg.BaseURL, cfg.ClientOpts...),
		limiter:   rate.NewLimiter(rate.Limit(cfg.QPS), cfg.Burst),
	}
}

Idempotency and Deterministic Naming​

Flyte ensures that a specific trigger at a specific scheduled time is only executed once, even if the scheduler attempts to fire it multiple times (e.g., due to a restart or a race condition).

The TriggerExecutor generates a deterministic run name using the trigger's project, domain, task name, trigger name, and the exact scheduled timestamp:

// From 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())
}

When Execute is called, it attempts to create a run with this name. If the RunService returns a connect.CodeAlreadyExists error, the executor treats this as a success and logs that the run was skipped.

Troubleshooting​

Blocking on Rate Limits​

If you observe that scheduled runs are delayed or the scheduler appears "stuck," check the ExecutionQPS and ExecutionBurst settings. Because the executor calls e.limiter.Wait(ctx), it will wait indefinitely (or until the context timeout) for a rate-limit token. If the number of triggers firing simultaneously exceeds the burst capacity and the QPS is too low to recover quickly, executions will queue up.

Missing Kickoff Time​

For triggers that require the scheduled time to be passed as an input argument (common in older SDK versions), the executor looks for a KickoffTimeInputArg in the TriggerAutomationSpec. If this is defined, it manually injects the scheduledAt time into the run inputs. If your workflow expects a kickoff time but isn't receiving it, ensure the AutomationSpec in the Trigger model correctly defines the Schedule.KickoffTimeInputArg.