Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Metabase ログを収集する

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

Google SecOps SIEM

このドキュメントでは、Google Cloud Storage V2 を使用して Metabase ログを Google Security Operations に取り込む方法について説明します。

Metabase は、組織がデータベースのクエリ、可視化の作成、ダッシュボードの構築を行えるオープンソースのビジネスインテリジェンスおよび分析プラットフォームです。Pro プランと Enterprise プランでは、Metabase は使用状況分析と監査データをアプリケーションデータベースに保存します。これには、アクティビティログ、ビューログ、クエリ実行レコードが含まれます。Cloud Run 関数は 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 プラン（使用状況分析は Open Source エディションまたは Cloud Starter エディションでは利用できません）
API キーを作成するための Metabase 管理者アクセス権
Metabase 内でデータソースとして追加された Metabase アプリケーションデータベース（API 経由で監査テーブルをクエリするために必要）

Metabase API アクセスを構成する

Google SecOps が使用状況分析データを取得できるようにするには、Metabase で API キーを作成し、アプリケーションデータベースをクエリ可能なデータソースとして追加する必要があります。

API キーを作成する

管理者として Metabase インスタンスにログインします。
右上隅にある歯車アイコンをクリックします。
[管理者設定] を選択します。
[設定] タブに移動します。
左側のメニューで [認証] タブをクリックします。
[API キー] までスクロールして、[管理] をクリックします。
[API キーを作成] ボタンをクリックします。
次の構成情報を指定します。
- キー名: わかりやすい名前を入力します（例: Google SecOps Integration）。
- グループ: アプリケーションデータベースの Usage Analytics コレクションとクエリ権限にアクセスできるグループを選択します。
[作成] をクリックします。
API キーをコピーして安全に保存します。

注: Metabase に API キーが表示されるのは 1 回のみです。秘密鍵を紛失した場合は、新しい鍵を再生成する必要があります。

アプリケーションデータベースをデータソースとして追加する

API を介して監査テーブルをクエリするには、Metabase 内で Metabase アプリケーションデータベースをデータソースとして追加する必要があります。

Metabase で、右上にある歯車アイコンをクリックします。
[管理者設定] を選択します。
[データベース] タブに移動します。
[データベースを追加] をクリックします。
データベース接続を構成します。
- データベースの種類: [PostgreSQL] または [MySQL] を選択します（アプリケーションデータベースによって異なります）。
- 名前: 「Metabase Application Database」と入力します。
- ホスト: アプリケーションデータベースのホスト名を入力します。
- ポート: データベースポートを入力します（デフォルト: PostgreSQL の場合は 5432、MySQL の場合は 3306）。
- データベース名: アプリケーションデータベース名（metabase など）を入力します。
- ユーザー名: audit_log、view_log、query_execution テーブルに対する SELECT 権限を持つデータベースユーザーを入力します。
- パスワード: データベースのパスワードを入力します。
[保存] をクリックします。
保存したら、URL からデータベース ID をメモします（たとえば、https://metabase.example.com/admin/databases/5 の場合、データベース ID は 5 です）。

権限を確認する

API キーに必要な権限があることを確認するには:

管理者として Metabase にログインします。
歯車アイコン > [Admin settings] > [People] をクリックします。
API キーに割り当てられているグループを確認します。
グループが使用状況分析コレクションにアクセスできることを確認します。
[管理設定] > [権限] > [データ] に移動します。
グループにアプリケーションデータベースへの制限なしまたはクエリのアクセス権があることを確認します。

テスト 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 と入力します。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
1. [ロールを選択] をクリックします。
2. [ストレージオブジェクト管理者] を検索して選択します。
3. [+ 別のロールを追加] をクリックします。
4. [Cloud Run 起動元] を検索して選択します。
5. [+ 別のロールを追加] をクリックします。
6. [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）を入力します。
- ロールを割り当てる: [ストレージオブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

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

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

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

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

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

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

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

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

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

変数名	値の例	説明
`GCS_BUCKET`	`metabase-audit-logs`	GCS バケット名
`GCS_PREFIX`	`metabase-audit`	ログファイルの接頭辞
`STATE_KEY`	`metabase-audit/state.json`	状態ファイルのパス
`METABASE_URL`	`https://metabase.yourcompany.com`	Metabase インスタンスのベース URL
`METABASE_API_KEY`	`mb_your_api_key_here`	Metabase API キー
`DATABASE_ID`	`5`	アプリケーションデータベース ID
`LOOKBACK_HOURS`	`24`	初期ルックバック期間
`PAGE_SIZE`	`2000`	API クエリあたりのレコード数

[変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 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 分）。

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

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）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[ログタイプ] として [METABASE] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントのメールアドレスが表示されます（例:）。
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーして、次のステップで使用します。

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

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

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

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

注: 削除オプションの [転送されたファイルを削除] または [転送されたファイルと空のディレクトリを削除] を使用する場合は、Storage オブジェクト閲覧者ではなく、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	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 から統合
ターゲット	ターゲット	target から名前が変更されました
metadata.product_name	metadata.product_name	「METABASE」に設定
metadata.vendor_name	metadata.vendor_name	「METABASE」に設定

変更ログ

このパーサーの変更ログを表示する

さらにサポートが必要な場合 コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。