Sentry ログを収集する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、Google Cloud Storage を使用して Sentry ログを Google Security Operations に取り込む方法について説明します。Sentry は、イベント、問題、パフォーマンスモニタリングデータ、エラートラッキング情報の形式で運用データを生成します。この統合により、これらのログを Google SecOps に送信して分析とモニタリングを行うことができます。これにより、Sentry でモニタリングされているアプリケーション内のアプリケーションエラー、パフォーマンスの問題、ユーザー操作を可視化できます。

始める前に

次の前提条件を満たしていることを確認します。

Google SecOps インスタンス
Cloud Storage API が有効になっている GCP プロジェクト
GCS バケットを作成および管理する権限
GCS バケットの IAM ポリシーを管理する権限
Cloud Run 関数、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
Sentry テナントへの特権アクセス（API スコープを含む認証トークン）

Sentry の前提条件（ID、API キー、組織 ID、トークン）を収集する

Sentry にログインします。
組織のスラッグを確認します。
- [設定] > [組織] > [設定] > [組織 ID]（組織名の横にスラッグが表示されます）に移動します。
認証トークンを作成します。
1. [Settings] > [Developer Settings] > [Personal Tokens] に移動します。
2. [Create New Token] をクリックします。
3. スコープ（最小）: org:read、project:read、event:read。
4. [トークンの作成] をクリックします。
5. トークンの値（1 回のみ表示）をコピーします。これは Authorization: Bearer <token> として使用されます。
（セルフホストの場合）ベース URL（例: https://<your-domain>）をメモします。それ以外の場合は https://sentry.io を使用します。

注: Sentry は、不正使用を防ぐためにすべての API リクエストをレート制限します。制限は呼び出し元とエンドポイントの固有の組み合わせごとに適用され、各エンドポイントには独自のリクエストの最大数とウィンドウサイズが設定されています。Cloud Run 関数は、レート制限を自動的に処理するために指数バックオフを実装します。
注: エラーログは、無料プランでは 30 日間、チーム/ビジネスプランでは 30 日間保存されます。Sentry は、ビジネスプランでサンプリングされたスパンとトランザクションも 13 か月間保持します。これにより、統合を初めて構成するときにバックフィルに使用できる履歴データが制限されます。

Google Cloud Storage バケットを作成する

Google Cloud Console に移動します。
プロジェクトを選択するか、新しいプロジェクトを作成します。
ナビゲーションメニューで、[Cloud Storage > バケット] に移動します。
[バケットを作成] をクリックします。

次の構成情報を提供してください。

設定	値
バケットに名前を付ける	グローバルに一意の名前（`sentry-logs` など）を入力します。
ロケーションタイプ	ニーズに基づいて選択します（リージョン、デュアルリージョン、マルチリージョン）。
ロケーション	ロケーションを選択します（例: `us-central1`）。
ストレージクラス	Standard（頻繁にアクセスされるログにおすすめ）
アクセス制御	均一（推奨）
保護ツール	省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

[作成] をクリックします。

Cloud Run functions のサービスアカウントを作成する

Cloud Run 関数には、GCS バケットに書き込み、Pub/Sub によって呼び出される権限を持つサービスアカウントが必要です。

サービスアカウントの作成

GCP Console で、[IAM と管理>サービスアカウント] に移動します。
[サービスアカウントを作成] をクリックします。
次の構成の詳細を指定します。
- サービスアカウント名: 「sentry-logs-collector-sa」と入力します。
- サービスアカウントの説明: 「Service account for Cloud Run function to collect Sentry logs」と入力します。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
1. [ロールを選択] をクリックします。
2. [ストレージオブジェクト管理者] を検索して選択します。
3. [+ 別のロールを追加] をクリックします。
4. [Cloud Run 起動元] を検索して選択します。
5. [+ 別のロールを追加] をクリックします。
6. [Cloud Functions 起動元] を検索して選択します。
[続行] をクリックします。
[完了] をクリックします。

これらのロールは、次の目的で必要です。

Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可する
Cloud Functions 起動元: 関数の呼び出しを許可する

GCS バケットに対する IAM 権限を付与する

GCS バケットに対する書き込み権限をサービスアカウントに付与します。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: サービスアカウントのメールアドレス（例: sentry-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
- ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run functions がサブスクライブする Pub/Sub トピックを作成します。

GCP Console で、[Pub/Sub> トピック] に移動します。
[トピックを作成] をクリックします。
次の構成の詳細を指定します。
- トピック ID: 「sentry-logs-trigger」と入力します。
- その他の設定はデフォルトのままにします。
[作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Sentry API からログを取得して GCS に書き込みます。

GCP Console で、[Cloud Run] に移動します。
[サービスを作成] をクリックします。
[関数] を選択します（インラインエディタを使用して関数を作成します）。

[構成] セクションで、次の構成の詳細を指定します。

設定	値
サービス名	`sentry-logs-collector`
リージョン	GCS バケットと一致するリージョンを選択します（例: `us-central1`）。
ランタイム	[Python 3.12] 以降を選択します。

[トリガー（省略可）] セクションで、次の操作を行います。
1. [+ トリガーを追加] をクリックします。
2. [Cloud Pub/Sub] を選択します。
3. [Cloud Pub/Sub トピックを選択してください] で、トピック（sentry-logs-trigger）を選択します。
4. [保存] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: サービスアカウントを選択します（sentry-logs-collector-sa）。

[コンテナ] タブに移動します。

[変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。

変数名	値の例	説明
`GCS_BUCKET`	`sentry-logs`	データが保存される GCS バケットの名前。
`GCS_PREFIX`	`sentry/events/`	オブジェクトの省略可能な GCS 接頭辞（サブフォルダ）。
`STATE_KEY`	`sentry/events/state.json`	状態/チェックポイントファイルのキー（省略可）。
`SENTRY_ORG`	`your-org-slug`	Sentry 組織のスラッグ。
`SENTRY_AUTH_TOKEN`	`sntrys_************************`	org:read、project:read、event:read を含む Sentry 認証トークン。
`SENTRY_API_BASE`	`https://sentry.io`	Sentry API のベース URL（セルフホスト: `https://<your-domain>`）。
`MAX_PROJECTS`	`100`	処理するプロジェクトの最大数。
`MAX_PAGES_PER_PROJECT`	`5`	1 回の実行あたりのプロジェクトあたりの最大ページ数。

[変数とシークレット] タブで [リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 600 秒（10 分）を入力します。
[コンテナ] の [設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
  - メモリ: 512 MiB 以上を選択します。
  - CPU: [1] を選択します。
- [完了] をクリックします。
[実行環境] までスクロールします。
- [デフォルト]（推奨）を選択します。
[リビジョンスケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数] に「0」と入力します。
- インスタンスの最大数: 100 と入力します（または、予想される負荷に基づいて調整します）。
[作成] をクリックします。
サービスが作成されるまで待ちます（1 ～ 2 分）。
サービスを作成すると、インラインコードエディタが自動的に開きます。

関数コードを追加する

[関数のエントリポイント] に「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 ```

[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます（2 ～ 3 分）。

注: Pub/Sub トリガー構成では、必要なサブスクリプションと権限が自動的に作成されます。

Cloud Scheduler ジョブの作成

Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。

GCP Console で、[Cloud Scheduler] に移動します。
[ジョブを作成] をクリックします。

次の構成情報を提供してください。

設定	値
名前	`sentry-logs-collector-hourly`
リージョン	Cloud Run functions と同じリージョンを選択する
周波数	`0 * * * *`（1 時間ごとに正時）
タイムゾーン	タイムゾーンを選択します（UTC を推奨）。
ターゲットタイプ	Pub/Sub
トピック	トピックを選択する（`sentry-logs-trigger`）
メッセージ本文	`{}`（空の JSON オブジェクト）

[作成] をクリックします。

スケジュールの頻度のオプション

ログの量とレイテンシの要件に基づいて頻度を選択します。

頻度	CRON 式	ユースケース
5 分毎	`/5 * * *`	大容量、低レイテンシ
15 分ごと	`/15 * * *`	検索量が普通
1 時間ごと	`0 * * * *`	標準（推奨）
6 時間ごと	`0 /6 * *`	少量、バッチ処理
毎日	`0 0 * * *`	履歴データの収集

スケジューラジョブをテストする

Cloud Scheduler コンソールで、ジョブを見つけます。
[強制実行] をクリックして手動でトリガーします。
数秒待ってから、[Cloud Run> サービス> sentry-logs-collector > ログ] に移動します。
関数が正常に実行されたことを確認します。
GCS バケットをチェックして、ログが書き込まれたことを確認します。

Google SecOps サービスアカウントを取得する

Google SecOps は、一意のサービスアカウントを使用して GCS バケットからデータを読み取ります。このサービスアカウントにバケットへのアクセス権を付与する必要があります。

サービスアカウントのメールアドレスを取得する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Sentry Logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [Sentry] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントメールアドレスが表示されます（例:）。
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーして、次のステップで使用します。

注: 各 Google SecOps インスタンスには一意のサービスアカウントがあります。他のドキュメントや例のサービスアカウントは使用しないでください。

Google SecOps サービスアカウントに IAM 権限を付与する

Google SecOps サービスアカウントには、GCS バケットに対する Storage オブジェクト閲覧者ロールが必要です。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: Google SecOps サービスアカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [ストレージオブジェクト閲覧者] を選択します。
[保存] をクリックします。

注: 削除オプションの [転送されたファイルを削除] または [転送されたファイルと空のディレクトリを削除] を使用する場合は、Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者ロールを付与します。

Sentry のログを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Sentry Logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [Sentry] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
```
gs://sentry-logs/sentry/events/
```
  - 次のように置き換えます。
    - sentry-logs: GCS バケット名。
    - sentry/events/: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
  - 例:
    - ルートバケット: gs://company-logs/
    - 接頭辞あり: gs://company-logs/sentry-logs/
    - サブフォルダあり: gs://company-logs/sentry/events/
  注: URI の末尾には常にスラッシュ（/）を付けます。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - なし: 転送後にファイルを削除しません（テストにおすすめ）。
  - 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
  - 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
    
    注: 削除オプションを選択する場合は、サービスアカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

ご不明な点がございましたら、コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。