fix: prevent silent fallback to default value in secret resolver

config/secret_k8s.go

@@ -2,10 +2,11 @@ package config
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
+	"maps"
 	"net/url"
 	"os"
-	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
@@ -102,38 +103,23 @@ func (p k8sSecretProvider) getDSN(ctx context.Context, ref *url.URL) (string, er
 		return "", fmt.Errorf("invalid k8ssecret URL format: expected k8ssecret://[namespace/]secret-name, got %s", ref.String())
 	}
 
-	// Extract the key from the secret data
-	key := ref.Query().Get("key")
-	if key == "" {
-		key = "data_source_name"
-	}
-
 	// Fetch the secret from Kubernetes API
 	secret, err := provider.clientset.CoreV1().Secrets(namespace).Get(ctx, secretName, metav1.GetOptions{})
 	if err != nil {
 		return "", fmt.Errorf("unable to fetch secret %q from namespace %q: %w", secretName, namespace, err)
 	}
-
-	// Extract the key from secret data - check both Data (binary) and StringData (string)
-	var secretValue string
-
-	// Check in Data field first (for binary/encoded data)
-	if data, ok := secret.Data[key]; ok {
-		secretValue = string(data)
-	} else if stringData, ok := secret.StringData[key]; ok {
-		// Check in StringData field (for direct string values)
-		secretValue = stringData
-	} else {
-		return "", fmt.Errorf("key %q not found in Kubernetes secret %s/%s", key, namespace, secretName)
+	// Return the raw secret payload (all keys) as JSON. Per-call '?key=' selection and '?template=' substitution are
+	// handled by secretResolver.extractKey, since this provider's result is cached (and shared) across all references
+	// to the same secret object regardless of query params.
+	raw := make(map[string]string, len(secret.Data)+len(secret.StringData))
+	for k, v := range secret.Data {
+		raw[k] = string(v)
 	}
+	maps.Copy(raw, secret.StringData)
 
-	// Apply template if provided
-	templateStr := ref.Query().Get("template")
-	if templateStr != "" {
-		// Simple string replacement - replace all occurrences of DSN_VALUE with the secret value
-		result := strings.ReplaceAll(templateStr, "DSN_VALUE", secretValue)
-		return result, nil
+	b, err := json.Marshal(raw)
+	if err != nil {
+		return "", fmt.Errorf("unable to marshal Kubernetes secret %s/%s: %w", namespace, secretName, err)
 	}
-
-	return secretValue, nil
+	return string(b), nil
 }

config/secret_k8s_test.go

@@ -0,0 +1,222 @@
+package config
+
+import (
+	"context"
+	"encoding/json"
+	"net/url"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+// newTestK8sProvider builds a k8sSecretProvider via a fake clientset, bypassing getK8sProvider in-cluster setup
+func newTestK8sProvider(namespace string, objects ...*corev1.Secret) k8sSecretProvider {
+	// k8sSecretProvider.getDSN calls getK8sProvider() internally; to keep this test hermetic we exercise the
+	// provider's logic via a lightweight wrapper.
+	return k8sSecretProvider{}
+}
+
+// newFakeClientsetWithSecret creates a fake Kubernetes clientset with a single Secret object for testing.
+func newFakeClientsetWithSecret(namespace, name string, data map[string][]byte, stringData map[string]string) *fake.Clientset {
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Data:       data,
+		StringData: stringData,
+	}
+	return fake.NewSimpleClientset(secret)
+}
+
+// TestK8sSecretProvider_GetDSN_ReturnsRawJSONPayload tests that getDSN returns the raw JSON payload of the secret.
+func TestK8sSecretProvider_GetDSN_ReturnsRawJSONPayload(t *testing.T) {
+	clientset := newFakeClientsetWithSecret("default", "my-db-secret",
+		map[string][]byte{
+			"writer": []byte("postgres://writer-dsn"),
+		},
+		map[string]string{
+			"reader": "postgres://reader-dsn",
+		},
+	)
+
+	k8sProviderInstance = &k8sSecretProvider{
+		clientset: clientset,
+		namespace: "default",
+	}
+	t.Cleanup(func() { k8sProviderInstance = nil })
+
+	ref, err := url.Parse("k8ssecret://default/my-db-secret")
+	if err != nil {
+		t.Fatalf("parse ref: %v", err)
+	}
+
+	p := k8sSecretProvider{}
+	raw, err := p.getDSN(context.Background(), ref)
+	if err != nil {
+		t.Fatalf("getDSN returned error: %v", err)
+	}
+
+	var payload map[string]string
+	if err := json.Unmarshal([]byte(raw), &payload); err != nil {
+		t.Fatalf("expected raw JSON payload, got %q (unmarshal err: %v)", raw, err)
+	}
+	if payload["writer"] != "postgres://writer-dsn" {
+		t.Fatalf("writer mismatch: got %q", payload["writer"])
+	}
+	if payload["reader"] != "postgres://reader-dsn" {
+		t.Fatalf("reader mismatch: got %q", payload["reader"])
+	}
+}
+
+// TestK8sSecretProvider_GetDSN_CurrentNamespaceShorthand tests that getDSN can infer the namespace when using the
+// shorthand URL format.
+func TestK8sSecretProvider_GetDSN_CurrentNamespaceShorthand(t *testing.T) {
+	clientset := newFakeClientsetWithSecret("current-ns", "db-creds",
+		map[string][]byte{"data_source_name": []byte("postgres://short-dsn")},
+		nil,
+	)
+
+	k8sProviderInstance = &k8sSecretProvider{
+		clientset: clientset,
+		namespace: "current-ns",
+	}
+	t.Cleanup(func() { k8sProviderInstance = nil })
+
+	// Single-segment URL: k8ssecret://db-creds (namespace inferred).
+	ref, err := url.Parse("k8ssecret://db-creds")
+	if err != nil {
+		t.Fatalf("parse ref: %v", err)
+	}
+
+	p := k8sSecretProvider{}
+	raw, err := p.getDSN(context.Background(), ref)
+	if err != nil {
+		t.Fatalf("getDSN returned error: %v", err)
+	}
+
+	var payload map[string]string
+	if err := json.Unmarshal([]byte(raw), &payload); err != nil {
+		t.Fatalf("expected raw JSON payload, got %q (unmarshal err: %v)", raw, err)
+	}
+	if payload["data_source_name"] != "postgres://short-dsn" {
+		t.Fatalf("data_source_name mismatch: got %q", payload["data_source_name"])
+	}
+}
+
+// TestK8sSecretProvider_GetDSN_SecretNotFound tests that getDSN returns an error when the secret is not found.
+func TestK8sSecretProvider_GetDSN_SecretNotFound(t *testing.T) {
+	clientset := fake.NewSimpleClientset() // no secrets
+
+	k8sProviderInstance = &k8sSecretProvider{
+		clientset: clientset,
+		namespace: "default",
+	}
+	t.Cleanup(func() { k8sProviderInstance = nil })
+
+	ref, err := url.Parse("k8ssecret://default/missing-secret")
+	if err != nil {
+		t.Fatalf("parse ref: %v", err)
+	}
+
+	p := k8sSecretProvider{}
+	_, err = p.getDSN(context.Background(), ref)
+	if err == nil {
+		t.Fatal("expected error for missing secret, got nil")
+	}
+	if !strings.Contains(err.Error(), "unable to fetch secret") {
+		t.Fatalf("unexpected error message: %v", err)
+	}
+}
+
+// TestK8sSecretProvider_GetDSN_InvalidURL tests that getDSN returns an error for an invalid k8ssecret URL.
+func TestK8sSecretProvider_GetDSN_InvalidURL(t *testing.T) {
+	k8sProviderInstance = &k8sSecretProvider{
+		clientset: fake.NewSimpleClientset(),
+		namespace: "default",
+	}
+	t.Cleanup(func() { k8sProviderInstance = nil })
+
+	ref, err := url.Parse("k8ssecret://")
+	if err != nil {
+		t.Fatalf("parse ref: %v", err)
+	}
+
+	p := k8sSecretProvider{}
+	_, err = p.getDSN(context.Background(), ref)
+	if err == nil {
+		t.Fatal("expected error for invalid k8ssecret URL, got nil")
+	}
+	if !strings.Contains(err.Error(), "invalid k8ssecret URL format") {
+		t.Fatalf("unexpected error message: %v", err)
+	}
+}
+
+// TestSecretResolver_K8sSecret_MultipleJobsDifferentKeys tests that the secretResolver correctly resolves different
+// keys from the same k8ssecret across multiple jobs without caching issues.
+func TestSecretResolver_K8sSecret_MultipleJobsDifferentKeys(t *testing.T) {
+	clientset := newFakeClientsetWithSecret("default", "my-db-secret",
+		map[string][]byte{
+			"writer": []byte("postgres://writer-dsn"),
+			"reader": []byte("postgres://reader-dsn"),
+		},
+		nil,
+	)
+
+	k8sProviderInstance = &k8sSecretProvider{
+		clientset: clientset,
+		namespace: "default",
+	}
+	t.Cleanup(func() { k8sProviderInstance = nil })
+
+	origProviders := secretProviders
+	defer func() { secretProviders = origProviders }()
+	secretProviders = map[string]secretProvider{
+		"k8ssecret": k8sSecretProvider{},
+	}
+
+	r := &secretResolver{}
+	ctx := context.Background()
+
+	// Simulates job1's static_config DSN.
+	gotWriter, err := r.resolve(ctx, "k8ssecret://default/my-db-secret?key=writer")
+	if err != nil {
+		t.Fatalf("resolve writer (job1) returned error: %v", err)
+	}
+	if gotWriter != "postgres://writer-dsn" {
+		t.Fatalf("job1 (writer) DSN mismatch: got %q want %q", gotWriter, "postgres://writer-dsn")
+	}
+
+	// Simulates job2's static_config DSN, same secret, different key. Before the fix, this would incorrectly return
+	// job1's writer DSN.
+	gotReader, err := r.resolve(ctx, "k8ssecret://default/my-db-secret?key=reader")
+	if err != nil {
+		t.Fatalf("resolve reader (job2) returned error: %v", err)
+	}
+	if gotReader != "postgres://reader-dsn" {
+		t.Fatalf("job2 (reader) DSN mismatch: got %q want %q (bug: got job1's cached DSN)", gotReader, "postgres://reader-dsn")
+	}
+
+	// A third job re-referencing job1's key should still work from cache.
+	gotWriterAgain, err := r.resolve(ctx, "k8ssecret://default/my-db-secret?key=writer")
+	if err != nil {
+		t.Fatalf("resolve writer (job3) returned error: %v", err)
+	}
+	if gotWriterAgain != "postgres://writer-dsn" {
+		t.Fatalf("job3 (writer, cached) DSN mismatch: got %q want %q", gotWriterAgain, "postgres://writer-dsn")
+	}
+
+	// Confirm only one underlying k8s API call was made (shared fetch), not one per job/key.
+	getActions := 0
+	for _, action := range clientset.Actions() {
+		if action.GetVerb() == "get" && action.GetResource().Resource == "secrets" {
+			getActions++
+		}
+	}
+	if getActions != 1 {
+		t.Fatalf("expected exactly 1 k8s secret fetch (shared across jobs), got %d", getActions)
+	}
+}

config/secret_resolver.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net/url"
+	"strings"
 	"sync"
 
 	"golang.org/x/sync/singleflight"
@@ -79,17 +80,24 @@ func (r *secretResolver) resolve(ctx context.Context, value string) (string, err
 
 // extractKey pulls the appropriate key from a raw secret value (JSON or plain string).
 func extractKey(raw string, u *url.URL, originalValue string) (string, error) {
+	var resolved string
 	var payload map[string]string
 	if jsonErr := json.Unmarshal([]byte(raw), &payload); jsonErr == nil {
 		key := u.Query().Get("key")
 		if key == "" {
 			key = "data_source_name"
 		}
-		val, ok := payload[key]
+		fieldValue, ok := payload[key]
 		if !ok {
 			return "", fmt.Errorf("key %q not found in secret %q", key, originalValue)
 		}
-		return val, nil
+		resolved = fieldValue
+	} else {
+		resolved = raw
 	}
-	return raw, nil
+
+	if tmpl := u.Query().Get("template"); tmpl != "" {
+		return strings.ReplaceAll(tmpl, "DSN_VALUE", resolved), nil
+	}
+	return resolved, nil
 }

config/secret_resolver_test.go

@@ -0,0 +1,61 @@
+package config
+
+import (
+	"context"
+	"encoding/json"
+	"net/url"
+	"testing"
+)
+
+// fakeSecretProvider is a mock implementation of secretProvider for testing purposes.
+type fakeSecretProvider struct {
+	payload map[string]string
+	calls   int
+}
+
+// getDSN simulates fetching a DSN from a secret provider. It returns the JSON-encoded payload and increments the call count.
+func (p *fakeSecretProvider) getDSN(_ context.Context, _ *url.URL) (string, error) {
+	p.calls++
+	b, err := json.Marshal(p.payload)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
+// TestSecretResolver_SharedFetch_DifferentKeys tests that the secretResolver fetches the secret only once when resolving different keys from the same secret.
+func TestSecretResolver_SharedFetch_DifferentKeys(t *testing.T) {
+	orig := secretProviders
+	defer func() { secretProviders = orig }()
+
+	fp := &fakeSecretProvider{
+		payload: map[string]string{
+			"writer": "postgres://writer-dsn",
+			"reader": "postgres://reader-dsn",
+		},
+	}
+	secretProviders = map[string]secretProvider{"k8ssecret": fp}
+
+	r := &secretResolver{}
+	ctx := context.Background()
+
+	gotWriter, err := r.resolve(ctx, "k8ssecret://default/my-db-secret?key=writer")
+	if err != nil {
+		t.Fatalf("resolve writer: %v", err)
+	}
+	if gotWriter != "postgres://writer-dsn" {
+		t.Fatalf("writer mismatch: got %q", gotWriter)
+	}
+
+	gotReader, err := r.resolve(ctx, "k8ssecret://default/my-db-secret?key=reader")
+	if err != nil {
+		t.Fatalf("resolve reader: %v", err)
+	}
+	if gotReader != "postgres://reader-dsn" {
+		t.Fatalf("reader mismatch: got %q", gotReader)
+	}
+
+	if fp.calls != 1 {
+		t.Fatalf("expected shared fetch (1 call), got %d", fp.calls)
+	}
+}