feat: add ClusterPlugin interface for shared-cluster plugins - #7460
Open
pingsutw wants to merge 18 commits into
Open
feat: add ClusterPlugin interface for shared-cluster plugins#7460pingsutw wants to merge 18 commits into
pingsutw wants to merge 18 commits into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a new Kubernetes plugin model for “job-on-shared-cluster” execution (ClusterPlugin + ClusterPluginManager) and migrates the Ray plugin to use a long-lived, hash-keyed RayCluster with per-run RayJob submission. This enables multiple identical Ray task specs to reuse a single cluster and avoid per-task cold-start costs.
Changes:
- Added
ClusterPlugininterface and extended plugin registration to support exactly one ofPluginvsClusterPlugin. - Implemented
ClusterPluginManagerin the executor to drive a cluster-create/wait + job-submit/watch state machine. - Migrated Ray plugin to split resources into
RayCluster+RayJob, including deterministic cluster naming and per-job identity injection viaruntime_env.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 8 comments.
Show a summary per file
| File | Description |
|---|---|
| flyteplugins/go/tasks/plugins/k8s/ray/ray.go | Migrates Ray to shared-cluster model; adds deterministic cluster naming, env stripping, runtime_env identity injection, and job/cluster split. |
| flyteplugins/go/tasks/plugins/k8s/ray/ray_test.go | Updates Ray tests for the new cluster/job split and adds coverage for naming/readiness/runtime_env injection. |
| flyteplugins/go/tasks/pluginmachinery/registry.go | Validates plugin registration for Plugin vs ClusterPlugin and enforces cluster resource watch type when needed. |
| flyteplugins/go/tasks/pluginmachinery/registry_test.go | Adds unit tests for the new registry validation rules. |
| flyteplugins/go/tasks/pluginmachinery/k8s/plugin.go | Introduces the ClusterPlugin interface and extends PluginEntry with cluster resource watch metadata. |
| executor/setup.go | Registers Ray + JobSet CRDs into the executor scheme and imports the Ray plugin package. |
| executor/pkg/plugin/registry.go | Instantiates ClusterPluginManager for cluster plugins and PluginManager for legacy plugins. |
| executor/pkg/plugin/k8s/plugin_manager.go | Refactors metadata injection to support injecting name/ownership conditionally (used by shared cluster resources). |
| executor/pkg/plugin/k8s/cluster_plugin_manager.go | Adds the new executor-side manager and its persisted state machine. |
| executor/pkg/plugin/k8s/cluster_plugin_manager_test.go | Adds unit tests validating the cluster/job lifecycle and that abort/finalize only act on the job. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
328
to
331
| // The cluster is shared across tasks, so its pods must not carry any single task's execution | ||
| // identity. Per-job identity travels with the RayJob instead. | ||
| stripRunScopedEnvVars(&headPodTemplate.Spec) | ||
|
|
Comment on lines
328
to
331
| // The cluster is shared across tasks, so its pods must not carry any single task's execution | ||
| // identity. Per-job identity travels with the RayJob instead. | ||
| stripRunScopedEnvVars(&headPodTemplate.Spec) | ||
|
|
Comment on lines
+113
to
+117
| func (m *ClusterPluginManager) addClusterMetadata(taskCtx pluginsCore.TaskExecutionMetadata, o client.Object, cfg *config.K8sPluginConfig, clusterName string) { | ||
| o.SetNamespace(taskCtx.GetNamespace()) | ||
| o.SetAnnotations(pluginsUtils.UnionMaps(cfg.DefaultAnnotations, o.GetAnnotations(), pluginsUtils.CopyMap(taskCtx.GetAnnotations()))) | ||
| o.SetLabels(pluginsUtils.UnionMaps(cfg.DefaultLabels, o.GetLabels(), pluginsUtils.CopyMap(taskCtx.GetLabels()))) | ||
| o.SetName(sanitizeName(clusterName)) |
Comment on lines
+277
to
+283
| jobSpec := rayv1.RayJobSpec{ | ||
| // The cluster is shared and standalone; the job must not own or shut it down. | ||
| RayClusterSpec: nil, | ||
| ClusterSelector: map[string]string{rayClusterLabelKey: clusterName}, | ||
| Entrypoint: entrypoint, | ||
| RuntimeEnvYAML: runtimeEnvYaml, | ||
| ShutdownAfterJobFinishes: false, |
Comment on lines
+1005
to
+1009
| ID: rayTaskType, | ||
| RegisteredTaskTypes: []pluginsCore.TaskType{rayTaskType}, | ||
| ResourceToWatch: &rayv1.RayJob{}, | ||
| ClusterResourceToWatch: &rayv1.RayCluster{}, | ||
| ClusterPlugin: rayJobResourceHandler{}, |
Comment on lines
+135
to
+144
| hash, err := pbhash.ComputeHashString(ctx, &rayJob) | ||
| if err != nil { | ||
| return "", flyteerr.Errorf(flyteerr.BadTaskSpecification, "failed to compute hash for ray job spec [%v]", err.Error()) | ||
| } | ||
|
|
||
| safe := utils.ConvertToDNS1123SubdomainCompatibleString(strings.ToLower(hash)) | ||
| if len(safe) > clusterNameHashLength { | ||
| safe = safe[:clusterNameHashLength] | ||
| } | ||
| return "ray-" + safe, nil |
Comment on lines
393
to
396
| // The cluster is shared across tasks, so its pods must not carry any single task's execution | ||
| // identity. Per-job identity travels with the RayJob instead. | ||
| stripRunScopedEnvVars(&headPodTemplate.Spec) | ||
|
|
| if err != nil { | ||
| return nil, err | ||
| } | ||
| stripRunScopedEnvVars(&workerPodTemplate.Spec) |
Comment on lines
+405
to
+406
| // Shared clusters should autoscale and idle-reap. | ||
| EnableInTreeAutoscaling: ptrBool(true), |
Comment on lines
+1075
to
+1086
| if existing, ok := doc["env_vars"]; ok { | ||
| switch m := existing.(type) { | ||
| case map[interface{}]interface{}: | ||
| for k, v := range m { | ||
| merged[fmt.Sprintf("%v", k)] = v | ||
| } | ||
| case map[string]interface{}: | ||
| for k, v := range m { | ||
| merged[k] = v | ||
| } | ||
| } | ||
| } |
pingsutw
changed the title
Add a ClusterPlugin interface to pluginmachinery, alongside Plugin, for plugins that run a job on top of a shared, long-lived cluster resource: the cluster is created once (keyed by a deterministic name, e.g. a hash of the job-spec proto), reused by every task whose spec maps to the same name, and the job is created only after the cluster is ready. The cluster is intentionally not owned by any task execution — no owner reference and no finalizer — so completing or aborting one task never deletes a cluster other tasks may still be using; the job resource is owned normally. PluginEntry gains ClusterPlugin and ClusterResourceToWatch, and RegisterK8sPlugin validates that exactly one of Plugin/ClusterPlugin is set. The executor plugin registry skips ClusterPlugin-only entries, which are driven by an external ClusterPluginManager rather than the executor. Signed-off-by: Kevin Su <pingsutw@apache.org>
pingsutw
force-pushed
the
feature/cluster-plugin-interface
branch
from
July 15, 2026 05:03
9a2c093 to
d7582d2
Compare
…ter pod The head and worker pod templates carry the task's execution metadata via flytek8s.ToK8sPodSpec, but the submitter pod template was built bare and KubeRay uses it verbatim — leaving the submitter pod invisible to anything that locates a task's pods by execution labels (e.g. log tailing). Signed-off-by: Kevin Su <pingsutw@apache.org>
Exported so wrappers that determine head readiness through a different lookup (e.g. jobs submitted to a pre-existing cluster via ClusterSelector) reuse the exact same link handling. Signed-off-by: Kevin Su <pingsutw@apache.org>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> Signed-off-by: Kevin Su <pingsutw@gmail.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 10 changed files in this pull request and generated 1 comment.
Files not reviewed (3)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
Comments suppressed due to low confidence (6)
flyteplugins/go/tasks/plugins/k8s/ray/ray.go:306
- The PR description says the in-tree Ray plugin is "byte-identical to main", but this file is modified (e.g., buildSubmitterPodTemplate now takes taskCtx). The PR description should be updated to reflect these changes (and the flyteidl proto/log context additions) or those changes should be split into a separate PR for clarity.
submitterPodTemplate := buildSubmitterPodTemplate(&rayClusterSpec, taskCtx)
flyteplugins/go/tasks/plugins/k8s/ray/ray.go:534
- New behavior: the submitter pod template now propagates task execution labels/annotations. There are existing Ray unit tests covering submitter resources/affinity, but none asserting labels/annotations are carried through; adding an assertion would prevent regressions in log tailing / pod discovery.
k8sCfg := config.GetK8sPluginConfig()
podTemplateSpec.SetLabels(utils.UnionMaps(k8sCfg.DefaultLabels, utils.CopyMap(taskCtx.TaskExecutionMetadata().GetLabels())))
podTemplateSpec.SetAnnotations(utils.UnionMaps(k8sCfg.DefaultAnnotations, utils.CopyMap(taskCtx.TaskExecutionMetadata().GetAnnotations())))
return podTemplateSpec
flyteplugins/go/tasks/pluginmachinery/registry_test.go:7
- This registry validation test only needs any
client.Objecttypes for ResourceToWatch/ClusterResourceToWatch, but it imports KubeRay CRD types (rayv1). Using core Kubernetes types here (e.g. Pod/Service) reduces unnecessary coupling and speeds up compilation for this package’s tests.
"testing"
rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
"github.com/stretchr/testify/assert"
flyteplugins/go/tasks/pluginmachinery/registry_test.go:25
- Use a core Kubernetes type here instead of a KubeRay CRD; the specific kind is irrelevant for this validation test.
ResourceToWatch: &rayv1.RayJob{},
flyteplugins/go/tasks/pluginmachinery/registry_test.go:38
- Use core Kubernetes types here instead of KubeRay CRDs; the test only verifies nil/non-nil validation.
ResourceToWatch: &rayv1.RayJob{},
ClusterResourceToWatch: &rayv1.RayCluster{},
flyteplugins/go/tasks/pluginmachinery/registry_test.go:52
- Use core Kubernetes types here instead of KubeRay CRDs; the test only verifies nil/non-nil validation.
ResourceToWatch: &rayv1.RayJob{},
ClusterResourceToWatch: &rayv1.RayCluster{},
…/flyteorg/flyte into feature/cluster-plugin-interface
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 16 changed files in this pull request and generated no new comments.
Files not reviewed (3)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
Comments suppressed due to low confidence (2)
flyteidl2/core/execution.proto:158
- The protos add
pod_name_prefixes, but other checked-in generated bindings (TypeScript/Python/Rust) still appear to only containpod_name_prefix(e.g. gen/ts/flyteidl2/core/execution_pb.ts and gen/python/flyteidl2/core/execution_pb2.pyi). To keep language bindings consistent (and make the new field usable by non-Go consumers), please regenerate and commit the updated artifacts for all supported languages.
// Deprecated: use pod_name_prefixes. Still populated with the task's own pod-name prefix for
// readers that predate pod_name_prefixes.
string pod_name_prefix = 4 [deprecated = true];
// Pod-name prefixes used by log sources to narrow stream/pod searches.
repeated string pod_name_prefixes = 5;
flyteidl2/logs/dataplane/payload.proto:41
pod_name_prefixeswas added here, but the repo’s checked-in generated bindings for other languages (TypeScript/Python/Rust) still only definepod_name_prefix(e.g. gen/ts/flyteidl2/logs/dataplane/payload_pb.ts and gen/python/flyteidl2/logs/dataplane/payload_pb2.pyi). Please regenerate and commit the updated artifacts so the change is complete and consumers outside Go can use the new field.
// +optional. Deprecated: use pod_name_prefixes. Still populated with the task's own pod-name
// prefix for readers that predate pod_name_prefixes.
string pod_name_prefix = 12 [deprecated = true];
// +optional, pod-name prefixes used by log sources to narrow stream/pod searches.
repeated string pod_name_prefixes = 13;
pingsutw
marked this pull request as ready for review
July 20, 2026 17:55
pingsutw
force-pushed
the
feature/cluster-plugin-interface
branch
from
July 20, 2026 17:58
d85c7da to
491c66c
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 11 out of 14 changed files in this pull request and generated no new comments.
Files not reviewed (3)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
Comments suppressed due to low confidence (1)
charts/flyte-devbox/templates/proxy/ingress.yaml:57
- The console ingress path is hard-coded to
/v2, but the local dev console is configured via.Values.sandbox.console.basePath(and the flyte-binary subchart supports.Values.flyte-binary.console.basePath). If either basePath is overridden, the console will serve/probe a different path than the ingress routes, breaking access/readiness. Consider deriving the ingress path from the same values used by the selected console implementation.
number: 80
{{- end }}
---
{{- if .Values.rustfs.enabled}}
apiVersion: networking.k8s.io/v1
pingsutw
force-pushed
the
feature/cluster-plugin-interface
branch
from
July 20, 2026 17:59
491c66c to
fc83037
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 10 changed files in this pull request and generated 2 comments.
Files not reviewed (3)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
Comments suppressed due to low confidence (1)
flyteidl2/core/execution.proto:158
- Proto changes add pod_name_prefixes, but several committed generated bindings are still stale (e.g. gen/ts/flyteidl2/core/execution_pb.ts, gen/python/.../execution_pb2.pyi, gen/rust/src/flyteidl2.core.rs still only expose pod_name_prefix). This leaves different language SDKs out of sync with the canonical .proto API and can break downstream builds that rely on the committed generated artifacts. Please re-run the repo’s protobuf generation for all supported languages (Go/TS/Python/Rust, and any others committed) so they include pod_name_prefixes and mark pod_name_prefix deprecated.
// Deprecated: use pod_name_prefixes. Still populated with the task's own pod-name prefix for
// readers that predate pod_name_prefixes.
string pod_name_prefix = 4 [deprecated = true];
// Pod-name prefixes used by log sources to narrow stream/pod searches.
repeated string pod_name_prefixes = 5;
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 10 changed files in this pull request and generated no new comments.
Files not reviewed (3)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
Comments suppressed due to low confidence (4)
flyteplugins/go/tasks/plugins/k8s/ray/ray.go:791
tl.Readyis set directly from thereadyparameter, but the accompanying reason treats the dashboard as “not ready” whenphaseInfo.Phase() < PhaseRunning. This can yield contradictory state (e.g.Ready=truewhile the phase reason says “Ray dashboard is not ready”). Consider gating theReadybit using the same phase condition used for the reason so readiness and reason stay consistent.
if tl != nil && tl.LinkType == core.TaskLog_DASHBOARD {
tl.Ready = ready
if !ready || phaseInfo.Phase() < pluginsCore.PhaseRunning {
phaseInfo.WithReason("Ray dashboard is not ready")
} else {
flyteplugins/go/tasks/plugins/k8s/ray/ray.go:798
- For IDE links, the reason marks the IDE as “not ready” unless
phaseInfo.Phase() == PhaseRunning, buttl.Readyis still set toreadyunconditionally. This can produceReady=truewhile the reason says the IDE is not ready (e.g. after the task succeeds but the pod is still ready). Consider makingtl.Readyfollow the same phase gating as the reason.
} else if tl != nil && tl.LinkType == core.TaskLog_IDE {
tl.Ready = ready
if !ready || phaseInfo.Phase() != pluginsCore.PhaseRunning {
phaseInfo.WithReason("Vscode server is not ready")
} else {
flyteidl2/logs/dataplane/payload.proto:41
- This proto adds
pod_name_prefixes, but the repository also checks in generated bindings for other languages (e.g.gen/python/.../payload_pb2.py,gen/ts/.../payload_pb.ts) and Go validation code. Those generated artifacts currently do not containpod_name_prefixes, so they appear out of date relative to this change. Please regenerate and commit all checked-in generated outputs for this proto to keep language bindings in sync.
// +optional. Deprecated: use pod_name_prefixes. Still populated with the task's own pod-name
// prefix for readers that predate pod_name_prefixes.
string pod_name_prefix = 12 [deprecated = true];
// +optional, pod-name prefixes used by log sources to narrow stream/pod searches.
repeated string pod_name_prefixes = 13;
flyteidl2/core/execution.proto:158
- This proto adds
pod_name_prefixes, but checked-in generated bindings for other languages (and Go validation code) should also be regenerated to include the new field and the deprecation marker. Right nowgen/python/.../execution_pb2.pyandgen/ts/.../execution_pb.tsdo not containpod_name_prefixes, suggesting the generated outputs are stale relative to this change.
// Deprecated: use pod_name_prefixes. Still populated with the task's own pod-name prefix for
// readers that predate pod_name_prefixes.
string pod_name_prefix = 4 [deprecated = true];
// Pod-name prefixes used by log sources to narrow stream/pod searches.
repeated string pod_name_prefixes = 5;
…anguages Signed-off-by: Kevin Su <pingsutw@apache.org>
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 18 changed files in this pull request and generated 1 comment.
Files not reviewed (5)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
- gen/python/flyteidl2/core/execution_pb2.py: Generated file
- gen/python/flyteidl2/logs/dataplane/payload_pb2.py: Generated file
Comments suppressed due to low confidence (1)
flyteplugins/go/tasks/pluginmachinery/k8s/plugin.go:125
- The
GetClusterNamedoc says “Callers should sanitize the result…”, but the framework is the caller and plugin implementations are the ones choosing/returning the name. To avoid ambiguity, document this as a requirement on implementations (e.g. the returned string must already be a DNS1123 subdomain) or explicitly state the framework will sanitize before using it.
// GetClusterName returns a deterministic name for the cluster backing this task. Implementations
// typically hash the task's plugin-spec proto so that identical specs collapse onto the same
// cluster. Callers should sanitize the result into a DNS1123 subdomain before use.
GetClusterName(ctx context.Context, taskCtx pluginsCore.TaskExecutionContext) (string, error)
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 18 changed files in this pull request and generated 1 comment.
Files not reviewed (5)
- flyteplugins/go/tasks/pluginmachinery/k8s/mocks/mocks.go: Generated file
- gen/go/flyteidl2/core/execution.pb.go: Generated file
- gen/go/flyteidl2/logs/dataplane/payload.pb.go: Generated file
- gen/python/flyteidl2/core/execution_pb2.py: Generated file
- gen/python/flyteidl2/logs/dataplane/payload_pb2.py: Generated file
cosmicBboy
pushed a commit
to flyteorg/flyte-sdk
that referenced
this pull request
Jul 24, 2026
…yClusters (#1308) ## Why `ReusePolicy` today only supports container/pod tasks (reusable actors). Ray tasks always pay full RayCluster cold-start cost on every run, even when many runs share an identical cluster spec. This PR lets users opt a Ray task environment into a shared, reusable RayCluster: ```python ray_env = flyte.TaskEnvironment( name="ray_env", plugin_config=RayJobConfig(...), reusable=flyte.ReusePolicy(replicas=1, idle_ttl=600), ) ``` ## What changed Core stays plugin-agnostic, and the task type is unchanged — reuse is signaled through the task's custom spec: - **Core**: plugins opt in to reuse via a `supports_reuse_policy` class attribute (checked by the `TaskEnvironment` reusable+plugin_config guard) and an `apply_reuse_policy(task_template)` hook that task serialization dispatches to instead of `add_reusable`. No Ray knowledge in `reuse.py`. - **Ray plugin**: validates the policy (`replicas` maps to the number of shared clusters; only 1 supported for now, otherwise `BadConfiguration`) and records it under a `reusePolicy` key in the task's custom (RayJob) spec. The field is unknown to the RayJob proto and ignored by its unmarshalling — the task type stays `ray`, so backends without shared-cluster support simply run the task as a normal ephemeral Ray job. - Backends with shared-cluster support route on the presence of `reusePolicy`: shared RayCluster (keyed by the Ray spec hash, so identical specs reuse one cluster) when present, ephemeral otherwise. The generic backend plugin interface lives in flyteorg/flyte#7460. - Example: `examples/plugins/ray_reusable_cluster.py`. ## Tests - `tests/flyte/internal/runtime/test_reuse.py` — core behavior unchanged (9 passing). - `plugins/ray/tests/test_task.py` — policy recorded in custom with type unchanged; multi-replica rejection (9 passing). --------- Signed-off-by: Kevin Su <pingsutw@apache.org>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Tracking issue
Related to shared-cluster execution for Ray (and future cluster-backed plugins).
Why are the changes needed?
Some plugins want to run a job on top of a shared, long-lived cluster instead of an ephemeral per-task resource: the cluster is created once (keyed by a deterministic name, e.g. the hash of the job-spec proto), reused by every task whose spec maps to the same name, and the job is created only after the cluster is ready. The existing
Plugininterface (singleBuildResource) cannot express this two-resource lifecycle.This PR adds the generic
ClusterPlugininterface and registry plumbing, plus a few supporting changes that let a shared-cluster consumer reuse the in-tree Ray plugin's job-shaping and link handling. Concrete consumers live downstream and driveClusterPluginwith their own evaluation loop.What changes were proposed in this pull request?
1. New
ClusterPlugininterface (flyteplugins/.../pluginmachinery/k8s/plugin.go)Sits alongside the existing
Plugininterface. Instead of oneBuildResource, it declares two resources, a readiness check, and cluster cleanup:The cluster resource is intentionally not owned by any task execution (no owner reference, no finalizer), so completing or aborting one task never deletes a cluster other tasks may still be using; the job resource is owned normally. Because cluster resources outlive individual tasks, cleanup is the plugin's own responsibility:
StartCleanupis called once at load time so a plugin can start a background loop that deletes clusters once they have been idle past their TTL.2. Registry validation (
pluginmachinery/registry.go)PluginEntrygainsClusterPlugin+ClusterResourceToWatch;RegisterK8sPluginvalidates exactly one ofPlugin/ClusterPluginis set.3. Executor guard (
executor/pkg/plugin/registry.go)The executor's plugin registry skips ClusterPlugin-only entries (it has no manager for them) instead of wrapping a nil
Plugin.4. Ray plugin: submitter-pod labels + reusable link-readiness helper (
plugins/k8s/ray/ray.go)buildSubmitterPodTemplatenow sources the submitter pod's labels/annotations from the task execution metadata, the same way the head and worker pod templates already do. KubeRay usesSubmitterPodTemplateverbatim, so previously the submitter pod carried none of the task's execution labels — leaving it invisible to anything that locates a task's pods by execution metadata (e.g. log tailing).UpdateLinkReadinessexported. The dashboard/IDE task-link readiness handling insideGetTaskPhaseis extracted into an exported helper, so a wrapper that determines head readiness through a different lookup (e.g. a job submitted to a pre-existing cluster viaClusterSelector) can reuse the identical link handling instead of duplicating it.5.
LogContext.pod_name_prefix→ repeatedpod_name_prefixes(flyteidl2/core/execution.proto,logs/dataplane/payload.proto)A task's pods can live under more than one name prefix — e.g. a job on a shared cluster has its own submitter pod plus the cluster's head/worker pods, which are named differently. The scalar
pod_name_prefixcan only describe one, so it is marked[deprecated = true](still populated with the task's own prefix for readers that predate the new field) and a repeatedpod_name_prefixesis added: element 0 is the task's own prefix; later elements cover pods living outside it. Additive and wire-compatible — existing readers keep using the scalar.How was this patch tested?
registry_test.gocovers registration/validation of ClusterPlugin entries.go test ./flyteplugins/...and./executor/...pass.main.