xds-k8s: Fix the issue with parsing Operation.metadata (#28736)

1. Solves an issue with unpacking Operation.metadata causing
   json_format.ParseError Can not find message descriptor by type_url
2. Improves the readability of the
   framework.infrastructure.gcp.api.OperationError

tools/run_tests/xds_k8s_test_driver/framework/infrastructure/gcp/api.py

@@ -14,6 +14,7 @@
 import abc
 import contextlib
 import functools
+import json
 import logging
 from typing import Any, Dict, List, Optional
 
@@ -21,9 +22,9 @@ from absl import flags
 from google.cloud import secretmanager_v1
 from google.longrunning import operations_pb2
 from google.protobuf import json_format
-from google.protobuf import text_format
 from google.rpc import code_pb2
 from google.rpc import error_details_pb2
+from google.rpc import status_pb2
 from googleapiclient import discovery
 import googleapiclient.errors
 import googleapiclient.http
@@ -260,29 +261,68 @@ class OperationError(Error):
     https://cloud.google.com/apis/design/design_patterns#long_running_operations
     https://github.com/googleapis/googleapis/blob/master/google/longrunning/operations.proto
     """
+    api_name: str
+    name: str
+    metadata: str
+    code_name: code_pb2.Code
+    error: status_pb2.Status
 
-    def __init__(self, api_name, operation_response, message=None):
+    def __init__(self, api_name: str, response: dict):
         self.api_name = api_name
-        operation = json_format.ParseDict(
-            operation_response,
-            Operation(),
-            ignore_unknown_fields=True,
-            descriptor_pool=error_details_pb2.DESCRIPTOR.pool)
+
+        # Operation.metadata field is Any specific to the API. It may not be
+        # present in the default descriptor pool, and that's expected.
+        # To avoid json_format.ParseError, handle it separately.
+        self.metadata = response.pop('metadata', {})
+
+        # Must be after removing metadata field.
+        operation: Operation = self._parse_operation_response(response)
         self.name = operation.name or 'unknown'
         self.code_name = code_pb2.Code.Name(operation.error.code)
         self.error = operation.error
-        # Collect error details packed as Any without parsing concrete types.
-        self.error_details = [
-            text_format.MessageToString(any_error, as_one_line=True)
-            for any_error in self.error.details
-        ]
-        if message is None:
-            message = (f'{api_name} operation "{self.name}" failed. Error '
-                       f'code: {self.error.code} ({self.code_name}), '
-                       f'message: {self.error.message}, '
-                       f'details: {self.error_details}')
-        self.message = message
-        super().__init__(message)
+        super().__init__()
+
+    @staticmethod
+    def _parse_operation_response(operation_response: dict) -> Operation:
+        try:
+            return json_format.ParseDict(
+                operation_response,
+                Operation(),
+                ignore_unknown_fields=True,
+                descriptor_pool=error_details_pb2.DESCRIPTOR.pool)
+        except (json_format.Error, TypeError) as e:
+            # Swallow parsing errors if any. Building correct OperationError()
+            # is more important than losing debug information. Details still
+            # can be extracted from the warning.
+            logger.warning(
+                ("Can't parse response while processing OperationError: '%r', "
+                 "error %r"), operation_response, e)
+            return Operation()
+
+    def __str__(self):
+        indent_l1 = ' ' * 2
+        indent_l2 = indent_l1 * 2
+
+        result = (f'{self.api_name} operation "{self.name}" failed.\n'
+                  f'{indent_l1}code: {self.error.code} ({self.code_name})\n'
+                  f'{indent_l1}message: "{self.error.message}"')
+
+        if self.error.details:
+            result += f'\n{indent_l1}details: [\n'
+            for any_error in self.error.details:
+                error_str = json_format.MessageToJson(any_error)
+                for line in error_str.splitlines():
+                    result += indent_l2 + line + '\n'
+            result += f'{indent_l1}]'
+
+        if self.metadata:
+            result += f'\n  metadata: \n'
+            metadata_str = json.dumps(self.metadata, indent=2)
+            for line in metadata_str.splitlines():
+                result += indent_l2 + line + '\n'
+            result = result.rstrip()
+
+        return result
 
 
 class GcpProjectApiResource: