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 キーを作成する

  1. 管理者として Metabase インスタンスにログインします。
  2. 右上隅の歯車アイコンをクリックします。
  3. [管理者設定] を選択します。
  4. [設定] タブに移動します。
  5. 左側のメニューで [Authentication] タブをクリックします。
  6. [API キー] までスクロールし、[管理] をクリックします。
  7. [API キーを作成] ボタンをクリックします。
  8. 次の構成情報を指定します。
    • 鍵名: わかりやすい名前(例: Google SecOps Integration)を入力します。
    • グループ: Usage Analytics コレクションへのアクセス権と、アプリケーション データベースに対するクエリ権限を持つグループを選択します。
  9. [作成] をクリックします。
  10. API キーをコピーして安全に保存します。

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

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

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

権限を確認する

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

  1. 管理者として Metabase にログインします。
  2. 歯車アイコン > [Admin settings] > [People] をクリックします。
  3. API キーに割り当てられているグループを見つけます。
  4. グループに使用状況分析コレクションへのアクセス権があることを確認します。
  5. [Admin settings] > [Permissions] > [Data] に移動します。
  6. グループにアプリケーション データベースへの無制限アクセス権またはクエリアクセス権があることを確認します。

テスト 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 バケットを作成する

  1. Google Cloud コンソールに移動します。
  2. プロジェクトを選択するか、新しいプロジェクトを作成します。
  3. ナビゲーション メニューで、[Cloud Storage] > [バケット] に移動します。
  4. [バケットを作成] をクリックします。
  5. 次の構成情報を提供してください。

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

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

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

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

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

これらのロールは、次の操作に必要です。

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

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

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

Pub/Sub トピックの作成

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

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

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

  1. GCP Console で Cloud Run に移動します。
  2. [サービスを作成] をクリックします。
  3. [関数](インライン編集を使用して関数を作成する)を選択します。
  4. [構成] セクションで、次の構成の詳細を指定します。

    設定 値
    サービス名 metabase-audit-collector
    リージョン GCS バケットに一致するリージョンを選択します(例: us-central1)。
    ランタイム [Python 3.12] 以降を選択します。
  5. [トリガー(省略可)] セクションで、次の操作を行います。

    1. [+ トリガーを追加] をクリックします。
    2. [Cloud Pub/Sub] を選択します。
    3. [Cloud Pub/Sub トピックを選択してください] で、metabase-audit-trigger を選択します。
    4. [保存] をクリックします。
  6. [認証] セクションで、次の操作を行います。

    1. [認証が必要] を選択します。
    2. Identity and Access Management(IAM)を確認します。
  7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

  8. [セキュリティ] タブに移動します。

    • サービス アカウント: metabase-audit-collector-sa を選択します。
  9. [コンテナ] タブに移動します。

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    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 クエリあたりのレコード数
  10. [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。

    • リクエストタイムアウト: 600秒(10 分)を入力してください
  11. [設定] タブに移動します。

    • [リソース] セクションで、次の操作を行います。
      • メモリ: 512 MiB 以上を選択します。
      • CPU: 1 を選択します。
  12. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数]: 「0」と入力します。
    • [インスタンスの最大数]: 「100」と入力します。
  13. [作成] をクリックします。

  14. サービスが作成されるまで待ちます(1 ~ 2 分)。

  15. サービスが作成されると、インライン コードエディタが自動的に開きます。

関数コードを追加する

  1. [エントリ ポイント] フィールドに「main」と入力します。
  2. インライン コードエディタで、次の 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
    
  3. [デプロイ] をクリックして関数を保存し、デプロイします。

  4. デプロイが完了するまで待ちます(2 ~ 3 分)。

Cloud Scheduler ジョブの作成

  1. GCP Console で、[Cloud Scheduler] に移動します。
  2. [ジョブを作成] をクリックします。
  3. 次の構成情報を提供してください。

    設定 値
    名前 metabase-audit-collector-hourly
    リージョン Cloud Run functions の関数と同じリージョンを選択します。
    周波数 0 * * * *(1 時間ごと、毎正時)
    タイムゾーン タイムゾーンを選択(UTC を推奨)
    ターゲット タイプ Pub/Sub
    トピック metabase-audit-trigger を選択
    メッセージ本文 {}(空の JSON オブジェクト)
  4. [作成] をクリックします。

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

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

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

統合をテストする

  1. Cloud Scheduler コンソールで、ジョブ(metabase-audit-collector-hourly)を見つけます。
  2. [強制実行] をクリックして、ジョブを手動でトリガーします。
  3. 数秒待ちます。
  4. [Cloud Run] > [サービス] に移動します。
  5. metabase-audit-collector をクリックします。
  6. [ログ] タブをクリックします。
  7. 関数が正常に実行されたことを確認します。以下のものを探します。

    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
    
  8. [Cloud Storage] > [バケット] に移動します。

  9. metabase-audit-logs をクリックします。

  10. metabase-audit/ フォルダに移動します。

  11. 現在のタイムスタンプで新しい .ndjson ファイルが作成されたことを確認します。

ログにエラーが表示される場合:

  • HTTP 401: METABASE_API_KEY 環境変数が正しいこと、キーが削除または再生成されていないことを確認します。
  • HTTP 403: API キーのグループにアプリケーション データベースのクエリ権限があることを確認します。
  • HTTP 429: レート制限。関数はページネーションを停止し、次のスケジュールされた実行で再開します。
  • 環境変数が不足している: 必要なすべての変数が Cloud Run functions の関数の構成に設定されていることを確認します。
  • 結果が空になる: DATABASE_ID が正しいこと、アプリケーション データベースがデータソースとして追加されていること、Metabase Pro または Enterprise を使用していることを確認します。

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

  1. [設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Metabase Audit Logs)。
  5. [Source type] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [METABASE] を選択します。
  7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます。例:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. このメールアドレスをコピーして、次のステップで使用します。

  9. [次へ] をクリックします。

  10. 次の入力パラメータの値を指定します。

    • ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。

      gs://metabase-audit-logs/metabase-audit/
      
    • Source deletion option: 必要に応じて削除オプションを選択します。

      • なし: 転送後にファイルを削除しません(テストに推奨)。
      • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
      • 転送されたファイルと空のディレクトリを削除する: 転送が成功した後にファイルと空のディレクトリを削除します。

    • ファイルの最長存続期間: 過去数日間に変更されたファイルを含めます(デフォルトは 180 日)。

    • Asset namespace: アセットの名前空間。

    • Ingestion labels: このフィードのイベントに適用されるラベル。

  11. [次へ] をクリックします。

  12. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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

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

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

使用状況分析テーブル

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 のプロフェッショナルから回答を得ることができます。