feat: add ClusterPlugin interface for shared-cluster plugins

executor/pkg/plugin/registry.go

@@ -53,6 +53,13 @@ func (r *Registry) Initialize(ctx context.Context) error {
 
 	// Load k8s plugins
 	for _, entry := range r.pluginRegistry.GetK8sPlugins() {
+		if entry.ClusterPlugin != nil {
+			logger.Infof(ctx, "Skipping k8s plugin %q: cluster plugin entries are not supported by the executor", entry.ID)
+			continue
+		}
+		if entry.Plugin == nil {
+			return fmt.Errorf("invalid k8s plugin entry %q: Plugin is nil", entry.ID)
+		}
 		pm := executorK8s.NewPluginManager(
 			entry.ID,
 			entry.Plugin,

flyteidl2/core/execution.proto

@@ -151,8 +151,11 @@ message LogContext {
   repeated PodLogContext pods = 1;
   string primary_pod_name = 2;
   ConnectorLogContext connector = 3;
-  // Pod-name prefix used by log sources to narrow stream/pod searches
-  string pod_name_prefix = 4;
+  // 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;
 }
 
 // ConnectorLogContext describes how to reach a connector that owns an action's logs.

flyteidl2/logs/dataplane/payload.proto

@@ -34,8 +34,11 @@ message LoggingContext {
   map<string, string> kubernetes_pod_labels = 9;
   common.Sort.Direction order = 10;
   uint64 number_of_batches = 11;
-  // +optional, Pod-name prefix used by log sources to narrow stream/pod searches.
-  string pod_name_prefix = 12;
+  // +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;
 }
 
 // Parameters of environment in which logs were collected. Should contain everything