Add Cloudwatch console link for container insight logs (#74)

* Add Cloudwatch console link for container insight logs

**Description**
When executing get-logs function, after printing all the logs, we also provide the cloudwatch link to container insight logs (if the CW observability add-on is enabled)

**Testing Done**
Tested in my personal account and verified the link, and added unit test

* fix unit tests

* Add validation to cw url

* Update cw url validation function

* Add try except for CW link generation

Author: zhaoqizqwang

src/hyperpod_cli/clients/kubernetes_client.py

@@ -275,6 +275,11 @@ def get_job(self, job_name: str, namespace: str):
             plural=PYTORCH_CUSTOM_OBJECT_PLURAL,
             name=job_name,
         )
+    
+    def get_pod_details(self, pod_name: str, namespace: str):
+        return client.CoreV1Api().read_namespaced_pod(
+            name=pod_name, namespace=namespace
+        )
 
     def delete_training_job(self, job_name: str, namespace: str):
         return client.CustomObjectsApi().delete_namespaced_custom_object(

src/hyperpod_cli/commands/pod.py

@@ -76,7 +76,13 @@ def get_log(
         sys.exit(
             f"Unexpected error happens when trying to get logs for training job {job_name} : {e}"
         )
-
+    
+    try:
+        cloudwatch_link = get_logs_service.generate_cloudwatch_link(pod, namespace=namespace)
+        if cloudwatch_link:
+            click.echo(cloudwatch_link)
+    except Exception as e:
+        click.echo(f"WARNING: Failed to generate container insights cloudwatch link: {e}")
 
 def _exec_command_required_option_pod_and_all_pods():
     class OptionRequiredClass(click.Command):

src/hyperpod_cli/service/get_logs.py

@@ -11,6 +11,7 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 from typing import Optional
+import boto3
 
 from hyperpod_cli.clients.kubernetes_client import (
     KubernetesClient,
@@ -19,8 +20,17 @@
 from hyperpod_cli.service.list_pods import (
     ListPods,
 )
+from hyperpod_cli.utils import (
+    get_eks_cluster_name,
+    get_hyperpod_cluster_region,
+    validate_region_and_cluster_name,
+)
 from kubernetes.client.rest import ApiException
 from kubernetes.client import V1ResourceAttributes
+import re
+
+AMAZON_ClOUDWATCH_OBSERVABILITY = "amazon-cloudwatch-observability"
+CONTAINER_INSIGHTS_LOG_REGEX_PATTERN = "https:\/\/([a-z0-9-]+).console.aws.amazon.com\/cloudwatch\/home\?region=([a-z0-9-]+)#logsV2:log-groups\/log-group\/\$252Faws\$252Fcontainerinsights\$252F([a-zA-Z0-9-]+)\$252Fapplication\/log-events\/([a-z0-9-]+)-application.var.log.containers.([a-z0-9-]+)_([a-z0-9-]+)_([a-z0-9-]+)-([a-z0-9-]+).log"
 
 class GetLogs:
     def __init__(self):
@@ -57,7 +67,70 @@ def get_training_job_logs(
                 raise RuntimeError(
                     f"Given pod name {pod_name} is not associated with training job {job_name} in namespace {namespace}"
                 )
-
             return k8s_client.get_logs_for_pod(pod_name, namespace)
         except ApiException as e:
             raise RuntimeError(f"Unexpected API error: {e.reason} ({e.status})")
+    
+    def generate_cloudwatch_link(
+        self,
+        pod_name: str,
+        namespace: Optional[str],
+    ):
+        eks_cluster_name = get_eks_cluster_name()
+        region = get_hyperpod_cluster_region()
+
+        if self.is_container_insights_addon_enabled(eks_cluster_name):
+            k8s_client = KubernetesClient()
+
+            # pod_details is a V1Pod object
+            pod_details = k8s_client.get_pod_details(pod_name, namespace)
+            
+            # get node name
+            if pod_details.spec and pod_details.spec.node_name:
+                node_name = pod_details.spec.node_name
+            else:
+                node_name = None
+
+            # get container name
+            if pod_details.spec and pod_details.spec.containers and pod_details.spec.containers[0].name:
+                container_name = pod_details.spec.containers[0].name
+            else:
+                container_name = None
+
+            # get container_id
+            if pod_details.status and pod_details.status.container_statuses and pod_details.status.container_statuses[0].container_id:
+                full_container_id = pod_details.status.container_statuses[0].container_id
+
+                # full_container_id has format "containerd://xxxxxxxxxx" 
+                container_id = full_container_id[13:] if full_container_id.startswith('containerd://') else None
+            else:
+                container_id = None         
+
+            # Cloudwatch container insight log groups should have the same pod log as API response
+            cloudwatch_url = self.get_log_url(eks_cluster_name, region, node_name, pod_name, namespace, container_name, container_id)
+
+            if not validate_region_and_cluster_name(region, eks_cluster_name):
+                raise ValueError('Eks cluster name or cluster region is invalid.')
+
+            if not re.match(CONTAINER_INSIGHTS_LOG_REGEX_PATTERN, cloudwatch_url):
+                raise ValueError("Failed to validate cloudwatch log url. Please make sure pod's node name, container name and container id are valid")
+            
+            cloudwatch_link = f'The pod cloudwatch log stream link is {cloudwatch_url}'
+        else:
+            cloudwatch_link = None
+
+        return cloudwatch_link
+
+    def get_log_url(self, eks_cluster_name, region, node_name, pod_name, namespace, container_name, container_id):
+        console_prefix = f'https://{region}.console.aws.amazon.com/cloudwatch/home?region={region}#'
+        log_group_prefix = f'logsV2:log-groups/log-group/$252Faws$252Fcontainerinsights$252F{eks_cluster_name}$252Fapplication/log-events/'
+        log_stream = f'{node_name}-application.var.log.containers.{pod_name}_{namespace}_{container_name}-{container_id}.log'
+        
+        return console_prefix + log_group_prefix + log_stream
+
+    def is_container_insights_addon_enabled(self, eks_cluster_name):
+        response = boto3.client("eks").list_addons(clusterName=eks_cluster_name, maxResults=50)
+        if AMAZON_ClOUDWATCH_OBSERVABILITY in response.get('addons', []):
+            return True
+        else:
+            return False

src/hyperpod_cli/utils.py

@@ -113,15 +113,15 @@ def _retrieve_current_hyperpod_context():
 
 
 def _validate_link(console_url):
-    pattern = "https:\/\/([a-z0-9-]+).console.aws.amazon.com\/sagemaker\/home\?region=([a-z0-9-]+)#\/cluster-management\/([a-z0-9-]+)"
+    pattern = "https:\/\/([a-z0-9-]+).console.aws.amazon.com\/sagemaker\/home\?region=([a-z0-9-]+)#\/cluster-management\/([a-zA-Z0-9-]+)"
     match = re.match(pattern, console_url)
     if match:
         return True
     else:
         return False
 
 
-def _validate_placeholders(region, cluster_name):
+def validate_region_and_cluster_name(region, cluster_name):
     output = False
     region_char_list = region.split("-")
 
@@ -135,6 +135,9 @@ def _validate_placeholders(region, cluster_name):
     region_prefix_length = len(region_char_list[0])
     region_length = len(region_char_list[1])
     region_suffix_length = len(region_char_list[2])
+
+    cluster_name_match = re.match("[a-zA-Z0-9-]+", cluster_name)
+
     if (
         region_prefix_match
         and region_match
@@ -143,6 +146,7 @@ def _validate_placeholders(region, cluster_name):
         and region_suffix_length == 1
         and region_length >= 4
         and region_length < 10
+        and cluster_name_match
         and len(cluster_name) >= 1
         and len(cluster_name) <= 63
     ):
@@ -165,6 +169,15 @@ def get_cluster_console_url():
             f"https://{region}.console.aws.amazon.com/sagemaker/"
             f"home?region={region}#/cluster-management/{cluster_name}"
         )
-        if _validate_link(console_url) and _validate_placeholders(region, cluster_name):
+        if _validate_link(console_url) and validate_region_and_cluster_name(region, cluster_name):
             return console_url
     return None
+
+def get_eks_cluster_name():
+    hyperpod_context_cluster = _retrieve_current_hyperpod_context()
+    eks_cluster_arn = hyperpod_context_cluster.get("Orchestrator", {}).get("Eks", {}).get("ClusterArn", '')
+    return eks_cluster_arn.split('cluster/')[-1]
+
+def get_hyperpod_cluster_region():
+    hyperpod_context_cluster = _retrieve_current_hyperpod_context()
+    return hyperpod_context_cluster.get("ClusterArn").split(":")[3]

test/unit_tests/clients/test_kubernetes_client.py

@@ -520,6 +520,15 @@ def test_logs_for_pods(self, mock_core_client: Mock):
         result = test_client.get_logs_for_pod("test", "kubeflow")
         self.assertIn("test log", result)
 
+    @patch(
+        "kubernetes.client.CoreV1Api",
+        return_value=Mock(read_namespaced_pod=Mock(return_value="pod details")),
+    )
+    def test_get_pod_details(self, mock_core_client: Mock):
+        test_client = KubernetesClient()
+        result = test_client.get_pod_details("test", "kubeflow")
+        self.assertIn("pod details", result)
+
     @patch(
         "kubernetes.client.CustomObjectsApi",
         return_value=Mock(

test/unit_tests/service/test_get_logs_service.py

@@ -125,3 +125,20 @@ def test_get_logs_auto_discover_namespace(
             "sample-job", "test-pod", None
         )
         self.assertIn("test logs", result)
+
+    def test_get_log_url(
+        self,
+    ):
+        eks_cluster_name = 'eks_cluster_name'
+        region = 'us-west-2'
+        node_name = 'node_name'
+        pod_name = 'pod_name'
+        namespace = 'namespace'
+        container_name = 'container_name'
+        container_id = 'container_id'
+        result_url = self.mock_get_logs.get_log_url(eks_cluster_name, region, node_name, pod_name, namespace, container_name, container_id)
+
+        self.assertEqual(
+            result_url,
+            'https://us-west-2.console.aws.amazon.com/cloudwatch/home?region=us-west-2#logsV2:log-groups/log-group/$252Faws$252Fcontainerinsights$252Feks_cluster_name$252Fapplication/log-events/node_name-application.var.log.containers.pod_name_namespace_container_name-container_id.log'
+        )

test/unit_tests/test_pod.py

@@ -35,11 +35,14 @@ def setUp(self):
 
     @mock.patch("hyperpod_cli.service.get_logs.GetLogs")
     @mock.patch("hyperpod_cli.service.get_logs.GetLogs.get_training_job_logs")
+    @mock.patch("hyperpod_cli.service.get_logs.GetLogs.generate_cloudwatch_link")
     def test_get_logs_happy_case(
         self,
+        mock_get_logs_service_and_generate_cloudwatch_link: mock.Mock,
         mock_get_logs_service_and_get_logs: mock.Mock,
         mock_get_logs_service: mock.Mock,
     ):
+        mock_get_logs_service_and_generate_cloudwatch_link.return_value = 'link'
         mock_get_logs_service.return_value = self.mock_get_job_log
         mock_get_logs_service_and_get_logs.return_value = "{}"
         result = self.runner.invoke(
@@ -55,15 +58,19 @@ def test_get_logs_happy_case(
 
     @mock.patch("hyperpod_cli.service.get_logs.GetLogs")
     @mock.patch("hyperpod_cli.service.get_logs.GetLogs.get_training_job_logs")
+    @mock.patch("hyperpod_cli.service.get_logs.GetLogs.generate_cloudwatch_link")
     @mock.patch("logging.Logger.debug")
     def test_get_logs_happy_case_debug_mode(
         self,
         mock_debug: mock.Mock,
+         mock_get_logs_service_and_generate_cloudwatch_link: mock.Mock,
         mock_get_logs_service_and_get_logs: mock.Mock,
         mock_get_logs_service: mock.Mock,
     ):
         mock_get_logs_service.return_value = self.mock_get_job_log
         mock_get_logs_service_and_get_logs.return_value = "{}"
+        mock_get_logs_service_and_generate_cloudwatch_link.return_value = 'link'
+
         result = self.runner.invoke(
             get_log,
             [
@@ -79,13 +86,17 @@ def test_get_logs_happy_case_debug_mode(
 
     @mock.patch("hyperpod_cli.service.get_logs.GetLogs")
     @mock.patch("hyperpod_cli.service.get_logs.GetLogs.get_training_job_logs")
+    @mock.patch("hyperpod_cli.service.get_logs.GetLogs.generate_cloudwatch_link")
     def test_describe_job_happy_case_with_namespace(
         self,
+        mock_get_logs_service_and_generate_cloudwatch_link: mock.Mock,
         mock_get_logs_service_and_get_logs: mock.Mock,
         mock_get_logs_service: mock.Mock,
     ):
         mock_get_logs_service.return_value = self.mock_get_job_log
         mock_get_logs_service_and_get_logs.return_value = "{}"
+        mock_get_logs_service_and_generate_cloudwatch_link.return_value = 'link'
+
         result = self.runner.invoke(
             get_log,
             [

test/unit_tests/test_utils.py

@@ -23,6 +23,8 @@
     setup_logger,
     get_cluster_console_url,
     store_current_hyperpod_context,
+    get_eks_cluster_name,
+    get_hyperpod_cluster_region,
 )
 
 DATA_JSON = {
@@ -36,7 +38,8 @@
     '{"ClusterArn": "arn:aws:sagemaker:us-west-2:1234567890:cluster/test",'
     ' "ClusterName": "hyperpod-eks-test",'
     ' "ClusterStatus": "InService",'
-    ' "CreationTime": "2024-08-17 01:26:35.921000+00:00"}'
+    ' "CreationTime": "2024-08-17 01:26:35.921000+00:00",'
+    ' "Orchestrator": {"Eks": {"ClusterArn":"arn:aws:eks:us-west-2:593793038179:cluster/hyperpod-eks-test"}}}'
 )
 
 INVALID_DATA = (
@@ -296,3 +299,24 @@ def test_get_cluster_console_url_longer_cluster_name_null(
             "/tmp/hyperpod_current_context.json",
             "r",
         )
+
+    def test_get_eks_cluster_name(self):
+        mock_read = mock_open(read_data=DATA)
+        with patch("builtins.open", mock_read):
+            result = get_eks_cluster_name()
+        self.assertEqual(result, "hyperpod-eks-test")
+        mock_read.assert_called_once_with(
+            "/tmp/hyperpod_current_context.json",
+            "r",
+        )
+    
+    def test_get_hyperpod_cluster_region(self):
+        mock_read = mock_open(read_data=DATA)
+        with patch("builtins.open", mock_read):
+            result = get_hyperpod_cluster_region()
+        self.assertEqual(result, "us-west-2")
+        mock_read.assert_called_once_with(
+            "/tmp/hyperpod_current_context.json",
+            "r",
+        )
+