Merge pull request #52 from kumarks1122/release-5.0.1-csp-checkpointing

LR-269 | CSP changes for Checkpointing added

ansible/inventory/env/group_vars/all.yml

@@ -85,8 +85,8 @@ search_es7_host: "{{ groups['es7']|join(':9200,')}}:9200"
 
 mlworkbench: "{{ groups['mlworkbench'][0]}}"
 
-cloud_account: "{{ sunbird_public_storage_account_name }}"
-cloud_secret: "{{ sunbird_public_storage_account_key }}"
+cloud_account: "{{ cloud_public_storage_accountname }}"
+cloud_secret: "{{ cloud_public_storage_accountname }}"
 dedup_redis_host: "{{ dp_redis_host }}"
 kp_redis_host: "{{ groups['redisall'][0] }}"
 neo4j_route_path: "bolt://{{ groups['learning-neo4j-node1'][0] }}:7687"
@@ -108,10 +108,10 @@ cert_rc_base_url: "http://{{private_ingressgateway_ip}}/registry-service/api/v1"
 cert_rc_entity: "TrainingCertificate"
 certificate_base_path: "https://{{domain_name}}/certs"
 cert_domain_url:  "{{proto}}://{{domain_name}}"
-cert_container_name: "{{cert_env_prefix}}-e-credentials"
+cert_container_name: "{{ cloud_storage_certservice_bucketname }}"
 cert_cloud_storage_type: "{{cloud_service_provider}}"
-cert_cloud_storage_secret: "{{sunbird_private_storage_account_key}}"
-cert_cloud_storage_key: "{{sunbird_private_storage_account_name}}"
+cert_cloud_storage_secret: "{{cloud_private_storage_secret}}"
+cert_cloud_storage_key: "{{cloud_private_storage_accountname}}"
 cloud_storage_base_url: "{{cloud_storage_base_url}}"
 
 default_channel: "org.sunbird"
@@ -151,4 +151,10 @@ kafka_group_collection_pre_processor: "{{env_name}}-collection-cert-pre-processo
 kafka_group_certificate_generator: "{{env_name}}-certificate-generator-group"
 kafka_group_merge_courses: "{{env_name}}-merge-courses-group"
 kafka_group_assessment_aggregator: "{{env_name}}-assessment-aggregator-group"
-kafka_group_lms_notification: "{{env_name}}-lms-notification"
\ No newline at end of file
+kafka_group_lms_notification: "{{env_name}}-lms-notification"
+
+## Checkpoint Config
+checkpoint_store_type: "{{cloud_service_provider}}"
+azure_cloud_account: "{{ cloud_public_storage_accountname }}"
+azure_cloud_secret: "{{ cloud_public_storage_secret }}"
+cloud_storage_flink_bucketname: "flink-check-points-store"
\ No newline at end of file

kubernetes/ansible/roles/flink-jobs-deploy/defaults/main.yml

@@ -29,7 +29,7 @@ taskmanager_heap_memory: 1024
 
 ### base-config related vars
 postgres_max_connections: 2
-flink_container_name: flink-check-points-store
+flink_container_name: "{{ cloud_storage_flink_bucketname }}"
 checkpoint_interval: 60000
 checkpoint_pause_between_seconds: 5000
 checkpoint_compression_enabled: true

kubernetes/helm_charts/datapipeline_jobs/templates/flink_job_deployment.yaml

@@ -109,8 +109,21 @@ spec:
         workingDir: /opt/flink
         command: ["/opt/flink/bin/standalone-job.sh"]
         args: ["start-foreground",
-               "--job-classname={{ .Values.job_classname }}", 
-               "-Dfs.azure.account.key.{{ .Values.cloud_account }}.blob.core.windows.net={{ .Values.cloud_secret }}",
+               "--job-classname={{ .Values.job_classname }}",
+{{- if eq .Values.checkpoint_store_type "azure" }}
+               "-Dfs.azure.account.key.{{ .Values.azure_cloud_account }}.blob.core.windows.net={{ .Values.azure_cloud_secret }}",
+{{- end }}
+{{- if eq .Values.checkpoint_store_type "s3" }}
+               "-Ds3.access-key={{ .Values.s3_access_key }}",
+               "-Ds3.secret-key={{ .Values.s3_secret_key }}",
+               "-Ds3.endpoint={{ .Values.s3_endpoint }}",
+               "-Ds3.path.style.access={{ .Values.s3_path_style_access }}",
+{{- end }}
+{{- if eq .Values.checkpoint_store_type "gcloud" }}
+               "-Dfs.gs.auth.client.id={{ .Values.gcloud_client_key }}",
+               "-Dfs.gs.auth.client.secret={{ .Values.gcloud_private_secret }}",
+               "-Dfs.gs.project.id={{ .Values.gcloud_project_id }}"
+{{- end }}
                "-Dweb.submit.enable=false",
                "-Dmetrics.reporter.prom.class=org.apache.flink.metrics.prometheus.PrometheusReporter",
                "-Dmetrics.reporter.prom.port={{ .Values.jobmanager.prom_port }}",
@@ -183,7 +196,20 @@ spec:
         workingDir: {{ .Values.taskmanager.flink_work_dir }}
         command: ["/opt/flink/bin/taskmanager.sh"]
         args: ["start-foreground",
-          "-Dfs.azure.account.key.{{ .Values.cloud_account }}.blob.core.windows.net={{ .Values.cloud_secret }}",
+{{- if eq .Values.checkpoint_store_type "azure" }}
+          "-Dfs.azure.account.key.{{ .Values.azure_cloud_account }}.blob.core.windows.net={{ .Values.azure_cloud_secret }}",
+{{- end }}
+{{- if eq .Values.checkpoint_store_type "s3" }}
+          "-Ds3.access-key={{ .Values.s3_access_key }}",
+          "-Ds3.secret-key={{ .Values.s3_secret_key }}",
+          "-Ds3.endpoint={{ .Values.s3_endpoint }}",
+          "-Ds3.path.style.access={{ .Values.s3_path_style_access }}",
+{{- end }}
+{{- if eq .Values.checkpoint_store_type "gcloud" }}
+          "-Dfs.gs.auth.client.id={{ .Values.gcloud_client_key }}",
+          "-Dfs.gs.auth.client.secret={{ .Values.gcloud_private_secret }}",
+          "-Dfs.gs.project.id={{ .Values.gcloud_project_id }}",
+{{- end }}
           "-Dweb.submit.enable=false",
           "-Dmetrics.reporter.prom.class=org.apache.flink.metrics.prometheus.PrometheusReporter",
           "-Dmetrics.reporter.prom.host={{ .Release.Name }}-taskmanager",

kubernetes/helm_charts/datapipeline_jobs/values.j2

@@ -3,8 +3,16 @@ imagepullsecrets: {{ imagepullsecrets }}
 dockerhub: {{ dockerhub }}
 repository: {{flink_repository|default('data-pipeline')}}
 image_tag: {{ image_tag }}
-cloud_account: {{ cloud_account }}
-cloud_secret: {{ cloud_secret }}
+checkpoint_store_type: {{ checkpoint_store_type }}
+azure_cloud_account: {{ azure_cloud_account }}
+azure_cloud_secret: {{ azure_cloud_secret }}
+s3_access_key: {{ s3_storage_key }}
+s3_secret_key: {{ s3_storage_secret }}
+s3_endpoint: {{ s3_storage_endpoint }}
+s3_path_style_access: {{ s3_path_style_access }}
+gcloud_client_key: {{ gcloud_client_key }}
+gcloud_private_secret: {{ gcloud_private_secret }}
+gcloud_project_id: {{ gcloud_project_id }}
 serviceMonitor:
   enabled: {{ service_monitor_enabled | lower}}
 
@@ -88,8 +96,8 @@ base_config: |
       statebackend {
         blob {
           storage {
-            account = "{{ cloud_account }}.blob.core.windows.net"
-            container = "{{ flink_container_name }}"
+            account = "{{ azure_cloud_account }}.blob.core.windows.net"
+            container = "{{ cloud_storage_flink_bucketname }}"
             checkpointing.dir = "checkpoint"
           }
         }