Rebase final plugin arch

Signed-off-by: Alyssa Goins <agoins@redhat.com>

Author: alyssacgoins

backend/src/apiserver/plugins/config.go

@@ -16,81 +16,23 @@ package plugins
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"time"
+	"strings"
 
 	apiv2beta1 "github.com/kubeflow/pipelines/backend/api/v2beta1/go_client"
-	"github.com/kubeflow/pipelines/backend/src/apiserver/model"
 	commonplugins "github.com/kubeflow/pipelines/backend/src/common/plugins"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
-	"google.golang.org/protobuf/encoding/protojson"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
 
-type PluginsOutputEnvelope struct {
-	Plugins map[string]json.RawMessage
-}
-
-func (e *PluginsOutputEnvelope) UnmarshalJSON(data []byte) error {
-	var all map[string]json.RawMessage
-	if err := json.Unmarshal(data, &all); err != nil {
-		return err
-	}
-	if len(all) > 0 {
-		e.Plugins = all
-	}
-	return nil
-}
-
-func (e PluginsOutputEnvelope) MarshalJSON() ([]byte, error) {
-	if len(e.Plugins) == 0 {
-		return []byte("{}"), nil
-	}
-	return json.Marshal(e.Plugins)
-}
-
-// set stores a plugin entry by name.
-func (e *PluginsOutputEnvelope) set(name string, data json.RawMessage) {
-	if e.Plugins == nil {
-		e.Plugins = make(map[string]json.RawMessage)
-	}
-	e.Plugins[name] = data
-}
-
-func (e *PluginsOutputEnvelope) forEachEntry(fn func(name string, payload json.RawMessage)) {
-	for name, payload := range e.Plugins {
-		fn(name, payload)
-	}
-}
-
-const (
-	EntryExperimentName = "experiment_name"
-	EntryExperimentID   = "experiment_id"
-	EntryRootRunID      = "root_run_id"
-	EntryRunURL         = "run_url"
-	EntryEndpoint       = "endpoint"
-)
-
 const (
 	LauncherConfigMapName   = "kfp-launcher"
 	LauncherConfigKeyPrefix = "plugins."
 )
 
-const (
-	// DefaultTimeout is the default HTTP request timeout for the plugin client.
-	DefaultTimeout = "30s"
-)
-
-type RunSyncMode string
-
-const (
-	RunSyncModeTerminal RunSyncMode = "terminal"
-	RunSyncModeRetry    RunSyncMode = "retry"
-)
-
 // KubeClientProvider abstracts Kubernetes clientset access.
 type KubeClientProvider interface {
 	GetClientSet() kubernetes.Interface
@@ -105,20 +47,20 @@ type PluginConfig struct {
 
 // InjectPluginRuntimeEnv upserts plugin-provided environment variables into the
 // driver and launcher containers of the execution spec.
-func InjectPluginRuntimeEnv(executionSpec util.ExecutionSpec, env map[string]string) error {
-	if len(env) == 0 || executionSpec == nil {
+func InjectPluginRuntimeEnv(executionSpec util.ExecutionSpec, envVars []corev1.EnvVar) error {
+	if len(envVars) == 0 || executionSpec == nil {
 		return nil
 	}
-	return executionSpec.UpsertRuntimeEnvVars(env,
+	return executionSpec.UpsertRuntimeEnvVars(envVars,
 		util.ExecutionRuntimeRoleDriver,
 		util.ExecutionRuntimeRoleLauncher,
 	)
 }
 
-// GetNamespacePluginConfig reads the namespace-level Plugin configuration
-// from the kfp-launcher ConfigMap.  Returns nil (no error) when the ConfigMap
-// or key is absent.
-func GetNamespacePluginConfig(ctx context.Context, clientSet kubernetes.Interface, pluginName, namespace string) (*PluginConfig, error) {
+// GetLauncherNamespacePluginConfigsMap retrieves a map of plugin configurations from a Kubernetes ConfigMap in a given namespace.
+// It requires a non-empty namespace and a valid Kubernetes clientset to access the ConfigMap.
+// Returns a map of plugin configurations or an error if the ConfigMap retrieval or parsing fails.
+func GetLauncherNamespacePluginConfigsMap(ctx context.Context, clientSet kubernetes.Interface, namespace string) (map[string]string, error) {
 	if namespace == "" {
 		return nil, util.NewInternalServerError(fmt.Errorf("namespace is empty"), "namespace must be specified when reading Plugin config")
 	}
@@ -130,129 +72,16 @@ func GetNamespacePluginConfig(ctx context.Context, clientSet kubernetes.Interfac
 		if apierrors.IsNotFound(err) {
 			return nil, nil
 		}
-		return nil, util.NewInternalServerError(err, "failed to read %s plugin namespace config from configmap %q in namespace %q", pluginName, LauncherConfigMapName, namespace)
-	}
-	launcherConfigKey := LauncherConfigKeyPrefix + pluginName
-	raw, ok := cm.Data[launcherConfigKey]
-	if !ok || raw == "" {
-		return nil, nil
-	}
-	var cfg PluginConfig
-	if err := json.Unmarshal([]byte(raw), &cfg); err != nil {
-		return nil, util.NewInternalServerError(err, "failed to parse %s plugin config from key %q in configmap %q/%q", pluginName, launcherConfigKey, namespace, LauncherConfigMapName)
+		return nil, util.NewInternalServerError(err, "failed to read MLflow namespace config from configmap %q in namespace %q", LauncherConfigMapName, namespace)
 	}
-	return &cfg, nil
-}
 
-// SetPendingRunPluginOutput serializes the given PluginOutput into PendingRun.PluginsOutput.
-func SetPendingRunPluginOutput(run *PendingRun, pluginName string, output *apiv2beta1.PluginOutput) error {
-	if run == nil || output == nil || pluginName == "" {
-		return nil
-	}
-	result, err := upsertPluginOutput(run.PluginsOutput, pluginName, output)
-	if err != nil {
-		return err
-	}
-	run.PluginsOutput = &result
-	return nil
-}
-
-// upsertPluginOutput merges a single plugin's output into an existing
-// plugins_output JSON string, returning the updated JSON.
-func upsertPluginOutput(existing *string, pluginName string, output *apiv2beta1.PluginOutput) (string, error) {
-	marshaledOutput, err := protojson.Marshal(output)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal plugin output for %q: %w", pluginName, err)
-	}
-	var envelope PluginsOutputEnvelope
-	if existing != nil && *existing != "" {
-		if err := json.Unmarshal([]byte(*existing), &envelope); err != nil {
-			return "", fmt.Errorf("failed to unmarshal existing plugins_output: %w", err)
+	cfgMap := make(map[string]string)
+	for key, value := range cm.Data {
+		if strings.HasPrefix(key, LauncherConfigKeyPrefix) {
+			cfgMap[strings.TrimPrefix(key, LauncherConfigKeyPrefix)] = value
 		}
 	}
-	envelope.set(pluginName, marshaledOutput)
-	marshaledMap, err := json.Marshal(envelope)
-	if err != nil {
-		return "", fmt.Errorf("failed to marshal plugins_output map: %w", err)
-	}
-	return string(marshaledMap), nil
-}
-
-func GetParentRunID(output *apiv2beta1.PluginOutput) string {
-	return GetStringEntry(output, EntryRootRunID)
-}
-
-func GetStringEntry(output *apiv2beta1.PluginOutput, key string) string {
-	if output == nil || output.Entries == nil || key == "" {
-		return ""
-	}
-	entry, ok := output.Entries[key]
-	if !ok || entry == nil || entry.Value == nil {
-		return ""
-	}
-	return entry.Value.GetStringValue()
-}
-
-// PersistPluginsOutput serializes the PersistedRun's PluginsOutput and writes
-// it to the database via the given store.
-func PersistPluginsOutput(run *PersistedRun, store RunPluginOutputStore) error {
-	lt, err := SerializePluginsOutput(run.PluginsOutput)
-	if err != nil {
-		return fmt.Errorf("failed to serialize plugins_output for run %q: %w", run.RunID, err)
-	}
-	return store.UpdateRunPluginsOutput(run.RunID, lt)
-}
-
-func SerializePluginsOutput(outputs map[string]*apiv2beta1.PluginOutput) (*model.LargeText, error) {
-	if len(outputs) == 0 {
-		return nil, nil
-	}
-	var envelope PluginsOutputEnvelope
-	for key, output := range outputs {
-		marshaledOutput, err := protojson.Marshal(output)
-		if err != nil {
-			return nil, fmt.Errorf("failed to marshal plugin output for %q: %w", key, err)
-		}
-		envelope.set(key, marshaledOutput)
-	}
-	marshaledMap, err := json.Marshal(envelope)
-	if err != nil {
-		return nil, fmt.Errorf("failed to marshal plugins_output map: %w", err)
-	}
-	lt := model.LargeText(string(marshaledMap))
-	return &lt, nil
-}
-
-// ResolvePluginRequestConfig builds the effective PluginConfig for a single
-// request by merging namespace-level overrides on top of the global config.
-// Returns (nil, nil) when neither source provides configuration, signaling
-// that the plugin is unconfigured rather than broken.
-func ResolvePluginRequestConfig(ctx context.Context, clientSet kubernetes.Interface, handler RunPluginHandler, namespace string) (*PluginConfig, error) {
-	if handler == nil {
-		return nil, fmt.Errorf("handler is nil")
-	}
-
-	globalCfg, err := handler.GetGlobalPluginConfig()
-	if err != nil {
-		return nil, err
-	}
-
-	namespaceCfg, err := GetNamespacePluginConfig(ctx, clientSet, handler.Name(), namespace)
-	if err != nil {
-		return nil, err
-	}
-
-	merged, err := MergePluginConfig(namespaceCfg, globalCfg)
-	if err != nil {
-		return nil, err
-	}
-	if merged == nil {
-		return nil, nil
-	}
-	if merged.Timeout == "" {
-		merged.Timeout = DefaultTimeout
-	}
-	return merged, nil
+	return cfgMap, nil
 }
 
 // MergePluginConfig merges namespace-level overrides into the global config.
@@ -298,47 +127,6 @@ func mergeSettings(ns, global map[string]interface{}) map[string]interface{} {
 	return merged
 }
 
-// ModelToPersistedRun converts a model.Run to a PersistedRun for the
-// post-run plugin hooks (OnRunEnd, OnRunRetry).
-func ModelToPersistedRun(m *model.Run, namespace string) (*PersistedRun, error) {
-	if m == nil {
-		return nil, fmt.Errorf("model.Run is nil")
-	}
-	pluginsOutput, err := DeserializePluginsOutput(m.PluginsOutputString)
-	if err != nil {
-		return nil, fmt.Errorf("failed to deserialize plugins_output for run %q: %w", m.UUID, err)
-	}
-	pr := &PersistedRun{
-		RunID:         m.UUID,
-		Namespace:     namespace,
-		State:         string(m.RunDetails.State), //nolint:staticcheck // QF1008
-		PluginsOutput: pluginsOutput,
-	}
-	if m.RunDetails.FinishedAtInSec > 0 { //nolint:staticcheck // QF1008
-		t := time.Unix(m.RunDetails.FinishedAtInSec, 0) //nolint:staticcheck // QF1008
-		pr.FinishedAt = &t
-	}
-	return pr, nil
-}
-
-func DeserializePluginsOutput(raw *model.LargeText) (map[string]*apiv2beta1.PluginOutput, error) {
-	result := make(map[string]*apiv2beta1.PluginOutput)
-	if raw == nil || *raw == "" {
-		return result, nil
-	}
-	var envelope PluginsOutputEnvelope
-	if err := json.Unmarshal([]byte(*raw), &envelope); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal plugins_output: %w", err)
-	}
-	envelope.forEachEntry(func(name string, payload json.RawMessage) {
-		output := &apiv2beta1.PluginOutput{}
-		if err := protojson.Unmarshal(payload, output); err == nil {
-			result[name] = output
-		}
-	})
-	return result, nil
-}
-
 func SetPluginOutputState(output *apiv2beta1.PluginOutput, state apiv2beta1.PluginState, stateMessage string) {
 	if output == nil {
 		return