Skip to main content

Kubernetes Resource Plugins

Flyte provides a specialized framework for building plugins that manage Kubernetes resources, such as Pods or Custom Resources (CRDs) like SparkApplications or RayJobs. By implementing the k8s.Plugin interface, you can delegate the low-level Kubernetes API interactions (Create, Get, Delete, and Watch) to the Flyte engine while focusing on the resource definition and status mapping.

Implementing the Kubernetes Plugin Interface​

To create a Kubernetes resource plugin, you must implement the k8s.Plugin interface defined in flyteplugins/go/tasks/pluginmachinery/k8s/plugin.go.

1. Define the Identity Resource​

The BuildIdentityResource method returns a "template" object used by the Flyte engine to query the Kubernetes API. It should only contain the TypeMeta (Kind and APIVersion) of the resource. Flyte automatically populates the Name and Namespace during execution.

func (sparkResourceHandler) BuildIdentityResource(ctx context.Context, taskCtx pluginsCore.TaskExecutionMetadata) (client.Object, error) {
	return &sparkOp.SparkApplication{
		TypeMeta: metav1.TypeMeta{
			Kind:       KindSparkApplication,
			APIVersion: sparkOp.SchemeGroupVersion.String(),
		},
	}, nil
}

2. Build the Full Resource​

The BuildResource method is responsible for creating the complete Kubernetes object that will be submitted to the cluster. You typically extract task-specific configuration from the TaskExecutionContext.

func (p plugin) BuildResource(ctx context.Context, taskCtx pluginsCore.TaskExecutionContext) (client.Object, error) {
	taskTemplate, err := taskCtx.TaskReader().Read(ctx)
	if err != nil {
		return nil, err
	}

	// Logic to construct the resource, e.g., a Pod
	podSpec, objectMeta, primaryContainerName, err := flytek8s.BuildRawPod(ctx, taskCtx)
	if err != nil {
		return nil, err
	}

	return &v1.Pod{
		ObjectMeta: objectMeta,
		Spec:       *podSpec,
	}, nil
}

3. Map Kubernetes Status to Flyte Phases​

The GetTaskPhase method analyzes the current state of the Kubernetes resource and maps it to a pluginsCore.PhaseInfo. This method is called frequently during the reconciliation loop and must be efficient.

func (sparkResourceHandler) GetTaskPhase(ctx context.Context, pluginContext k8s.PluginContext, resource client.Object) (pluginsCore.PhaseInfo, error) {
	app := resource.(*sparkOp.SparkApplication)
	
	info := pluginsCore.TaskInfo{
		OccurredAt: &app.CreationTimestamp.Time,
	}

	switch app.Status.AppState.State {
	case sparkOp.CompletedState:
		return pluginsCore.PhaseInfoSuccess(&info), nil
	case sparkOp.FailedState:
		reason := fmt.Sprintf("Spark application failed: %s", app.Status.AppState.ErrorMessage)
		return pluginsCore.PhaseInfoRetryableFailure(errors.DownstreamSystemError, reason, &info), nil
	case sparkOp.RunningState:
		return pluginsCore.PhaseInfoRunning(pluginsCore.DefaultPhaseVersion, &info), nil
	}
	
	return pluginsCore.PhaseInfoQueued(pluginsCore.DefaultPhaseVersion, "Waiting for SparkApplication to start", &info), nil
}

Registering the Plugin​

Once the plugin is implemented, register it using a k8s.PluginEntry in your package's init() function. This associates your plugin with specific Flyte task types and tells the engine which Kubernetes resource type to watch.

func init() {
	pluginmachinery.PluginRegistry().RegisterK8sPlugin(
		k8s.PluginEntry{
			ID:                  "spark",
			RegisteredTaskTypes: []pluginsCore.TaskType{"spark"},
			ResourceToWatch:     &sparkOp.SparkApplication{},
			Plugin:              sparkResourceHandler{},
			IsDefault:           false,
		})
}

Key fields in PluginEntry:

  • ID: A unique identifier for the plugin.
  • RegisteredTaskTypes: The list of Flyte task types this plugin handles.
  • ResourceToWatch: An instance of the Kubernetes resource (e.g., &v1.Pod{}) used to initialize the informer.
  • IsDefault: If true, this plugin handles any task type not explicitly registered to another plugin.

Global Configuration and Defaults​

Flyte uses K8sPluginConfig (found in flyteplugins/go/tasks/pluginmachinery/flytek8s/config/config.go) to apply platform-wide defaults to Kubernetes resources. The PluginManager automatically injects these values into the resources created by your plugin.

Common configuration options include:

  • DefaultCPURequest / DefaultMemoryRequest: Applied if the task does not specify resource requirements.
  • DefaultLabels / DefaultAnnotations: Injected into every resource created by Flyte.
  • InterruptibleTolerations: Automatically added to pods when a task is marked as interruptible.
  • InjectFinalizer: If enabled, Flyte adds a finalizer to the Kubernetes resource to ensure clean-up.

Troubleshooting and Best Practices​

Phase Versioning​

If the Kubernetes resource transitions between sub-states that both map to the same Flyte phase (e.g., Running), use k8s.MaybeUpdatePhaseVersion to ensure the Flyte engine detects the change and updates the task execution metadata.

Resource Cleanup​

By default, Flyte injects owner references into created resources so they are garbage collected by Kubernetes when the parent execution is deleted. If your plugin manages resources in a remote cluster where owner references are not applicable, set DisableInjectOwnerReferences in the PluginProperties returned by GetProperties().

Performance​

GetTaskPhase is executed within the main FlytePropeller reconciliation loop. Avoid making network calls or performing heavy computations inside this method. If you need to fetch additional data, consider using a cache or offloading the work to a background process.