Add log streaming for container app jobs

src/containerapp/HISTORY.rst

@@ -4,6 +4,8 @@ Release History
 ===============
 upcoming
 ++++++
+* 'az containerapp job logs show': Support log streaming for job execution
+* 'az containerapp job replica list': Support list replicas of a job execution
 * 'az containerapp env update': Fix logs configuration about removing destination with `--logs-destination none`
 
 0.3.52

src/containerapp/azext_containerapp/_clients.py

@@ -47,6 +47,38 @@ class ContainerAppPreviewClient(ContainerAppClient):
 
 class ContainerAppsJobPreviewClient(ContainerAppsJobClient):
     api_version = PREVIEW_API_VERSION
+    LOG_STREAM_API_VERSION = "2023-11-02-preview"
+
+    @classmethod
+    def get_replicas(cls, cmd, resource_group_name, name, execution_name):
+        management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
+        sub_id = get_subscription_id(cmd.cli_ctx)
+        url_fmt = "{}/subscriptions/{}/resourceGroups/{}/providers/Microsoft.App/jobs/{}/executions/{}/replicas?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            name,
+            execution_name,
+            cls.LOG_STREAM_API_VERSION)
+
+        r = send_raw_request(cmd.cli_ctx, "GET", request_url)
+        return r.json()
+
+    @classmethod
+    def get_auth_token(cls, cmd, resource_group_name, name):
+        management_hostname = cmd.cli_ctx.cloud.endpoints.resource_manager
+        sub_id = get_subscription_id(cmd.cli_ctx)
+        url_fmt = "{}/subscriptions/{}/resourceGroups/{}/providers/Microsoft.App/jobs/{}/getAuthToken?api-version={}"
+        request_url = url_fmt.format(
+            management_hostname.strip('/'),
+            sub_id,
+            resource_group_name,
+            name,
+            cls.LOG_STREAM_API_VERSION)
+
+        r = send_raw_request(cmd.cli_ctx, "POST", request_url)
+        return r.json()
 
 
 class ContainerAppsResiliencyPreviewClient():

src/containerapp/azext_containerapp/_help.py

@@ -1745,6 +1745,40 @@
           az containerapp env telemetry otlp list -n MyContainerappEnvironment -g MyResourceGroup
 """
 
+helps['containerapp job logs'] = """
+    type: group
+    short-summary: Show container app job logs
+"""
+
+helps['containerapp job logs show'] = """
+    type: command
+    short-summary: Show past logs and/or print logs in real time (with the --follow parameter). Note that the logs are only taken from one execution, replica, and container.
+    examples:
+    - name: Fetch the past 20 lines of logs from a job and return
+      text: |
+          az containerapp job logs show -n my-containerappjob -g MyResourceGroup --container MyContainer
+    - name: Fetch 30 lines of past logs logs from a job and print logs as they come in
+      text: |
+          az containerapp job logs show -n my-containerappjob -g MyResourceGroup --container MyContainer --follow --tail 30
+    - name: Fetch logs for a particular execution, replica, and container
+      text: |
+          az containerapp job logs show -n my-containerappjob -g MyResourceGroup --execution MyExecution --replica MyReplica --container MyContainer
+"""
+
+helps['containerapp job replica'] = """
+    type: group
+    short-summary: Manage container app replicas
+"""
+
+helps['containerapp job replica list'] = """
+    type: command
+    short-summary: List a container app job execution's replica
+    examples:
+    - name: List a container app job's replicas in a particular execution
+      text: |
+          az containerapp job replica list -n my-containerappjob -g MyResourceGroup --execution MyExecution
+"""
+
 # SessionPool Commands
 helps['containerapp sessionpool'] = """
     type: group

src/containerapp/azext_containerapp/_params.py

@@ -359,6 +359,22 @@ def load_arguments(self, _):
         c.argument('unbind_service_bindings', nargs='*', options_list=['--unbind'], help="Space separated list of services, bindings or Java components to be removed from this Java Component. e.g. BIND_NAME1...")
         c.argument('configuration', nargs="*", help="Java component configuration. Configuration must be in format \"<propertyName>=<value>\" \"<propertyName>=<value>\"...")
 
+    with self.argument_context('containerapp job logs show') as c:
+        c.argument('follow', help="Print logs in real time if present.", arg_type=get_three_state_flag())
+        c.argument('tail', help="The number of past logs to print (0-300)", type=int, default=20)
+        c.argument('container', help="The name of the container")
+        c.argument('output_format', options_list=["--format"], help="Log output format", arg_type=get_enum_type(["json", "text"]), default="json")
+        c.argument('replica', help="The name of the replica. List replicas with 'az containerapp job replica list'. A replica may not exist if the job pod has been cleaned up.")
+        c.argument('execution', help="The name of the container app execution. Defaults to the latest execution.")
+        c.argument('name', name_type, id_part=None, help="The name of the Containerapp job.")
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+
+    with self.argument_context('containerapp job replica') as c:
+        c.argument('replica', help="The name of the replica. ")
+        c.argument('execution', help="The name of the container app execution. Defaults to the latest execution.")
+        c.argument('name', name_type, id_part=None, help="The name of the Containerapp.")
+        c.argument('resource_group_name', arg_type=resource_group_name_type, id_part=None)
+
     with self.argument_context('containerapp env dotnet-component') as c:
         c.argument('dotnet_component_name', options_list=['--name', '-n'], help="The DotNet component name.")
         c.argument('environment_name', options_list=['--environment'], help="The environment name.")

src/containerapp/azext_containerapp/commands.py

@@ -205,6 +205,12 @@ def load_command_table(self, args):
         g.custom_show_command('show', 'show_eureka_server_for_spring')
         g.custom_command('delete', 'delete_eureka_server_for_spring', confirmation=True, supports_no_wait=True)
 
+    with self.command_group('containerapp job logs',  is_preview=True) as g:
+        g.custom_show_command('show', 'stream_job_logs')
+
+    with self.command_group('containerapp job replica', is_preview=True) as g:
+        g.custom_show_command('list', 'list_replica_containerappsjob')
+
     with self.command_group('containerapp env java-component nacos') as g:
         g.custom_command('create', 'create_nacos', supports_no_wait=True)
         g.custom_command('update', 'update_nacos', supports_no_wait=True)
@@ -236,7 +242,6 @@ def load_command_table(self, args):
         g.custom_command('update', 'update_session_pool', supports_no_wait=True)
         g.custom_command('delete', 'delete_session_pool', confirmation=True, supports_no_wait=True)
 
-
     with self.command_group('containerapp session code-interpreter', is_preview=True) as g:
         g.custom_command('execute', 'execute_session_code_interpreter', supports_no_wait=True)
         g.custom_command('upload-file', 'upload_session_code_interpreter', supports_no_wait=True)

src/containerapp/azext_containerapp/custom.py

@@ -8,6 +8,7 @@
 import time
 from urllib.parse import urlparse
 import json
+import requests
 import subprocess
 from concurrent.futures import ThreadPoolExecutor
 from ._constants import DOTNET_COMPONENT_RESOURCE_TYPE
@@ -2689,6 +2690,73 @@ def list_environment_telemetry_otlp(cmd,
     return containerapp_env_def
 
 
+def list_replica_containerappsjob(cmd, resource_group_name, name, execution=None):
+    if execution is None:
+        executions = ContainerAppsJobPreviewClient.get_executions(cmd=cmd, resource_group_name=resource_group_name, name=name)
+        execution = executions['value'][0]['name']
+        logger.warning('No execution specified. Using the latest execution: %s', execution)
+    try:
+        replicas = ContainerAppsJobPreviewClient.get_replicas(cmd, resource_group_name, name, execution)
+        return replicas['value']
+    except CLIError as e:
+        handle_raw_exception(e)
+
+
+def stream_job_logs(cmd, resource_group_name, name, container, execution=None, replica=None, follow=False, tail=None, output_format=None):
+    if tail:
+        if tail < 0 or tail > 300:
+            raise ValidationError("--tail must be between 0 and 300.")
+
+    sub = get_subscription_id(cmd.cli_ctx)
+    token_response = ContainerAppsJobPreviewClient.get_auth_token(cmd, resource_group_name, name)
+    token = token_response["properties"]["token"]
+
+    job = ContainerAppsJobPreviewClient.show(cmd, resource_group_name, name)
+    base_url = job["properties"]["eventStreamEndpoint"]
+    base_url = base_url[:base_url.index("/subscriptions/")]
+
+    if execution is None and replica is not None:
+        raise ValidationError("Cannot specify a replica without an execution")
+
+    if execution is None:
+        executions = ContainerAppsJobPreviewClient.get_executions(cmd, resource_group_name, name)['value']
+        if not executions:
+            raise ValidationError("No executions found for this job")
+        execution = executions[0]["name"]
+        logger.warning("No execution provided, defaulting to latest execution: %s", execution)
+
+    if replica is None:
+        replicas = ContainerAppsJobPreviewClient.get_replicas(cmd, resource_group_name, name, execution)['value']
+        if not replicas:
+            raise ValidationError("No replicas found for execution")
+        replica = replicas[0]["name"]
+        logger.warning("No replica provided, defaulting to latest replica: %s", replica)
+
+    url = (f"{base_url}/subscriptions/{sub}/resourceGroups/{resource_group_name}/jobs/{name}"
+           f"/executions/{execution}/replicas/{replica}/containers/{container}/logstream")
+
+    logger.info("connecting to : %s", url)
+    request_params = {"follow": str(follow).lower(),
+                      "output": output_format,
+                      "tailLines": tail}
+    headers = {"Authorization": f"Bearer {token}"}
+    resp = requests.get(url,
+                        timeout=None,
+                        stream=True,
+                        params=request_params,
+                        headers=headers)
+
+    if not resp.ok:
+        raise ValidationError(f"Got bad status from the logstream API: {resp.status_code}. Error: {str(resp.content)}")
+
+    for line in resp.iter_lines():
+        if line:
+            logger.info("received raw log line: %s", line)
+            # these .replaces are needed to display color/quotations properly
+            # for some reason the API returns garbled unicode special characters (may need to add more in the future)
+            print(line.decode("utf-8").replace("\\u0022", "\u0022").replace("\\u001B", "\u001B").replace("\\u002B", "\u002B").replace("\\u0027", "\u0027"))
+
+
 def create_or_update_java_logger(cmd, logger_name, logger_level, name, resource_group_name, no_wait=False):
     raw_parameters = locals()
     containerapp_java_logger_set_decorator = ContainerappJavaLoggerSetDecorator(
@@ -3021,5 +3089,4 @@ def create_dotnet_component(cmd, dotnet_component_name, environment_name, resour
     if component_type == DOTNET_COMPONENT_RESOURCE_TYPE:
         aspire_dashboard_url = dotnet_component_decorator._get_aspire_dashboard_url(environment_name, resource_group_name, dotnet_component_name)
         logger.warning("Access your Aspire Dashboard at %s.", aspire_dashboard_url)
-
-    return
+    return