add Azure storage support

Author: ggtrd

Dockerfile

@@ -5,9 +5,6 @@ COPY LICENSE.md /app
 COPY requirements.txt /app
 COPY src /app/src
 
-# RUN pip freeze > requirements.txt \
-#  && pip install -r requirements.txt
-
 RUN pip install -r requirements.txt
 
 CMD ["python", "-m", "src.main", "--job"]

README.md

@@ -88,29 +88,10 @@ just run the image
 ```
 clustersnap/
 ├── src/
-│   ├── main.py               # CLI entrypoint (--mode export | import)
-│   ├── config.py             # Pydantic v2 environment validation
-│   ├── storage/
-│   │   ├── base.py           # Abstract BaseStorageProvider
-│   │   ├── s3.py             # AWS S3 / MinIO / SeaweedFS
-│   │   └── azure.py          # Azure Blob Storage (shared-key + SP)
-│   ├── backup/
-│   │   ├── exporter.py       # Zip → Upload → Cleanup
-│   │   └── importer.py       # Download → Validate → Extract
-│   └── telemetry/
-│       ├── logger.py         # JSON formatter for Grafana Loki
-│       └── metrics.py        # Prometheus counters, histograms, gauges
-├── deploy/
-│   └── clustersnap/          # Helm chart
-│       ├── Chart.yaml
-│       ├── values.yaml
-│       └── templates/
-│           ├── configmap.yaml
-│           ├── secret.yaml
-│           └── cronjob.yaml
-├── Dockerfile
-├── requirements.txt
-└── README.md
+│   ├── main.py               # Entrypoint for CLI & Kubernetes main pod
+│   ├── worker.py             # Entrypoint for Kubernetes job pods
+│   └── helpers/
+│       ├── kubernetes.py     # Kubernetes class
 ```
 
 ### Workflow of export

src/helpers/kubernetes.py

@@ -134,7 +134,7 @@ def check_secret_exists(self, secret: str) -> bool:
             self.client_corev1.read_namespaced_secret(name=secret, namespace=current_ns)
             return True
         except Exception as e:
-            if e.status == 404:
+            if hasattr(e, 'status') and e.status == 404:
                 return False
             print(f"error: {e}")
             return False
@@ -156,16 +156,6 @@ def create_backup_job(self, pvc_name: str, mounted_path: str) -> bool:
         print(f"starting backup job: {job_name}")
 
 
-        # Remove existing jobs
-        try:
-            batch_api = client.BatchV1Api(self.client_corev1.api_client)
-            batch_api.delete_namespaced_job(name=job_name, namespace=current_ns, propagation_policy="Background")
-            time.sleep(2)
-        except ApiException as e:
-            if e.status != 404: # detect other errors that "not found"
-                print(f"error: {e}")
-
-
         # Apply YAML template file
         project_root = Path(__file__).resolve().parents[2]
         template_path = f"{project_root}/templates/backup.job.yaml"
@@ -203,7 +193,7 @@ def delete_backup_job(self, pvc_name: str) -> bool:
         try:
             batch_api = client.BatchV1Api(self.client_corev1.api_client)
             batch_api.delete_namespaced_job(name=job_name, namespace=current_ns, propagation_policy="Background")
-            print(f"  🧹 Old job '{job_name}' successfully removed.")
+            print(f"  Old job '{job_name}' successfully removed.")
             return True
         except ApiException as e:
             if e.status == 404:
@@ -219,44 +209,39 @@ def wait_for_jobs_deletion(self, pvc_names: list, timeout: int = 60) -> bool:
         current_ns = self.get_namespace()
         batch_api = client.BatchV1Api(self.client_corev1.api_client)
 
-        # On construit la liste des noms de jobs qu'on attend de voir disparaître
         jobs_to_wait = [f"backup-{pvc}"[-63:].lower().strip("-") for pvc in pvc_names]
-        
-        print(f"  ⏳ Waiting for {len(jobs_to_wait)} old jobs to be fully purged from cluster...", end="", flush=True)
+        print(f"  Waiting for {len(jobs_to_wait)} old jobs to be fully purged from cluster...", end="", flush=True)
 
         start_time = time.time()
         while time.time() - start_time < timeout:
             active_jobs = []
 
             for job_name in jobs_to_wait:
                 try:
-                    # On tente de lire le job
                     batch_api.read_namespaced_job(name=job_name, namespace=current_ns)
-                    # Si on arrive ici, c'est que le job existe encore sur le cluster
                     active_jobs.append(job_name)
                 except ApiException as e:
                     if e.status == 404:
-                        # Le job est bien supprimé, on ne fait rien
                         continue
                     else:
-                        print(f"\n  ❌ Error checking job {job_name}: {e}")
+                        print(f"\n  error: checking job {job_name}: {e}")
                         return False
 
-            # Si la liste des jobs encore actifs est vide, c'est qu'ils ont tous disparu !
             if not active_jobs:
-                print(" Done! 🟢")
+                print(" Done!")
                 return True
 
             print(".", end="", flush=True)
-            time.sleep(2) # On réinterroge toutes les 2 secondes
+            time.sleep(2)
 
-        print("\n  ❌ Timeout reached while waiting for old jobs deletion.")
+        print("\n  error: timeout reached while waiting for old jobs deletion.")
         return False
 
 
-
-    # Get replicas quantity of a Statefulset or Deployment
     def get_replicas(self, resource_type: str, name: str) -> int:
+        """
+        Get replicas quantity of a Statefulset or Deployment
+        """
         current_ns = self.get_namespace()
         try:
             if resource_type.lower() == "statefulset":
@@ -273,8 +258,10 @@ def get_replicas(self, resource_type: str, name: str) -> int:
             return 0
 
 
-    # Scale replicas quantity of a Statefulset or Deployment
     def scale_replicas(self, resource_type: str, name: str, replicas: int) -> bool:
+        """
+        Scale replicas quantity of a Statefulset or Deployment
+        """
         current_ns = self.get_namespace()
         body = {"spec": {"replicas": replicas}}
 
@@ -293,8 +280,10 @@ def scale_replicas(self, resource_type: str, name: str, replicas: int) -> bool:
             return False
 
 
-    # Hook to ensure replicas are scaled to 0
     def wait_for_scale_down(self, resource_type: str, name: str, timeout: int = 120) -> bool:
+        """
+        Hook to ensure replicas are scaled to 0
+        """
         current_ns = self.get_namespace()
         resource = f"{resource_type}/{name}"
 
@@ -323,4 +312,4 @@ def wait_for_scale_down(self, resource_type: str, name: str, timeout: int = 120)
                 return False
 
         print(f"error: timeout reached while waiting for scale down of {resource}")
-        return False
+        return False

src/main.py

@@ -4,8 +4,10 @@
 from src.helpers.kubernetes import Kubernetes
 
 
-# Get a YAML map of what will be backuped
 def get_backup_map(kubernetes: Kubernetes):
+    """
+    Get a YAML map of what will be backuped
+    """
     current_ns = kubernetes.get_namespace()
 
     print(f"context: {kubernetes.get_context()}")
@@ -17,7 +19,6 @@ def get_backup_map(kubernetes: Kubernetes):
         print("error: no PVC found or unauthorized access.")
         return 0
 
-    # Loop to display the backup map
     print(f"  claims:")
     for pvc in kubernetes.list_pvc():
         print(f"    - name: {pvc['name']}")
@@ -36,101 +37,59 @@ def get_backup_map(kubernetes: Kubernetes):
             print(f"      mounts: {mounts}")
 
 
-# Run a backup job
 def run_backup_job(kubernetes: Kubernetes):
-    # Display the map before running the job
-    get_backup_map(kubernetes)
-
-    # for pvc in kubernetes.list_pvc():
-    #     pvc_name = pvc['name']
-
-    #     targets = kubernetes.get_pvc(pvc_name)
-    #     if not targets:
-    #         print(f"info: no active workload config found for {pvc_name}. Skipping.")
-    #         continue
-
-    #     target = targets[0]
-    #     mounted_path = target['mount_path']
-    #     resource = target['resource']
-    #     resource_type, resource_name = resource.split('/')
-    #     resource_initial_replicas = kubernetes.get_replicas(resource_type, resource_name)
-
-    #     print(f"replicas quantity found for {resource}: {resource_initial_replicas}")
-
-    #     is_running = resource_initial_replicas > 0
-    #     if is_running:
-    #         kubernetes.scale_replicas(resource_type, resource_name, replicas=0)
-    #         if not kubernetes.wait_for_scale_down(resource_type, resource_name):
-    #             print(f"error: scale down fail for {resource}. Aborting backup of {pvc_name}")
-    #             continue 
-    #     else:
-    #         print(f"info: {resource} is already scaled down. Safe to proceed.")
-
-    #     job_success = kubernetes.create_backup_job(pvc=pvc_name, mounted_path=mounted_path)
-    #     if is_running:
-    #         kubernetes.scale_replicas(resource_type, resource_name, replicas=resource_initial_replicas)
-    #         print(f"restored replicas of {resource} {resource_initial_replicas}")
-
-    #     print('')
-
-
     """
-    Orchestrateur séquentiel global :
-    Phase 0: Nettoyage + Attente de suppression des vieux Jobs
-    Phase 1: Collecte des infos + Ordre de Scale Down (0)
-    Phase 2: Attente de l'extinction complète de la prod
-    Phase 3: Lancement de tous les nouveaux Jobs en parallèle via K8s
-    Phase 4: Restauration (Scale Up) de la prod à son état initial
+    Global sequential orchestrator:
+    Phase 0: Cleanup + wait for deletion of old Jobs
+    Phase 1: Gather configuration + Trigger Scale Down (0)
+    Phase 2: Wait for complete production workload termination
+    Phase 3: Trigger all new backup Jobs concurrently via K8s control plane
+    Phase 4: Restore (Scale Up) production workloads to original layout states
     """
-    # Récupération de la liste des PVCs du namespace
+    get_backup_map(kubernetes)
+
     pvcs = kubernetes.list_pvc()
     if not pvcs:
         print("error: no PVC found to backup.")
         return
 
     print(f"\n==================================================")
-    print(f"⚡ Starting Sequential Backup Orchestration for {len(pvcs)} PVCs")
+    print(f"Starting Sequential Backup Orchestration for {len(pvcs)} PVCs")
     print(f"==================================================")
 
     # ==================================================
-    # PHASE 0 : NETTOYAGE GLOBAL DES ANCIENS JOBS
+    # PHASE 0 : GLOBAL JOB CLEANUP
     # ==================================================
-    print("\n🧹 PHASE 0: Purging any existing/stale backup jobs...")
+    print("\nPHASE 0: Purging any existing/stale backup jobs...")
     pvc_names = []
 
     for pvc in pvcs:
         pvc_name = pvc['name']
         pvc_names.append(pvc_name)
-        # Envoie l'ordre de suppression à l'API (non-bloquant)
         kubernetes.delete_backup_job(pvc_name=pvc_name)
 
-    # Validation dynamique de la disparition des anciens Jobs
     if not kubernetes.wait_for_jobs_deletion(pvc_names=pvc_names, timeout=45):
-        print("  ⚠️ Warning: Some old jobs might still be terminating. Proceeding anyway...")
+        print("  warn: Some old jobs might still be terminating. Proceeding anyway...")
 
-    # Structure pour mémoriser l'état initial de chaque workload
     backup_workloads = {}
 
     # ==================================================
-    # PHASE 1 : ANALYSE ET ENVOI DU SCALE DOWN (0)
+    # PHASE 1 : WORKLOAD ASSESSMENT AND SCALE DOWN
     # ==================================================
-    print("\n🛑 PHASE 1: Capturing states and scaling down workloads...")
+    print("\nPHASE 1: Capturing states and scaling down workloads...")
     for pvc in pvcs:
         pvc_name = pvc['name']
         targets = kubernetes.get_pvc(pvc_name)
         if not targets:
-            print(f"  ⚠️ No active workload config found for {pvc_name}. Skipping.")
+            print(f"  warn: No active workload config found for {pvc_name}. Skipping.")
             continue
 
-        # On prend la première configuration trouvée liée au PVC
         target = targets[0]
-        resource_key = target['resource'] # Ex: "StatefulSet/postgresql"
+        resource_key = target['resource']
         resource_type, resource_name = resource_key.split('/')
         mounted_path = target['mount_path']
 
-        # Évite de traiter deux fois le même StatefulSet si celui-ci a plusieurs PVCs
         if resource_key not in backup_workloads:
-            # On lit et on stocke le nombre de répliques en cours (ex: 1, 2, 3...)
             initial_replicas = kubernetes.get_replicas(resource_type, resource_name)
 
             backup_workloads[resource_key] = {
@@ -141,56 +100,49 @@ def run_backup_job(kubernetes: Kubernetes):
                 "mounted_path": mounted_path
             }
 
-            # Si l'application tourne, on demande à K8s de l'éteindre
             if initial_replicas > 0:
                 kubernetes.scale_replicas(resource_type, resource_name, replicas=0)
             else:
-                print(f"  🌙 {resource_key} is already stopped (Night mode).")
+                print(f"  info: {resource_key} is already stopped (Night mode).")
 
     # ==================================================
-    # PHASE 2 : ATTENTE DE L'EXTINCTION GLOBALE
+    # PHASE 2 : WAIT FOR COMPLETE EXCLUSIVITY
     # ==================================================
-    print("\n⏳ PHASE 2: Waiting for all workloads to be fully stopped...")
+    print("\nPHASE 2: Waiting for all workloads to be fully stopped...")
     for resource_key, info in backup_workloads.items():
         if info["initial_replicas"] > 0:
-            # Bloque le script tant que les conteneurs de l'application ne sont pas à 0
             kubernetes.wait_for_scale_down(info["resource_type"], info["resource_name"])
 
     # ==================================================
-    # PHASE 3 : LANCEMENT DE TOUS LES BACKUPS D'UN COUP
+    # PHASE 3 : TRIGGER CONCURRENT BACKUPS
     # ==================================================
-    print("\n🚀 PHASE 3: Triggering all Kubernetes backup jobs simultaneously...")
+    print("\nPHASE 3: Triggering all Kubernetes backup jobs simultaneously...")
     for resource_key, info in backup_workloads.items():
-        # K8s instancie les Jobs en tâche de fond (le traitement disque se fait en parallèle sur le cluster)
         kubernetes.create_backup_job(pvc_name=info["pvc_name"], mounted_path=info["mounted_path"])
 
-    # Temporisation de sécurité pour laisser le temps aux Pods de backup de démarrer,
-    # d'accrocher le volume et de poser leurs verrous de lecture.
-    print("  🛌 Allowing a 5s safety buffer for backup containers to initialize...")
+    print("  info: Allowing a 5s safety buffer for backup containers to initialize...")
     time.sleep(5)
 
     # ==================================================
-    # PHASE 4 : RESTAURATION (SCALE UP) DE LA PRODUCTION
+    # PHASE 4 : PRODUCTION WORKLOAD RESTORATION
     # ==================================================
-    print("\n🔄 PHASE 4: Restoring production workloads to initial states...")
+    print("\nPHASE 4: Restoring production workloads to initial states...")
     for resource_key, info in backup_workloads.items():
         if info["initial_replicas"] > 0:
-            # On remet l'application exactement dans son état de départ (ex: replicas=1)
             kubernetes.scale_replicas(
                 info["resource_type"], 
                 info["resource_name"], 
                 replicas=info["initial_replicas"]
             )
-            print(f"  ✅ {resource_key} restored to {info['initial_replicas']} replicas.")
+            print(f"  info: {resource_key} restored to {info['initial_replicas']} replicas.")
 
     print(f"\n==================================================")
-    print(f"🏁 Sequential orchestration process finished successfully.")
+    print(f"Sequential orchestration process finished successfully.")
     print(f"==================================================")
 
 
 def main():
     parser = argparse.ArgumentParser(description="Backup Kubernetes volumes")
-    parser.add_index = False
     parser.add_argument("-c", "--context", type=str, default=None, help="Local kubectl context")
     parser.add_argument("-n", "--namespace", type=str, default=None, help="Kubernetes namespace")
     parser.add_argument("-j", "--job", action="store_true", help="Run a backup job")
@@ -210,4 +162,4 @@ def main():
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())