Configuring the Execution Controller

Configuring the Flyte Execution Controller involves defining how the manager handles network endpoints, leader election, service integrations, and the lifecycle of task executions. These settings are managed through the Config and GCConfig structures in the executor/pkg/config package.

Basic Controller Manager Configuration

The Execution Controller uses sigs.k8s.io/controller-runtime to manage its lifecycle. You can configure the network addresses for metrics and health probes, as well as enable leader election for high-availability deployments.

import (
    "github.com/flyteorg/flyte/v2/executor/pkg/config"
    metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
    ctrl "sigs.k8s.io/controller-runtime"
)

func InitializeManager() {
    cfg := config.GetConfig()

    mgr, err := ctrl.NewManager(k8sConfig, ctrl.Options{
        Scheme:                 scheme,
        Metrics: metricsserver.Options{
            BindAddress:   cfg.MetricsBindAddress, // Default: ":10254"
            SecureServing: cfg.MetricsSecure,      // Default: true
        },
        HealthProbeBindAddress: cfg.HealthProbeBindAddress, // Default: ":8081"
        LeaderElect:            cfg.LeaderElect,            // Default: false
        LeaderElectionID:       "abf369a8.flyte.org",
    })
}

Service Integration

The Executor interacts with external Flyte services to report events and manage task caching. These are configured via service URLs:

• EventsServiceURL: The endpoint for reporting TaskAction state updates (default: http://localhost:8090).
• CacheServiceURL: The endpoint for task catalog and caching operations (default: http://localhost:8094).
• Cluster: A unique identifier attached to action events to distinguish between different Flyte clusters.

// Example configuration values
// cfg.EventsServiceURL = "http://flyteadmin.flyte.svc.cluster.local:80"
// cfg.CacheServiceURL = "http://datacatalog.flyte.svc.cluster.local:80"
// cfg.Cluster = "production-us-east-1"

Garbage Collection for TaskActions

Flyte uses a garbage collector to clean up terminal TaskAction resources. This prevents the Kubernetes API server from being overwhelmed by completed tasks. The GCConfig struct defines the behavior of this collector.

// In executor/setup.go, the GC is initialized based on the config:
if cfg.GC.Interval.Duration > 0 {
    // Ensure MaxTTL is positive if GC is enabled
    if cfg.GC.MaxTTL.Duration <= 0 {
        log.Fatalf("gc.maxTTL must be positive when gc is enabled")
    }
    
    gc := controller.NewGarbageCollector(
        mgr.GetClient(), 
        cfg.GC.Interval.Duration, 
        cfg.GC.MaxTTL.Duration,
    )
    mgr.Add(gc)
}

Parameter	Description	Default
gc.interval	How often the garbage collector runs. Set to 0 to disable.	30m
gc.maxTTL	The time-to-live for terminal TaskActions before they are deleted.	1h

Task Execution Defaults

You can define global defaults for how tasks are executed within the cluster, including security contexts and failure handling.

Default Service Account

The DefaultK8sServiceAccount is assigned to task pods when the task's own security context does not specify one. If this is also empty, Kubernetes assigns the default ServiceAccount of the pod's namespace.

// executor/pkg/plugin/task_exec_metadata.go
func resolveServiceAccount(securityContext *core.SecurityContext, defaultSA string) string {
    if sa := securityContext.GetRunAs().GetK8SServiceAccount(); sa != "" {
        return sa
    }
    return defaultSA
}

System Failure Threshold

MaxSystemFailures (default: 3) bounds the number of consecutive system-level failures (such as plugin errors or system-retryable failures) before a TaskAction is forced into a permanent failure state.

Security and TLS Configuration

If MetricsSecure is enabled (which is the default), you must provide certificates for the metrics server. Similarly, webhooks require certificate configuration.

// Configuring TLS for Metrics in executor/setup.go
if cfg.MetricsSecure && len(cfg.MetricsCertPath) > 0 {
    metricsServerOptions.CertDir = cfg.MetricsCertPath
    metricsServerOptions.CertName = cfg.MetricsCertName // Default: "tls.crt"
    metricsServerOptions.KeyName = cfg.MetricsCertKey   // Default: "tls.key"
}

For webhooks, use the following fields in Config:

• WebhookCertPath: Directory containing the certificates.
• WebhookCertName: Name of the certificate file (default: tls.crt).
• WebhookCertKey: Name of the key file (default: tls.key).
• EnableHTTP2: Enables HTTP/2 for both metrics and webhook servers (default: false).

Troubleshooting

• Executor Fails to Start: If you enable garbage collection by setting gc.interval to a non-zero value, you must also set gc.maxTTL to a positive duration. Failure to do so will cause the Setup function in executor/setup.go to return an error.
• Metrics Connection Refused: If MetricsSecure is true (default), ensure your monitoring system is configured to use HTTPS and trust the certificates provided in MetricsCertPath.
• Task Pods using 'default' SA: If tasks are not using the expected ServiceAccount, check if DefaultK8sServiceAccount is set in the executor config section and that the task itself hasn't overridden it in its SecurityContext.