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,20 @@
+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="Pass the path of your stages directory generated after rendering QHub configurations before deployment",
+        required=False,
+    )
+    subparser.set_defaults(func=handle_cost_report)
+
+
+def handle_cost_report(args):
+    infracost_report(args.path)

qhub/cost.py

@@ -0,0 +1,128 @@
+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
+    except AttributeError:
+        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)
+            if data:
+                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)
+                try:
+                    print(f"Access the dashboard here: {data['shareUrl']}")
+                except KeyError:
+                    print(
+                        "Dashboard is not available. Enable it via: infracost configure set enable_dashboard true"
+                    )
+            else:
+                print(
+                    "No data was generated. Please check your QHub configuration and generated stages."
+                )
+    else:
+        print("Infracost is not installed or the API key is not configured")