[Spring] Support managed component log stream (Azure#7021)

* Add support for managed component log stream in Azure Spring Apps

* Fix CI failure

src/spring/HISTORY.md

@@ -1,5 +1,9 @@
 Release History
 ===============
+1.19.0
+---
+* Add new commands for managed component log streaming `az spring component list`, `az spring component instance list` and `az spring component logs`.
+
 1.18.0
 ---
 * Add arguments `--bind-service-registry` in `spring app create`.

src/spring/azext_spring/_help.py

@@ -1541,3 +1541,45 @@
         - name: Disable an APM globally.
           text: az spring apm disable-globally --name first-apm --service MyCluster --resource-group MyResourceGroup
 """
+
+helps['spring component'] = """
+    type: group
+    short-summary: (Enterprise Tier Only) Commands to handle managed components.
+"""
+
+helps['spring component logs'] = """
+    type: command
+    short-summary: (Enterprise Tier Only) Show logs for managed components. Logs will be streamed when setting '-f/--follow'. For now, only supports subcomponents of (a) Application Configuration Service (b) Spring Cloud Gateway
+    examples:
+        - name: Show logs for all instances of flux in Application Configuration Serice (Gen2)
+          text: az spring component logs --name flux-source-controller --service MyAzureSpringAppsInstance --resource-group MyResourceGroup --all-instances
+        - name: Show logs for a specific instance of application-configuration-service in Application Configuration Serice
+          text: az spring component logs --name application-configuration-service --service MyAzureSpringAppsInstance --resource-group MyResourceGroup --instance InstanceName
+        - name: Stream and watch logs for all instances of spring-cloud-gateway
+          text: az spring component logs --name spring-cloud-gateway --service MyAzureSpringAppsInstance --resource-group MyResourceGroup --all-instances --follow
+        - name: Show logs for a specific instance without specify the component name
+          text: az spring component logs --service MyAzureSpringAppsInstance --resource-group MyResourceGroup --instance InstanceName
+"""
+
+helps['spring component list'] = """
+    type: command
+    short-summary: (Enterprise Tier Only) List managed components.
+    examples:
+        - name: List all managed components
+          text: az spring component list --service MyAzureSpringAppsInstance --resource-group MyResourceGroup
+"""
+
+helps['spring component instance'] = """
+    type: group
+    short-summary: (Enterprise Tier Only) Commands to handle instances of a managed component.
+"""
+
+helps['spring component instance list'] = """
+    type: command
+    short-summary: (Enterprise Tier Only) List all available instances of a specific managed component in an Azure Spring Apps instance.
+    examples:
+        - name: List instances for spring-cloud-gateway of Spring Cloud Gateway
+          text: az spring component instance list --component spring-cloud-gateway --service MyAzureSpringAppsInstance --resource-group MyResourceGroup
+        - name: List instances for spring-cloud-gateway-operator of Spring Cloud Gateway
+          text: az spring component instance list --component spring-cloud-gateway-operator --service MyAzureSpringAppsInstance --resource-group MyResourceGroup
+"""

src/spring/azext_spring/_params.py

@@ -284,7 +284,7 @@ def load_arguments(self, _):
             TestKeyType), help='Type of test-endpoint key')
 
     with self.argument_context('spring list-support-server-versions') as c:
-            c.argument('service', service_name_type, validator=not_support_enterprise)
+        c.argument('service', service_name_type, validator=not_support_enterprise)
 
     with self.argument_context('spring app') as c:
         c.argument('service', service_name_type)
@@ -1095,3 +1095,36 @@ def prepare_logs_argument(c):
             c.argument('private_key', help='Private SSH Key algorithm of git repository.')
             c.argument('host_key', help='Public SSH Key of git repository.')
             c.argument('host_key_algorithm', help='SSH Key algorithm of git repository.')
+
+    for scope in ['spring component']:
+        with self.argument_context(scope) as c:
+            c.argument('service', service_name_type)
+
+    with self.argument_context('spring component logs') as c:
+        c.argument('name', options_list=['--name', '-n'],
+                   help="Name of the component. Find component names from command `az spring component list`")
+        c.argument('all_instances',
+                   help='The flag to indicate get logs for all instances of the component.',
+                   action='store_true')
+        c.argument('instance',
+                   options_list=['--instance', '-i'],
+                   help='Name of an existing instance of the component.')
+        c.argument('follow',
+                   options_list=['--follow ', '-f'],
+                   help='The flag to indicate logs should be streamed.',
+                   action='store_true')
+        c.argument('lines',
+                   type=int,
+                   help='Number of lines to show. Maximum is 10000. Default is 50.')
+        c.argument('since',
+                   help='Only return logs newer than a relative duration like 5s, 2m, or 1h. Maximum is 1h')
+        c.argument('limit',
+                   type=int,
+                   help='Maximum kibibyte of logs to return. Ceiling number is 2048.')
+        c.argument('max_log_requests',
+                   type=int,
+                   help="Specify maximum number of concurrent logs to follow when get logs by all-instances.")
+
+    with self.argument_context('spring component instance') as c:
+        c.argument('component', options_list=['--component', '-c'],
+                   help="Name of the component. Find components from command `az spring component list`")

src/spring/azext_spring/app.py

@@ -8,7 +8,7 @@
 from knack.log import get_logger
 from azure.cli.core.util import sdk_no_wait
 from azure.cli.core.azclierror import (ValidationError, ArgumentUsageError)
-from .custom import app_get, _get_app_log
+from .custom import app_get
 from ._utils import (get_spring_sku, wait_till_end, convert_argument_to_parameter_list)
 from ._deployment_factory import (deployment_selector,
                                   deployment_settings_options_from_resource,
@@ -20,6 +20,7 @@
 from .custom import app_tail_log_internal
 import datetime
 from time import sleep
+from .log_stream.log_stream_operations import log_stream_from_url
 
 logger = get_logger(__name__)
 DEFAULT_DEPLOYMENT_NAME = "default"
@@ -516,7 +517,7 @@ def _get_deployment_ignore_exception(client, resource_group, service, app_name,
 
 def _get_app_log_deploy_phase(url, auth, format_json, exceptions):
     try:
-        _get_app_log(url, auth, format_json, exceptions, chunk_size=10 * 1024, stderr=True)
+        log_stream_from_url(url, auth, format_json, exceptions, chunk_size=10 * 1024, stderr=True)
     except Exception:
         pass
 

src/spring/azext_spring/commands.py

@@ -31,7 +31,8 @@
                             transform_support_server_versions_output)
 from ._validators import validate_app_insights_command_not_supported_tier
 from ._marketplace import (transform_marketplace_plan_output)
-from ._validators_enterprise import (validate_gateway_update, validate_api_portal_update, validate_dev_tool_portal, validate_customized_accelerator, validate_central_build_instance)
+from ._validators_enterprise import (validate_gateway_update, validate_api_portal_update, validate_dev_tool_portal, validate_customized_accelerator)
+from .managed_components.validators_managed_component import (validate_component_logs, validate_component_list, validate_instance_list)
 from ._app_managed_identity_validator import (validate_app_identity_remove_or_warning,
                                               validate_app_identity_assign_or_warning)
 
@@ -118,6 +119,11 @@ def load_command_table(self, _):
         client_factory=cf_spring
     )
 
+    managed_component_cmd_group = CliCommandType(
+        operations_tmpl='azext_spring.managed_components.managed_component_operations#{}',
+        client_factory=cf_spring
+    )
+
     with self.command_group('spring', custom_command_type=spring_routing_util,
                             exception_handler=handle_asc_exception) as g:
         g.custom_command('create', 'spring_create', supports_no_wait=True)
@@ -459,5 +465,16 @@ def load_command_table(self, _):
         g.custom_command('update', 'update_build_service', supports_no_wait=True)
         g.custom_show_command('show', 'build_service_show')
 
+    with self.command_group('spring component',
+                            custom_command_type=managed_component_cmd_group,
+                            exception_handler=handle_asc_exception) as g:
+        g.custom_command('logs', 'managed_component_logs', validator=validate_component_logs)
+        g.custom_command('list', 'managed_component_list', validator=validate_component_list)
+
+    with self.command_group('spring component instance',
+                            custom_command_type=managed_component_cmd_group,
+                            exception_handler=handle_asc_exception) as g:
+        g.custom_command('list', 'managed_component_instance_list', validator=validate_instance_list)
+
     with self.command_group('spring', exception_handler=handle_asc_exception):
         pass

src/spring/azext_spring/custom.py

@@ -37,6 +37,7 @@
 from collections import defaultdict
 from ._log_stream import LogStream
 from ._build_service import _update_default_build_agent_pool
+from .log_stream.log_stream_operations import log_stream_from_url
 
 logger = get_logger(__name__)
 DEFAULT_DEPLOYMENT_NAME = "default"
@@ -512,7 +513,7 @@ def app_get_build_log(cmd, client, resource_group, service, name, deployment=Non
 def app_tail_log(cmd, client, resource_group, service, name,
                  deployment=None, instance=None, follow=False, lines=50, since=None, limit=2048, format_json=None):
     app_tail_log_internal(cmd, client, resource_group, service, name, deployment, instance, follow, lines, since, limit,
-                          format_json, get_app_log=_get_app_log)
+                          format_json, get_app_log=log_stream_from_url)
 
 
 def app_tail_log_internal(cmd, client, resource_group, service, name,
@@ -1167,136 +1168,6 @@ def _get_redis_primary_key(cli_ctx, resource_id):
     return keys.primary_key
 
 
-# pylint: disable=bare-except, too-many-statements
-def _get_app_log(url, auth, format_json, exceptions, chunk_size=None, stderr=False):
-    logger_seg_regex = re.compile(r'([^\.])[^\.]+\.')
-
-    def build_log_shortener(length):
-        if length <= 0:
-            raise InvalidArgumentValueError('Logger length in `logger{length}` should be positive')
-
-        def shortener(record):
-            '''
-            Try shorten the logger property to the specified length before feeding it to the formatter.
-            '''
-            logger_name = record.get('logger', None)
-            if logger_name is None:
-                return record
-
-            # first, try to shorten the package name to one letter, e.g.,
-            #     org.springframework.cloud.netflix.eureka.config.DiscoveryClientOptionalArgsConfiguration
-            # to: o.s.c.n.e.c.DiscoveryClientOptionalArgsConfiguration
-            while len(logger_name) > length:
-                logger_name, count = logger_seg_regex.subn(r'\1.', logger_name, 1)
-                if count < 1:
-                    break
-
-            # then, cut off the leading packages if necessary
-            logger_name = logger_name[-length:]
-            record['logger'] = logger_name
-            return record
-
-        return shortener
-
-    def build_formatter():
-        '''
-        Build the log line formatter based on the format_json argument.
-        '''
-        nonlocal format_json
-
-        def identity(o):
-            return o
-
-        if format_json is None or len(format_json) == 0:
-            return identity
-
-        logger_regex = re.compile(r'\blogger\{(\d+)\}')
-        match = logger_regex.search(format_json)
-        pre_processor = identity
-        if match:
-            length = int(match[1])
-            pre_processor = build_log_shortener(length)
-            format_json = logger_regex.sub('logger', format_json, 1)
-
-        first_exception = True
-
-        def format_line(line):
-            nonlocal first_exception
-            try:
-                log_record = json.loads(line)
-                # Add n=\n so that in Windows CMD it's easy to specify customized format with line ending
-                # e.g., "{timestamp} {message}{n}"
-                # (Windows CMD does not escape \n in string literal.)
-                return format_json.format_map(pre_processor(defaultdict(str, n="\n", **log_record)))
-            except:
-                if first_exception:
-                    # enable this format error logging only with --verbose
-                    logger.info("Failed to format log line '{}'".format(line), exc_info=sys.exc_info())
-                    first_exception = False
-                return line
-
-        return format_line
-
-    def iter_lines(response, limit=2 ** 20, chunk_size=None):
-        '''
-        Returns a line iterator from the response content. If no line ending was found and the buffered content size is
-        larger than the limit, the buffer will be yielded directly.
-        '''
-        buffer = []
-        total = 0
-        for content in response.iter_content(chunk_size=chunk_size):
-            if not content:
-                if len(buffer) > 0:
-                    yield b''.join(buffer)
-                break
-
-            start = 0
-            while start < len(content):
-                line_end = content.find(b'\n', start)
-                should_print = False
-                if line_end < 0:
-                    next = (content if start == 0 else content[start:])
-                    buffer.append(next)
-                    total += len(next)
-                    start = len(content)
-                    should_print = total >= limit
-                else:
-                    buffer.append(content[start:line_end + 1])
-                    start = line_end + 1
-                    should_print = True
-
-                if should_print:
-                    yield b''.join(buffer)
-                    buffer.clear()
-                    total = 0
-
-    with requests.get(url, stream=True, auth=auth) as response:
-        try:
-            if response.status_code != 200:
-                failure_reason = response.reason
-                if response.content:
-                    if isinstance(response.content, bytes):
-                        failure_reason = "{}:{}".format(failure_reason, response.content.decode('utf-8'))
-                    else:
-                        failure_reason = "{}:{}".format(failure_reason, response.content)
-                raise CLIError("Failed to connect to the server with status code '{}' and reason '{}'".format(
-                    response.status_code, failure_reason))
-            std_encoding = sys.stdout.encoding
-
-            formatter = build_formatter()
-
-            for line in iter_lines(response, chunk_size=chunk_size):
-                decoded = (line.decode(encoding='utf-8', errors='replace')
-                           .encode(std_encoding, errors='replace')
-                           .decode(std_encoding, errors='replace'))
-                if stderr:
-                    print(formatter(decoded), end='', file=sys.stderr)
-                else:
-                    print(formatter(decoded), end='')
-        except CLIError as e:
-            exceptions.append(e)
-
-
 def storage_callback(pipeline_response, deserialized, headers):
     return models.StorageResource.deserialize(json.loads(pipeline_response.http_response.text()))
 

src/spring/azext_spring/log_stream/__init__.py

@@ -0,0 +1,4 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------