Agiloft のログを収集する

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

Google SecOps SIEM

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

Agiloft は、契約の作成、承認ワークフロー、コンプライアンスの追跡を自動化するノーコードの契約ライフサイクル管理（CLM）プラットフォームです。Agiloft アクティビティ ログテーブルには、ユーザーのログイン、ログアウト、レコードの変更、ルールの変更、管理操作などのシステム使用イベントが記録されます。これらのアクティビティ ログは、Agiloft REST API を介してエクスポートし、Google SecOps による取り込みのために GCS バケットに書き込むことができます。

始める前に

次の前提条件を満たしていることを確認してください。

• Google SecOps インスタンス
• Cloud Storage API が有効になっている GCP プロジェクト
• GCS バケットを作成および管理する権限
• Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
• Advanced、Premium、Platform エディションのいずれかの Agiloft インスタンス（REST API アクセスにはこれらのエディションのいずれかが必要です）
• REST 対応グループに属し、アクティビティ ログ テーブルへのアクセス権を持つ Agiloft ユーザー アカウント
• Agiloft ナレッジベース（KB）でアクティビティ ロギングが有効になっている

Agiloft アクティビティ ログを構成する

1. 管理者アカウントで Agiloft KB にログインします。
2. 右上にある設定（歯車）アイコンをクリックします。
3. [システム>アクティビティ ログを構成する] に移動します。
4. [新規] をクリックして、監査ルール ウィザードを開きます。
5. 次の構成の詳細を入力します。
   • 名前: ルールの名前を入力します（例: All Activity for SIEM）。
   • 言語: KB の言語を選択します。
6. どのユーザーのアクションをログに記録するかを決定する保存済み検索を作成または選択します。
7. 追跡するイベントを選択します。包括的なセキュリティ モニタリングを行うには、次の項目を選択します。
   • ログイン
   • ログアウト
   • ログイン失敗
   • レコードの作成
   • レコードの編集
   • レコードの削除
   • ファイルのダウンロード
   • ルールの編集
   • ワークフローの編集
   • グループ編集
   • チーム編集
   • 表の編集
   • Column Edit
8. 監査レコードの保持期間（1 year など）を定義します。

9. [完了] をクリックします。

   注: デフォルトでは、Agiloft にはシステム イベント、すべてのログイン、管理者アクション、削除の監査ルールが含まれています。重複を避けるため、新しいルールを作成する前に既存のルールを確認してください。

アクティビティ ログの表を再表示する

1. Agiloft KB で、[Setup > Tables] に移動します。
2. テーブル リストから [アクティビティ ログ] を選択します。

3. [非表示を解除] をクリックして、アクティビティ ログテーブルを API クエリで使用できるようにします。

   注: REST API を介してクエリを実行するには、アクティビティ ログ テーブルが表示（非表示解除）されている必要があります。

Agiloft API 認証情報を収集する

Agiloft インスタンスの詳細を取得する

1. Agiloft インスタンスにログインします。

2. ブラウザのアドレスバーから次の値をメモします。

   • ホスト名: サーバー アドレス（yourcompany.agiloft.com など）
   • KB 名: ナレッジベースの名前。KB の右上にあるヘルプアイコンの横に表示されます（例: Production）。
   注: KB 名では大文字と小文字が区別されます。Agiloft インターフェースに表示されている正確な名前を使用します。

専用の API ユーザーを作成する

1. Agiloft の KB で、[Setup> Access> People] に移動します。
2. API アクセス専用の新しいユーザー アカウントを作成します。
   • ログイン: わかりやすいログイン名を入力します（例: siem_api_user）。
   • パスワード: 安全なパスワードを設定する

3. REST API アクセスが有効になっており、アクティビティ ログ テーブルに対する読み取りアクセス権があるグループにユーザーを追加します。

   注: API ユーザーはネイティブの Agiloft アカウントである必要があります。LDAP アカウントは REST API 認証と互換性がありません。

権限を確認する

アカウントに必要な権限があることを確認するには:

1. API ユーザー認証情報を使用して Agiloft にログインします。
2. [設定 > テーブル] に移動します。
3. [アクティビティ ログ] テーブルが表示され、アクセスできることを確認します。
4. アクティビティ ログの表が表示されない場合は、Agiloft 管理者に連絡して、必要なグループ権限を付与してもらってください。

テスト API へのアクセス

• 統合に進む前に、認証情報をテストします。

  AGILOFT_HOST="https://yourcompany.agiloft.com"
  AGILOFT_KB="YourKBName"
  AGILOFT_LOGIN="siem_api_user"
  AGILOFT_PASSWORD="your-password"
  
  # Test login and get JWT token
  TOKEN=$(curl -s "${AGILOFT_HOST}/ewws/EWLogin?$KB=${AGILOFT_KB}&\$login=${AGILOFT_LOGIN}&\$password=${AGILOFT_PASSWORD}&\$lang=en" \
    | python3 -c "import sys,json; data=json.load(sys.stdin); print(data.get('token',''))")
  
  echo "Token: ${TOKEN}"
  
  # Test Activity Log table access with JSON format
  curl -v "${AGILOFT_HOST}/ewws/EWSearch/.json?\$KB=${AGILOFT_KB}&\$table=activity_log&\$login=${AGILOFT_LOGIN}&\$password=${AGILOFT_PASSWORD}&\$lang=en&limit=1&page=0&field=id&field=action&field=description&field=login&field=date" \
    -H "Accept: application/json"
  

  注: デフォルトの JWT トークンの有効期限は 15 分です。Cloud Run 関数では、トークンの更新の複雑さを回避するために、代わりに認証情報ベースの認証方法（URL パラメータのログインとパスワード）が使用されます。

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

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

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

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

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

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

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

サービス アカウントを作成する

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

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

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

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

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

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

Pub/Sub トピックの作成

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

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

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

Cloud Run functions は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Agiloft REST API からアクティビティ ログを取得して GCS に書き込みます。

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

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

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

5. [トリガー（省略可）] セクションで、次の操作を行います。

   1. [+ トリガーを追加] をクリックします。
   2. [Cloud Pub/Sub] を選択します。
   3. [Cloud Pub/Sub トピックを選択してください] で、トピック agiloft-logs-trigger を選択します。
   4. [保存] をクリックします。

6. [認証] セクションで次の操作を行います。

   1. [認証が必要] を選択します。
   2. Identity and Access Management（IAM）を確認します。
   注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。

7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

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

   • サービス アカウント: サービス アカウント agiloft-logs-collector-sa を選択します。

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

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

   変数名	値の例	説明
   GCS_BUCKET	agiloft-activity-logs	GCS バケット名
   GCS_PREFIX	agiloft	ログファイルの接頭辞
   STATE_KEY	agiloft/state.json	状態ファイルのパス
   AGILOFT_HOST	https://yourcompany.agiloft.com	Agiloft インスタンスの URL
   AGILOFT_KB	YourKBName	Agiloft KB 名（大文字と小文字が区別されます）
   AGILOFT_LOGIN	siem_api_user	API ユーザーのログイン
   AGILOFT_PASSWORD	your-password	API ユーザーのパスワード
   PAGE_SIZE	100	API ごとのレコード数ページ
   MAX_RECORDS	10000	実行あたりの最大レコード数
   LOOKBACK_HOURS	2	最初のルックバック期間

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
     import urllib.parse
     from datetime import datetime, timezone, timedelta
     import time
     
     # Initialize HTTP client with timeouts
     http = urllib3.PoolManager(
       timeout=urllib3.Timeout(connect=5.0, read=30.0),
       retries=False,
     )
     
     # Initialize Storage client
     storage_client = storage.Client()
     
     # Environment variables
     GCS_BUCKET = os.environ.get('GCS_BUCKET')
     GCS_PREFIX = os.environ.get('GCS_PREFIX', 'agiloft').strip('/')
     STATE_KEY = os.environ.get('STATE_KEY') or f"{GCS_PREFIX}/state.json"
     AGILOFT_HOST = os.environ.get('AGILOFT_HOST', '').rstrip('/')
     AGILOFT_KB = os.environ.get('AGILOFT_KB')
     AGILOFT_LOGIN = os.environ.get('AGILOFT_LOGIN')
     AGILOFT_PASSWORD = os.environ.get('AGILOFT_PASSWORD')
     PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
     MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
     LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '2'))
     
     ACTIVITY_LOG_FIELDS = [
       'id', 'action', 'action_duration', 'login', 'user_name',
       'user_primary_team', 'company', 'cookie', 'date',
       'description', 'groups', 'interface', 'ip',
       'record_id', 'rule_id', 'session_duration',
       'session_id', 'table_label'
     ]
     
     def parse_datetime(value):
       """Parse ISO datetime string to datetime object."""
       if not value:
         return None
       if value.endswith("Z"):
         value = value[:-1] + "+00:00"
       try:
         return datetime.fromisoformat(value)
       except Exception:
         return None
     
     @functions_framework.cloud_event
     def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch
       Agiloft Activity Log records and write to GCS.
     
       Args:
         cloud_event: CloudEvent object containing Pub/Sub message
       """
     
       if not all([GCS_BUCKET, AGILOFT_HOST, AGILOFT_KB, AGILOFT_LOGIN, AGILOFT_PASSWORD]):
         print('Error: Missing required environment variables')
         return
     
       try:
         bucket = storage_client.bucket(GCS_BUCKET)
     
         # Load state
         state = load_state(bucket, STATE_KEY)
     
         # Determine time window
         now = datetime.now(timezone.utc)
         last_time = None
     
         if isinstance(state, dict) and state.get("last_event_time"):
           try:
             last_time = parse_datetime(state["last_event_time"])
             # Overlap by 2 minutes to catch delayed events
             if last_time:
               last_time = last_time - timedelta(minutes=2)
           except Exception as e:
             print(f"Warning: Could not parse last_event_time: {e}")
     
         if last_time is None:
           last_time = now - timedelta(hours=LOOKBACK_HOURS)
     
         print(f"Fetching activity logs from {last_time.isoformat()} to {now.isoformat()}")
     
         # Fetch activity logs
         records, newest_event_time = fetch_activity_logs(
           start_time=last_time,
           end_time=now,
         )
     
         if not records:
           print("No new activity log records found.")
           save_state(bucket, STATE_KEY, now.isoformat())
           return
     
         # Write to GCS as NDJSON
         timestamp = now.strftime('%Y%m%d_%H%M%S')
         object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
         blob = bucket.blob(object_key)
     
         ndjson = '\n'.join(
           [json.dumps(record, ensure_ascii=False) for record in records]
         ) + '\n'
         blob.upload_from_string(ndjson, content_type='application/x-ndjson')
     
         print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")
     
         # Update state
         if newest_event_time:
           save_state(bucket, STATE_KEY, newest_event_time)
         else:
           save_state(bucket, STATE_KEY, now.isoformat())
     
         print(f"Successfully processed {len(records)} records")
     
       except Exception as e:
         print(f'Error processing activity 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)
       except Exception as e:
         print(f"Warning: Could not load state: {e}")
       return {}
     
     def save_state(bucket, key, last_event_time_iso):
       """Save the last event timestamp to GCS state file."""
       try:
         state = {'last_event_time': last_event_time_iso}
         blob = bucket.blob(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}")
     
     def fetch_activity_logs(start_time, end_time):
       """
       Fetch activity logs from Agiloft REST API with pagination.
     
       Args:
         start_time: Start time for log query
         end_time: End time for log query
     
       Returns:
         Tuple of (records list, newest_event_time ISO string)
       """
       base_url = f"{AGILOFT_HOST}/ewws/EWSearch/.json"
     
       # Build field parameters
       field_params = '&'.join(
         [f"field={urllib.parse.quote(f)}" for f in ACTIVITY_LOG_FIELDS]
       )
     
       # Build date filter query
       start_str = start_time.strftime('%d %m %y %H:%M:%S')
       end_str = end_time.strftime('%d %m %y %H:%M:%S')
       query_filter = urllib.parse.quote(
         f"date >= '{start_str}' AND date <= '{end_str}'"
       )
     
       records = []
       newest_time = None
       page_num = 0
       backoff = 1.0
     
       while True:
         if len(records) >= MAX_RECORDS:
           print(f"Reached max_records limit ({MAX_RECORDS})")
           break
     
         url = (
           f"{base_url}"
           f"?$KB={urllib.parse.quote(AGILOFT_KB)}"
           f"&$table=activity_log"
           f"&$login={urllib.parse.quote(AGILOFT_LOGIN)}"
           f"&$password={urllib.parse.quote(AGILOFT_PASSWORD)}"
           f"&$lang=en"
           f"&query={query_filter}"
           f"&limit={PAGE_SIZE}"
           f"&page={page_num}"
           f"&{field_params}"
         )
     
         try:
           response = http.request('GET', url)
     
           # Handle rate limiting with exponential backoff
           if response.status == 429:
             retry_after = int(
               response.headers.get('Retry-After', str(int(backoff)))
             )
             print(f"Rate limited (429). Retrying after {retry_after}s...")
             time.sleep(retry_after)
             backoff = min(backoff * 2, 30.0)
             continue
     
           backoff = 1.0
     
           if response.status != 200:
             print(f"HTTP Error: {response.status}")
             response_text = response.data.decode('utf-8')
             print(f"Response body: {response_text}")
             return [], None
     
           data = json.loads(response.data.decode('utf-8'))
     
           if not data.get('success', False):
             print(f"API error: {data.get('message', 'Unknown error')}")
             return [], None
     
           result = data.get('result', [])
     
           # Handle single record vs list
           if isinstance(result, dict):
             result = [result]
     
           if not result:
             print(f"No more results (empty page {page_num})")
             break
     
           print(f"Page {page_num}: Retrieved {len(result)} events")
           records.extend(result)
     
           # Track newest event time
           for event in result:
             try:
               event_time = event.get('date')
               if event_time:
                 if newest_time is None:
                   newest_time = event_time
                 else:
                   current_dt = parse_datetime(event_time)
                   newest_dt = parse_datetime(newest_time)
                   if current_dt and newest_dt and current_dt > newest_dt:
                     newest_time = event_time
             except Exception as e:
               print(f"Warning: Could not parse event time: {e}")
     
           # Check pagination
           if len(result) < PAGE_SIZE:
             print(
               f"Reached last page (size={len(result)} < limit={PAGE_SIZE})"
             )
             break
     
           page_num += 1
     
         except Exception as e:
           print(f"Error fetching activity logs: {e}")
           return [], None
     
       print(f"Retrieved {len(records)} total records from {page_num + 1} pages")
       return records, newest_time
     

   • 2 つ目のファイル - requirements.txt:

     functions-framework==3.*
     google-cloud-storage==2.*
     urllib3>=2.0.0
     

3. [デプロイ] をクリックして、関数を保存してデプロイします。

4. デプロイが完了するまで待ちます（2 ～ 3 分）。

   注: Pub/Sub トリガー構成では、必要なサブスクリプションと権限が自動的に作成されます。Agiloft REST API では、デフォルトで呼び出し間に 1 秒の遅延が発生します（WSDelay グローバル変数で構成可能）。この関数は、順次ページネーションを通じてこれをネイティブに処理します。

Cloud Scheduler ジョブを作成する

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

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

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

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

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

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

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

頻度	CRON 式	ユースケース
1 時間ごと	0 * * * *	標準（推奨）
2 時間ごと	0 */2 * * *	音量を下げる
6 時間ごと	0 */6 * * *	少量、バッチ処理

統合をテストする

1. Cloud Scheduler コンソールで、ジョブを見つけます。
2. [強制実行] をクリックして、ジョブを手動でトリガーします。
3. 数秒待ちます。
4. Cloud Run > サービスに移動します。
5. 関数名 agiloft-logs-collector をクリックします。
6. [Logs] タブをクリックします。

7. 関数が正常に実行されたことを確認します。以下のものを探します。

   Fetching activity logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
   Page 0: Retrieved X events
   Wrote X records to gs://agiloft-activity-logs/agiloft/logs_YYYYMMDD_HHMMSS.ndjson
   Successfully processed X records
   

8. [Cloud Storage] > [バケット] に移動します。

9. バケット名をクリックします。

10. プレフィックス フォルダ agiloft/ に移動します。

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

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

• HTTP 401: 環境変数の API 認証情報を確認します。ログインとパスワードが正しいことを確認します。
• HTTP 403: API ユーザーがアクティビティ ログテーブルに対する読み取りアクセス権を持ち、REST 対応グループに属していることを確認します。
• HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます。
• 「success」: false: エラー メッセージを確認します。一般的な原因としては、KB 名（大文字と小文字が区別される）またはテーブル名が正しくないことが考えられます。
• 環境変数が不足している: 必要な変数がすべて設定されていることを確認します。

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

1. [SIEM 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [単一フィードを設定] をクリックします。
4. [フィード名] フィールドに、フィードの名前を入力します（例: Agiloft Activity Logs）。
5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
6. [ログタイプ] として [Agiloft] を選択します。

7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます（例:）。

   [e-mail]
   

8. このメールアドレスをコピーして、次のステップで使用します。

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

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

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

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

      gs://agiloft-activity-logs/agiloft/
      

      • 次のように置き換えます。
        • agiloft-activity-logs: GCS バケット名。
        • agiloft: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
    注: URI の末尾には常にスラッシュ（/）を付けてください。

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

      • なし: 転送後にファイルを削除しません（テストにおすすめ）。
      • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
      • 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
      注: 削除オプションを選択する場合は、サービス アカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。

    • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます（デフォルトは 180 日）。

    • アセットの名前空間: アセットの名前空間

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

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

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

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

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

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

6. [保存] をクリックします。

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

UDM マッピング テーブル

ログフィールド	UDM マッピング	ロジック
_1880_full_name_label	additional.fields	統合済み
_1888_full_name_label	additional.fields	統合済み
action_duration_label	additional.fields	統合済み
cookie_label	additional.fields	統合済み
creator_login_label	additional.fields	統合済み
creator_team_label	additional.fields	統合済み
deleteable_label	additional.fields	統合済み
demo_data_label	additional.fields	統合済み
interface_label	additional.fields	統合済み
login_label	additional.fields	統合済み
record_id_label	additional.fields	統合済み
table_label_label	additional.fields	統合済み
type_label	additional.fields	統合済み
description	metadata.description	直接マッピングされます。
date	metadata.event_timestamp	MMM dd yyyy HH:mm:ss として解析
date_created	metadata.event_timestamp	MMM dd yyyy HH:mm:ss として解析
date_updated	metadata.event_timestamp	MMM dd yyyy HH:mm:ss として解析
timestamp	metadata.event_timestamp	dd/MMM/yyyy:HH:mm:ss Z として解析
has_principal	metadata.event_type	マッピング: true → STATUS_UPDATE
has_principal_user	metadata.event_type	マッピング: true → USER_UNCATEGORIZED
activity_log_id	metadata.product_deployment_id	直接マッピングされます。
id	metadata.product_log_id	直接マッピングされます。
method	network.http.method	直接マッピングされます。
url	network.http.referral_url	直接マッピングされます。
response_code	network.http.response_code	直接マッピングされます。
byte_transferred	network.received_bytes	直接マッピングされます。
session_duration	network.session_duration.seconds	直接マッピングされます。
client_ip	principal.asset.ip	統合済み
ip	principal.asset.ip	統合済み
client_ip	principal.ip	統合済み
ip	principal.ip	統合済み
user_primary_team_label	principal.user.attribute.labels	統合済み
company	principal.user.company_name	直接マッピングされます。
groups	principal.user.group_identifiers	統合済み
user_name	principal.user.user_display_name	直接マッピングされます。
user	principal.user.userid	直接マッピングされます。
user_login	principal.user.userid	直接マッピングされます。
なし	metadata.event_type	定数: USER_UNCATEGORIZED
なし	metadata.product_name	定数: Agiloft
なし	metadata.vendor_name	定数: Agiloft

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