Metabase のログを収集する
このドキュメントでは、Google Cloud Storage V2 を使用して Metabase ログを Google Security Operations に取り込む方法について説明します。
Metabase は、組織がデータベースのクエリ、可視化の作成、ダッシュボードの構築を行えるようにする、オープンソースのビジネス インテリジェンスおよび分析プラットフォームです。Pro プランと Enterprise プランでは、Metabase は使用状況分析と監査データをアプリケーション データベースに保存します。これには、アクティビティ ログ、ビューログ、クエリ実行レコードが含まれます。Cloud Run functions の関数が Metabase API にクエリを実行してこのデータを取得し、Google SecOps による取り込みのために GCS バケットに書き込みます。
始める前に
次の前提条件を満たしていることを確認します。
- Google SecOps インスタンス
- Cloud Storage、Cloud Run、Pub/Sub、Cloud Scheduler API が有効になっている GCP プロジェクト
- GCS バケットを作成、管理する権限
- GCS バケットの IAM ポリシーを管理する権限
- Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
- Metabase Pro または Enterprise プラン(使用状況分析は、オープンソース エディションまたは Cloud Starter エディションでは利用できません)
- API キーを作成するための Metabase 管理者アクセス権
- Metabase 内にデータソースとして追加された Metabase アプリケーション データベース(API を介して監査テーブルをクエリするために必要)
Metabase API アクセスを構成する
Google SecOps で使用状況分析データを取得できるようにするには、Metabase で API キーを作成し、クエリ可能なデータソースとしてアプリケーション データベースを追加する必要があります。
API キーを作成する
- 管理者として Metabase インスタンスにログインします。
- 右上隅の歯車アイコンをクリックします。
- [管理者設定] を選択します。
- [設定] タブに移動します。
- 左側のメニューで [Authentication] タブをクリックします。
- [API キー] までスクロールし、[管理] をクリックします。
- [API キーを作成] ボタンをクリックします。
- 次の構成情報を指定します。
- 鍵名: わかりやすい名前(例:
Google SecOps Integration)を入力します。 - グループ: Usage Analytics コレクションへのアクセス権と、アプリケーション データベースに対するクエリ権限を持つグループを選択します。
- 鍵名: わかりやすい名前(例:
- [作成] をクリックします。
API キーをコピーして安全に保存します。
アプリケーション データベースをデータソースとして追加する
API を介して監査テーブルをクエリするには、Metabase アプリケーション データベースを Metabase 内のデータソースとして追加する必要があります。
- Metabase で、右上にある歯車アイコンをクリックします。
- [管理者設定] を選択します。
- [データベース] タブに移動します。
- [データベースを追加] をクリックします。
- データベース接続を構成します。
- データベースの種類: [PostgreSQL] または [MySQL] を選択します(アプリケーション データベースによって異なります)。
- 名前: 「
Metabase Application Database」と入力します。 - ホスト: アプリケーション データベースのホスト名を入力します。
- ポート: データベース ポートを入力します(デフォルト: PostgreSQL の場合は
5432、MySQL の場合は3306)。 - データベース名: アプリケーション データベース名を入力します(例:
metabase)。 - ユーザー名:
audit_log、view_log、query_executionテーブルに対する SELECT 権限を持つデータベース ユーザーを入力します。 - [Password]: データベースのパスワードを入力します。
- [保存] をクリックします。
- 保存したら、URL からデータベース ID をメモします(たとえば、
https://metabase.example.com/admin/databases/5の場合、データベース ID は5です)。
権限を確認する
API キーに必要な権限があることを確認するには:
- 管理者として Metabase にログインします。
- 歯車アイコン > [Admin settings] > [People] をクリックします。
- API キーに割り当てられているグループを見つけます。
- グループに使用状況分析コレクションへのアクセス権があることを確認します。
- [Admin settings] > [Permissions] > [Data] に移動します。
- グループにアプリケーション データベースへの無制限アクセス権またはクエリアクセス権があることを確認します。
テスト API へのアクセス
統合に進む前に、認証情報をテストします。
# Replace with your actual values METABASE_URL="https://your-metabase-instance.com" API_KEY="mb_your_api_key_here" DATABASE_ID="5" # Test API key authentication curl -s -H "x-api-key: ${API_KEY}" \ "${METABASE_URL}/api/database" \ | python3 -m json.tool # Test querying the application database for audit_log records curl -s -X POST "${METABASE_URL}/api/dataset/json" \ -H "x-api-key: ${API_KEY}" \ -H "Content-Type: application/json" \ -d "{\"database\": ${DATABASE_ID}, \"type\": \"native\", \"native\": {\"query\": \"SELECT * FROM audit_log ORDER BY id DESC LIMIT 5\", \"template-tags\": {}}, \"parameters\": []}" \ | python3 -m json.tool
成功すると、最近の監査ログレコードの JSON 配列が返されます。
Google Cloud Storage バケットを作成する
- Google Cloud コンソールに移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage] > [バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( metabase-audit-logsなど)を入力します。ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション ロケーションを選択します(例: us-central1)。ストレージ クラス Standard(頻繁にアクセスされるログに推奨) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Cloud Run functions のサービス アカウントを作成する
Cloud Run functions の関数には、GCS バケットへの書き込み権限と Pub/Sub による呼び出し権限を持つサービス アカウントが必要です。
サービス アカウントの作成
- GCP コンソールで、[IAM と管理] > [サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成情報を指定します。
- サービス アカウント名: 「
metabase-audit-collector-sa」と入力します。 - サービス アカウントの説明: 「
Service account for Cloud Run function to collect Metabase audit logs」と入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [Storage オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の操作に必要です。
- Storage オブジェクト管理者: GCS バケットにログを書き込み、状態ファイルを管理します。
- Cloud Run 起動元: Pub/Sub が関数を呼び出せるようにします。
- Cloud Functions 起動元: 関数の呼び出しを許可します。
GCS バケットに対する IAM 権限を付与する
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
metabase-audit-logs)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: サービス アカウントのメールアドレス(
metabase-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
- プリンシパルを追加: サービス アカウントのメールアドレス(
- [保存] をクリックします。
Pub/Sub トピックの作成
- GCP コンソールで、[Pub/Sub] > [トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成情報を指定します。
- トピック ID: 「
metabase-audit-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Metabase API から監査データを取得して GCS に書き込みます。
- GCP Console で Cloud Run に移動します。
- [サービスを作成] をクリックします。
- [関数](インライン編集を使用して関数を作成する)を選択します。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 metabase-audit-collectorリージョン GCS バケットに一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、
metabase-audit-triggerを選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント:
metabase-audit-collector-saを選択します。
- サービス アカウント:
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 説明 GCS_BUCKETmetabase-audit-logsGCS バケット名 GCS_PREFIXmetabase-auditログファイルの接頭辞 STATE_KEYmetabase-audit/state.json状態ファイルのパス METABASE_URLhttps://metabase.yourcompany.comMetabase インスタンスのベース URL METABASE_API_KEYmb_your_api_key_hereMetabase API キー DATABASE_ID5アプリケーション データベース ID LOOKBACK_HOURS24最初のルックバック期間 PAGE_SIZE2000API クエリあたりのレコード数 [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストタイムアウト:
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, timedelta http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=10.0, read=60.0), retries=False, ) storage_client = storage.Client() GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'metabase-audit') STATE_KEY = os.environ.get('STATE_KEY', 'metabase-audit/state.json') METABASE_URL = os.environ.get('METABASE_URL', '').rstrip('/') API_KEY = os.environ.get('METABASE_API_KEY') DATABASE_ID = int(os.environ.get('DATABASE_ID', '1')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '2000')) TABLES = ['audit_log', 'view_log', 'query_execution'] TABLE_TIME_COLUMNS = { 'audit_log': 'timestamp', 'view_log': 'timestamp', 'query_execution': 'started_at', } @functions_framework.cloud_event def main(cloud_event): if not all([GCS_BUCKET, METABASE_URL, API_KEY]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) state = load_state(bucket) now = datetime.now(timezone.utc) if isinstance(state, dict) and state.get('last_event_time'): try: last_val = state['last_event_time'] if last_val.endswith('Z'): last_val = last_val[:-1] + '+00:00' last_time = datetime.fromisoformat(last_val) last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") last_time = now - timedelta(hours=LOOKBACK_HOURS) else: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}") all_records = [] newest_time = None for table in TABLES: time_col = TABLE_TIME_COLUMNS[table] records = fetch_table(table, time_col, last_time, now) for r in records: r['_metabase_table'] = table t = r.get(time_col) if t and (newest_time is None or str(t) > str(newest_time)): newest_time = t all_records.extend(records) print(f"Table {table}: fetched {len(records)} records") if not all_records: print("No new records found.") save_state(bucket, now.isoformat()) return timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/metabase_audit_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join( [json.dumps(r, ensure_ascii=False, default=str) for r in all_records] ) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}") save_state(bucket, str(newest_time) if newest_time else now.isoformat()) print(f"Successfully processed {len(all_records)} records") except Exception as e: print(f'Error processing logs: {str(e)}') raise def fetch_table(table, time_col, start_time, end_time): start_str = start_time.strftime('%Y-%m-%d %H:%M:%S') end_str = end_time.strftime('%Y-%m-%d %H:%M:%S') all_records = [] offset = 0 max_pages = 50 for page in range(max_pages): sql = ( f"SELECT * FROM {table} " f"WHERE {time_col} >= '{start_str}' " f"AND {time_col} < '{end_str}' " f"ORDER BY {time_col} ASC " f"LIMIT {PAGE_SIZE} OFFSET {offset}" ) query_body = { "database": DATABASE_ID, "type": "native", "native": { "query": sql, "template-tags": {} }, "parameters": [] } url = f"{METABASE_URL}/api/dataset/json" response = http.request( 'POST', url, body=json.dumps(query_body).encode('utf-8'), headers={ 'x-api-key': API_KEY, 'Content-Type': 'application/json' } ) if response.status == 429: print(f"Rate limited on {table} query. Stopping pagination.") break if response.status != 200: print(f"{table} query failed: {response.status} - " f"{response.data.decode('utf-8')}") break page_results = json.loads(response.data.decode('utf-8')) if not page_results: break all_records.extend(page_results) print(f"{table} page {page + 1}: {len(page_results)} records " f"(total: {len(all_records)})") if len(page_results) < PAGE_SIZE: break offset += PAGE_SIZE return all_records def load_state(bucket): try: blob = bucket.blob(STATE_KEY) if blob.exists(): return json.loads(blob.download_as_text()) except Exception as e: print(f"Warning: Could not load state: {e}") return {} def save_state(bucket, last_event_time_iso): try: state = { 'last_event_time': last_event_time_iso, 'last_run': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(STATE_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}")requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0[デプロイ] をクリックして関数を保存し、デプロイします。
デプロイが完了するまで待ちます(2 ~ 3 分)。
Cloud Scheduler ジョブの作成
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 metabase-audit-collector-hourlyリージョン Cloud Run functions の関数と同じリージョンを選択します。 周波数 0 * * * *(1 時間ごと、毎正時)タイムゾーン タイムゾーンを選択(UTC を推奨) ターゲット タイプ Pub/Sub トピック metabase-audit-triggerを選択メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
スケジュール頻度のオプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
| 頻度 | CRON 式 | ユースケース |
|---|---|---|
| 5 分おき | */5 * * * * |
大容量、低レイテンシ |
| 15 分ごと | */15 * * * * |
検索量が普通 |
| 1 時間ごと | 0 * * * * |
標準(推奨) |
| 6 時間ごと | 0 */6 * * * |
少量のバッチ処理 |
| 毎日 | 0 0 * * * |
過去のデータの収集 |
統合をテストする
- Cloud Scheduler コンソールで、ジョブ(
metabase-audit-collector-hourly)を見つけます。 - [強制実行] をクリックして、ジョブを手動でトリガーします。
- 数秒待ちます。
- [Cloud Run] > [サービス] に移動します。
metabase-audit-collectorをクリックします。- [ログ] タブをクリックします。
関数が正常に実行されたことを確認します。以下のものを探します。
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Table audit_log: fetched X records Table view_log: fetched X records Table query_execution: fetched X records Wrote X records to gs://metabase-audit-logs/metabase-audit/metabase_audit_YYYYMMDD_HHMMSS.ndjson Successfully processed X records[Cloud Storage] > [バケット] に移動します。
metabase-audit-logsをクリックします。metabase-audit/フォルダに移動します。現在のタイムスタンプで新しい
.ndjsonファイルが作成されたことを確認します。
ログにエラーが表示される場合:
- HTTP 401:
METABASE_API_KEY環境変数が正しいこと、キーが削除または再生成されていないことを確認します。 - HTTP 403: API キーのグループにアプリケーション データベースのクエリ権限があることを確認します。
- HTTP 429: レート制限。関数はページネーションを停止し、次のスケジュールされた実行で再開します。
- 環境変数が不足している: 必要なすべての変数が Cloud Run functions の関数の構成に設定されていることを確認します。
- 結果が空になる:
DATABASE_IDが正しいこと、アプリケーション データベースがデータソースとして追加されていること、Metabase Pro または Enterprise を使用していることを確認します。
Google SecOps サービス アカウントを取得する
- [設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
Metabase Audit Logs)。 - [Source type] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [METABASE] を選択します。
[サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます。例:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーして、次のステップで使用します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://metabase-audit-logs/metabase-audit/Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストに推奨)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
転送されたファイルと空のディレクトリを削除する: 転送が成功した後にファイルと空のディレクトリを削除します。
ファイルの最長存続期間: 過去数日間に変更されたファイルを含めます(デフォルトは 180 日)。
Asset namespace: アセットの名前空間。
Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
Google SecOps サービス アカウントに IAM 権限を付与する
Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。
- [Cloud Storage] > [バケット] に移動します。
metabase-audit-logsをクリックします。- [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
[保存] をクリックします。
使用状況分析テーブル
Cloud Run functions は、Metabase アプリケーション データベースから次のテーブルをクエリします。
| テーブル名 | 説明 |
|---|---|
| audit_log | 構成変更、ユーザー操作、システム イベントの記録 |
| view_log | カード(質問 / モデル)、ダッシュボード、テーブルのビューを追跡します。 |
| query_execution | すべてのダッシュボードで実行されたすべてのクエリに関する情報 |
データの保持
デフォルトでは、Metabase は使用状況分析データを 720 日間(約 2 年間)保持します。Metabase は、このしきい値より古い行を 1 日に 2 回自動的に削除します。保持期間を変更するには、Metabase の実行時に環境変数 MB_AUDIT_MAX_RETENTION_DAYS を設定します。
UDM マッピング テーブル
| ログフィールド | UDM マッピング | ロジック |
|---|---|---|
| 追加の | 追加の | 「追加」から名前を変更 |
| locale | additional.fields | それぞれのラベルと統合(例: locale の場合は locale_label、is_active の場合は is_active_label など) |
| is_active | additional.fields | |
| is_qbnewb | additional.fields | |
| group_ids | additional.fields | |
| is_superuser | additional.fields | |
| login_attributes | additional.fields | |
| id | additional.fields | |
| sso_source | additional.fields | |
| personal_collection_id | additional.fields | |
| metadata | metadata | メタデータから名前を変更 |
| updated_at | metadata.event_timestamp | ISO8601、RFC 3339、または yyyy-MM-ddTHH:mm:ss.SSSSSSZ を使用して updated_at から解析された日付 |
| has_userid | metadata.event_type | has_userid == "true" の場合は「USER_UNCATEGORIZED」に設定され、それ以外の場合は「GENERIC_EVENT」に設定されます。 |
| プリンシパル | プリンシパル | プリンシパルから名前を変更 |
| metabase_host | principal.url | 値が直接コピーされる |
| date_joined | principal.user.attribute.creation_time | ISO8601、RFC 3339、または yyyy-MM-ddTHH:mm:ss.SSSSSSZ を使用して date_joined から解析された日付 |
| メール | principal.user.email_addresses | メールから統合 |
| first_name | principal.user.first_name | 値が直接コピーされる |
| last_login | principal.user.last_login_time | ISO8601、RFC 3339、または yyyy-MM-ddTHH:mm:ss.SSSSSSZ を使用して last_login から解析された日付 |
| last_name | principal.user.last_name | 値が直接コピーされる |
| common_name | principal.user.user_display_name | 値が直接コピーされる |
| メール | principal.user.userid | メールから直接コピーされた値 |
| security_result | security_result | security_result から統合 |
| ターゲット | ターゲット | ターゲットから名前が変更されました |
| metadata.product_name | metadata.product_name | 「METABASE」に設定します。 |
| metadata.vendor_name | metadata.vendor_name | 「METABASE」に設定します。 |
変更ログ
さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。