收集 Sentry 記錄

支援的國家/地區：

Google SecOps SIEM

注意： 這項功能受《Google Security Operations 服務專屬條款》的《正式發布前產品條款》規範。正式發布前的功能或許只提供部分支援，而且這類功能的變更可能與其他正式發布前版本不相容。詳情請參閱《Google SecOps 技術支援服務指南》和《Google SecOps 服務專屬條款》。

本文說明如何使用 Google Cloud Storage，將 Sentry 記錄檔擷取至 Google Security Operations。Sentry 會以事件、問題、效能監控資料和錯誤追蹤資訊的形式，產生作業資料。整合後，您可以將這些記錄傳送至 Google SecOps 進行分析和監控，進一步瞭解 Sentry 監控應用程式中的應用程式錯誤、效能問題和使用者互動。

事前準備

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

• Google SecOps 執行個體
• 已啟用 Cloud Storage API 的 GCP 專案
• 建立及管理 GCS 值區的權限
• 管理 Google Cloud Storage 值區 IAM 政策的權限
• 建立 Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限
• Sentry 租戶的特殊存取權 (具有 API 範圍的驗證權杖)

收集 Sentry 必要條件 (ID、API 金鑰、機構 ID、權杖)

1. 登入 Sentry。
2. 找出機構代碼：
   • 依序前往「設定」>「機構」>「設定」>「機構 ID」 (機構名稱旁邊會顯示機構 ID)。
3. 建立「驗證權杖」：
   1. 依序前往「設定」>「開發人員設定」>「個人權杖」。
   2. 按一下「建立新權杖」。
   3. 範圍 (最低)：org:read、project:read、event:read。
   4. 按一下「建立權杖」。
   5. 複製權杖值 (只會顯示一次)。這項資訊會做為：Authorization: Bearer <token>。

4. (如果是自行代管)：請記下基準網址 (例如 https://<your-domain>)；否則請使用 https://sentry.io。

   注意： 為防止濫用，Sentry 會限制每項 API 要求的速率，並針對每個呼叫端和端點的專屬組合套用限制。每個端點都有自己的要求數量上限和時間範圍。Cloud Run 函式會實作指數輪詢，自動處理速率限制。
   注意： 錯誤記錄在免費方案中會保留 30 天，在 Team/Business 方案中則會保留 30 天。此外，Sentry 也會在 Business 方案中保留取樣範圍和交易 13 個月。這會限制首次設定整合時可用的回填歷史資料。

建立 Google Cloud Storage 值區

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

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

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

6. 點選「建立」。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

這些角色適用於：

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

授予 GCS 值區的 IAM 權限

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

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

建立 Pub/Sub 主題

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

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

建立 Cloud Run 函式來收集記錄

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

1. 前往 GCP Console 的「Cloud Run」。
2. 按一下「Create service」(建立服務)。
3. 選取「函式」 (使用內嵌編輯器建立函式)。

4. 在「設定」部分，提供下列設定詳細資料：

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

5. 在「Trigger (optional)」(觸發條件 (選用)) 專區：

   1. 按一下「+ 新增觸發條件」。
   2. 選取「Cloud Pub/Sub」。
   3. 在「選取 Cloud Pub/Sub 主題」中，選擇主題 (sentry-logs-trigger)。
   4. 按一下 [儲存]。

6. 在「Authentication」(驗證) 部分：

   1. 選取「需要驗證」。
   2. 檢查 Identity and Access Management (IAM)。
   注意： 叫用函式時，Pub/Sub 會自動處理驗證。

7. 向下捲動並展開「Containers, Networking, Security」。

8. 前往「安全性」分頁：

   • 服務帳戶：選取服務帳戶 (sentry-logs-collector-sa)。

9. 前往「容器」分頁：

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

   變數名稱	範例值	說明
   GCS_BUCKET	sentry-logs	要儲存資料的 GCS bucket 名稱。
   GCS_PREFIX	sentry/events/	物件的選填 GCS 前置字串 (子資料夾)。
   STATE_KEY	sentry/events/state.json	選用的狀態/檢查點檔案金鑰。
   SENTRY_ORG	your-org-slug	Sentry 機構代碼。
   SENTRY_AUTH_TOKEN	sntrys_************************	Sentry 驗證權杖，具有 org:read、project:read、event:read。
   SENTRY_API_BASE	https://sentry.io	Sentry API 基礎網址 (自架式：https://<your-domain>)。
   MAX_PROJECTS	100	要處理的專案數量上限。
   MAX_PAGES_PER_PROJECT	5	每個專案每次執行作業的頁數上限。

10. 在「變數與密鑰」分頁中向下捲動至「要求」：

    • 要求逾時：輸入 600 秒 (10 分鐘)。

11. 前往「容器」中的「設定」分頁：

    • 在「資源」部分：
      • 記憶體：選取 512 MiB 以上。
      • CPU：選取 1。
    • 按一下 [完成]。

12. 捲動至「執行環境」：

    • 選取「預設」 (建議選項)。

13. 在「修訂版本資源調度」部分：

    • 執行個體數量下限：輸入 0。
    • 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。

14. 點選「建立」。

15. 等待服務建立完成 (1 到 2 分鐘)。

16. 服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

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

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

   • 第一個檔案：main.py:

    import functions_framework
    from google.cloud import storage
    import json
    import os
    import urllib3
    from datetime import datetime, timezone
    import time
   
    # Initialize HTTP client
    http = urllib3.PoolManager()
   
    # Initialize Storage client
    storage_client = storage.Client()
   
    @functions_framework.cloud_event
    def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch Sentry events and write to GCS.
   
        Args:
            cloud_event: CloudEvent object containing Pub/Sub message
        """
   
        # Get environment variables
        bucket_name = os.environ.get('GCS_BUCKET')
        prefix = os.environ.get('GCS_PREFIX', 'sentry/events/')
        state_key = os.environ.get('STATE_KEY', 'sentry/events/state.json')
        org = os.environ.get('SENTRY_ORG', '').strip()
        token = os.environ.get('SENTRY_AUTH_TOKEN', '').strip()
        api_base = os.environ.get('SENTRY_API_BASE', 'https://sentry.io').rstrip('/')
        max_projects = int(os.environ.get('MAX_PROJECTS', '100'))
        max_pages_per_project = int(os.environ.get('MAX_PAGES_PER_PROJECT', '5'))
   
        if not all([bucket_name, org, token]):
            print('Error: Missing required environment variables')
            return
   
        try:
            # Get GCS bucket
            bucket = storage_client.bucket(bucket_name)
   
            # Load state
            state = load_state(bucket, state_key)
            state.setdefault('projects', {})
   
            # Get list of projects
            projects = list_projects(api_base, org, token, max_projects)
            print(f'Found {len(projects)} projects')
   
            summary = []
   
            # Process each project
            for slug in projects:
                start_prev = state['projects'].get(slug, {}).get('prev_cursor')
                res = fetch_project_events(
                    api_base, org, token, slug, start_prev, 
                    max_pages_per_project, bucket, prefix
                )
   
                if res.get('store_prev_cursor'):
                    state['projects'][slug] = {'prev_cursor': res['store_prev_cursor']}
   
                summary.append(res)
   
            # Save state
            save_state(bucket, state_key, state)
   
            print(f'Successfully processed {len(projects)} projects')
            print(f'Summary: {json.dumps(summary)}')
   
        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) if state_data else {'projects': {}}
        except Exception as e:
            print(f'Warning: Could not load state: {str(e)}')
        return {'projects': {}}
   
    def save_state(bucket, key, state):
        """Save state to GCS."""
        try:
            blob = bucket.blob(key)
            blob.upload_from_string(
                json.dumps(state, separators=(',', ':')),
                content_type='application/json'
            )
        except Exception as e:
            print(f'Warning: Could not save state: {str(e)}')
   
    def sentry_request(api_base, token, path, params=None):
        """Make request to Sentry API."""
        url = f"{api_base}{path}"
        if params:
            url = f"{url}?{urllib3.request.urlencode(params)}"
   
        headers = {
            'Authorization': f'Bearer {token}',
            'Accept': 'application/json',
            'User-Agent': 'chronicle-gcs-sentry-function/1.0'
        }
   
        response = http.request('GET', url, headers=headers, timeout=60.0)
        data = json.loads(response.data.decode('utf-8'))
        link = response.headers.get('Link')
   
        return data, link
   
    def parse_link_header(link_header):
        """Parse Link header to extract cursors."""
        if not link_header:
            return None, False, None, False
   
        prev_cursor, next_cursor = None, None
        prev_more, next_more = False, False
   
        parts = [p.strip() for p in link_header.split(',')]
        for p in parts:
            if '<' not in p or '>' not in p:
                continue
   
            url = p.split('<', 1)[1].split('>', 1)[0]
            rel = 'previous' if 'rel="previous"' in p else ('next' if 'rel="next"' in p else None)
            has_more = 'results="true"' in p
   
            try:
                from urllib.parse import urlparse, parse_qs
                q = urlparse(url).query
                cur = parse_qs(q).get('cursor', [None])[0]
            except Exception:
                cur = None
   
            if rel == 'previous':
                prev_cursor, prev_more = cur, has_more
            elif rel == 'next':
                next_cursor, next_more = cur, has_more
   
        return prev_cursor, prev_more, next_cursor, next_more
   
    def write_page(bucket, prefix, project_slug, payload, page_idx):
        """Write page of events to GCS."""
        ts = time.gmtime()
        key = f"{prefix.rstrip('/')}/{time.strftime('%Y/%m/%d', ts)}/sentry-{project_slug}-{page_idx:05d}.json"
   
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(payload, separators=(',', ':')),
            content_type='application/json'
        )
   
        return key
   
    def list_projects(api_base, org, token, max_projects):
        """List Sentry projects."""
        projects, cursor = [], None
   
        while len(projects) < max_projects:
            params = {'cursor': cursor} if cursor else {}
            data, link = sentry_request(api_base, token, f'/api/0/organizations/{org}/projects/', params)
   
            for p in data:
                slug = p.get('slug')
                if slug:
                    projects.append(slug)
                if len(projects) >= max_projects:
                    break
   
            _, _, next_cursor, next_more = parse_link_header(link)
            cursor = next_cursor if next_more else None
            if not next_more:
                break
   
        return projects
   
    def fetch_project_events(api_base, org, token, project_slug, start_prev_cursor, max_pages, bucket, prefix):
        """Fetch events for a project."""
        pages = 0
        total = 0
        latest_prev_cursor_to_store = None
   
        def fetch_one(cursor):
            nonlocal pages, total, latest_prev_cursor_to_store
   
            params = {'cursor': cursor} if cursor else {}
            data, link = sentry_request(api_base, token, f'/api/0/projects/{org}/{project_slug}/events/', params)
   
            write_page(bucket, prefix, project_slug, data, pages)
            total += len(data) if isinstance(data, list) else 0
   
            prev_c, prev_more, next_c, next_more = parse_link_header(link)
            latest_prev_cursor_to_store = prev_c or latest_prev_cursor_to_store
            pages += 1
   
            return prev_c, prev_more, next_c, next_more
   
        if start_prev_cursor:
            # Poll new pages toward "previous" until no more
            cur = start_prev_cursor
            while pages < max_pages:
                prev_c, prev_more, _, _ = fetch_one(cur)
                if not prev_more:
                    break
                cur = prev_c
        else:
            # First run: start at newest, then backfill older pages
            prev_c, _, next_c, next_more = fetch_one(None)
            cur = next_c
            while next_more and pages < max_pages:
                _, _, next_c, next_more = fetch_one(cur)
                cur = next_c
   
        return {
            'project': project_slug,
            'pages': pages,
            'written': total,
            'store_prev_cursor': latest_prev_cursor_to_store
        }
    ```
   
   * Second file: **requirements.txt:**
   
   

   functions-framework3.* google-cloud-storage2.* urllib3>=2.0.0 ```

3. 點選「部署」來儲存並部署函式。

4. 等待部署作業完成 (2 到 3 分鐘)。

   注意： Pub/Sub 觸發條件設定會自動建立必要的訂閱項目和權限。

建立 Cloud Scheduler 工作

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

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

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

   設定	值
   名稱	sentry-logs-collector-hourly
   區域	選取與 Cloud Run 函式相同的區域
   頻率	0 * * * * (每小時整點)
   時區	選取時區 (建議使用世界標準時間)
   目標類型	Pub/Sub
   主題	選取主題 (sentry-logs-trigger)
   郵件內文	{} (空白 JSON 物件)

4. 點選「建立」。

排程頻率選項

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

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

測試排程器工作

1. 在 Cloud Scheduler 控制台中找出您的工作。
2. 按一下「強制執行」即可手動觸發。
3. 等待幾秒鐘，然後依序前往「Cloud Run」>「Services」(服務) &gt「sentry-logs-collector」&gt「Logs」(記錄)。
4. 確認函式是否已順利執行。
5. 檢查 GCS 值區，確認是否已寫入記錄。

擷取 Google SecOps 服務帳戶

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

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

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「新增動態消息」。
3. 按一下「設定單一動態饋給」。
4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Sentry Logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 選取「Sentry」做為「記錄類型」。

7. 按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：

   [e-mail]
   

8. 複製這個電子郵件地址，以便在下一步中使用。

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

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

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

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

6. 按一下 [儲存]。

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

在 Google SecOps 中設定動態饋給，擷取 Sentry 記錄

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「新增動態消息」。
3. 按一下「設定單一動態饋給」。
4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Sentry Logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 選取「Sentry」做為「記錄類型」。
7. 點選 [下一步]。

8. 指定下列輸入參數的值：

   • 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：

     gs://sentry-logs/sentry/events/
     

     • 取代：

       • sentry-logs：您的 GCS bucket 名稱。
       • sentry/events/：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。

     • 範例：

       • 根層級 bucket：gs://company-logs/
       • 前置字串：gs://company-logs/sentry-logs/
       • 有子資料夾：gs://company-logs/sentry/events/
     注意： URI 結尾一律要加上尾端斜線 (/)。

   • 來源刪除選項：根據偏好設定選取刪除選項：

     • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
     • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。

     • 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。

       注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。

   • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。

   • 資產命名空間：資產命名空間。

   • 擷取標籤：要套用至這個動態饋給事件的標籤。

9. 點選 [下一步]。

10. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

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