fix: import log to destination project (GoogleCloudPlatform#11005)

* fix: import log to destination project

patch logName to have destination project.
store original logName as user label.
refactor code to replace double quotes with single quotes in all literals.
refactor method comments.
refactor handling partial errors of type WriteLogEntriesPartialErrors.

* fix: update README to address code changes

Add clarifications about the changes to log's logName field.
Update references to the lines in code with the new commit hashes.

* fix: update mock object

add .project property to mocked logging client to support changes in code

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

* chore(fix): cosmetic changes

fix english in the code comment
fix line numbers in README to reflect recent code refactoring.

* chore(fix): convert quote to double quote

* Update logging/import-logs/main.py

---------

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
Co-authored-by: Maciej Strzelczyk <[email protected]>

Author: 3 people

logging/import-logs/README.md

@@ -9,6 +9,19 @@ solution in [documentation].
 
 [documentation]: (LINK-TO-REFERENCE-ARCHITECTURE-ARTICLE)-->
 
+All logs will be imported to the project that runs Cloud Run job.
+Alternatively you can explicitly configure the project ID where you want logs to be imported.
+The logs will be imported with log name `imported_logs`.
+It means that the `logName` field of the imported logs will be:
+
+```text
+projects/PROJECT_ID/logs/imported_logs
+```
+
+where `PROJECT_ID` is the project ID of the project where you import the logs.
+
+The original log name will be stored in the user labels under the key `original_logName` (mind the casing).
+
 ## Requirements
 
 You have to grant the following permissions to a service account that you will
@@ -47,6 +60,7 @@ have to be configured when a new Cloud Run job for the solution is created:
 | END_DATE | A string in format `MM/DD/YYYY` that defines the last day of the import range |
 | LOG_ID | A string that identifies the particular log to be imported. See [documentation][logid] for more details. |
 | STORAGE_BUCKET_NAME | A name of the storage bucket where the exported logs are stored. |
+| PROJECT_ID | (Optional) If you want to explicitly define destination project other than one your import job is deployed |
 
 <!--Read [documentation] for more information about Cloud Run job setup.-->
 
@@ -94,33 +108,25 @@ nox -s py-3.11
 
 ## Importing log entries with timestamps older than 30 days
 
-The import logs implementation does not currently support logs with the timestamp older than 30 days
-because the incoming log entries must not be older than default retention period (see [documentation][retention]).
-To prevent such logs from being ignored the implementation returns errors if the timestamp is older than 29 days.
+The incoming log entries that are older than default retention period (i.e. 30 days) are not ingested.
+You can see [documentation][retention] for more details. The code takes a safety margin of an extra day
+to prevent importing old logs that cannot be ingested.
 To import the older logs you have to modified the [current] code by performing the following modifications:
 
-* remove the fencing condition [lines][code1]:
-
-  <https://github.com/GoogleCloudPlatform/python-docs-samples/blob/95dd4f53ff96470a1f842d3134d56b017a85ac27/logging/import-logs/main.py#L91-L93>
+* comment the fencing condition [lines][code1]:
 
-* in [`import_logs`][code2] add the following block after the call to [`_patch_reserved_log_ids`][code3]:
-  
-  ```python
-  log.labels['original_timestamp'] = log.timestamp
-  log.timestamp = None
-  ```
+  <https://github.com/GoogleCloudPlatform/python-docs-samples/blob/86f12a752a4171e137adaa855c7247be9d5d39a2/logging/import-logs/main.py#L81-L83>
 
-  to keep the original timestamp as a user metadata labeled `original_timestamp` and to ingest the log entry using _current_ timestamp.
+* uncomment [2 lines][code2] to keep the original timestamp as a user label and to reset the `timestamp` field of the imported log entries.
 
 > [!IMPORTANT]  
 >
-> 1. After this change the logs should be queried using the `timestamp` field.
+> 1. To query the imported logs by timestamp you will have to use the label `original_timestamp` instead of the `timestamp` field.
 > 1. If the same log entry is imported multiple times, the query response may include more than one line.
 
 After applying the changes, [build](#build) a custom container image and use it when creating an import job.
 
 [retention]: https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#FIELDS.timestamp
 [current]: https://github.com/GoogleCloudPlatform/python-docs-samples/blob/e2709a218072c86ec1a9b9101db45057ebfdbff0/logging/import-logs/main.py
-[code1]: https://github.com/GoogleCloudPlatform/python-docs-samples/blob/95dd4f53ff96470a1f842d3134d56b017a85ac27/logging/import-logs/main.py#L91-L93
-[code2]: https://github.com/GoogleCloudPlatform/python-docs-samples/blob/95dd4f53ff96470a1f842d3134d56b017a85ac27/logging/import-logs/main.py#L196
-[code3]: https://github.com/GoogleCloudPlatform/python-docs-samples/blob/95dd4f53ff96470a1f842d3134d56b017a85ac27/logging/import-logs/main.py#L206
+[code1]: https://github.com/GoogleCloudPlatform/python-docs-samples/blob/86f12a752a4171e137adaa855c7247be9d5d39a2/logging/import-logs/main.py#L81-L83
+[code2]: https://github.com/GoogleCloudPlatform/python-docs-samples/blob/86f12a752a4171e137adaa855c7247be9d5d39a2/logging/import-logs/main.py#L186-L187

logging/import-logs/main.py

@@ -19,7 +19,6 @@
 import json
 import math
 import os
-import re
 import sys
 
 from typing import List, Tuple, TypedDict
@@ -30,19 +29,6 @@
 # Logging limits (https://cloud.google.com/logging/quotas#api-limits)
 _LOGS_MAX_SIZE_BYTES = 9 * 1024 * 1024  # < 10MB
 
-_RESERVED_LOG_IDS = ["cloudaudit.googleapis.com"]
-_LOGGER_NAME_TEMPLATE = re.compile(
-    r"""
-    (projects/)     # static prefix group (1)
-    ([^/]+)         # initial letter, wordchars group (2) for project ID
-    (/logs/)        # static midfix group (3)
-    (?P<name>[^/]+) # initial letter, wordchars group for LOG_ID
-""",
-    re.VERBOSE,
-)
-
-_LOG_ID_PREFIX = "_"  # allowed characters are r"[a-zA-Z_\-\.\/\\]
-
 # Read Cloud Run environment variables
 TASK_INDEX = int(os.getenv("CLOUD_RUN_TASK_INDEX", "0"))
 TASK_COUNT = int(os.getenv("CLOUD_RUN_TASK_COUNT", "1"))
@@ -70,24 +56,28 @@ def eprint(*objects: str, **kwargs: TypedDict) -> None:
 
 
 def _day(blob_name: str) -> int:
-    """Parse day number from Blob's name
-    using the following Blob name convention:
-    <LOG_ID>/YYYY/MM/DD/<OBJECT_NAME>
+    """Parse day number from Blob's path
+
+    Use the known Blob path convention to parse the day part from the path.
+    The path convention is <LOG_ID>/YYYY/MM/DD/<OBJECT_NAME>
     """
     # calculated in function to allow test to set LOG_ID
     offset = len(LOG_ID) + 1 + 4 + 1 + 2 + 1
     return int(blob_name[offset : offset + 2])
 
 
 def _is_valid_import_range() -> bool:
-    """Check the import range dates to ensure that
+    """Validate the import date range
+
+    Checks the import range dates to ensure that
     - start date is earlier than end date
-    - no dates in the range is older than 29 days
-    (for reason see https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#FIELDS.timestamp)
+    - no dates in the range are older than 29 days
+     due to https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#FIELDS.timestamp
     """
     if START_DATE > END_DATE:
         eprint("Start date of the import time range should be earlier than end date")
         return False
+    # comment the following 3 lines if import range includes dates older than 29 days from now
     if (date.today() - START_DATE).days > 29:
         eprint("Import range includes dates older than 29 days from today.")
         return False
@@ -170,27 +160,31 @@ def _write_logs(logs: List[dict], client: logging_v2.Client) -> None:
     try:
         client.logging_api.write_entries(logs)
     except exceptions.PermissionDenied as err2:
-        partialerrors = logging_v2.types.WriteLogEntriesPartialErrors()
         for detail in err2.details:
-            if detail.Unpack(partialerrors):
+            if isinstance(detail, logging_v2.types.WriteLogEntriesPartialErrors):
                 # partialerrors.log_entry_errors is a dictionary
                 # keyed by the logs' zero-based index in the logs.
                 # consider implementing custom error handling
-                eprint(json.dumps(partialerrors.log_entry_errors))
+                eprint(f"{detail}")
         raise
 
 
-def _patch_reserved_log_ids(log: dict) -> None:
-    """Replaces first character in LOG_ID with underscore for reserved LOG_ID prefixes"""
+def _patch_entry(log: dict, project_id: str) -> None:
+    """Modify entry fields to allow importing entry to destination project.
+
+    Save logName as a user label.
+    Replace logName with the fixed value "projects/PROJECT_ID/logs/imported_logs"
+    """
     log_name = log.get("logName")
-    if log_name:
-        match = _LOGGER_NAME_TEMPLATE.match(log_name)
-        log_id = match.group("name")
-        if log_id and log_id.startswith(tuple(_RESERVED_LOG_IDS)):
-            log_name = _LOGGER_NAME_TEMPLATE.sub(
-                f"\\g<1>\\g<2>\\g<3>{_LOG_ID_PREFIX + log_id[1:]}", log_name
-            )
-            log["logName"] = log_name
+    labels = log.get("labels")
+    log["logName"] = f"projects/{project_id}/logs/imported_logs"
+    if not labels:
+        labels = dict()
+        log["labels"] = labels
+    labels["original_logName"] = log_name
+    # uncomment the following 2 lines if import range includes dates older than 29 days from now
+    # labels["original_timestamp"] = log["timestamp"]
+    # log["timestamp"] = None
 
 
 def import_logs(
@@ -203,7 +197,7 @@ def import_logs(
         data = _read_logs(file_path, bucket)
         for entry in data:
             log = json.loads(entry)
-            _patch_reserved_log_ids(log)
+            _patch_entry(log, logging_client.project)
             size = sys.getsizeof(log)
             if total_size + size >= _LOGS_MAX_SIZE_BYTES:
                 _write_logs(logs, logging_client)
@@ -244,5 +238,5 @@ def main() -> None:
     try:
         main()
     except Exception as err:
-        eprint(f"Task #{TASK_INDEX}, failed: {str(err)}")
+        eprint(f"Task #{TASK_INDEX+1}, failed: {err}")
         sys.exit(1)

logging/import-logs/main_test.py

@@ -33,6 +33,7 @@
 TEST_LOG_ID = "test-log"
 TEST_BUCKET = "test-bucket"
 TEST_BUCKET_NAME = f"gs://{TEST_BUCKET}"
+TEST_PROJECT_ID = "test-project-id"
 
 
 def _setup_environment(
@@ -339,6 +340,7 @@ def test_import_logs(
     mocked_bucket.blob = MagicMock(side_effect=_args_based_blob_return)
     mocked_logging_client = MagicMock(spec=logging_v2.Client)
     mocked_logging_client.logging_api = MagicMock()
+    mocked_logging_client.project = TEST_PROJECT_ID
     mocked_write_entries = mocked_logging_client.logging_api.write_entries = MagicMock()
 
     main.import_logs(log_files, mocked_storage_client, mocked_logging_client)