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

MuleSoft Anypoint プラットフォームのログを収集する

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

Google SecOps SIEM

このドキュメントでは、Cloud Storage を使用して MuleSoft Anypoint プラットフォームのログから Google Security Operations に監査証跡イベントを取り込む方法について説明します。

始める前に

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

Google SecOps インスタンス
Cloud Storage API が有効になっている Google Cloud プロジェクト
Cloud Storage バケットを作成して管理する権限
Cloud Storage バケットで Identity and Access Management（IAM）ポリシーを管理する権限
Cloud Run functions、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
サービスアカウントを作成する権限
MuleSoft Anypoint Platform への特権アクセス

MuleSoft 組織 ID を取得する

Anypoint Platform にログインします。
[アクセス管理] > [組織] に移動します。
[ビジネスグループ] テーブルで、組織の名前をクリックします。
組織 ID（例: 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f）をコピーします。

または、MuleSoft Business Groups に移動して、URL から ID をコピーします。

MuleSoft 接続アプリを作成する

Anypoint Platform にログインします。
[アクセス管理> 接続済みアプリ> アプリを作成] に移動します。
次の構成情報を指定します。
- アプリ名: 一意の名前を入力します（例: Google SecOps export）。
- [App acts on its own behalf (client credentials)] を選択します。
[Add scopes] > [監査ログビューア] > [Next] をクリックします。
ログが必要なすべてのビジネスグループを選択します。
[次へ> スコープを追加] をクリックします。
[保存] をクリックして、クライアント ID とクライアントシークレットをコピーします。

注: 組織の権限によっては、追加のスコープが必要になる場合があります。権限エラーが発生した場合は、接続アプリに必要なすべてのビジネスグループへのアクセス権が付与されていることを確認します。

Cloud Storage バケットを作成する

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

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

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

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

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

Cloud Run 関数には、Cloud Storage バケットに書き込む権限を持つサービスアカウントが必要です。

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

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

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

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

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

次のように、Cloud Storage バケットに対する write 権限をサービスアカウントに付与します。

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

Pub/Sub トピックの作成

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

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

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

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

GCP Console で、[Cloud Run] に移動します。
[関数を作成] をクリックします。

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

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

[トリガー] セクションで [トリガーを追加] をクリックし、[Pub/Sub トリガー] を選択します。
[Eventarc トリガー] ペインで、次の構成の詳細を指定します。
- トリガー名: 生成された名前をそのまま使用するか、トリガーの名前を入力します。
- トリガーのタイプ: [Google のソース] を選択します。
- イベントプロバイダ: [Pub/Sub] を選択します。
- イベントタイプ: [google.cloud.pubsub.topic.v1.messagePublished] を選択します。
- Cloud Pub/Sub トピックを選択してください: トピック mulesoft-audit-trigger を選択します。
- リージョン: 関数と同じリージョンを選択します。
- サービスアカウント: サービスアカウント mulesoft-logs-collector-sa を選択します。
[トリガーを保存] をクリックします。

注: トリガーオプションは Eventarc から取得されます。[Pub/Sub トリガー] が表示されない場合は、プロジェクトの Eventarc API を有効にして、ページを再読み込みします。サービスアカウントに Eventarc イベントレシーバまたは Cloud Run 起動元のロールがないことがペインに表示されたら、[付与] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数の呼び出し時に認証を自動的に処理します。
[コンテナ、ネットワーキング、セキュリティ] に移動して開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: mulesoft-logs-collector-sa サービスアカウントを選択します。

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

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

変数名	値の例	説明
`GCS_BUCKET`	`mulesoft-audit-logs`	Cloud Storage バケットの名前
`GCS_PREFIX`	`mulesoft-audit`	ログファイルの接頭辞
`STATE_KEY`	`mulesoft-audit-state.json`	状態パス（ログプレフィックスの外側）
`MULE_ORG_ID`	`your_org_id`	MuleSoft 組織 ID
`CLIENT_ID`	`your_client_id`	接続されているアプリのクライアント ID
`CLIENT_SECRET`	`your_client_secret`	接続されているアプリのクライアントシークレット
`MAX_RECORDS`	`10000`	実行あたりの最大レコード数
`PAGE_SIZE`	`200`	ページあたりのエントリ数（API の最大値は 200）
`LOOKBACK_HOURS`	`24`	初回実行時の初期ルックバック期間

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

関数のコードを追加する

[関数のエントリポイント] に「main」と入力します。

インラインコードエディタで、次の 2 つのファイルを作成します。

最初のファイル - main.py:

import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import urllib3
from urllib.parse import urlencode
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()

# MuleSoft API endpoints
TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
AUDIT_URL = "https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query?cursorPagination=true"

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mulesoft-audit')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'mulesoft-audit-state.json')
MULE_ORG_ID = os.environ.get('MULE_ORG_ID')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
# The Audit Log Query API returns at most 200 entries per JSON query.
PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '200')), 200)
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events the
# platform records late are still collected. Re-read events are removed by hash.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

class FetchError(Exception):
    """Raised when the MuleSoft API call fails.

    The watermark must never advance on a failed fetch, otherwise the events in
    the failed window are skipped permanently.
    """

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.astimezone(timezone.utc).timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """Parse an ISO 8601 datetime string into an aware datetime."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt

def event_keys(event: dict) -> set:
    """Return the identity this event can be recognised by.

    Audit log entries carry no unique event ID, so a record counts as already
    collected when its full JSON content was seen before. The overlap window
    returns the same JSON again, which this catches without any knowledge of
    the payload schema.
    """
    return {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}

def event_time(event: dict) -> str:
    """Return the event timestamp as an ISO 8601 string, or an empty string.

    Audit log entries carry the event time in `timestamp` as Unix milliseconds.
    """
    raw = event.get('timestamp')
    if raw is None or raw == '':
        return ''
    try:
        return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc).isoformat()
    except (TypeError, ValueError):
        return ''

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch MuleSoft Anypoint audit logs and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, MULE_ORG_ID, CLIENT_ID, CLIENT_SECRET]):
        # Raise rather than return: a bare return acks the Pub/Sub message and
        # reports the run as successful, silently discarding the schedule tick.
        raise RuntimeError('Missing required environment variables')

    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket, STATE_KEY)

    now = datetime.now(timezone.utc)
    watermark = None
    if state.get('last_event_time'):
        watermark = parse_datetime(state['last_event_time'])
    seen_keys = set(state.get('seen_keys', []))

    if watermark is None:
        start_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
        start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

    print(f"Fetching audit logs from {start_time.isoformat()} to {now.isoformat()}")

    token = get_token(CLIENT_ID, CLIENT_SECRET)

    # A FetchError here propagates: the run fails, the watermark is untouched,
    # and the next invocation retries the same window.
    records, newest_event_time = fetch_audit(
        org_id=MULE_ORG_ID,
        token=token,
        start_time=start_time,
        end_time=now,
        page_size=PAGE_SIZE,
        max_records=MAX_RECORDS,
    )

    # Remove events already written by an earlier run. Without this, the overlap
    # window re-emits its events on every invocation.
    fresh = [r for r in records if not (event_keys(r) & seen_keys)]
    print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

    if not fresh:
        print("No new audit log records found. Watermark left unchanged.")
        return

    if not newest_event_time:
        # A watermark that cannot be derived from the data is not a watermark.
        # Advancing on wall-clock here would skip every late-recorded event.
        raise FetchError('Records were returned but no event timestamp could be parsed')

    timestamp = now.strftime('%Y%m%dT%H%M%SZ')
    object_key = f"{GCS_PREFIX}/audit_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')

    print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")

    # Advance the watermark only after the data is durably written.
    new_watermark = parse_datetime(newest_event_time)
    if watermark and new_watermark < watermark:
        new_watermark = watermark

    # Retain only the hashes still inside the overlap window, so the state file
    # stays small while covering every event the next query can re-read.
    cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
    retained = set()
    for record in records:
        stamp = event_time(record)
        if not stamp:
            continue
        if parse_datetime(stamp) >= cutoff:
            retained.update(event_keys(record))

    save_state(bucket, STATE_KEY, {
        'last_event_time': new_watermark.isoformat(),
        'seen_keys': sorted(retained),
    })

    print(f"Successfully processed {len(fresh)} records")

def load_state(bucket, key):
    """Read the collector state from Cloud Storage.

    Only a missing object is treated as a cold start. Any other error is raised:
    swallowing it would silently reset collection to the full lookback window
    and re-ingest that entire period.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the lookback window.')
        return {}

def save_state(bucket, key, state: dict):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. If the state write fails after the data was
    uploaded, the next run repeats the same window and duplicates it.
    """
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(state, indent=2),
        content_type='application/json',
    )
    print(f"Saved state: last_event_time={state.get('last_event_time')}")

def retry_after_seconds(response, backoff: float) -> int:
    """Parse Retry-After defensively; it may be an HTTP date, which int() cannot parse."""
    raw = response.headers.get('Retry-After')
    try:
        return int(raw) if raw else int(backoff)
    except (TypeError, ValueError):
        return int(backoff)

def get_token(client_id: str, client_secret: str) -> str:
    """Get an OAuth 2.0 access token from MuleSoft using client credentials."""
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
    }).encode('utf-8')

    backoff = 1.0
    for attempt in range(5):
        try:
            response = http.request(
                'POST',
                TOKEN_URL,
                body=body,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
            )
        except Exception as e:
            raise FetchError(f'Token request failed: {e}') from e

        if response.status == 429:
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429) on token request. Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(f'Failed to get token: HTTP {response.status} {response.data.decode("utf-8", "replace")}')

        try:
            return json.loads(response.data.decode('utf-8'))['access_token']
        except (json.JSONDecodeError, KeyError) as e:
            raise FetchError(f'Malformed token response: {e}') from e

    raise FetchError('Rate limited repeatedly on token request; giving up without advancing the watermark')

def fetch_audit(org_id: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """Fetch audit log entries from the MuleSoft Audit Log Query API.

    Uses cursor pagination. The API returns entries in ascending timestamp order,
    which a time watermark requires: with newest-first ordering a truncated run
    would advance the watermark past unread events.

    Returns:
        Tuple of (records list, newest_event_time ISO string).

    Raises:
        FetchError: on any API or transport failure, so that the caller does not
            mistake a failed fetch for an empty result and advance the watermark.
    """
    endpoint = AUDIT_URL.format(org_id=org_id)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }

    records = []
    newest_time = None
    cursor = None
    page_num = 0
    backoff = 1.0
    rate_limit_retries = 0
    MAX_RATE_LIMIT_RETRIES = 5

    while True:
        page_num += 1

        if len(records) >= max_records:
            # Stop cleanly. The remaining events stay ahead of the watermark and
            # are collected by the next run.
            print(f"Reached max_records limit ({max_records})")
            break

        # Rebuild the full query on every page; the time window must never be dropped.
        body = {
            'startDate': to_unix_millis(start_time),
            'endDate': to_unix_millis(end_time),
            'limit': page_size,
            'ascending': True,
        }
        if cursor:
            body['cursor'] = cursor

        try:
            response = http.request(
                'POST',
                endpoint,
                body=json.dumps(body).encode('utf-8'),
                headers=headers,
            )
        except Exception as e:
            raise FetchError(f'Request to {endpoint} failed: {e}') from e

        if response.status == 429:
            rate_limit_retries += 1
            if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429). Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        backoff = 1.0
        rate_limit_retries = 0

        if response.status != 200:
            raise FetchError(f'HTTP {response.status} from {endpoint}: {response.data.decode("utf-8", "replace")}')

        try:
            data = json.loads(response.data.decode('utf-8'))
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response from {endpoint}: {e}') from e

        page_results = data.get('data') or []
        if not page_results:
            print("No more results (empty page)")
            break

        print(f"Page {page_num}: Retrieved {len(page_results)} entries")
        records.extend(page_results)

        for event in page_results:
            stamp = event_time(event)
            if not stamp:
                continue
            if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                newest_time = stamp

        cursor = data.get('cursor')
        if not cursor:
            break

    print(f"Retrieved {len(records)} total entries from {page_num} pages")
    return records, newest_time

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

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

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

重要な考慮事項

レート制限: 監査ログクエリエンドポイントは、3 つのコントロールプレーンで IP ごとにレート制限を適用します。米国のコントロールプレーンでは IP あたり 1 分間に 700 件のリクエストが許可されますが、EU と Gov のコントロールプレーンでは IP あたり 1 分間に 40 件のリクエストが許可されます。この関数は、レート制限を自動的に処理するために指数バックオフを実装します。
トークンの有効期限: 通常、アクセストークンは発行されてから 30 ～ 60 分後に期限切れになります。この関数は実行ごとに新しいトークンをリクエストしますが、これはそのウィンドウ内に収まります。
重複除去と状態: 関数は、最新のイベント時間のウォーターマークと、重複ウィンドウ内のイベントのハッシュを状態ファイルに保持します。各実行では、ウォーターマークの 2 分前に再読み取りを行い、遅れて記録されたエントリをキャッチして、すでに書き込まれたエントリを削除します。そのため、再試行、手動実行、スケジューラ遅延によって重複するイベントが生成されることはありません。API 呼び出しが失敗すると、実行が失敗し、ウォーターマークが変更されないため、次の実行で同じウィンドウが再試行されます。
監査ログの保持: 監査ログのデフォルトの保持期間は 1 年です。組織が 2023 年 7 月 10 日より前に作成され、保持期間を手動で変更していない場合、保持期間は 6 年間です。構成された保持期間を超えてログを保持する必要がある場合は、ログを定期的にダウンロードします。

Cloud Scheduler ジョブを作成する

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

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

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

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

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

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

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

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

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

注: 各 Google SecOps インスタンスには一意のサービスアカウントがあります。他のドキュメントや例のサービスアカウントは使用しないでください。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: Cloud Storage バケットの URI を入力します。
```
gs://mulesoft-audit-logs/
```
  - mulesoft-audit-logs は、バケットの実際の名前に置き換えます。
  注: URI の末尾には常にスラッシュ（/）を含めてください。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - ファイルを削除しない: ソースからファイルを削除しません（テストに推奨）。
  - 転送されたファイルと空のディレクトリを削除する: 取得が正常に完了した後、ソースからファイルと空のディレクトリを削除します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- Asset namespace: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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

Google SecOps サービスアカウントには、Cloud Storage バケットに対する 2 つのロールが必要です。ログオブジェクトを読み取るための Storage オブジェクト閲覧者と、バケットメタデータを読み取るためのバケットレベルのロールです。

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービスアカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: 次の両方を選択します。
- Storage オブジェクト閲覧者: ログオブジェクトを読み取ります。
- Storage レガシーバケット読み取り: バケットメタデータを読み取ります。[転送されたファイルと空のディレクトリを削除する] 削除オプションを選択した場合は、代わりに [ストレージレガシーバケット書き込み] を選択します。これにより、削除権限も付与されます。
[保存] をクリックします。

注: 両方のロールを付与します。Storage オブジェクト閲覧者は、フィードに必要なバケットレベルの権限を付与しません。また、Storage レガシーバケット ロールは、オブジェクトの内容を読み取る権限を付与しません。

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