[Console] Make changes for filesystem snapshots

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/console_link/environment.py

@@ -1,17 +1,24 @@
 import logging
-from typing import Optional
+from typing import Optional, Dict
 from console_link.models.cluster import Cluster
 from console_link.models.metrics_source import MetricsSource
 from console_link.logic.metrics import get_metrics_source
 from console_link.logic.backfill import get_backfill
 from console_link.models.backfill_base import Backfill
-from console_link.models.snapshot import Snapshot, S3Snapshot
+from console_link.models.snapshot import FileSystemSnapshot, Snapshot, S3Snapshot
 import yaml
 from cerberus import Validator
 
 
 logger = logging.getLogger(__name__)
 
+
+def get_snapshot(config: Dict, source_cluster: Cluster, target_cluster: Cluster):
+    if 'fs' in config:
+        return FileSystemSnapshot(config, source_cluster, target_cluster)
+    return S3Snapshot(config, source_cluster, target_cluster)
+
+
 SCHEMA = {
     "source_cluster": {"type": "dict", "required": False},
     "target_cluster": {"type": "dict", "required": True},
@@ -66,9 +73,9 @@ def __init__(self, config_file: str):
             logger.info("No backfill provided")
 
         if 'snapshot' in self.config:
-            self.snapshot: Snapshot = S3Snapshot(self.config["snapshot"],
-                                                 source_cluster=self.source_cluster,
-                                                 target_cluster=self.target_cluster)
+            self.snapshot: Snapshot = get_snapshot(self.config["snapshot"],
+                                                   source_cluster=self.source_cluster,
+                                                   target_cluster=self.target_cluster)
             logger.info(f"Snapshot initialized: {self.snapshot}")
         else:
             logger.info("No snapshot provided")

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/console_link/models/snapshot.py

@@ -3,10 +3,12 @@
 import logging
 import subprocess
 from typing import Dict, Optional
-from console_link.models.cluster import Cluster
+from console_link.models.cluster import AuthMethod, Cluster
 from console_link.models.command_result import CommandResult
 from cerberus import Validator
 
+from console_link.models.schema_tools import contains_one_of
+
 logger = logging.getLogger(__name__)
 
 SnapshotStatus = Enum(
@@ -17,6 +19,30 @@
         "FAILED"])
 
 
+SNAPSHOT_SCHEMA = {
+    'snapshot': {
+        'type': 'dict',
+        'schema': {
+            'snapshot_name': {'type': 'string', 'required': True},
+            's3': {
+                'type': 'dict',
+                'schema': {
+                    'repo_uri': {'type': 'string', 'required': True},
+                    'aws_region': {'type': 'string', 'required': True},
+                }
+            },
+            'fs': {
+                'type': 'dict',
+                'schema': {
+                    'repo_path': {'type': 'string', 'required': True},
+                }
+            }
+        },
+        'check_with': contains_one_of({'s3', 'fs'})
+    }
+}
+
+
 class Snapshot(ABC):
     """
     Interface for creating and managing snapshots.
@@ -25,6 +51,9 @@
         self.config = config
         self.source_cluster = source_cluster
         self.target_cluster = target_cluster
+        v = Validator(SNAPSHOT_SCHEMA)
+        if not v.validate({'snapshot': config}):
+            raise ValueError("Invalid config file for snapshot", v.errors)
 
     @abstractmethod
     def create(self, *args, **kwargs) -> CommandResult:
@@ -37,40 +66,69 @@
         pass
 
 
-S3_SNAPSHOT_SCHEMA = {
-    'snapshot_name': {
-        'type': 'string',
-        'required': True
-    },
-    's3_repo_uri': {
-        'type': 'string',
-        'required': True
-    },
-    's3_region': {
-        'type': 'string',
-        'required': True
-    }
-}
-
-
 class S3Snapshot(Snapshot):
-    def __init__(self, config: Dict, source_cluster: Cluster, target_cluster: Optional[Cluster] = None) -> None:
+    def __init__(self, config: Dict, source_cluster: Cluster, target_cluster: Cluster) -> None:
         super().__init__(config, source_cluster, target_cluster)
-        v = Validator(S3_SNAPSHOT_SCHEMA)
-        if not v.validate(config):
-            raise ValueError("Invalid config file for snapshot", v.errors)
+
         self.snapshot_name = config['snapshot_name']
-        self.s3_repo_uri = config['s3_repo_uri']
-        self.s3_region = config['s3_region']
+        self.s3_repo_uri = config['s3']['repo_uri']
+        self.s3_region = config['s3']['aws_region']
 
     def create(self, *args, **kwargs) -> CommandResult:
+        assert isinstance(self.target_cluster, Cluster)
+        if self.source_cluster.auth_type != AuthMethod.NO_AUTH:
+            raise NotImplementedError("Source cluster authentication is not supported for creating snapshots")
+
+        if self.target_cluster.auth_type != AuthMethod.NO_AUTH:
+            raise NotImplementedError("Target cluster authentication is not supported for creating snapshots")
+
         command = [
             "/root/createSnapshot/bin/CreateSnapshot",
             "--snapshot-name", self.snapshot_name,
             "--s3-repo-uri", self.s3_repo_uri,
             "--s3-region", self.s3_region,
             "--source-host", self.source_cluster.endpoint,
-            "--target-host", self.target_cluster.endpoint,
+        ]
+
+        if self.source_cluster.allow_insecure:
+            command.append("--source-insecure")
+        if self.target_cluster.allow_insecure:
+            command.append("--target-insecure")
+
+        logger.info(f"Creating snapshot with command: {' '.join(command)}")
+        try:
+            # Pass None to stdout and stderr to not capture output and show in terminal
+            subprocess.run(command, stdout=None, stderr=None, text=True, check=True)
+            logger.info(f"Snapshot {self.config['snapshot_name']} created successfully")
+            return CommandResult(success=True, value=f"Snapshot {self.config['snapshot_name']} created successfully")
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Failed to create snapshot: {str(e)}")
+            return CommandResult(success=False, value=f"Failed to create snapshot: {str(e)}")
+
+    def status(self, *args, **kwargs) -> CommandResult:
+        return CommandResult(success=False, value="Command not implemented")
+
+
+class FileSystemSnapshot(Snapshot):
+    def __init__(self, config: Dict, source_cluster: Cluster, target_cluster: Cluster) -> None:
+        super().__init__(config, source_cluster, target_cluster)
+        self.snapshot_name = config['snapshot_name']
+        self.repo_path = config['fs']['repo_path']
+
+    def create(self, *args, **kwargs) -> CommandResult:
+        assert isinstance(self.target_cluster, Cluster)
+
+        if self.source_cluster.auth_type != AuthMethod.NO_AUTH:
+            raise NotImplementedError("Source cluster authentication is not supported for creating snapshots")
+
+        if self.target_cluster.auth_type != AuthMethod.NO_AUTH:
+            raise NotImplementedError("Target cluster authentication is not supported for creating snapshots")
+
+        command = [
+            "/root/createSnapshot/bin/CreateSnapshot",
+            "--snapshot-name", self.snapshot_name,
+            "--file-system-repo-path", self.repo_path,
+            "--source-host", self.source_cluster.endpoint,
         ]
 
         if self.source_cluster.allow_insecure:

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/services.yaml

@@ -2,21 +2,21 @@ source_cluster:
   endpoint: "https://capture-proxy:9200"
   allow_insecure: true
   basic_auth:
-      username: "admin"
-      password: "admin"
+    username: "admin"
+    password: "admin"
 target_cluster:
   endpoint: "https://opensearchtarget:9200"
   allow_insecure: true
   basic_auth:
-      username: "admin"
-      password: "myStrongPassword123!"
+    username: "admin"
+    password: "myStrongPassword123!"
 metrics_source:
   prometheus:
     endpoint: "http://prometheus:9090"
 backfill:
-    reindex_from_snapshot:
-        docker:
+  reindex_from_snapshot:
+    docker:
 snapshot:
-  snapshot_name: "snapshot_2023_01_01"
-  s3_repo_uri: "s3://my-snapshot-bucket"
-  s3_region: "us-east-2"
+  snapshot_name: "snapshot_2024_06_21"
+  fs:
+    repo_path: "/snapshot/test-console"

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/tests/data/services.yaml

@@ -2,21 +2,22 @@ source_cluster:
   endpoint: "https://capture-proxy-es:9200"
   allow_insecure: true
   basic_auth:
-      username: "admin"
-      password: "admin"
+    username: "admin"
+    password: "admin"
 target_cluster:
   endpoint: "https://opensearchtarget:9200"
   allow_insecure: true
   basic_auth:
-      username: "admin"
-      password: "myStrongPassword123!"
+    username: "admin"
+    password: "myStrongPassword123!"
 metrics_source:
   prometheus:
     endpoint: "http://prometheus:9090"
 backfill:
-    reindex_from_snapshot:
-        docker:
+  reindex_from_snapshot:
+    docker:
 snapshot:
-  snapshot_name: "test_snapshot"
-  s3_repo_uri: "s3://test-bucket"
-  s3_region: "us-east-2"
+  snapshot_name: "snapshot_2023_01_01"
+  s3:
+    repo_uri: "s3://my-snapshot-bucket"
+    aws_region: "us-east-2"