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
 - [12165](https://github.com/grafana/loki/pull/12165) **JoaoBraveCoding**: Change attribute value used for CCO-based credential mode
 - [12157](https://github.com/grafana/loki/pull/12157) **periklis**: Fix managed auth features annotation for community-openshift bundle
 - [12104](https://github.com/grafana/loki/pull/12104) **periklis**: Upgrade build and runtime dependencies

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 empty.
+	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 empty.</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"
 
@@ -85,11 +86,20 @@ func CreateOrUpdateLokiStack(
 
 	timeoutConfig, err := manifests.NewTimeoutConfig(stack.Spec.Limits)
 	if err != nil {
-		ll.Error(err, "failed to parse query timeout")
-		return "", &status.DegradedError{
-			Message: fmt.Sprintf("Error parsing query timeout: %s", err),
-			Reason:  lokiv1.ReasonQueryTimeoutInvalid,
-			Requeue: false,
+		if errors.Is(err, manifests.ErrInvalidPerTenantConfig) {
+			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,
+			}
+		} else {
+			ll.Error(err, "failed to parse query timeout")
+			return "", &status.DegradedError{
+				Message: fmt.Sprintf("Error parsing query timeout: %s", err),
+				Reason:  lokiv1.ReasonQueryTimeoutInvalid,
+				Requeue: false,
+			}
 		}
 	}
 

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: invalid per-tenant config",
+		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,7 @@
 package manifests
 
 import (
+	"errors"
 	"strings"
 	"time"
 
@@ -126,6 +127,8 @@ type TLSProfileSpec struct {
 	MinTLSVersion string
 }
 
+var ErrInvalidPerTenantConfig error = errors.New("invalid per-tenant config")
+
 // TLSCipherSuites transforms TLSProfileSpec.Ciphers from a slice
 // to a string of elements joined with a comma.
 func (o Options) TLSCipherSuites() string {
@@ -142,6 +145,13 @@ func NewTimeoutConfig(s *lokiv1.LimitsSpec) (TimeoutConfig, error) {
 		return defaultTimeoutConfig, nil
 	}
 
+	for _, config := range s.Tenants {
+		c := lokiv1.PerTenantLimitsTemplateSpec{}
+		if config == c {
+			return TimeoutConfig{}, ErrInvalidPerTenantConfig
+		}
+	}
+
 	queryTimeout := lokiDefaultQueryTimeout
 	if s.Global != nil && s.Global.QueryLimits != nil && s.Global.QueryLimits.QueryTimeout != "" {
 		var err error