Issue 489: Added support for custom storage option in long term storage

deploy/crds/pravega.pravega.io_pravegaclusters_crd.yaml

@@ -2210,6 +2210,20 @@ spec:
                       that a PersistentVolumeClaim called "pravega-longterm" is present
                       and it will use it as Tier 2
                     properties:
+                      custom:
+                        description: Custom Storage as a Tier2 backend
+                        properties:
+                          env:
+                            additionalProperties:
+                              type: string
+                            type: object
+                          options:
+                            additionalProperties:
+                              type: string
+                            type: object
+                        required:
+                        - env
+                        type: object
                       ecs:
                         description: Ecs is used to configure a Dell EMC ECS system
                           as a Tier 2 backend

doc/longtermstorage.md

@@ -6,6 +6,7 @@ The following LongTermStorage storage providers are supported:
 - [Filesystem: Google Filestore](#use-google-filestore-storage-as-longtermstorage)
 - [S3: Dell EMC ECS](#use-dell-emc-ecs-as-longtermstorage)
 - [HDFS](#use-hdfs-as-longtermstorage)
+- [Custom Storage](#use-custom-storage-as-longtermstorage)
 
 ### Use NFS as LongTermStorage
 
@@ -238,3 +239,34 @@ spec:
       root: /example
       replicationFactor: 3
 ```
+
+### Use Custom Storage as LongTermStorage
+
+Pravega can also use Custom storage such as `S3` for LongTermStorage.
+
+Modify the Pravega manifest to include custom storage as LongTermStorage in [helm charts](https://github.com/pravega/charts/blob/master/charts/pravega/values.yaml).
+
+**NOTE:** The type of tier2storage has to be passed as env variable `TIER2_STORAGE`, as shown in the below example.
+```
+storage:
+
+  longtermStorage:
+    ## configure the long term storage backend type
+    ## accepted values : filesystem/ecs/hdfs
+    ## default option is filesystem
+    type: custom
+    custom:
+      options:
+        pravegaservice.storage.layout: "CHUNKED_STORAGE"
+        pravegaservice.storage.impl.name: "S3"
+        s3.bucket: "aws-sdk-test"
+        s3.prefix: "10-11-1"
+        s3.connect.config.uri.override: "false"
+        s3.connect.config.uri: <uri>
+        s3.connect.config.access.key: <access key>
+        s3.connect.config.secret.key: <secret key>
+      env:
+        TIER2_STORAGE: "S3"
+        AWS_ACCESS_KEY_ID: "key"
+        AWS_SECRET_ACCESS_KEY: "secret"
+```

pkg/apis/pravega/v1beta1/pravega.go

@@ -456,6 +456,9 @@ type LongTermStorageSpec struct {
 
 	// Hdfs is used to configure an HDFS system as a Tier 2 backend
 	Hdfs *HDFSSpec `json:"hdfs,omitempty"`
+
+	// Custom Storage as a Tier2 backend
+	Custom *CustomSpec `json:"custom,omitempty"`
 }
 
 // AuthImpemenationSpec helps to inject plugins to contoller pod
@@ -471,7 +474,7 @@ type AuthHandlerSpec struct {
 }
 
 func (s *LongTermStorageSpec) withDefaults() (changed bool) {
-	if s.FileSystem == nil && s.Ecs == nil && s.Hdfs == nil {
+	if s.FileSystem == nil && s.Ecs == nil && s.Hdfs == nil && s.Custom == nil {
 		changed = true
 		fs := &FileSystemSpec{
 			PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
@@ -511,3 +514,9 @@ type HDFSSpec struct {
 	// +optional
 	ReplicationFactor int32 `json:"replicationFactor"`
 }
+
+// CustomSpec contains the options and env variables to be passed to segmentstore
+type CustomSpec struct {
+	Options map[string]string `json:"options,omitempty"`
+	Env     map[string]string `json:"env"`
+}

pkg/apis/pravega/v1beta1/zz_generated_deepcopy_test.go

@@ -160,6 +160,19 @@ var _ = Describe("PravegaCluster DeepCopy", func() {
 				},
 			}
 			p2.Spec.Pravega.LongTermStorage.Hdfs = p1.Spec.Pravega.LongTermStorage.Hdfs.DeepCopy()
+
+			p2.Spec.Pravega.LongTermStorage.Custom = p1.Spec.Pravega.LongTermStorage.Custom.DeepCopy()
+			p1.Spec.Pravega.LongTermStorage = &v1beta1.LongTermStorageSpec{
+				Custom: &v1beta1.CustomSpec{
+					Options: map[string]string{
+						"key": "dummy-value",
+					},
+					Env: map[string]string{
+						"AWS_KEY": "key",
+					},
+				},
+			}
+			p2.Spec.Pravega.LongTermStorage.Custom = p1.Spec.Pravega.LongTermStorage.Custom.DeepCopy()
 		})
 		It("value of str1 and str2 should be equal", func() {
 			Ω(str2).To(Equal(str1))

pkg/controller/pravega/pravega_segmentstore.go

@@ -284,6 +284,12 @@ func MakeSegmentstoreConfigMap(p *api.PravegaCluster) *corev1.ConfigMap {
 		javaOpts = append(javaOpts, fmt.Sprintf("-D%v=%v", name, value))
 	}
 
+	if p.Spec.Pravega.LongTermStorage.Custom != nil {
+		for name, value := range p.Spec.Pravega.LongTermStorage.Custom.Options {
+			javaOpts = append(javaOpts, fmt.Sprintf("-D%v=%v", name, value))
+		}
+	}
+
 	sort.Strings(javaOpts)
 
 	authEnabledStr := fmt.Sprint(p.Spec.Authentication.IsEnabled())
@@ -309,7 +315,6 @@ func MakeSegmentstoreConfigMap(p *api.PravegaCluster) *corev1.ConfigMap {
 	for k, v := range getTier2StorageOptions(p.Spec.Pravega) {
 		configData[k] = v
 	}
-
 	return &corev1.ConfigMap{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "ConfigMap",
@@ -362,6 +367,12 @@ func getTier2StorageOptions(pravegaSpec *api.PravegaSpec) map[string]string {
 		}
 	}
 
+	if pravegaSpec.LongTermStorage.Custom != nil {
+		if pravegaSpec.LongTermStorage.Custom.Env != nil {
+			return pravegaSpec.LongTermStorage.Custom.Env
+		}
+	}
+
 	return make(map[string]string)
 }
 

pkg/controller/pravega/pravega_segmentstore_test.go

@@ -416,6 +416,176 @@ var _ = Describe("PravegaSegmentstore", func() {
 				})
 			})
 
+			Context("SegmentStore", func() {
+
+				It("should create a headless service", func() {
+					_ = pravega.MakeSegmentStoreHeadlessService(p)
+					Ω(err).Should(BeNil())
+				})
+
+				It("should create a pod disruption budget", func() {
+					_ = pravega.MakeSegmentstorePodDisruptionBudget(p)
+					Ω(err).Should(BeNil())
+				})
+
+				It("should create a config-map and set the JVM options given by user", func() {
+					cm := pravega.MakeSegmentstoreConfigMap(p)
+					javaOpts := cm.Data["JAVA_OPTS"]
+					Ω(strings.Contains(javaOpts, "-Dpravegaservice.clusterName=default")).Should(BeTrue())
+					Ω(strings.Contains(javaOpts, "-XX:MaxDirectMemorySize=1g")).Should(BeTrue())
+					Ω(strings.Contains(javaOpts, "-XX:MaxRAMPercentage=50.0")).Should(BeTrue())
+					Ω(strings.Contains(javaOpts, "-Dpravegaservice.service.listener.port=12345")).Should(BeTrue())
+					Ω(err).Should(BeNil())
+				})
+				It("should create a stateful set", func() {
+					sts := pravega.MakeSegmentStoreStatefulSet(p)
+					mounthostpath0 := sts.Spec.Template.Spec.Containers[0].VolumeMounts[0].MountPath
+					Ω(mounthostpath0).Should(Equal("/tmp/dumpfile/heap"))
+					mounthostpath1 := sts.Spec.Template.Spec.Containers[0].VolumeMounts[1].MountPath
+					Ω(mounthostpath1).Should(Equal("/opt/pravega/logs"))
+					mounthostpath2 := sts.Spec.Template.Spec.Containers[0].VolumeMounts[2].MountPath
+					Ω(mounthostpath2).Should(Equal("/tmp/dumpfile/heap"))
+					mounthostpath3 := sts.Spec.Template.Spec.Containers[0].VolumeMounts[3].MountPath
+					Ω(mounthostpath3).Should(Equal("/opt/pravega/logs"))
+					mounthostpath4 := sts.Spec.Template.Spec.Containers[0].VolumeMounts[4].MountPath
+					Ω(mounthostpath4).Should(Equal("/opt/pravega/conf/logback.xml"))
+					Ω(err).Should(BeNil())
+				})
+				It("should set external access service type to LoadBalancer", func() {
+					_ = pravega.MakeSegmentStoreExternalServices(p)
+					Ω(err).Should(BeNil())
+				})
+			})
+			Context("Create External service with external service type and access type empty", func() {
+				BeforeEach(func() {
+					p.Spec.Pravega.SegmentStoreExternalServiceType = ""
+					p.Spec.ExternalAccess.Type = ""
+					p.Spec.ExternalAccess.DomainName = "example"
+
+				})
+				It("should create external service with access type loadbalancer", func() {
+					svc := pravega.MakeSegmentStoreExternalServices(p)
+					Ω(svc[0].Spec.Type).To(Equal(corev1.ServiceTypeLoadBalancer))
+					Ω(err).Should(BeNil())
+				})
+			})
+			Context("Create External service with SegmentStoreExternalTrafficPolicy as cluster", func() {
+				BeforeEach(func() {
+					p.Spec.Pravega.SegmentStoreExternalTrafficPolicy = "cluster"
+				})
+				It("should create external service with ExternalTrafficPolicy type cluster", func() {
+					svc := pravega.MakeSegmentStoreExternalServices(p)
+					Ω(svc[0].Spec.ExternalTrafficPolicy).To(Equal(corev1.ServiceExternalTrafficPolicyTypeCluster))
+				})
+			})
+			Context("Create External service with LoadBalancerIP", func() {
+				BeforeEach(func() {
+					p.Spec.Pravega.SegmentStoreLoadBalancerIP = "10.240.12.18"
+				})
+				It("should create external service with LoadBalancerIP", func() {
+					svc := pravega.MakeSegmentStoreExternalServices(p)
+					Ω(svc[0].Spec.LoadBalancerIP).To(Equal("10.240.12.18"))
+				})
+			})
+			Context("Create External service with external service type empty", func() {
+				BeforeEach(func() {
+					m := make(map[string]string)
+					p.Spec.Pravega.SegmentStoreServiceAnnotations = m
+					p.Spec.Pravega.SegmentStoreExternalServiceType = ""
+				})
+				It("should create the service with external access type ClusterIP", func() {
+					svc := pravega.MakeSegmentStoreExternalServices(p)
+					Ω(svc[0].Spec.Type).To(Equal(corev1.ServiceTypeClusterIP))
+					Ω(err).Should(BeNil())
+				})
+			})
+		})
+		Context("With CustomStorage as Tier2", func() {
+			var (
+				customReq *corev1.ResourceRequirements
+				err       error
+			)
+			BeforeEach(func() {
+				annotationsMap := map[string]string{
+					"service.beta.kubernetes.io/aws-load-balancer-type": "nlb",
+				}
+				customReq = &corev1.ResourceRequirements{
+					Requests: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("2"),
+						corev1.ResourceMemory: resource.MustParse("4Gi"),
+					},
+					Limits: corev1.ResourceList{
+						corev1.ResourceCPU:    resource.MustParse("4"),
+						corev1.ResourceMemory: resource.MustParse("6Gi"),
+					},
+				}
+				p.Spec = v1beta1.ClusterSpec{
+					Version: "0.5.0",
+					ExternalAccess: &v1beta1.ExternalAccess{
+						Enabled:    true,
+						Type:       corev1.ServiceTypeClusterIP,
+						DomainName: "pravega.com.",
+					},
+					BookkeeperUri: v1beta1.DefaultBookkeeperUri,
+					Pravega: &v1beta1.PravegaSpec{
+						ControllerReplicas:              2,
+						SegmentStoreReplicas:            4,
+						ControllerServiceAccountName:    "pravega-components",
+						SegmentStoreServiceAccountName:  "pravega-components",
+						ControllerResources:             customReq,
+						SegmentStoreResources:           customReq,
+						ControllerServiceAnnotations:    annotationsMap,
+						ControllerPodLabels:             annotationsMap,
+						SegmentStoreServiceAnnotations:  annotationsMap,
+						SegmentStorePodLabels:           annotationsMap,
+						SegmentStoreExternalServiceType: corev1.ServiceTypeLoadBalancer,
+						SegmentStoreSecret: &v1beta1.SegmentStoreSecret{
+							Secret:    "seg-secret",
+							MountPath: "/tmp/mount",
+						},
+						Image: &v1beta1.ImageSpec{
+							Repository: "bar/pravega",
+						},
+						ControllerJvmOptions:   []string{"-XX:MaxDirectMemorySize=1g", "-XX:MaxRAMPercentage=50.0"},
+						SegmentStoreJVMOptions: []string{"-XX:MaxDirectMemorySize=1g", "-XX:MaxRAMPercentage=50.0"},
+						Options: map[string]string{
+							"dummy-key":             "dummy-value",
+							"configMapVolumeMounts": "prvg-logback:logback.xml=/opt/pravega/conf/logback.xml",
+							"emptyDirVolumeMounts":  "heap-dump=/tmp/dumpfile/heap,log=/opt/pravega/logs",
+							"hostPathVolumeMounts":  "heap-dump=/tmp/dumpfile/heap,log=/opt/pravega/logs",
+						},
+						LongTermStorage: &v1beta1.LongTermStorageSpec{
+							Custom: &v1beta1.CustomSpec{
+								Options: map[string]string{
+									"key": "dummy-value",
+								},
+								Env: map[string]string{
+									"AWS_KEY": "key",
+								},
+							},
+						},
+					},
+					TLS: &v1beta1.TLSPolicy{
+						Static: &v1beta1.StaticTLS{
+							ControllerSecret:   "controller-secret",
+							SegmentStoreSecret: "segmentstore-secret",
+							CaBundle:           "ecs-cert",
+						},
+					},
+					Authentication: &v1beta1.AuthenticationParameters{
+						Enabled:            true,
+						PasswordAuthSecret: "authentication-secret",
+					},
+				}
+				p.WithDefaults()
+			})
+
+			Context("First reconcile", func() {
+				It("shouldn't error", func() {
+					Ω(err).Should(BeNil())
+				})
+			})
+
 			Context("SegmentStore", func() {
 
 				It("should create a headless service", func() {

test/e2e/resources/crd.yaml

@@ -2212,6 +2212,20 @@ spec:
                       that a PersistentVolumeClaim called "pravega-longterm" is present
                       and it will use it as Tier 2
                     properties:
+                      custom:
+                        description: Custom Storage as a Tier2 backend
+                        properties:
+                          env:
+                            additionalProperties:
+                              type: string
+                            type: object
+                          options:
+                            additionalProperties:
+                              type: string
+                            type: object
+                        required:
+                        - env
+                        type: object
                       ecs:
                         description: Ecs is used to configure a Dell EMC ECS system
                           as a Tier 2 backend