Generate QHub Costs via infracost

qhub/cli/__init__.py

@@ -4,6 +4,7 @@
 
 from pydantic.error_wrappers import ValidationError
 
+from qhub.cli.cost import create_cost_subcommand
 from qhub.cli.deploy import create_deploy_subcommand
 from qhub.cli.destroy import create_destroy_subcommand
 from qhub.cli.initialize import create_init_subcommand
@@ -37,6 +38,7 @@ def cli(args):
     create_support_subcommand(subparser)
     create_upgrade_subcommand(subparser)
     create_keycloak_subcommand(subparser)
+    create_cost_subcommand(subparser)
 
     args = parser.parse_args(args)
 

qhub/cli/cost.py

@@ -0,0 +1,15 @@
+import logging
+
+from qhub.cost import infracost_report
+
+logger = logging.getLogger(__name__)
+
+
+def create_cost_subcommand(subparser):
+    subparser = subparser.add_parser("cost-estimate")
+    subparser.add_argument("-p", "--path", help="qhub stages path", required=False)
+    subparser.set_defaults(func=handle_cost_report)
+
+
+def handle_cost_report(args):
+    infracost_report(args.path)

qhub/cost.py

@@ -0,0 +1,112 @@
+import json
+import logging
+import os
+import re
+import subprocess
+
+from rich.console import Console
+from rich.table import Table
+
+logger = logging.getLogger(__name__)
+
+
+def _check_infracost():
+    """
+    Check if infracost is installed
+    """
+    try:
+        subprocess.check_output(["infracost", "--version"])
+        return True
+    except subprocess.CalledProcessError:
+        return False
+
+
+def _check_infracost_api_key():
+    """
+    Check if infracost API key is configured
+    """
+    try:
+        subprocess.check_output(["infracost", "configure", "get", "api_key"])
+        return True
+    except subprocess.CalledProcessError:
+        return False
+
+
+def _run_infracost(path):
+    """
+    Run infracost on the given path and return the JSON output
+    """
+    try:
+        process = subprocess.Popen(
+            ["infracost", "breakdown", "--path", path, "--format", "json"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        stdout, stderr = process.communicate()
+        infracost_data = json.loads(
+            re.search("({.+})", stdout.decode("UTF-8"))
+            .group(0)
+            .replace("u'", '"')
+            .replace("'", '"')
+        )
+        return infracost_data
+    except subprocess.CalledProcessError:
+        return None
+
+
+def infracost_report(path):
+    """
+    Generate a report of the infracost cost of the given path
+    args:
+        path: path to the qhub stages directory
+    """
+    if not path:
+        path = os.path.join(os.getcwd(), "stages")
+
+    if _check_infracost() and _check_infracost_api_key():
+        if not os.path.exists(path):
+            print("Deployment is not available")
+        else:
+            data = _run_infracost(path)
+
+            cost_table = Table(title="Cost Breakdown")
+            cost_table.add_column("Name", justify="right", style="cyan", no_wrap=True)
+            cost_table.add_column("Cost", justify="right", style="cyan", no_wrap=True)
+
+            cost_table.add_row("Total Monthly Cost", data["totalMonthlyCost"])
+            cost_table.add_row("Total Hourly Cost", data["totalHourlyCost"])
+
+            resource_table = Table(title="Resource Breakdown")
+            resource_table.add_column(
+                "Name", justify="right", style="cyan", no_wrap=True
+            )
+            resource_table.add_column(
+                "Number", justify="right", style="cyan", no_wrap=True
+            )
+
+            resource_table.add_row(
+                "Total Detected Costs", str(data["summary"]["totalDetectedResources"])
+            )
+            resource_table.add_row(
+                "Total Supported Resources",
+                str(data["summary"]["totalSupportedResources"]),
+            )
+            resource_table.add_row(
+                "Total Un-Supported Resources",
+                str(data["summary"]["totalUnsupportedResources"]),
+            )
+            resource_table.add_row(
+                "Total Non-Priced Resources",
+                str(data["summary"]["totalNoPriceResources"]),
+            )
+            resource_table.add_row(
+                "Total Usage-Priced Resources",
+                str(data["summary"]["totalUsageBasedResources"]),
+            )
+
+            console = Console()
+            console.print(cost_table)
+            console.print(resource_table)
+            print(f"Access the dashboard here: {data['shareUrl']}")
+    else:
+        print("Infracost is not installed or the API key is not configured")