fix(operator): Add validation on empty per-tenant limits config

operator/CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Main
 
+- [12199](https://github.com/grafana/loki/pull/12199) **btaani**: Add validation on empty per-tenant limits config
 - [12333](https://github.com/grafana/loki/pull/12333) **periklis**: Bump max OpenShift version to next release
 
 ## 0.6.0 (2024-03-19)

operator/apis/loki/v1/lokistack_types.go

@@ -1109,6 +1109,8 @@ const (
 	ReasonFailedCertificateRotation LokiStackConditionReason = "FailedCertificateRotation"
 	// ReasonQueryTimeoutInvalid when the QueryTimeout can not be parsed.
 	ReasonQueryTimeoutInvalid LokiStackConditionReason = "ReasonQueryTimeoutInvalid"
+	// ReasonInvalidPerTenantLimitsConfig when the per-tenant limits configuration is invalid.
+	ReasonInvalidPerTenantLimitsConfig LokiStackConditionReason = "ReasonInvalidPerTenantLimitsConfig"
 	// ReasonZoneAwareNodesMissing when the cluster does not contain any nodes with the labels needed for zone-awareness.
 	ReasonZoneAwareNodesMissing LokiStackConditionReason = "ReasonZoneAwareNodesMissing"
 	// ReasonZoneAwareEmptyLabel when the node-label used for zone-awareness has an empty value.

operator/docs/operator/api.md

@@ -1769,6 +1769,9 @@ PodStatusMap
 </tr><tr><td><p>&#34;InvalidObjectStorageSecret&#34;</p></td>
 <td><p>ReasonInvalidObjectStorageSecret when the format of the secret is invalid.</p>
 </td>
+</tr><tr><td><p>&#34;ReasonInvalidPerTenantLimitsConfig&#34;</p></td>
+<td><p>ReasonInvalidPerTenantLimitsConfig when the per-tenant limits configuration is invalid.</p>
+</td>
 </tr><tr><td><p>&#34;InvalidReplicationConfiguration&#34;</p></td>
 <td><p>ReasonInvalidReplicationConfiguration when the configurated replication factor is not valid
 with the select cluster size.</p>

operator/internal/handlers/lokistack_create_or_update.go

@@ -2,6 +2,7 @@ package handlers
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 
@@ -82,6 +83,18 @@ func CreateOrUpdateLokiStack(
 		certRotationRequiredAt = stack.Annotations[manifests.AnnotationCertRotationRequiredAt]
 	}
 
+	err = manifests.ValidatePerTenantConfig(stack.Spec.Limits)
+	if err != nil {
+		if errors.Is(err, manifests.ErrPerTenantConfigInvalid) {
+			ll.Error(err, "invalid per-tenant config")
+			return "", &status.DegradedError{
+				Message: fmt.Sprintf("Invalid per-tenant limits config: %s", err),
+				Reason:  lokiv1.ReasonInvalidPerTenantLimitsConfig,
+				Requeue: false,
+			}
+		}
+	}
+
 	timeoutConfig, err := manifests.NewTimeoutConfig(stack.Spec.Limits)
 	if err != nil {
 		ll.Error(err, "failed to parse query timeout")

operator/internal/handlers/lokistack_create_or_update_test.go

@@ -740,3 +740,73 @@ func TestCreateOrUpdateLokiStack_WhenInvalidQueryTimeout_SetDegraded(t *testing.
 	require.Error(t, err)
 	require.Equal(t, degradedErr, err)
 }
+
+func TestCreateOrUpdateLokiStack_WhenInvalidPerTenantLimits_SetDegraded(t *testing.T) {
+	sw := &k8sfakes.FakeStatusWriter{}
+	k := &k8sfakes.FakeClient{}
+	r := ctrl.Request{
+		NamespacedName: types.NamespacedName{
+			Name:      "my-stack",
+			Namespace: "some-ns",
+		},
+	}
+
+	degradedErr := &status.DegradedError{
+		Message: "Invalid per-tenant limits config: per-tenant config cannot contain empty keys: invalid",
+		Reason:  lokiv1.ReasonInvalidPerTenantLimitsConfig,
+		Requeue: false,
+	}
+
+	stack := lokiv1.LokiStack{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "LokiStack",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-stack",
+			Namespace: "some-ns",
+			UID:       "b23f9a38-9672-499f-8c29-15ede74d3ece",
+		},
+		Spec: lokiv1.LokiStackSpec{
+			Size: lokiv1.SizeOneXExtraSmall,
+			Storage: lokiv1.ObjectStorageSpec{
+				Schemas: []lokiv1.ObjectStorageSchema{
+					{
+						Version:       lokiv1.ObjectStorageSchemaV11,
+						EffectiveDate: "2020-10-11",
+					},
+				},
+				Secret: lokiv1.ObjectStorageSecretSpec{
+					Name: defaultSecret.Name,
+					Type: lokiv1.ObjectStorageSecretS3,
+				},
+			},
+			Tenants: &lokiv1.TenantsSpec{
+				Mode: "openshift",
+			},
+			Limits: &lokiv1.LimitsSpec{
+				Tenants: map[string]lokiv1.PerTenantLimitsTemplateSpec{
+					"invalid": {},
+				},
+			},
+		},
+	}
+
+	// Create looks up the CR first, so we need to return our fake stack
+	k.GetStub = func(_ context.Context, name types.NamespacedName, object client.Object, _ ...client.GetOption) error {
+		_, isLokiStack := object.(*lokiv1.LokiStack)
+		if r.Name == name.Name && r.Namespace == name.Namespace && isLokiStack {
+			k.SetClientObject(object, &stack)
+		}
+		if defaultSecret.Name == name.Name {
+			k.SetClientObject(object, &defaultSecret)
+		}
+		return nil
+	}
+
+	k.StatusStub = func() client.StatusWriter { return sw }
+
+	_, err := CreateOrUpdateLokiStack(context.TODO(), logger, r, k, scheme, featureGates)
+
+	require.Error(t, err)
+	require.Equal(t, degradedErr, err)
+}

operator/internal/manifests/options.go

@@ -1,6 +1,8 @@
 package manifests
 
 import (
+	"errors"
+	"fmt"
 	"strings"
 	"time"
 
@@ -126,6 +128,8 @@ type TLSProfileSpec struct {
 	MinTLSVersion string
 }
 
+var ErrPerTenantConfigInvalid error = errors.New("per-tenant config cannot contain empty keys")
+
 // TLSCipherSuites transforms TLSProfileSpec.Ciphers from a slice
 // to a string of elements joined with a comma.
 func (o Options) TLSCipherSuites() string {
@@ -195,3 +199,15 @@ func calculateHTTPTimeouts(queryTimeout time.Duration) TimeoutConfig {
 		},
 	}
 }
+
+func ValidatePerTenantConfig(s *lokiv1.LimitsSpec) error {
+	if s != nil && s.Tenants != nil {
+		for key, config := range s.Tenants {
+			c := lokiv1.PerTenantLimitsTemplateSpec{}
+			if config == c {
+				return fmt.Errorf("%w: %s", ErrPerTenantConfigInvalid, key)
+			}
+		}
+	}
+	return nil
+}