operator: Add support for custom S3 CA (#6198)

operator/CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Main
 
+- [6198](https://github.com/grafana/loki/pull/6198) **periklis**: Add support for custom S3 CA
 - [6199](https://github.com/grafana/loki/pull/6199) **Red-GV**: Update GCP secret volume path
 - [6125](https://github.com/grafana/loki/pull/6125) **sasagarw**: Add method to get authenticated from GCP
 - [5986](https://github.com/grafana/loki/pull/5986) **periklis**: Add support for Loki Rules reconciliation

operator/api/v1beta1/lokistack_types.go

@@ -350,15 +350,33 @@ type ObjectStorageSecretSpec struct {
 	Name string `json:"name"`
 }
 
+// ObjectStorageTLSSpec is the TLS configuration for reaching the object storage endpoint.
+type ObjectStorageTLSSpec struct {
+	// CA is the name of a ConfigMap containing a CA certificate.
+	// It needs to be in the same namespace as the LokiStack custom resource.
+	//
+	// +optional
+	// +kubebuilder:validation:optional
+	// +operator-sdk:csv:customresourcedefinitions:type=spec,xDescriptors="urn:alm:descriptor:io.kubernetes:ConfigMap",displayName="CA ConfigMap Name"
+	CA string `json:"caName,omitempty"`
+}
+
 // ObjectStorageSpec defines the requirements to access the object
 // storage bucket to persist logs by the ingester component.
 type ObjectStorageSpec struct {
 	// Secret for object storage authentication.
-	// Name of a secret in the same namespace as the cluster logging operator.
+	// Name of a secret in the same namespace as the LokiStack custom resource.
 	//
 	// +required
 	// +kubebuilder:validation:Required
 	Secret ObjectStorageSecretSpec `json:"secret"`
+
+	// TLS configuration for reaching the object storage endpoint.
+	//
+	// +optional
+	// +kubebuilder:validation:Optional
+	// +operator-sdk:csv:customresourcedefinitions:type=spec,displayName="TLS Config"
+	TLS *ObjectStorageTLSSpec `json:"tls,omitempty"`
 }
 
 // QueryLimitSpec defines the limits applies at the query path.
@@ -612,6 +630,11 @@ const (
 	ReasonMissingObjectStorageSecret LokiStackConditionReason = "MissingObjectStorageSecret"
 	// ReasonInvalidObjectStorageSecret when the format of the secret is invalid.
 	ReasonInvalidObjectStorageSecret LokiStackConditionReason = "InvalidObjectStorageSecret"
+	// ReasonMissingObjectStorageCAConfigMap when the required configmap to verify object storage
+	// certificates is missing.
+	ReasonMissingObjectStorageCAConfigMap LokiStackConditionReason = "MissingObjectStorageCAConfigMap"
+	// ReasonInvalidObjectStorageCAConfigMap when the format of the CA configmap is invalid.
+	ReasonInvalidObjectStorageCAConfigMap LokiStackConditionReason = "InvalidObjectStorageCAConfigMap"
 	// ReasonInvalidReplicationConfiguration when the configurated replication factor is not valid
 	// with the select cluster size.
 	ReasonInvalidReplicationConfiguration LokiStackConditionReason = "InvalidReplicationConfiguration"

operator/api/v1beta1/zz_generated.deepcopy.go

@@ -520,7 +520,7 @@ func (in *LokiStackList) DeepCopyObject() runtime.Object {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *LokiStackSpec) DeepCopyInto(out *LokiStackSpec) {
 	*out = *in
-	out.Storage = in.Storage
+	in.Storage.DeepCopyInto(&out.Storage)
 	if in.Rules != nil {
 		in, out := &in.Rules, &out.Rules
 		*out = new(RulesSpec)
@@ -685,6 +685,11 @@ func (in *ObjectStorageSecretSpec) DeepCopy() *ObjectStorageSecretSpec {
 func (in *ObjectStorageSpec) DeepCopyInto(out *ObjectStorageSpec) {
 	*out = *in
 	out.Secret = in.Secret
+	if in.TLS != nil {
+		in, out := &in.TLS, &out.TLS
+		*out = new(ObjectStorageTLSSpec)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ObjectStorageSpec.
@@ -697,6 +702,21 @@ func (in *ObjectStorageSpec) DeepCopy() *ObjectStorageSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ObjectStorageTLSSpec) DeepCopyInto(out *ObjectStorageTLSSpec) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ObjectStorageTLSSpec.
+func (in *ObjectStorageTLSSpec) DeepCopy() *ObjectStorageTLSSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(ObjectStorageTLSSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in PodStatusMap) DeepCopyInto(out *PodStatusMap) {
 	{

operator/bundle/manifests/loki-operator.clusterserviceversion.yaml

@@ -394,6 +394,15 @@ spec:
         - urn:alm:descriptor:com.tectonic.ui:select:gcs
         - urn:alm:descriptor:com.tectonic.ui:select:s3
         - urn:alm:descriptor:com.tectonic.ui:select:swift
+      - description: TLS configuration for reaching the object storage endpoint.
+        displayName: TLS Config
+        path: storage.tls
+      - description: CA is the name of a ConfigMap containing a CA certificate. It
+          needs to be in the same namespace as the LokiStack custom resource.
+        displayName: CA ConfigMap Name
+        path: storage.tls.caName
+        x-descriptors:
+        - urn:alm:descriptor:io.kubernetes:ConfigMap
       - description: Storage class name defines the storage class for ingester/querier
           PVCs.
         displayName: Storage Class Name

operator/bundle/manifests/loki.grafana.com_lokistacks.yaml

@@ -316,7 +316,7 @@ spec:
                 properties:
                   secret:
                     description: Secret for object storage authentication. Name of
-                      a secret in the same namespace as the cluster logging operator.
+                      a secret in the same namespace as the LokiStack custom resource.
                     properties:
                       name:
                         description: Name of a secret in the namespace configured
@@ -334,6 +334,16 @@ spec:
                     - name
                     - type
                     type: object
+                  tls:
+                    description: TLS configuration for reaching the object storage
+                      endpoint.
+                    properties:
+                      caName:
+                        description: CA is the name of a ConfigMap containing a CA
+                          certificate. It needs to be in the same namespace as the
+                          LokiStack custom resource.
+                        type: string
+                    type: object
                 required:
                 - secret
                 type: object

operator/config/crd/bases/loki.grafana.com_lokistacks.yaml

@@ -311,7 +311,7 @@ spec:
                 properties:
                   secret:
                     description: Secret for object storage authentication. Name of
-                      a secret in the same namespace as the cluster logging operator.
+                      a secret in the same namespace as the LokiStack custom resource.
                     properties:
                       name:
                         description: Name of a secret in the namespace configured
@@ -329,6 +329,16 @@ spec:
                     - name
                     - type
                     type: object
+                  tls:
+                    description: TLS configuration for reaching the object storage
+                      endpoint.
+                    properties:
+                      caName:
+                        description: CA is the name of a ConfigMap containing a CA
+                          certificate. It needs to be in the same namespace as the
+                          LokiStack custom resource.
+                        type: string
+                    type: object
                 required:
                 - secret
                 type: object

operator/config/manifests/bases/loki-operator.clusterserviceversion.yaml

@@ -308,6 +308,15 @@ spec:
         - urn:alm:descriptor:com.tectonic.ui:select:gcs
         - urn:alm:descriptor:com.tectonic.ui:select:s3
         - urn:alm:descriptor:com.tectonic.ui:select:swift
+      - description: TLS configuration for reaching the object storage endpoint.
+        displayName: TLS Config
+        path: storage.tls
+      - description: CA is the name of a ConfigMap containing a CA certificate. It
+          needs to be in the same namespace as the LokiStack custom resource.
+        displayName: CA ConfigMap Name
+        path: storage.tls.caName
+        x-descriptors:
+        - urn:alm:descriptor:io.kubernetes:ConfigMap
       - description: Storage class name defines the storage class for ingester/querier
           PVCs.
         displayName: Storage Class Name

operator/internal/handlers/internal/gateway/tenant_secrets.go

@@ -8,7 +8,6 @@ import (
 
 	lokiv1beta1 "github.com/grafana/loki/operator/api/v1beta1"
 	"github.com/grafana/loki/operator/internal/external/k8s"
-	"github.com/grafana/loki/operator/internal/handlers/internal/secrets"
 	"github.com/grafana/loki/operator/internal/manifests"
 	"github.com/grafana/loki/operator/internal/status"
 
@@ -48,7 +47,7 @@ func GetTenantSecrets(
 		}
 
 		var ts *manifests.TenantSecrets
-		ts, err := secrets.ExtractGatewaySecret(&gatewaySecret, tenant.TenantName)
+		ts, err := extractSecret(&gatewaySecret, tenant.TenantName)
 		if err != nil {
 			return nil, &status.DegradedError{
 				Message: "Invalid gateway tenant secret contents",
@@ -61,3 +60,27 @@ func GetTenantSecrets(
 
 	return tenantSecrets, nil
 }
+
+// extractSecret reads a k8s secret into a manifest tenant secret struct if valid.
+func extractSecret(s *corev1.Secret, tenantName string) (*manifests.TenantSecrets, error) {
+	// Extract and validate mandatory fields
+	clientID := s.Data["clientID"]
+	if len(clientID) == 0 {
+		return nil, kverrors.New("missing clientID field", "field", "clientID")
+	}
+	clientSecret := s.Data["clientSecret"]
+	if len(clientSecret) == 0 {
+		return nil, kverrors.New("missing clientSecret field", "field", "clientSecret")
+	}
+	issuerCAPath := s.Data["issuerCAPath"]
+	if len(issuerCAPath) == 0 {
+		return nil, kverrors.New("missing issuerCAPath field", "field", "issuerCAPath")
+	}
+
+	return &manifests.TenantSecrets{
+		TenantName:   tenantName,
+		ClientID:     string(clientID),
+		ClientSecret: string(clientSecret),
+		IssuerCAPath: string(issuerCAPath),
+	}, nil
+}

operator/internal/handlers/internal/gateway/tenant_secrets_test.go

@@ -142,3 +142,66 @@ func TestGetTenantSecrets_DynamicMode(t *testing.T) {
 	}
 	require.ElementsMatch(t, ts, expected)
 }
+
+func TestExtractSecret(t *testing.T) {
+	type test struct {
+		name       string
+		tenantName string
+		secret     *corev1.Secret
+		wantErr    bool
+	}
+	table := []test{
+		{
+			name:       "missing clientID",
+			tenantName: "tenant-a",
+			secret:     &corev1.Secret{},
+			wantErr:    true,
+		},
+		{
+			name:       "missing clientSecret",
+			tenantName: "tenant-a",
+			secret: &corev1.Secret{
+				Data: map[string][]byte{
+					"clientID": []byte("test"),
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name:       "missing issuerCAPath",
+			tenantName: "tenant-a",
+			secret: &corev1.Secret{
+				Data: map[string][]byte{
+					"clientID":     []byte("test"),
+					"clientSecret": []byte("test"),
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name:       "all set",
+			tenantName: "tenant-a",
+			secret: &corev1.Secret{
+				Data: map[string][]byte{
+					"clientID":     []byte("test"),
+					"clientSecret": []byte("test"),
+					"issuerCAPath": []byte("/tmp/test"),
+				},
+			},
+		},
+	}
+	for _, tst := range table {
+		tst := tst
+		t.Run(tst.name, func(t *testing.T) {
+			t.Parallel()
+
+			_, err := extractSecret(tst.secret, tst.tenantName)
+			if !tst.wantErr {
+				require.NoError(t, err)
+			}
+			if tst.wantErr {
+				require.NotNil(t, err)
+			}
+		})
+	}
+}

operator/internal/handlers/internal/storage/ca_configmap.go

@@ -0,0 +1,9 @@
+package storage
+
+import corev1 "k8s.io/api/core/v1"
+
+// IsValidCAConfigMap checks if the given CA configMap has an
+// non-empty entry for key `service-ca.crt`
+func IsValidCAConfigMap(cm *corev1.ConfigMap) bool {
+	return cm.Data["service-ca.crt"] != ""
+}

operator/internal/handlers/internal/storage/ca_configmap_test.go

@@ -0,0 +1,49 @@
+package storage_test
+
+import (
+	"testing"
+
+	"github.com/grafana/loki/operator/internal/handlers/internal/storage"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func TestIsValidConfigMap(t *testing.T) {
+	type test struct {
+		name  string
+		cm    *corev1.ConfigMap
+		valid bool
+	}
+	table := []test{
+		{
+			name: "valid CA configmap",
+			cm: &corev1.ConfigMap{
+				Data: map[string]string{
+					"service-ca.crt": "has-some-data",
+				},
+			},
+			valid: true,
+		},
+		{
+			name: "missing `service-ca.crt` key",
+			cm:   &corev1.ConfigMap{},
+		},
+		{
+			name: "missing CA content",
+			cm: &corev1.ConfigMap{
+				Data: map[string]string{
+					"service-ca.crt": "",
+				},
+			},
+		},
+	}
+	for _, tst := range table {
+		tst := tst
+		t.Run(tst.name, func(t *testing.T) {
+			t.Parallel()
+
+			ok := storage.IsValidCAConfigMap(tst.cm)
+			require.Equal(t, tst.valid, ok)
+		})
+	}
+}

operator/internal/handlers/internal/storage/secrets.go

@@ -1,42 +1,20 @@
-package secrets
+package storage
 
 import (
 	"github.com/ViaQ/logerr/v2/kverrors"
 	lokiv1beta1 "github.com/grafana/loki/operator/api/v1beta1"
-	"github.com/grafana/loki/operator/internal/manifests"
 	"github.com/grafana/loki/operator/internal/manifests/storage"
 
 	corev1 "k8s.io/api/core/v1"
 )
 
-// ExtractGatewaySecret reads a k8s secret into a manifest tenant secret struct if valid.
-func ExtractGatewaySecret(s *corev1.Secret, tenantName string) (*manifests.TenantSecrets, error) {
-	// Extract and validate mandatory fields
-	clientID, ok := s.Data["clientID"]
-	if !ok {
-		return nil, kverrors.New("missing clientID field", "field", "clientID")
-	}
-	clientSecret, ok := s.Data["clientSecret"]
-	if !ok {
-		return nil, kverrors.New("missing clientSecret field", "field", "clientSecret")
-	}
-	issuerCAPath, ok := s.Data["issuerCAPath"]
-	if !ok {
-		return nil, kverrors.New("missing issuerCAPath field", "field", "issuerCAPath")
-	}
-
-	return &manifests.TenantSecrets{
-		TenantName:   tenantName,
-		ClientID:     string(clientID),
-		ClientSecret: string(clientSecret),
-		IssuerCAPath: string(issuerCAPath),
-	}, nil
-}
-
-// ExtractStorageSecret reads a k8s secret into a manifest object storage struct if valid.
-func ExtractStorageSecret(s *corev1.Secret, secretType lokiv1beta1.ObjectStorageSecretType) (*storage.Options, error) {
+// ExtractSecret reads a k8s secret into a manifest object storage struct if valid.
+func ExtractSecret(s *corev1.Secret, secretType lokiv1beta1.ObjectStorageSecretType) (*storage.Options, error) {
 	var err error
-	storageOpts := storage.Options{SharedStore: secretType}
+	storageOpts := storage.Options{
+		SecretName:  s.Name,
+		SharedStore: secretType,
+	}
 
 	switch secretType {
 	case lokiv1beta1.ObjectStorageSecretAzure:
@@ -59,20 +37,20 @@ func ExtractStorageSecret(s *corev1.Secret, secretType lokiv1beta1.ObjectStorage
 
 func extractAzureConfigSecret(s *corev1.Secret) (*storage.AzureStorageConfig, error) {
 	// Extract and validate mandatory fields
-	env, ok := s.Data["environment"]
-	if !ok {
+	env := s.Data["environment"]
+	if len(env) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "environment")
 	}
-	container, ok := s.Data["container"]
-	if !ok {
+	container := s.Data["container"]
+	if len(container) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "container")
 	}
-	name, ok := s.Data["account_name"]
-	if !ok {
+	name := s.Data["account_name"]
+	if len(name) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "account_name")
 	}
-	key, ok := s.Data["account_key"]
-	if !ok {
+	key := s.Data["account_key"]
+	if len(key) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "account_key")
 	}
 
@@ -86,14 +64,14 @@ func extractAzureConfigSecret(s *corev1.Secret) (*storage.AzureStorageConfig, er
 
 func extractGCSConfigSecret(s *corev1.Secret) (*storage.GCSStorageConfig, error) {
 	// Extract and validate mandatory fields
-	bucket, ok := s.Data["bucketname"]
-	if !ok {
+	bucket := s.Data["bucketname"]
+	if len(bucket) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "bucketname")
 	}
 
 	// Check if google authentication credentials is provided
-	_, ok = s.Data["key.json"]
-	if !ok {
+	keyJSON := s.Data["key.json"]
+	if len(keyJSON) == 0 {
 		return nil, kverrors.New("missing google authentication credentials", "field", "key.json")
 	}
 
@@ -104,21 +82,21 @@ func extractGCSConfigSecret(s *corev1.Secret) (*storage.GCSStorageConfig, error)
 
 func extractS3ConfigSecret(s *corev1.Secret) (*storage.S3StorageConfig, error) {
 	// Extract and validate mandatory fields
-	endpoint, ok := s.Data["endpoint"]
-	if !ok {
+	endpoint := s.Data["endpoint"]
+	if len(endpoint) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "endpoint")
 	}
-	buckets, ok := s.Data["bucketnames"]
-	if !ok {
+	buckets := s.Data["bucketnames"]
+	if len(buckets) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "bucketnames")
 	}
 	// TODO buckets are comma-separated list
-	id, ok := s.Data["access_key_id"]
-	if !ok {
+	id := s.Data["access_key_id"]
+	if len(id) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "access_key_id")
 	}
-	secret, ok := s.Data["access_key_secret"]
-	if !ok {
+	secret := s.Data["access_key_secret"]
+	if len(secret) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "access_key_secret")
 	}
 
@@ -136,40 +114,40 @@ func extractS3ConfigSecret(s *corev1.Secret) (*storage.S3StorageConfig, error) {
 
 func extractSwiftConfigSecret(s *corev1.Secret) (*storage.SwiftStorageConfig, error) {
 	// Extract and validate mandatory fields
-	url, ok := s.Data["auth_url"]
-	if !ok {
+	url := s.Data["auth_url"]
+	if len(url) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "auth_url")
 	}
-	username, ok := s.Data["username"]
-	if !ok {
+	username := s.Data["username"]
+	if len(username) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "username")
 	}
-	userDomainName, ok := s.Data["user_domain_name"]
-	if !ok {
+	userDomainName := s.Data["user_domain_name"]
+	if len(userDomainName) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "user_domain_name")
 	}
-	userDomainID, ok := s.Data["user_domain_id"]
-	if !ok {
+	userDomainID := s.Data["user_domain_id"]
+	if len(userDomainID) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "user_domain_id")
 	}
-	userID, ok := s.Data["user_id"]
-	if !ok {
+	userID := s.Data["user_id"]
+	if len(userID) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "user_id")
 	}
-	password, ok := s.Data["password"]
-	if !ok {
+	password := s.Data["password"]
+	if len(password) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "password")
 	}
-	domainID, ok := s.Data["domain_id"]
-	if !ok {
+	domainID := s.Data["domain_id"]
+	if len(domainID) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "domain_id")
 	}
-	domainName, ok := s.Data["domain_name"]
-	if !ok {
+	domainName := s.Data["domain_name"]
+	if len(domainName) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "domain_name")
 	}
-	containerName, ok := s.Data["container_name"]
-	if !ok {
+	containerName := s.Data["container_name"]
+	if len(containerName) == 0 {
 		return nil, kverrors.New("missing secret field", "field", "container_name")
 	}
 

operator/internal/handlers/internal/storage/secrets_test.go

@@ -1,10 +1,10 @@
-package secrets_test
+package storage_test
 
 import (
 	"testing"
 
 	lokiv1beta1 "github.com/grafana/loki/operator/api/v1beta1"
-	"github.com/grafana/loki/operator/internal/handlers/internal/secrets"
+	"github.com/grafana/loki/operator/internal/handlers/internal/storage"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -68,7 +68,7 @@ func TestAzureExtract(t *testing.T) {
 		t.Run(tst.name, func(t *testing.T) {
 			t.Parallel()
 
-			_, err := secrets.ExtractStorageSecret(tst.secret, lokiv1beta1.ObjectStorageSecretAzure)
+			_, err := storage.ExtractSecret(tst.secret, lokiv1beta1.ObjectStorageSecretAzure)
 			if !tst.wantErr {
 				require.NoError(t, err)
 			}
@@ -115,7 +115,7 @@ func TestGCSExtract(t *testing.T) {
 		t.Run(tst.name, func(t *testing.T) {
 			t.Parallel()
 
-			_, err := secrets.ExtractStorageSecret(tst.secret, lokiv1beta1.ObjectStorageSecretGCS)
+			_, err := storage.ExtractSecret(tst.secret, lokiv1beta1.ObjectStorageSecretGCS)
 			if !tst.wantErr {
 				require.NoError(t, err)
 			}
@@ -185,7 +185,7 @@ func TestS3Extract(t *testing.T) {
 		t.Run(tst.name, func(t *testing.T) {
 			t.Parallel()
 
-			_, err := secrets.ExtractStorageSecret(tst.secret, lokiv1beta1.ObjectStorageSecretS3)
+			_, err := storage.ExtractSecret(tst.secret, lokiv1beta1.ObjectStorageSecretS3)
 			if !tst.wantErr {
 				require.NoError(t, err)
 			}
@@ -330,70 +330,7 @@ func TestSwiftExtract(t *testing.T) {
 		t.Run(tst.name, func(t *testing.T) {
 			t.Parallel()
 
-			_, err := secrets.ExtractStorageSecret(tst.secret, lokiv1beta1.ObjectStorageSecretSwift)
-			if !tst.wantErr {
-				require.NoError(t, err)
-			}
-			if tst.wantErr {
-				require.NotNil(t, err)
-			}
-		})
-	}
-}
-
-func TestExtractGatewaySecret(t *testing.T) {
-	type test struct {
-		name       string
-		tenantName string
-		secret     *corev1.Secret
-		wantErr    bool
-	}
-	table := []test{
-		{
-			name:       "missing clientID",
-			tenantName: "tenant-a",
-			secret:     &corev1.Secret{},
-			wantErr:    true,
-		},
-		{
-			name:       "missing clientSecret",
-			tenantName: "tenant-a",
-			secret: &corev1.Secret{
-				Data: map[string][]byte{
-					"clientID": []byte("test"),
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name:       "missing issuerCAPath",
-			tenantName: "tenant-a",
-			secret: &corev1.Secret{
-				Data: map[string][]byte{
-					"clientID":     []byte("test"),
-					"clientSecret": []byte("test"),
-				},
-			},
-			wantErr: true,
-		},
-		{
-			name:       "all set",
-			tenantName: "tenant-a",
-			secret: &corev1.Secret{
-				Data: map[string][]byte{
-					"clientID":     []byte("test"),
-					"clientSecret": []byte("test"),
-					"issuerCAPath": []byte("/tmp/test"),
-				},
-			},
-		},
-	}
-	for _, tst := range table {
-		tst := tst
-		t.Run(tst.name, func(t *testing.T) {
-			t.Parallel()
-
-			_, err := secrets.ExtractGatewaySecret(tst.secret, tst.tenantName)
+			_, err := storage.ExtractSecret(tst.secret, lokiv1beta1.ObjectStorageSecretSwift)
 			if !tst.wantErr {
 				require.NoError(t, err)
 			}

operator/internal/handlers/lokistack_create_or_update.go

@@ -9,8 +9,9 @@ import (
 	"github.com/grafana/loki/operator/internal/external/k8s"
 	"github.com/grafana/loki/operator/internal/handlers/internal/gateway"
 	"github.com/grafana/loki/operator/internal/handlers/internal/rules"
-	"github.com/grafana/loki/operator/internal/handlers/internal/secrets"
+	"github.com/grafana/loki/operator/internal/handlers/internal/storage"
 	"github.com/grafana/loki/operator/internal/manifests"
+	storageoptions "github.com/grafana/loki/operator/internal/manifests/storage"
 	"github.com/grafana/loki/operator/internal/metrics"
 	"github.com/grafana/loki/operator/internal/status"
 
@@ -68,7 +69,7 @@ func CreateOrUpdateLokiStack(
 		return kverrors.Wrap(err, "failed to lookup lokistack storage secret", "name", key)
 	}
 
-	storage, err := secrets.ExtractStorageSecret(&storageSecret, stack.Spec.Storage.Secret.Type)
+	objstorage, err := storage.ExtractSecret(&storageSecret, stack.Spec.Storage.Secret.Type)
 	if err != nil {
 		return &status.DegradedError{
 			Message: fmt.Sprintf("Invalid object storage secret contents: %s", err),
@@ -77,6 +78,31 @@ func CreateOrUpdateLokiStack(
 		}
 	}
 
+	if stack.Spec.Storage.TLS != nil {
+		var cm corev1.ConfigMap
+		key := client.ObjectKey{Name: stack.Spec.Storage.TLS.CA, Namespace: stack.Namespace}
+		if err = k.Get(ctx, key, &cm); err != nil {
+			if apierrors.IsNotFound(err) {
+				return &status.DegradedError{
+					Message: "Missing object storage CA config map",
+					Reason:  lokiv1beta1.ReasonMissingObjectStorageCAConfigMap,
+					Requeue: false,
+				}
+			}
+			return kverrors.Wrap(err, "failed to lookup lokistack object storage CA config map", "name", key)
+		}
+
+		if !storage.IsValidCAConfigMap(&cm) {
+			return &status.DegradedError{
+				Message: "Invalid object storage CA configmap contents: missing key `service-ca.crt` or no contents",
+				Reason:  lokiv1beta1.ReasonInvalidObjectStorageCAConfigMap,
+				Requeue: false,
+			}
+		}
+
+		objstorage.TLS = &storageoptions.TLSConfig{CA: cm.Name}
+	}
+
 	var (
 		baseDomain    string
 		tenantSecrets []*manifests.TenantSecrets
@@ -138,7 +164,7 @@ func CreateOrUpdateLokiStack(
 		GatewayBaseDomain: baseDomain,
 		Stack:             stack.Spec,
 		Flags:             flags,
-		ObjectStorage:     *storage,
+		ObjectStorage:     *objstorage,
 		AlertingRules:     alertingRules,
 		RecordingRules:    recordingRules,
 		Tenants: manifests.Tenants{

operator/internal/handlers/lokistack_create_or_update_test.go

@@ -75,6 +75,14 @@ var (
 		},
 		Data: map[string][]byte{},
 	}
+
+	invalidCAConfigMap = corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "some-stack-ca-configmap",
+			Namespace: "some-ns",
+		},
+		Data: map[string]string{},
+	}
 )
 
 func TestMain(m *testing.M) {
@@ -724,6 +732,137 @@ func TestCreateOrUpdateLokiStack_WhenInvalidSecret_SetDegraded(t *testing.T) {
 	require.Equal(t, degradedErr, err)
 }
 
+func TestCreateOrUpdateLokiStack_WhenMissingCAConfigMap_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: "Missing object storage CA config map",
+		Reason:  lokiv1beta1.ReasonMissingObjectStorageCAConfigMap,
+		Requeue: false,
+	}
+
+	stack := &lokiv1beta1.LokiStack{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "LokiStack",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-stack",
+			Namespace: "some-ns",
+			UID:       "b23f9a38-9672-499f-8c29-15ede74d3ece",
+		},
+		Spec: lokiv1beta1.LokiStackSpec{
+			Size: lokiv1beta1.SizeOneXExtraSmall,
+			Storage: lokiv1beta1.ObjectStorageSpec{
+				Secret: lokiv1beta1.ObjectStorageSecretSpec{
+					Name: defaultSecret.Name,
+					Type: lokiv1beta1.ObjectStorageSecretS3,
+				},
+				TLS: &lokiv1beta1.ObjectStorageTLSSpec{
+					CA: "not-existing",
+				},
+			},
+		},
+	}
+
+	// GetStub looks up the CR first, so we need to return our fake stack
+	// return NotFound for everything else to trigger create.
+	k.GetStub = func(_ context.Context, name types.NamespacedName, object client.Object) error {
+		if r.Name == name.Name && r.Namespace == name.Namespace {
+			k.SetClientObject(object, stack)
+			return nil
+		}
+
+		if name.Name == defaultSecret.Name {
+			k.SetClientObject(object, &defaultSecret)
+			return nil
+		}
+
+		return apierrors.NewNotFound(schema.GroupResource{}, "something is not found")
+	}
+
+	k.StatusStub = func() client.StatusWriter { return sw }
+
+	err := handlers.CreateOrUpdateLokiStack(context.TODO(), logger, r, k, scheme, flags)
+
+	// make sure error is returned
+	require.Error(t, err)
+	require.Equal(t, degradedErr, err)
+}
+
+func TestCreateOrUpdateLokiStack_WhenInvalidCAConfigMap_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 object storage CA configmap contents: missing key `service-ca.crt` or no contents",
+		Reason:  lokiv1beta1.ReasonInvalidObjectStorageCAConfigMap,
+		Requeue: false,
+	}
+
+	stack := &lokiv1beta1.LokiStack{
+		TypeMeta: metav1.TypeMeta{
+			Kind: "LokiStack",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "my-stack",
+			Namespace: "some-ns",
+			UID:       "b23f9a38-9672-499f-8c29-15ede74d3ece",
+		},
+		Spec: lokiv1beta1.LokiStackSpec{
+			Size: lokiv1beta1.SizeOneXExtraSmall,
+			Storage: lokiv1beta1.ObjectStorageSpec{
+				Secret: lokiv1beta1.ObjectStorageSecretSpec{
+					Name: defaultSecret.Name,
+					Type: lokiv1beta1.ObjectStorageSecretS3,
+				},
+				TLS: &lokiv1beta1.ObjectStorageTLSSpec{
+					CA: invalidCAConfigMap.Name,
+				},
+			},
+		},
+	}
+
+	// GetStub looks up the CR first, so we need to return our fake stack
+	// return NotFound for everything else to trigger create.
+	k.GetStub = func(_ context.Context, name types.NamespacedName, object client.Object) error {
+		if r.Name == name.Name && r.Namespace == name.Namespace {
+			k.SetClientObject(object, stack)
+			return nil
+		}
+		if name.Name == defaultSecret.Name {
+			k.SetClientObject(object, &defaultSecret)
+			return nil
+		}
+
+		if name.Name == invalidCAConfigMap.Name {
+			k.SetClientObject(object, &invalidCAConfigMap)
+			return nil
+		}
+		return apierrors.NewNotFound(schema.GroupResource{}, "something is not found")
+	}
+
+	k.StatusStub = func() client.StatusWriter { return sw }
+
+	err := handlers.CreateOrUpdateLokiStack(context.TODO(), logger, r, k, scheme, flags)
+
+	// make sure error is returned
+	require.Error(t, err)
+	require.Equal(t, degradedErr, err)
+}
+
 func TestCreateOrUpdateLokiStack_WhenInvalidTenantsConfiguration_SetDegraded(t *testing.T) {
 	sw := &k8sfakes.FakeStatusWriter{}
 	k := &k8sfakes.FakeClient{}

operator/internal/manifests/compactor.go

@@ -5,6 +5,7 @@ import (
 	"path"
 
 	"github.com/grafana/loki/operator/internal/manifests/internal/config"
+	"github.com/grafana/loki/operator/internal/manifests/storage"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -25,9 +26,7 @@ func BuildCompactor(opts Options) ([]client.Object, error) {
 		}
 	}
 
-	storageType := opts.Stack.Storage.Secret.Type
-	secretName := opts.Stack.Storage.Secret.Name
-	if err := configureStatefulSetForStorageType(statefulSet, storageType, secretName); err != nil {
+	if err := storage.ConfigureStatefulSet(statefulSet, opts.ObjectStorage); err != nil {
 		return nil, err
 	}
 

operator/internal/manifests/indexgateway.go

@@ -5,6 +5,7 @@ import (
 	"path"
 
 	"github.com/grafana/loki/operator/internal/manifests/internal/config"
+	"github.com/grafana/loki/operator/internal/manifests/storage"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -24,9 +25,7 @@ func BuildIndexGateway(opts Options) ([]client.Object, error) {
 		}
 	}
 
-	storageType := opts.Stack.Storage.Secret.Type
-	secretName := opts.Stack.Storage.Secret.Name
-	if err := configureStatefulSetForStorageType(statefulSet, storageType, secretName); err != nil {
+	if err := storage.ConfigureStatefulSet(statefulSet, opts.ObjectStorage); err != nil {
 		return nil, err
 	}
 

operator/internal/manifests/ingester.go

@@ -5,6 +5,7 @@ import (
 	"path"
 
 	"github.com/grafana/loki/operator/internal/manifests/internal/config"
+	"github.com/grafana/loki/operator/internal/manifests/storage"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -24,9 +25,7 @@ func BuildIngester(opts Options) ([]client.Object, error) {
 		}
 	}
 
-	storageType := opts.Stack.Storage.Secret.Type
-	secretName := opts.Stack.Storage.Secret.Name
-	if err := configureStatefulSetForStorageType(statefulSet, storageType, secretName); err != nil {
+	if err := storage.ConfigureStatefulSet(statefulSet, opts.ObjectStorage); err != nil {
 		return nil, err
 	}
 

operator/internal/manifests/object_storage.go

@@ -1,31 +0,0 @@
-package manifests
-
-import (
-	lokiv1beta1 "github.com/grafana/loki/operator/api/v1beta1"
-	"github.com/grafana/loki/operator/internal/manifests/storage"
-	appsv1 "k8s.io/api/apps/v1"
-)
-
-func configureDeploymentForStorageType(
-	d *appsv1.Deployment,
-	t lokiv1beta1.ObjectStorageSecretType,
-	secretName string) error {
-	switch t {
-	case lokiv1beta1.ObjectStorageSecretGCS:
-		return storage.ConfigureDeployment(d, secretName)
-	default:
-		return nil
-	}
-}
-
-func configureStatefulSetForStorageType(
-	d *appsv1.StatefulSet,
-	t lokiv1beta1.ObjectStorageSecretType,
-	secretName string) error {
-	switch t {
-	case lokiv1beta1.ObjectStorageSecretGCS:
-		return storage.ConfigureStatefulSet(d, secretName)
-	default:
-		return nil
-	}
-}

operator/internal/manifests/querier.go

@@ -5,6 +5,7 @@ import (
 	"path"
 
 	"github.com/grafana/loki/operator/internal/manifests/internal/config"
+	"github.com/grafana/loki/operator/internal/manifests/storage"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -23,9 +24,7 @@ func BuildQuerier(opts Options) ([]client.Object, error) {
 		}
 	}
 
-	storageType := opts.Stack.Storage.Secret.Type
-	secretName := opts.Stack.Storage.Secret.Name
-	if err := configureDeploymentForStorageType(deployment, storageType, secretName); err != nil {
+	if err := storage.ConfigureDeployment(deployment, opts.ObjectStorage); err != nil {
 		return nil, err
 	}
 

operator/internal/manifests/storage/configure.go

@@ -1,9 +1,11 @@
 package storage
 
 import (
+	"fmt"
 	"path"
 
 	"github.com/ViaQ/logerr/v2/kverrors"
+	lokiv1beta1 "github.com/grafana/loki/operator/api/v1beta1"
 	"github.com/imdario/mergo"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -17,11 +19,56 @@ const (
 	GCSFileName = "key.json"
 
 	secretDirectory = "/etc/storage/secrets"
+	caDirectory     = "/etc/storage/ca"
 )
 
+// ConfigureDeployment appends additional pod volumes and container env vars, args, volume mounts
+// based on the object storage type. Currently supported amendments:
+// - GCS: Ensure env var GOOGLE_APPLICATION_CREDENTIALS in container
+// - S3: Ensure mounting custom CA configmap if any TLSConfig given
+func ConfigureDeployment(
+	d *appsv1.Deployment,
+	opts Options,
+) error {
+	switch opts.SharedStore {
+	case lokiv1beta1.ObjectStorageSecretGCS:
+		return configureDeployment(d, opts.SecretName)
+	case lokiv1beta1.ObjectStorageSecretS3:
+		if opts.TLS == nil {
+			return nil
+		}
+
+		return configureDeploymentCA(d, opts.TLS.CA)
+	default:
+		return nil
+	}
+}
+
+// ConfigureStatefulSet appends additional pod volumes and container env vars, args, volume mounts
+// based on the object storage type. Currently supported amendments:
+// - GCS: Ensure env var GOOGLE_APPLICATION_CREDENTIALS in container
+// - S3: Ensure mounting custom CA configmap if any TLSConfig given
+func ConfigureStatefulSet(
+	d *appsv1.StatefulSet,
+	opts Options,
+) error {
+	switch opts.SharedStore {
+	case lokiv1beta1.ObjectStorageSecretGCS:
+		return configureStatefulSet(d, opts.SecretName)
+	case lokiv1beta1.ObjectStorageSecretS3:
+		if opts.TLS == nil {
+			return nil
+		}
+
+		return configureStatefulSetCA(d, opts.TLS.CA)
+	default:
+		return nil
+	}
+}
+
 // ConfigureDeployment merges a GCS Object Storage volume into the deployment spec.
 // With this, the deployment will expose an environment variable for Google authentication.
-func ConfigureDeployment(d *appsv1.Deployment, secretName string) error {
+func configureDeployment(d *appsv1.Deployment, secretName string) error {
 	p := ensureCredentialsForGCS(&d.Spec.Template.Spec, secretName)
 
 	if err := mergo.Merge(&d.Spec.Template.Spec, p, mergo.WithOverride); err != nil {
@@ -31,9 +78,20 @@ func ConfigureDeployment(d *appsv1.Deployment, secretName string) error {
 	return nil
 }
 
+// ConfigureDeploymentCA merges a S3 CA ConfigMap volume into the deployment spec.
+func configureDeploymentCA(d *appsv1.Deployment, cmName string) error {
+	p := ensureCAForS3(&d.Spec.Template.Spec, cmName)
+
+	if err := mergo.Merge(&d.Spec.Template.Spec, p, mergo.WithOverride); err != nil {
+		return kverrors.Wrap(err, "failed to merge s3 object storage ca options ")
+	}
+
+	return nil
+}
+
 // ConfigureStatefulSet merges a GCS Object Storage volume into the statefulset spec.
 // With this, the statefulset will expose an environment variable for Google authentication.
-func ConfigureStatefulSet(s *appsv1.StatefulSet, secretName string) error {
+func configureStatefulSet(s *appsv1.StatefulSet, secretName string) error {
 	p := ensureCredentialsForGCS(&s.Spec.Template.Spec, secretName)
 
 	if err := mergo.Merge(&s.Spec.Template.Spec, p, mergo.WithOverride); err != nil {
@@ -43,6 +101,17 @@ func ConfigureStatefulSet(s *appsv1.StatefulSet, secretName string) error {
 	return nil
 }
 
+// ConfigureStatefulSetCA merges a S3 CA ConfigMap volume into the statefulset spec.
+func configureStatefulSetCA(s *appsv1.StatefulSet, cmName string) error {
+	p := ensureCAForS3(&s.Spec.Template.Spec, cmName)
+
+	if err := mergo.Merge(&s.Spec.Template.Spec, p, mergo.WithOverride); err != nil {
+		return kverrors.Wrap(err, "failed to merge s3 object storage ca options ")
+	}
+
+	return nil
+}
+
 func ensureCredentialsForGCS(p *corev1.PodSpec, secretName string) corev1.PodSpec {
 	container := p.Containers[0].DeepCopy()
 	volumes := p.Volumes
@@ -74,3 +143,36 @@ func ensureCredentialsForGCS(p *corev1.PodSpec, secretName string) corev1.PodSpe
 		Volumes: volumes,
 	}
 }
+
+func ensureCAForS3(p *corev1.PodSpec, cmName string) corev1.PodSpec {
+	container := p.Containers[0].DeepCopy()
+	volumes := p.Volumes
+
+	volumes = append(volumes, corev1.Volume{
+		Name: cmName,
+		VolumeSource: corev1.VolumeSource{
+			ConfigMap: &corev1.ConfigMapVolumeSource{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: cmName,
+				},
+			},
+		},
+	})
+
+	container.VolumeMounts = append(container.VolumeMounts, corev1.VolumeMount{
+		Name:      cmName,
+		ReadOnly:  false,
+		MountPath: caDirectory,
+	})
+
+	container.Args = append(container.Args,
+		fmt.Sprintf("-s3.http.ca-file=%s", path.Join(caDirectory, "service-ca.crt")),
+	)
+
+	return corev1.PodSpec{
+		Containers: []corev1.Container{
+			*container,
+		},
+		Volumes: volumes,
+	}
+}

operator/internal/manifests/storage/configure_test.go

@@ -1,10 +1,9 @@
-package manifests
+package storage_test
 
 import (
 	"testing"
 
 	lokiv1beta1 "github.com/grafana/loki/operator/api/v1beta1"
-	"github.com/grafana/loki/operator/internal/manifests/internal/config"
 	"github.com/grafana/loki/operator/internal/manifests/storage"
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
@@ -13,18 +12,19 @@ import (
 
 func TestConfigureDeploymentForStorageType(t *testing.T) {
 	type tt struct {
-		desc        string
-		storageType lokiv1beta1.ObjectStorageSecretType
-		secretName  string
-		dpl         *appsv1.Deployment
-		want        *appsv1.Deployment
+		desc string
+		opts storage.Options
+		dpl  *appsv1.Deployment
+		want *appsv1.Deployment
 	}
 
 	tc := []tt{
 		{
-			desc:        "object storage other than GCS",
-			storageType: lokiv1beta1.ObjectStorageSecretS3,
-			secretName:  "test",
+			desc: "object storage other than GCS",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretS3,
+			},
 			dpl: &appsv1.Deployment{
 				Spec: appsv1.DeploymentSpec{
 					Template: corev1.PodTemplateSpec{
@@ -32,26 +32,6 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 							Containers: []corev1.Container{
 								{
 									Name: "loki-ingester",
-									VolumeMounts: []corev1.VolumeMount{
-										{
-											Name:      configVolumeName,
-											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
-										},
-									},
-								},
-							},
-							Volumes: []corev1.Volume{
-								{
-									Name: configVolumeName,
-									VolumeSource: corev1.VolumeSource{
-										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
-											},
-										},
-									},
 								},
 							},
 						},
@@ -65,26 +45,6 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 							Containers: []corev1.Container{
 								{
 									Name: "loki-ingester",
-									VolumeMounts: []corev1.VolumeMount{
-										{
-											Name:      configVolumeName,
-											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
-										},
-									},
-								},
-							},
-							Volumes: []corev1.Volume{
-								{
-									Name: configVolumeName,
-									VolumeSource: corev1.VolumeSource{
-										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
-											},
-										},
-									},
 								},
 							},
 						},
@@ -93,9 +53,11 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 			},
 		},
 		{
-			desc:        "object storage GCS",
-			storageType: lokiv1beta1.ObjectStorageSecretGCS,
-			secretName:  "test",
+			desc: "object storage GCS",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretGCS,
+			},
 			dpl: &appsv1.Deployment{
 				Spec: appsv1.DeploymentSpec{
 					Template: corev1.PodTemplateSpec{
@@ -103,26 +65,6 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 							Containers: []corev1.Container{
 								{
 									Name: "loki-ingester",
-									VolumeMounts: []corev1.VolumeMount{
-										{
-											Name:      configVolumeName,
-											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
-										},
-									},
-								},
-							},
-							Volumes: []corev1.Volume{
-								{
-									Name: configVolumeName,
-									VolumeSource: corev1.VolumeSource{
-										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
-											},
-										},
-									},
 								},
 							},
 						},
@@ -137,11 +79,7 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 								{
 									Name: "loki-ingester",
 									VolumeMounts: []corev1.VolumeMount{
-										{
-											Name:      configVolumeName,
-											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
-										},
+
 										{
 											Name:      "test",
 											ReadOnly:  false,
@@ -157,17 +95,6 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 								},
 							},
 							Volumes: []corev1.Volume{
-								{
-									Name: configVolumeName,
-									VolumeSource: corev1.VolumeSource{
-										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
-											},
-										},
-									},
-								},
 								{
 									Name: "test",
 									VolumeSource: corev1.VolumeSource{
@@ -188,7 +115,7 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 		tc := tc
 		t.Run(tc.desc, func(t *testing.T) {
 			t.Parallel()
-			err := configureDeploymentForStorageType(tc.dpl, tc.storageType, tc.secretName)
+			err := storage.ConfigureDeployment(tc.dpl, tc.opts)
 			require.NoError(t, err)
 			require.Equal(t, tc.want, tc.dpl)
 		})
@@ -197,18 +124,19 @@ func TestConfigureDeploymentForStorageType(t *testing.T) {
 
 func TestConfigureStatefulSetForStorageType(t *testing.T) {
 	type tt struct {
-		desc        string
-		storageType lokiv1beta1.ObjectStorageSecretType
-		secretName  string
-		sts         *appsv1.StatefulSet
-		want        *appsv1.StatefulSet
+		desc string
+		opts storage.Options
+		sts  *appsv1.StatefulSet
+		want *appsv1.StatefulSet
 	}
 
 	tc := []tt{
 		{
-			desc:        "object storage other than GCS",
-			storageType: lokiv1beta1.ObjectStorageSecretS3,
-			secretName:  "test",
+			desc: "object storage other than GCS",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretS3,
+			},
 			sts: &appsv1.StatefulSet{
 				Spec: appsv1.StatefulSetSpec{
 					Template: corev1.PodTemplateSpec{
@@ -216,26 +144,39 @@ func TestConfigureStatefulSetForStorageType(t *testing.T) {
 							Containers: []corev1.Container{
 								{
 									Name: "loki-ingester",
-									VolumeMounts: []corev1.VolumeMount{
-										{
-											Name:      configVolumeName,
-											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
-										},
-									},
 								},
 							},
-							Volumes: []corev1.Volume{
+						},
+					},
+				},
+			},
+			want: &appsv1.StatefulSet{
+				Spec: appsv1.StatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
 								{
-									Name: configVolumeName,
-									VolumeSource: corev1.VolumeSource{
-										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
-											},
-										},
-									},
+									Name: "loki-ingester",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			desc: "object storage GCS",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretGCS,
+			},
+			sts: &appsv1.StatefulSet{
+				Spec: appsv1.StatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-ingester",
 								},
 							},
 						},
@@ -251,22 +192,25 @@ func TestConfigureStatefulSetForStorageType(t *testing.T) {
 									Name: "loki-ingester",
 									VolumeMounts: []corev1.VolumeMount{
 										{
-											Name:      configVolumeName,
+											Name:      "test",
 											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
+											MountPath: "/etc/storage/secrets",
+										},
+									},
+									Env: []corev1.EnvVar{
+										{
+											Name:  storage.EnvGoogleApplicationCredentials,
+											Value: "/etc/storage/secrets/key.json",
 										},
 									},
 								},
 							},
 							Volumes: []corev1.Volume{
 								{
-									Name: configVolumeName,
+									Name: "test",
 									VolumeSource: corev1.VolumeSource{
-										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
-											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
-											},
+										Secret: &corev1.SecretVolumeSource{
+											SecretName: "test",
 										},
 									},
 								},
@@ -276,34 +220,109 @@ func TestConfigureStatefulSetForStorageType(t *testing.T) {
 				},
 			},
 		},
+	}
+
+	for _, tc := range tc {
+		tc := tc
+		t.Run(tc.desc, func(t *testing.T) {
+			t.Parallel()
+			err := storage.ConfigureStatefulSet(tc.sts, tc.opts)
+			require.NoError(t, err)
+			require.Equal(t, tc.want, tc.sts)
+		})
+	}
+}
+
+func TestConfigureDeploymentForStorageCA(t *testing.T) {
+	type tt struct {
+		desc string
+		opts storage.Options
+		dpl  *appsv1.Deployment
+		want *appsv1.Deployment
+	}
+
+	tc := []tt{
 		{
-			desc:        "object storage GCS",
-			storageType: lokiv1beta1.ObjectStorageSecretGCS,
-			secretName:  "test",
-			sts: &appsv1.StatefulSet{
-				Spec: appsv1.StatefulSetSpec{
+			desc: "object storage other than S3",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretAzure,
+			},
+			dpl: &appsv1.Deployment{
+				Spec: appsv1.DeploymentSpec{
 					Template: corev1.PodTemplateSpec{
 						Spec: corev1.PodSpec{
 							Containers: []corev1.Container{
 								{
-									Name: "loki-ingester",
+									Name: "loki-querier",
+								},
+							},
+						},
+					},
+				},
+			},
+			want: &appsv1.Deployment{
+				Spec: appsv1.DeploymentSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-querier",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			desc: "object storage S3",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretS3,
+				TLS: &storage.TLSConfig{
+					CA: "test",
+				},
+			},
+			dpl: &appsv1.Deployment{
+				Spec: appsv1.DeploymentSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-querier",
+								},
+							},
+						},
+					},
+				},
+			},
+			want: &appsv1.Deployment{
+				Spec: appsv1.DeploymentSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-querier",
 									VolumeMounts: []corev1.VolumeMount{
 										{
-											Name:      configVolumeName,
+											Name:      "test",
 											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
+											MountPath: "/etc/storage/ca",
 										},
 									},
+									Args: []string{
+										"-s3.http.ca-file=/etc/storage/ca/service-ca.crt",
+									},
 								},
 							},
 							Volumes: []corev1.Volume{
 								{
-									Name: configVolumeName,
+									Name: "test",
 									VolumeSource: corev1.VolumeSource{
 										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
 											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
+												Name: "test",
 											},
 										},
 									},
@@ -313,6 +332,87 @@ func TestConfigureStatefulSetForStorageType(t *testing.T) {
 					},
 				},
 			},
+		},
+	}
+
+	for _, tc := range tc {
+		tc := tc
+		t.Run(tc.desc, func(t *testing.T) {
+			t.Parallel()
+			err := storage.ConfigureDeployment(tc.dpl, tc.opts)
+			require.NoError(t, err)
+			require.Equal(t, tc.want, tc.dpl)
+		})
+	}
+}
+
+func TestConfigureStatefulSetForStorageCA(t *testing.T) {
+	type tt struct {
+		desc string
+		opts storage.Options
+		sts  *appsv1.StatefulSet
+		want *appsv1.StatefulSet
+	}
+
+	tc := []tt{
+		{
+			desc: "object storage other than S3",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretAzure,
+				TLS: &storage.TLSConfig{
+					CA: "test",
+				},
+			},
+			sts: &appsv1.StatefulSet{
+				Spec: appsv1.StatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-ingester",
+								},
+							},
+						},
+					},
+				},
+			},
+			want: &appsv1.StatefulSet{
+				Spec: appsv1.StatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-ingester",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			desc: "object storage S3",
+			opts: storage.Options{
+				SecretName:  "test",
+				SharedStore: lokiv1beta1.ObjectStorageSecretS3,
+				TLS: &storage.TLSConfig{
+					CA: "test",
+				},
+			},
+			sts: &appsv1.StatefulSet{
+				Spec: appsv1.StatefulSetSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "loki-ingester",
+								},
+							},
+						},
+					},
+				},
+			},
 			want: &appsv1.StatefulSet{
 				Spec: appsv1.StatefulSetSpec{
 					Template: corev1.PodTemplateSpec{
@@ -321,45 +421,28 @@ func TestConfigureStatefulSetForStorageType(t *testing.T) {
 								{
 									Name: "loki-ingester",
 									VolumeMounts: []corev1.VolumeMount{
-										{
-											Name:      configVolumeName,
-											ReadOnly:  false,
-											MountPath: config.LokiConfigMountDir,
-										},
 										{
 											Name:      "test",
 											ReadOnly:  false,
-											MountPath: "/etc/storage/secrets",
+											MountPath: "/etc/storage/ca",
 										},
 									},
-									Env: []corev1.EnvVar{
-										{
-											Name:  storage.EnvGoogleApplicationCredentials,
-											Value: "/etc/storage/secrets/key.json",
-										},
+									Args: []string{
+										"-s3.http.ca-file=/etc/storage/ca/service-ca.crt",
 									},
 								},
 							},
 							Volumes: []corev1.Volume{
 								{
-									Name: configVolumeName,
+									Name: "test",
 									VolumeSource: corev1.VolumeSource{
 										ConfigMap: &corev1.ConfigMapVolumeSource{
-											DefaultMode: &defaultConfigMapMode,
 											LocalObjectReference: corev1.LocalObjectReference{
-												Name: lokiConfigMapName("stack-name"),
+												Name: "test",
 											},
 										},
 									},
 								},
-								{
-									Name: "test",
-									VolumeSource: corev1.VolumeSource{
-										Secret: &corev1.SecretVolumeSource{
-											SecretName: "test",
-										},
-									},
-								},
 							},
 						},
 					},
@@ -372,7 +455,7 @@ func TestConfigureStatefulSetForStorageType(t *testing.T) {
 		tc := tc
 		t.Run(tc.desc, func(t *testing.T) {
 			t.Parallel()
-			err := configureStatefulSetForStorageType(tc.sts, tc.storageType, tc.secretName)
+			err := storage.ConfigureStatefulSet(tc.sts, tc.opts)
 			require.NoError(t, err)
 			require.Equal(t, tc.want, tc.sts)
 		})

operator/internal/manifests/storage/options.go

@@ -7,11 +7,13 @@ import (
 // Options is used to configure Loki to integrate with
 // supported object storages.
 type Options struct {
+	SecretName  string
 	SharedStore lokiv1beta1.ObjectStorageSecretType
 	Azure       *AzureStorageConfig
 	GCS         *GCSStorageConfig
 	S3          *S3StorageConfig
 	Swift       *SwiftStorageConfig
+	TLS         *TLSConfig
 }
 
 // AzureStorageConfig for Azure storage config
@@ -53,3 +55,9 @@ type SwiftStorageConfig struct {
 	Region            string
 	Container         string
 }
+
+// TLSConfig for object storage endpoints. Currently supported only by:
+// - S3
+type TLSConfig struct {
+	CA string
+}