[Console] Implement start/stop for RFS backfill on ECS

TrafficCapture/dockerSolution/src/main/docker/migrationConsole/lib/console_link/console_link/logic/backfill.py

@@ -75,8 +75,8 @@
         return ExitCode.FAILURE, f"Failure when creating backfill: {type(e).__name__} {e}"
 
     if result.success:
-        return ExitCode.SUCCESS, "Backfill created successfully." + "\n" + result.value
-    return ExitCode.FAILURE, "Backfill creation failed." + "\n" + result.value
+        return ExitCode.SUCCESS, "Backfill created successfully." + "\n" + result.display()
+    return ExitCode.FAILURE, "Backfill creation failed." + "\n" + result.display()
 
 
 def start(backfill: Backfill, *args, **kwargs) -> Tuple[ExitCode, str]:
@@ -90,8 +90,8 @@
         return ExitCode.FAILURE, f"Failure when starting backfill: {type(e).__name__} {e}"
 
     if result.success:
-        return ExitCode.SUCCESS, "Backfill started successfully." + "\n" + result.value
-    return ExitCode.FAILURE, "Backfill start failed." + "\n" + result.value
+        return ExitCode.SUCCESS, "Backfill started successfully." + "\n" + result.display()
+    return ExitCode.FAILURE, "Backfill start failed." + "\n" + result.display()
 
 
 def stop(backfill: Backfill, *args, **kwargs) -> Tuple[ExitCode, str]:
@@ -105,8 +105,8 @@
         logger.error(f"Failed to stop backfill: {e}")
         return ExitCode.FAILURE, f"Failure when stopping backfill: {type(e).__name__} {e}"
     if result.success:
-        return ExitCode.SUCCESS, "Backfill stopped successfully." + "\n" + result.value
-    return ExitCode.FAILURE, "Backfill stop failed." + "\n" + result.value
+        return ExitCode.SUCCESS, "Backfill stopped successfully." + "\n" + result.display()
+    return ExitCode.FAILURE, "Backfill stop failed." + "\n" + result.display()
 
 
 def scale(backfill: Backfill, units: int, *args, **kwargs) -> Tuple[ExitCode, str]:
@@ -120,20 +120,20 @@
         logger.error(f"Failed to scale backfill: {e}")
         return ExitCode.FAILURE, f"Failure when scaling backfill: {type(e).__name__} {e}"
     if result.success:
-        return ExitCode.SUCCESS, "Backfill scaled successfully." + "\n" + result.value
-    return ExitCode.FAILURE, "Backfill scale failed." + "\n" + result.value
+        return ExitCode.SUCCESS, "Backfill scaled successfully." + "\n" + result.display()
+    return ExitCode.FAILURE, "Backfill scale failed." + "\n" + result.display()
 
 
 def status(backfill: Backfill, *args, **kwargs) -> Tuple[ExitCode, str]:
     logger.info("Getting backfill status")
     try:
-        result = backfill.get_status(*args, **kwargs)
+        status = backfill.get_status(*args, **kwargs)
     except NotImplementedError:
         logger.error(f"Status is not implemented for backfill {type(backfill).__name__}")
         return ExitCode.FAILURE, f"Status is not implemented for backfill: {type(backfill).__name__}"
     except Exception as e:
         logger.error(f"Failed to get status of backfill: {e}")
         return ExitCode.FAILURE, f"Failure when getting status of backfill: {type(e).__name__} {e}"
-    if result.success:
-        return ExitCode.SUCCESS, result.value
-    return ExitCode.FAILURE, "Backfill status retrieval failed." + "\n" + result.value
+    if status:
+        return ExitCode.SUCCESS, status.value
+    return ExitCode.FAILURE, "Backfill status retrieval failed." + "\n" + status

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

@@ -3,10 +3,13 @@
 from console_link.models.cluster import Cluster
 from console_link.models.schema_tools import contains_one_of
 from console_link.models.command_result import CommandResult
+from console_link.models.ecs_service import ECSService
 
 from cerberus import Validator
-import boto3
 
+import logging
+
+logger = logging.getLogger(__name__)
 
 DOCKER_RFS_SCHEMA = {
     "type": "dict",
@@ -76,5 +79,16 @@ class ECSRFSBackfill(RFSBackfill):
     def __init__(self, config: Dict, target_cluster: Cluster) -> None:
         super().__init__(config)
         self.target_cluster = target_cluster
+        self.default_scale = self.config["reindex_from_snapshot"].get("scale", 1)
+
         self.ecs_config = self.config["reindex_from_snapshot"]["ecs"]
-        self.client = boto3.client("ecs", region_name=self.ecs_config.get("aws_region", None))
+        self.ecs_client = ECSService(self.ecs_config["cluster_name"], self.ecs_config["service_name"],
+                                     self.ecs_config.get("aws_region", None))
+
+    def start(self, *args, **kwargs) -> CommandResult:
+        logger.info(f"Starting RFS backfill by setting desired count to {self.default_scale} instances")
+        return self.ecs_client.set_desired_count(self.default_scale)
+
+    def stop(self, *args, **kwargs) -> CommandResult:
+        logger.info("Stopping RFS backfill by setting desired count to 0 instances")
+        return self.ecs_client.set_desired_count(0)

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

@@ -4,3 +4,8 @@
 class CommandResult(NamedTuple):
     success: bool
     value: Any
+
+    def display(self) -> str:
+        if self.value:
+            return str(self.value)
+        return ""

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

@@ -0,0 +1,40 @@
+import logging
+
+import boto3
+
+from console_link.models.command_result import CommandResult
+from console_link.models.utils import AWSAPIError, raise_for_aws_api_error
+
+
+logger = logging.getLogger(__name__)
+
+
+class ECSService:
+    def __init__(self, cluster_name, service_name, aws_region=None):
+        self.cluster_name = cluster_name
+        self.service_name = service_name
+        self.aws_region = aws_region
+
+        logger.info(f"Creating ECS client for region {aws_region}, if specified")
+        self.client = boto3.client("ecs", region_name=self.aws_region)
+
+    def set_desired_count(self, desired_count: int) -> CommandResult:
+        logger.info(f"Setting desired count for service {self.service_name} to {desired_count}")
+        response = self.client.update_service(
+            cluster=self.cluster_name,
+            service=self.service_name,
+            desiredCount=desired_count
+        )
+        logger.debug(f"Response from update_service: {response}")
+
+        try:
+            raise_for_aws_api_error(response)
+        except AWSAPIError as e:
+            return CommandResult(False, e)
+
+        logger.info(f"Service status: {response['service']['status']}")
+        running_count = response["service"]["runningCount"]
+        pending_count = response["service"]["pendingCount"]
+        desired_count = response["service"]["desiredCount"]
+        return CommandResult(True, f"Service {self.service_name} set to {desired_count} desired count."
+                             f" Currently {running_count} running and {pending_count} pending.")

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

@@ -5,7 +5,7 @@
 import boto3
 import botocore
 from cerberus import Validator
-from console_link.logic.utils import raise_for_aws_api_error
+from console_link.models.utils import raise_for_aws_api_error
 import requests
 import logging