收集 Swimlane 平台記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage，將 Swimlane Platform 記錄檔擷取至 Google Security Operations。Swimlane Platform 是一套自動化安全性調度管理與應變 (SOAR) 平台，提供稽核記錄功能，可追蹤帳戶和租戶的使用者活動、設定變更和系統事件。

事前準備

請確認您已完成下列事前準備事項：

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
具備帳戶管理員權限，可存取稽核記錄的 Swimlane 平台高權限存取權
Swimlane Platform 執行個體網址和帳戶憑證

收集 Swimlane Platform 憑證

取得 Swimlane Platform 執行個體網址

登入 Swimlane Platform 執行個體。
記下瀏覽器網址列中的執行個體網址。
- 格式：https://<region>.swimlane.app (例如 https://us.swimlane.app 或 https://eu.swimlane.app)
- 範例：如果您透過 https://us.swimlane.app/workspace 存取 Swimlane，則基準網址為 https://us.swimlane.app

建立個人存取權杖

以帳戶管理員身分登入 Swimlane 平台。
前往「設定檔選項」。
按一下「個人資料」開啟個人資料編輯器。
前往「Personal Access Token」專區。
按一下「產生權杖」，建立新的個人存取權杖。
請立即複製並妥善儲存權杖 (系統不會再顯示)。

取得帳戶 ID

如果您不知道帳戶 ID，請與 Swimlane 管理員聯絡。稽核記錄 API 路徑需要帳戶 ID。

記錄整合項目的下列詳細資料：

個人存取權杖 (PAT)：用於 API 呼叫的 Private-Token 標頭。
帳戶 ID：稽核記錄 API 路徑 /api/public/audit/account/{ACCOUNT_ID}/auditlogs 的必要欄位。
基準網址：您的 Swimlane 網域 (例如 https://eu.swimlane.app、https://us.swimlane.app)。

驗證權限

如要確認帳戶是否具備存取稽核記錄的必要權限，請按照下列步驟操作：

登入 Swimlane Platform。
確認您擁有帳戶管理員存取權。
如果無法存取稽核記錄功能，請與 Swimlane 管理員聯絡。

注意： 您必須在 Swimlane 平台中具備帳戶管理員權限，才能存取稽核記錄。只有帳戶管理員可以存取稽核記錄 API。

測試 API 存取權

請先確認 API 憑證可正常運作，再繼續進行整合：

# Replace with your actual credentials
SWIMLANE_BASE_URL="https://<region>.swimlane.app"
SWIMLANE_ACCOUNT_ID="<your-account-id>"
SWIMLANE_PAT_TOKEN="<your-personal-access-token>"

# Test API access
curl -v -X GET "${SWIMLANE_BASE_URL}/api/public/audit/account/${SWIMLANE_ACCOUNT_ID}/auditlogs?pageNumber=1&pageSize=10" \
  -H "Private-Token: ${SWIMLANE_PAT_TOKEN}" \
  -H "Accept: application/json"

預期回應：HTTP 200，其中包含稽核記錄的 JSON。

如果收到錯誤訊息，請按照下列步驟操作：

HTTP 401：確認個人存取權杖是否正確
HTTP 403：確認帳戶具備帳戶管理員權限
HTTP 404：確認帳戶 ID 和基本網址是否正確

建立 Google Cloud Storage 值區

前往 Google Cloud 控制台。
選取專案或建立新專案。
在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。

請提供下列設定詳細資料：

設定	值
為 bucket 命名	輸入全域不重複的名稱 (例如 `swimlane-audit`)
位置類型	根據需求選擇 (區域、雙區域、多區域)
位置	選取位置 (例如 `us-central1`)
儲存空間級別	標準 (建議用於經常存取的記錄)
存取控管	統一 (建議)
保護工具	選用：啟用物件版本管理或保留政策

點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
按一下 [Create Service Account] (建立服務帳戶)。
請提供下列設定詳細資料：
- 服務帳戶名稱：輸入 swimlane-audit-collector-sa。
- 服務帳戶說明：輸入 Service account for Cloud Run function to collect Swimlane Platform logs。
按一下「建立並繼續」。
在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
1. 按一下「選擇角色」。
2. 搜尋並選取「Storage 物件管理員」。
3. 點選「+ 新增其他角色」。
4. 搜尋並選取「Cloud Run Invoker」。
5. 點選「+ 新增其他角色」。
6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
按一下「繼續」。
按一下 [完成]。

這些角色適用於：

Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
Cloud Run 叫用者：允許 Pub/Sub 叫用函式
Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

授予服務帳戶 GCS bucket 的寫入權限：

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如 swimlane-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
- 指派角色：選取「Storage 物件管理員」。
按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
按一下「建立主題」。
請提供下列設定詳細資料：
- 主題 ID：輸入 swimlane-audit-trigger。
- 其他設定保留預設值。
點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Swimlane Platform API 擷取記錄，並將記錄寫入 GCS。

前往 GCP Console 的「Cloud Run」。
按一下「Create service」(建立服務)。
選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：

設定值

服務名稱 swimlane-audit-collector

區域選取與 GCS bucket 相符的區域 (例如 us-central1)

執行階段 選取「Python 3.12」以上版本
在「Trigger (optional)」(觸發條件 (選用)) 專區：
1. 按一下「+ 新增觸發條件」。
2. 選取「Cloud Pub/Sub」。
3. 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (swimlane-audit-trigger)。
4. 按一下 [儲存]。
在「Authentication」(驗證) 部分：
1. 選取「需要驗證」。
2. 檢查 Identity and Access Management (IAM)。
注意： 叫用函式時，Pub/Sub 會自動處理驗證。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (swimlane-audit-collector-sa)。

設定	值
服務名稱	`swimlane-audit-collector`
區域	選取與 GCS bucket 相符的區域 (例如 `us-central1`)
執行階段	選取「Python 3.12」以上版本

前往「容器」分頁：

按一下「變數與密鑰」。
針對每個環境變數，按一下「+ 新增變數」：

變數名稱	範例值	說明
`GCS_BUCKET`	`swimlane-audit`	GCS bucket 名稱
`GCS_PREFIX`	`swimlane/audit/`	記錄檔的前置字串
`STATE_KEY`	`swimlane/audit/state.json`	狀態檔案路徑
`SWIMLANE_BASE_URL`	`https://us.swimlane.app`	Swimlane Platform 基準網址
`SWIMLANE_PAT_TOKEN`	`your-personal-access-token`	Swimlane 個人存取權杖
`SWIMLANE_ACCOUNT_ID`	`your-account-id`	Swimlane 帳戶 ID
`SWIMLANE_TENANT_LIST`	``	以半形逗號分隔的房客 ID (選用，留空即代表適用於所有房客)
`INCLUDE_ACCOUNT`	`true`	是否要納入帳戶層級記錄 (是/否)
`PAGE_SIZE`	`100`	每頁記錄數 (最多 100 筆)
`LOOKBACK_HOURS`	`24`	初始回溯期
`TIMEOUT`	`30`	API 要求逾時時間 (以秒為單位)

在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入 600 秒 (10 分鐘)。
前往「設定」分頁：
- 在「資源」部分：
  - 記憶體：選取 512 MiB 以上。
  - CPU：選取 1。
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入 0。
- 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

在「Function entry point」(函式進入點) 中輸入 main

在內嵌程式碼編輯器中建立兩個檔案：

第一個檔案：main.py:

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time
import uuid
import gzip
import io

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'swimlane/audit/')
STATE_KEY = os.environ.get('STATE_KEY', 'swimlane/audit/state.json')
SWIMLANE_BASE_URL = os.environ.get('SWIMLANE_BASE_URL', '').rstrip('/')
SWIMLANE_PAT_TOKEN = os.environ.get('SWIMLANE_PAT_TOKEN')
SWIMLANE_ACCOUNT_ID = os.environ.get('SWIMLANE_ACCOUNT_ID')
SWIMLANE_TENANT_LIST = os.environ.get('SWIMLANE_TENANT_LIST', '')
INCLUDE_ACCOUNT = os.environ.get('INCLUDE_ACCOUNT', 'true').lower() == 'true'
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
TIMEOUT = int(os.environ.get('TIMEOUT', '30'))

def parse_datetime(value: str) -> datetime:
    """Parse ISO datetime string to datetime object."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Swimlane Platform logs and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    if not all([GCS_BUCKET, SWIMLANE_BASE_URL, SWIMLANE_PAT_TOKEN, SWIMLANE_ACCOUNT_ID]):
        print('Error: Missing required environment variables (GCS_BUCKET, SWIMLANE_BASE_URL, SWIMLANE_PAT_TOKEN, SWIMLANE_ACCOUNT_ID)')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(GCS_BUCKET)

        # Load state
        state = load_state(bucket, STATE_KEY)

        # Determine time window
        now = datetime.now(timezone.utc)
        last_time = None

        if isinstance(state, dict) and state.get("last_event_time"):
            try:
                last_time = parse_datetime(state["last_event_time"])
                # Overlap by 2 minutes to catch any delayed events
                last_time = last_time - timedelta(minutes=2)
            except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")

        if last_time is None:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

        # Fetch logs
        records, newest_event_time = fetch_logs(
            base_url=SWIMLANE_BASE_URL,
            pat_token=SWIMLANE_PAT_TOKEN,
            account_id=SWIMLANE_ACCOUNT_ID,
            tenant_list=SWIMLANE_TENANT_LIST,
            include_account=INCLUDE_ACCOUNT,
            start_time=last_time,
            end_time=now,
            page_size=PAGE_SIZE,
        )

        if not records:
            print("No new log records found.")
            save_state(bucket, STATE_KEY, now.isoformat())
            return

        # Write to GCS as gzipped NDJSON
        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}{now:%Y/%m/%d}/swimlane-audit-{uuid.uuid4()}.json.gz"

        buf = io.BytesIO()
        with gzip.GzipFile(fileobj=buf, mode='w') as gz:
            for record in records:
                gz.write((json.dumps(record, ensure_ascii=False) + '\n').encode())

        buf.seek(0)
        blob = bucket.blob(object_key)
        blob.upload_from_file(buf, content_type='application/gzip')

        print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")

        # Update state with newest event time
        if newest_event_time:
            save_state(bucket, STATE_KEY, newest_event_time)
        else:
            save_state(bucket, STATE_KEY, now.isoformat())

        print(f"Successfully processed {len(records)} records")

    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

def load_state(bucket, key):
    """Load state from GCS."""
    try:
        blob = bucket.blob(key)
        if blob.exists():
            state_data = blob.download_as_text()
            return json.loads(state_data)
    except Exception as e:
        print(f"Warning: Could not load state: {e}")

    return {}

def save_state(bucket, key, last_event_time_iso: str):
    """Save the last event timestamp to GCS state file."""
    try:
        state = {
            'last_event_time': last_event_time_iso,
            'updated_at': datetime.now(timezone.utc).isoformat() + 'Z'
        }
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: last_event_time={last_event_time_iso}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

def fetch_logs(base_url: str, pat_token: str, account_id: str, tenant_list: str, include_account: bool, start_time: datetime, end_time: datetime, page_size: int):
    """
    Fetch logs from Swimlane Platform API with pagination and rate limiting.

    Args:
        base_url: Swimlane Platform base URL
        pat_token: Personal Access Token
        account_id: Swimlane account identifier
        tenant_list: Comma-separated tenant IDs (optional)
        include_account: Include account-level logs
        start_time: Start time for log query
        end_time: End time for log query
        page_size: Number of records per page (max 100)

    Returns:
        Tuple of (records list, newest_event_time ISO string)
    """

    endpoint = f"{base_url}/api/public/audit/account/{account_id}/auditlogs"

    headers = {
        'Private-Token': pat_token,
        'Accept': 'application/json',
        'Content-Type': 'application/json',
        'User-Agent': 'GoogleSecOps-SwimlaneCollector/1.0'
    }

    records = []
    newest_time = None
    page_num = 1
    backoff = 1.0

    while True:
        params = []
        params.append(f"pageNumber={page_num}")
        params.append(f"pageSize={min(page_size, 100)}")
        params.append(f"fromdate={start_time.isoformat()}")
        params.append(f"todate={end_time.isoformat()}")

        if tenant_list:
            params.append(f"tenantList={tenant_list}")

        params.append(f"includeAccount={'true' if include_account else 'false'}")

        url = f"{endpoint}?{'&'.join(params)}"

        try:
            response = http.request('GET', url, headers=headers, timeout=TIMEOUT)

            # Handle rate limiting with exponential backoff
            if response.status == 429:
                retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                print(f"Rate limited (429). Retrying after {retry_after}s...")
                time.sleep(retry_after)
                backoff = min(backoff * 2, 30.0)
                continue

            backoff = 1.0

            if response.status == 401:
                print(f"Authentication failed (401). Verify SWIMLANE_PAT_TOKEN is correct.")
                return [], None

            if response.status == 403:
                print(f"Access forbidden (403). Verify account has Account Admin permissions to access audit logs.")
                return [], None

            if response.status == 400:
                print(f"Bad request (400). Verify account_id and query parameters are correct.")
                response_text = response.data.decode('utf-8')
                print(f"Response body: {response_text}")
                return [], None

            if response.status != 200:
                print(f"HTTP Error: {response.status}")
                response_text = response.data.decode('utf-8')
                print(f"Response body: {response_text}")
                return [], None

            data = json.loads(response.data.decode('utf-8'))

            page_results = data.get('auditlogs', [])

            if not page_results:
                print(f"No more results (empty page)")
                break

            print(f"Page {page_num}: Retrieved {len(page_results)} events")
            records.extend(page_results)

            # Track newest event time
            for event in page_results:
                try:
                    event_time = event.get('eventTime') or event.get('EventTime')
                    if event_time:
                        if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                            newest_time = event_time
                except Exception as e:
                    print(f"Warning: Could not parse event time: {e}")

            # Check for more results
            has_next = data.get('next')
            total_count = data.get('totalCount', 0)

            if not has_next:
                print(f"Reached last page (no next link)")
                break

            # Check if we've hit the 10,000 log limit
            if total_count > 10000 and len(records) >= 10000:
                print(f"Warning: Reached Swimlane API limit of 10,000 logs. Consider narrowing the time range.")
                break

            page_num += 1

        except Exception as e:
            print(f"Error fetching logs: {e}")
            return [], None

    print(f"Retrieved {len(records)} total records from {page_num} pages")
    return records, newest_time

第二個檔案：requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

點選「部署」來儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。

注意： Swimlane 稽核記錄 API 的限制為每項要求 10,000 筆記錄。如果記錄檔量很大，請調整 LOOKBACK_HOURS 和 Cloud Scheduler 的頻率，確保不超過這個限制。
注意： Swimlane 稽核記錄的資料保留期限為 7 天。請確保 Cloud Scheduler 執行頻率夠高，可在所有記錄過期前擷取。
注意： Pub/Sub 觸發條件設定會自動建立必要的訂閱項目和權限。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。

前往 GCP 主控台的「Cloud Scheduler」。
點選「建立工作」。

請提供下列設定詳細資料：

設定	值
名稱	`swimlane-audit-schedule-15min`
區域	選取與 Cloud Run 函式相同的區域
頻率	`/15 * * *` (每 15 分鐘)
時區	選取時區 (建議使用世界標準時間)
目標類型	Pub/Sub
主題	選取 Pub/Sub 主題 (`swimlane-audit-trigger`)
郵件內文	`{}` (空白 JSON 物件)

點選「建立」。

排程頻率選項

根據記錄檔量和延遲時間要求選擇頻率：

頻率	Cron 運算式	用途
每 5 分鐘	`/5 * * *`	高容量、低延遲
每 15 分鐘檢查一次	`/15 * * *`	標準 (建議採用)
每小時	`0 * * * *`	普通量
每 6 小時	`0 /6 * *`	少量、批次處理
每日	`0 0 * * *`	歷來資料集合

測試整合項目

在 Cloud Scheduler 控制台中找出您的工作。
按一下「強制執行」，手動觸發工作。
稍等幾秒鐘。
前往「Cloud Run」>「Services」。
按一下函式名稱 (swimlane-audit-collector)。
按一下 [Logs] (記錄) 分頁標籤。

確認函式是否已順利執行。請找出以下項目：

Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X events
Wrote X records to gs://bucket-name/swimlane/audit/YYYY/MM/DD/swimlane-audit-UUID.json.gz
Successfully processed X records

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往前置字元資料夾 (swimlane/audit/)。
確認是否已建立含有目前時間戳記的新 .json.gz 檔案。

如果在記錄中發現錯誤：

HTTP 401：檢查環境變數中的 SWIMLANE_PAT_TOKEN，並確認個人存取權杖是否正確
HTTP 403：確認帳戶是否具備存取稽核記錄的帳戶管理員權限
HTTP 400：確認 SWIMLANE_ACCOUNT_ID 正確無誤，且查詢參數有效
HTTP 404：確認 SWIMLANE_BASE_URL 和 API 端點路徑是否正確
HTTP 429：頻率限制 - 函式會自動重試並延遲
缺少環境變數：檢查是否已設定所有必要變數 (GCS_BUCKET、SWIMLANE_BASE_URL、SWIMLANE_PAT_TOKEN、SWIMLANE_ACCOUNT_ID)
連線錯誤：確認與 Swimlane 平台的網路連線和防火牆規則
記錄上限為 10,000 筆的警告：減少 LOOKBACK_HOURS 或提高 Cloud Scheduler 頻率，確保不超過 Swimlane 的 API 限制

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Swimlane Platform logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Swimlane Platform」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
複製這個電子郵件地址，以便在下一步中使用。

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。

注意： 如果打算使用「刪除已轉移的檔案」或「刪除已轉移的檔案並清空目錄」刪除選項，請授予「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。

在 Google SecOps 中設定資訊提供，擷取 Swimlane Platform 記錄

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Swimlane Platform logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Swimlane Platform」做為「記錄類型」。
點選 [下一步]。
指定下列輸入參數的值：
- 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
```
gs://swimlane-audit/swimlane/audit/
```
  - 取代：
    - swimlane-audit：您的 GCS bucket 名稱。
    - swimlane/audit/：儲存記錄的前置字元/資料夾路徑。
  注意： URI 結尾一律要加上尾端斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項：
  - 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
  - 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
  - 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
    
    注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。