feat: Add support for ACL from PVC (#1582)

Add possibility to use ACL files mounted from PVC, not k8s secret.
Implement validation webhook to ensure only one ACL storage type is used.
Update E2E tests with the example that uses ACL from prepopulated PVC.

Signed-off-by: Damian Seredyn <[email protected]>
Co-authored-by: Damian Seredyn <[email protected]>

Author: naimadswdn

api/common/v1beta2/common_types.go

@@ -1,6 +1,8 @@
 package v1beta2
 
 import (
+	"fmt"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -272,5 +274,27 @@ type InitContainer struct {
 
 // +k8s:deepcopy-gen=true
 type ACLConfig struct {
+	// Secret-based ACL configuration.
+	// Adapts a Secret into a volume containing ACL rules.
+	// The contents of the target Secret's Data field will be presented in a volume
+	// as files using the keys in the Data field as the file names.
+	// Secret volumes support ownership management and SELinux relabeling.
 	Secret *corev1.SecretVolumeSource `json:"secret,omitempty"`
+	// PersistentVolumeClaim-based ACL configuration
+	// Specify the PVC name to mount ACL file from persistent storage
+	// The operator will automatically mount /etc/redis/user.acl from the PVC
+	PersistentVolumeClaim *string `json:"persistentVolumeClaim,omitempty"`
+}
+
+// Validate checks that only one ACL source is specified
+func (a *ACLConfig) Validate() error {
+	if a == nil {
+		return nil
+	}
+
+	if a.Secret != nil && a.PersistentVolumeClaim != nil {
+		return fmt.Errorf("only one of 'secret' or 'persistentVolumeClaim' can be specified in ACL configuration")
+	}
+
+	return nil
 }

api/common/v1beta2/common_types_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/utils/ptr"
 )
 
@@ -84,3 +85,150 @@ func TestKubernetesConfig_ShouldCreateAdditionalService(t *testing.T) {
 		})
 	}
 }
+
+func TestACLConfig_PersistentVolumeClaim(t *testing.T) {
+	tests := []struct {
+		name                     string
+		aclConfig                *ACLConfig
+		expectedPVCName          *string
+		expectedSecretConfigured bool
+	}{
+		{
+			name:                     "nil ACL config",
+			aclConfig:                nil,
+			expectedPVCName:          nil,
+			expectedSecretConfigured: false,
+		},
+		{
+			name:                     "empty ACL config",
+			aclConfig:                &ACLConfig{},
+			expectedPVCName:          nil,
+			expectedSecretConfigured: false,
+		},
+		{
+			name: "ACL with secret only",
+			aclConfig: &ACLConfig{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: "redis-acl-secret",
+				},
+			},
+			expectedPVCName:          nil,
+			expectedSecretConfigured: true,
+		},
+		{
+			name: "ACL with PVC only",
+			aclConfig: &ACLConfig{
+				PersistentVolumeClaim: ptr.To("redis-acl-pvc"),
+			},
+			expectedPVCName:          ptr.To("redis-acl-pvc"),
+			expectedSecretConfigured: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.aclConfig == nil {
+				assert.Nil(t, tt.aclConfig)
+				return
+			}
+
+			// Verify PVC configuration
+			if tt.expectedPVCName != nil {
+				assert.NotNil(t, tt.aclConfig.PersistentVolumeClaim)
+				assert.Equal(t, *tt.expectedPVCName, *tt.aclConfig.PersistentVolumeClaim)
+			} else {
+				assert.Nil(t, tt.aclConfig.PersistentVolumeClaim)
+			}
+
+			// Verify Secret configuration
+			if tt.expectedSecretConfigured {
+				assert.NotNil(t, tt.aclConfig.Secret)
+			} else {
+				assert.Nil(t, tt.aclConfig.Secret)
+			}
+		})
+	}
+}
+
+func TestACLConfig_BothSourcesConfigured(t *testing.T) {
+	// Test scenario where both Secret and PVC are configured
+	// Validation should fail
+	aclConfig := &ACLConfig{
+		Secret: &corev1.SecretVolumeSource{
+			SecretName: "redis-acl-secret",
+		},
+		PersistentVolumeClaim: ptr.To("redis-acl-pvc"),
+	}
+
+	// Both should be settable (struct level)
+	assert.NotNil(t, aclConfig.Secret)
+	assert.NotNil(t, aclConfig.PersistentVolumeClaim)
+	assert.Equal(t, "redis-acl-secret", aclConfig.Secret.SecretName)
+	assert.Equal(t, "redis-acl-pvc", *aclConfig.PersistentVolumeClaim)
+
+	// But validation should fail
+	err := aclConfig.Validate()
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "only one of 'secret' or 'persistentVolumeClaim' can be specified")
+}
+
+func TestACLConfig_Validate(t *testing.T) {
+	tests := []struct {
+		name      string
+		aclConfig *ACLConfig
+		wantErr   bool
+		errMsg    string
+	}{
+		{
+			name:      "nil config is valid",
+			aclConfig: nil,
+			wantErr:   false,
+		},
+		{
+			name:      "empty config is valid",
+			aclConfig: &ACLConfig{},
+			wantErr:   false,
+		},
+		{
+			name: "only secret is valid",
+			aclConfig: &ACLConfig{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: "redis-acl-secret",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "only PVC is valid",
+			aclConfig: &ACLConfig{
+				PersistentVolumeClaim: ptr.To("redis-acl-pvc"),
+			},
+			wantErr: false,
+		},
+		{
+			name: "both secret and PVC is invalid",
+			aclConfig: &ACLConfig{
+				Secret: &corev1.SecretVolumeSource{
+					SecretName: "redis-acl-secret",
+				},
+				PersistentVolumeClaim: ptr.To("redis-acl-pvc"),
+			},
+			wantErr: true,
+			errMsg:  "only one of 'secret' or 'persistentVolumeClaim' can be specified",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.aclConfig.Validate()
+			if tt.wantErr {
+				assert.Error(t, err)
+				if tt.errMsg != "" {
+					assert.Contains(t, err.Error(), tt.errMsg)
+				}
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

api/common/v1beta2/zz_generated.deepcopy.go

@@ -17,13 +17,80 @@ limitations under the License.
 package v1beta2
 
 import (
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	ctrl "sigs.k8s.io/controller-runtime"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 )
 
+const (
+	webhookPath = "/validate-redis-redis-opstreelabs-in-v1beta2-redis"
+)
+
+// log is for logging in this package.
+var redislog = logf.Log.WithName("redis-v1beta2-validation")
+
+// +kubebuilder:webhook:path=/validate-redis-redis-opstreelabs-in-v1beta2-redis,mutating=false,failurePolicy=fail,sideEffects=None,groups=redis.redis.opstreelabs.in,resources=redis,verbs=create;update,versions=v1beta2,name=validate-redis.redis.opstreelabs.in,admissionReviewVersions=v1
+
 func (r *Redis) SetupWebhookWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewWebhookManagedBy(mgr).
 		For(r).
 		Complete()
 }
 
-// TODO(user): EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
+var _ webhook.Validator = &Redis{}
+
+// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
+func (r *Redis) ValidateCreate() (admission.Warnings, error) {
+	redislog.Info("validate create", "name", r.Name)
+
+	return r.validate(nil)
+}
+
+// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
+func (r *Redis) ValidateUpdate(old runtime.Object) (admission.Warnings, error) {
+	redislog.Info("validate update", "name", r.Name)
+
+	return r.validate(old.(*Redis))
+}
+
+// ValidateDelete implements webhook.Validator so a webhook will be registered for the type
+func (r *Redis) ValidateDelete() (admission.Warnings, error) {
+	redislog.Info("validate delete", "name", r.Name)
+
+	return nil, nil
+}
+
+// validate validates the Redis CR
+func (r *Redis) validate(_ *Redis) (admission.Warnings, error) {
+	var errors field.ErrorList
+
+	// Validate ACL configuration
+	if r.Spec.ACL != nil {
+		if err := r.Spec.ACL.Validate(); err != nil {
+			errors = append(errors, field.Invalid(
+				field.NewPath("spec").Child("acl"),
+				r.Spec.ACL,
+				err.Error(),
+			))
+		}
+	}
+
+	if len(errors) == 0 {
+		return nil, nil
+	}
+
+	return nil, apierrors.NewInvalid(
+		schema.GroupKind{Group: "redis.redis.opstreelabs.in", Kind: "Redis"},
+		r.Name,
+		errors,
+	)
+}
+
+func (r *Redis) WebhookPath() string {
+	return webhookPath
+}

api/redis/v1beta2/redis_webhook.go

@@ -0,0 +1,68 @@
+package v1beta2_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	common "github.com/OT-CONTAINER-KIT/redis-operator/api/common/v1beta2"
+	v1beta2 "github.com/OT-CONTAINER-KIT/redis-operator/api/redis/v1beta2"
+	"github.com/OT-CONTAINER-KIT/redis-operator/internal/testutil/webhook"
+	"github.com/stretchr/testify/require"
+	admissionv1beta1 "k8s.io/api/admission/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
+)
+
+func TestRedisWebhook(t *testing.T) {
+	cases := []webhook.ValidationWebhookTestCase{
+		{
+			Name:      "failed-create-v1beta2-redis-acl-both-sources",
+			Operation: admissionv1beta1.Create,
+			Object: func(t *testing.T, uid string) []byte {
+				t.Helper()
+				redis := mkRedis(uid)
+				redis.Spec.ACL = mkACLWithBothSources()
+				return marshal(t, redis)
+			},
+			Check: webhook.ValidationWebhookFailed("only one of 'secret' or 'persistentVolumeClaim' can be specified"),
+		},
+	}
+
+	gvk := metav1.GroupVersionKind{
+		Group:   "redis.redis.opstreelabs.in",
+		Version: "v1beta2",
+		Kind:    "Redis",
+	}
+
+	redis := &v1beta2.Redis{}
+	webhook.RunValidationWebhookTests(t, gvk, redis, cases...)
+}
+
+func mkRedis(uid string) *v1beta2.Redis {
+	return &v1beta2.Redis{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("test-%s", uid),
+			UID:  types.UID(fmt.Sprintf("test-%s", uid)),
+		},
+		Spec: v1beta2.RedisSpec{},
+	}
+}
+
+func marshal(t *testing.T, obj interface{}) []byte {
+	t.Helper()
+	bytes, err := json.Marshal(obj)
+	require.NoError(t, err)
+	return bytes
+}
+
+func mkACLWithBothSources() *common.ACLConfig {
+	return &common.ACLConfig{
+		Secret: &corev1.SecretVolumeSource{
+			SecretName: "test-secret",
+		},
+		PersistentVolumeClaim: ptr.To("test-pvc"),
+	}
+}

api/redis/v1beta2/redis_webhook_test.go

@@ -84,6 +84,17 @@ func (r *RedisCluster) validate(_ *RedisCluster) (admission.Warnings, error) {
 		))
 	}
 
+	// Validate ACL configuration
+	if r.Spec.ACL != nil {
+		if err := r.Spec.ACL.Validate(); err != nil {
+			errors = append(errors, field.Invalid(
+				field.NewPath("spec").Child("acl"),
+				r.Spec.ACL,
+				err.Error(),
+			))
+		}
+	}
+
 	if len(errors) == 0 {
 		return nil, nil
 	}