Repository: kedacore/keda

Issue:
Keda ignoring prefixes when resolving envvars
### Report

Prefixes on `envFrom` elements in a deployment spec aren't being interpreted by the Keda env resolver. For example in our deployment spec we have this:

```
      envFrom:
        - prefix: KAFKA_
          secretRef:
            name: kafka-cluster-details
            optional: false
```

Where the `kafka-cluster-details` secret has a key value pair for `BOOTSTRAP_SERVERS`.

In our pods this means an envvar `KAFKA_BOOTSTRAP_SERVERS`, but if I set `bootstrapServersFromEnv` to `KAFKA_BOOTSTRAP_SERVERS` in my Scaler config it doesn't work. If I set it to `BOOTSTRAP_SERVERS` it works as expected.

The current setup is not only confusing, but it is also likely to cause issues if secrets/config maps have overlapping keys.

### Expected Behavior

The env resolver should correctly interpret envFrom prefixes. 

In this example setting `bootstrapServersFromEnv` to `KAFKA_BOOTSTRAP_SERVERS` should correctly configure the scaler.

If the specified envvar cannot be found in the deployment configuration, then the scaler should emit a more specific error, rather than interpreting it as an empty string.


### Actual Behavior

The scaler interprets the value of the KAFKA_BOOTSTRAP_SERVERS env var as an empty string, and doesn't give a useful error to suggest the envvar isn't actually set at all. To make it work we need to set `bootstrapServersFromEnv` to `BOOTSTRAP_SERVERS`

### Steps to Reproduce the Problem

1. Create a deployment which uses envFrom with a prefix
2. Create a scaled object that uses any *fromEnv value to use on of the prefixed envvars
3. Note that the envar isn't correctly interpreted


### Logs from KEDA operator




### KEDA Version

< 2.15.0

### Kubernetes Version

1.32

### Platform

Amazon Web Services

### Scaler Details

Kafka, but checking the code this impacts all scalers

### Anything else?

_No response_

Hints from issue comments:

Solution patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d440cba35..05f6a29c9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -75,6 +75,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 ### Fixes
 
 - **General**: Admission Webhook blocks ScaledObject without metricType with fallback ([#6696](https://github.com/kedacore/keda/issues/6696))
+- **General**: Fix prefixes on envFrom elements in a deployment spec aren't being interpreted and Environment variables are not prefixed with the prefix ([#6728](https://github.com/kedacore/keda/issues/6728))
 - **General**: ScalerCache gets the lock before operate the scalers to prevent panics ([#6739](https://github.com/kedacore/keda/pull/6739))
 - **AWS SQS Queue Scaler**: Fix AWS SQS Queue queueURLFromEnv not working ([#6712](https://github.com/kedacore/keda/issues/6712))
 - **Azure Service Bus scaler**: Fix Azure Service Bus scaler add default Operation ([#6730](https://github.com/kedacore/keda/issues/6730))
diff --git a/pkg/scaling/resolver/scale_resolvers.go b/pkg/scaling/resolver/scale_resolvers.go
index 15f564f6e..6e772d951 100644
--- a/pkg/scaling/resolver/scale_resolvers.go
+++ b/pkg/scaling/resolver/scale_resolvers.go
@@ -392,12 +392,15 @@ func resolveEnv(ctx context.Context, client client.Client, logger logr.Logger, c
 	accessSecrets := readSecrets(secretAccessRestricted, namespace)
 	if container.EnvFrom != nil {
 		for _, source := range container.EnvFrom {
+			// prefix is used to prefix environment variables, prefix is empty string if not set
+			// if prefix is set, all environment variables will be prefixed with the prefix
+			envPrefix := source.Prefix
 			if source.ConfigMapRef != nil {
 				configMap, err := resolveConfigMap(ctx, client, source.ConfigMapRef, namespace)
 				switch {
 				case err == nil:
 					for k, v := range configMap {
-						resolved[k] = v
+						resolved[envPrefix+k] = v
 					}
 				case source.ConfigMapRef.Optional != nil && *source.ConfigMapRef.Optional:
 					// ignore error when ConfigMap is marked as optional
@@ -410,7 +413,7 @@ func resolveEnv(ctx context.Context, client client.Client, logger logr.Logger, c
 				switch {
 				case err == nil:
 					for k, v := range secretsMap {
-						resolved[k] = v
+						resolved[envPrefix+k] = v
 					}
 				case source.SecretRef.Optional != nil && *source.SecretRef.Optional:
 					// ignore error when Secret is marked as optional



Test patch:
diff --git a/pkg/scaling/resolver/scale_resolvers_test.go b/pkg/scaling/resolver/scale_resolvers_test.go
index a642b53cc..82db4bef7 100644
--- a/pkg/scaling/resolver/scale_resolvers_test.go
+++ b/pkg/scaling/resolver/scale_resolvers_test.go
@@ -917,6 +917,7 @@ func TestEnvWithRestrictSecretAccess(t *testing.T) {
 }
 
 func TestEnvWithRestrictedNamespace(t *testing.T) {
+	envPrefix := "PREFIX_"
 	tests := []struct {
 		name      string
 		expected  map[string]string
@@ -952,6 +953,20 @@ func TestEnvWithRestrictedNamespace(t *testing.T) {
 			},
 			expected: map[string]string{secretKey: secretData},
 		},
+		{
+			name: "env reference secret key with prefix",
+			container: &corev1.Container{
+				EnvFrom: []corev1.EnvFromSource{{
+					SecretRef: &corev1.SecretEnvSource{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: secretName,
+						},
+					},
+					Prefix: envPrefix,
+				}},
+			},
+			expected: map[string]string{envPrefix + secretKey: secretData},
+		},
 	}
 	secret := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{