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

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

@@ -72,6 +72,10 @@ def get_log(
             job_name, pod, namespace=namespace
         )
         click.echo(result)
+
+        cloudwatch_link = get_logs_service.generate_cloudwatch_link(pod, namespace=namespace)
+        if cloudwatch_link:
+            click.echo(cloudwatch_link)
     except Exception as e:
         sys.exit(
             f"Unexpected error happens when trying to get logs for training job {job_name} : {e}"

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,9 +20,15 @@
 from hyperpod_cli.service.list_pods import (
     ListPods,
 )
+from hyperpod_cli.utils import (
+    get_eks_cluster_name,
+    get_hyperpod_cluster_region,
+)
 from kubernetes.client.rest import ApiException
 from kubernetes.client import V1ResourceAttributes
 
+AMAZON_ClOUDWATCH_OBSERVABILITY = "amazon-cloudwatch-observability"
+
 class GetLogs:
     def __init__(self):
         return
@@ -57,7 +64,67 @@ 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()
+
+        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
+            if node_name and pod_name and namespace and container_name and container_id:
+                region = get_hyperpod_cluster_region()
+
+                cloudwatch_url = self.get_log_url(eks_cluster_name, region, node_name, pod_name, namespace, container_name, container_id)
+                cloudwatch_link = f'The pod cloudwatch log stream link is {cloudwatch_url}'
+            else:
+                cloudwatch_link = 'Failed to load container insights CloudWatch Link!'
+        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,7 +113,7 @@ 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
@@ -168,3 +168,12 @@ def get_cluster_console_url():
         if _validate_link(console_url) and _validate_placeholders(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_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",
+        )
+