OpenTelemetry Netflow レシーバのログを収集する

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

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

OpenTelemetry Netflow レシーバーは、netflow、sflow、IPFIX UDP トラフィックをリッスンして OpenTelemetry ログレコードに変換するオープンソース コンポーネントです。これにより、プロトコルの識別、トラフィック量の分析、ポート使用状況の追跡、バイト/パケットの統計情報など、ネットワーク トラフィックのモニタリングと分析が可能になります。

始める前に

次の前提条件を満たしていることを確認します。

  • Google SecOps インスタンス
  • Cloud Storage API が有効になっている GCP プロジェクト
  • GCS バケットを作成および管理する権限
  • GCS バケットの IAM ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • NetFlow、sFlow、IPFIX データを送信できるネットワーク デバイス
  • ネットワーク デバイスのエクスポート設定を構成するためのアクセス権

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

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

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

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

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

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

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

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

  1. GCP コンソールで、[IAM と管理>サービス アカウント] に移動します。
  2. [サービス アカウントを作成] をクリックします。
  3. 次の構成情報を指定します。
    • サービス アカウント名: 「netflow-collector-sa」と入力します。
    • サービス アカウントの説明: Service account for Cloud Run function to collect netflow 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. バケット名（netflow-logs-bucket など）をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成情報を指定します。
    • プリンシパルを追加: サービス アカウントのメールアドレス（例: netflow-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
    • ロールを割り当てる: [ストレージ オブジェクト管理者] を選択します。
  6. [保存] をクリックします。

Pub/Sub トピックの作成

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

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

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

Cloud Run 関数は、ネットフロー データを受信して GCS にエクスポートする OpenTelemetry コレクタを実行します。

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

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

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

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

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

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

    1. [認証が必要] を選択します。
    2. Identity and Access Management（IAM）を確認します。

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

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

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

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

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    GCS_BUCKET netflow-logs-bucket GCS バケット名
    GCS_PREFIX netflow ログファイルの接頭辞
    NETFLOW_PORT 2055 NetFlow レシーバのポート
    NETFLOW_SCHEME netflow スキームタイプ: netflow、sflow、ipfix
    NETFLOW_SOCKETS 4 UDP ソケットの数
    NETFLOW_WORKERS 8 デコーダ ワーカーの数

  10. [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。

    • リクエストのタイムアウト: 600 秒（10 分）を入力します。

  11. [設定] タブに移動します。

    • [リソース] セクションで、次の操作を行います。
      • メモリ: 1 GiB 以上を選択します。
      • CPU: [2] を選択します。

  12. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数]: 「1」と入力します。
    • [インスタンスの最大数] に「10」と入力します。

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

  14. サービスが作成されるまで待ちます（1 ～ 2 分）。

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

関数コードを追加する

  1. [エントリ ポイント] フィールドに「main」と入力します。

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

    • main.py:
    import functions_framework
from google.cloud import storage
import json
import os
import subprocess
import tempfile
import signal
import time
from datetime import datetime, timezone

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netflow')
NETFLOW_PORT = os.environ.get('NETFLOW_PORT', '2055')
NETFLOW_SCHEME = os.environ.get('NETFLOW_SCHEME', 'netflow')
NETFLOW_SOCKETS = os.environ.get('NETFLOW_SOCKETS', '4')
NETFLOW_WORKERS = os.environ.get('NETFLOW_WORKERS', '8')

# Global process handle
collector_process = None

def create_collector_config():
  """Create OpenTelemetry Collector configuration."""
  timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
  file_path = f"/tmp/netflow_{timestamp}.ndjson"

  config = {
    'receivers': {
      'netflow': {
        'scheme': NETFLOW_SCHEME,
        'hostname': '0.0.0.0',
        'port': int(NETFLOW_PORT),
        'sockets': int(NETFLOW_SOCKETS),
        'workers': int(NETFLOW_WORKERS),
        'queue_size': 5000
      }
    },
    'processors': {
      'batch': {
        'timeout': '10s',
        'send_batch_size': 1000
      }
    },
    'exporters': {
      'file': {
        'path': file_path,
        'format': 'json'
      }
    },
    'service': {
      'pipelines': {
        'logs': {
          'receivers': ['netflow'],
          'processors': ['batch'],
          'exporters': ['file']
        }
      },
      'telemetry': {
        'logs': {
          'level': 'info'
        }
      }
    }
  }

  config_path = '/tmp/otel-config.yaml'
  with open(config_path, 'w') as f:
    import yaml
    yaml.dump(config, f)

  return config_path, file_path

def upload_to_gcs(file_path):
  """Upload netflow logs to GCS."""
  if not os.path.exists(file_path) or os.path.getsize(file_path) == 0:
    print(f"No data to upload from {file_path}")
    return

  bucket = storage_client.bucket(GCS_BUCKET)
  timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
  object_key = f"{GCS_PREFIX}/netflow_{timestamp}.ndjson"
  blob = bucket.blob(object_key)

  blob.upload_from_filename(file_path, content_type='application/x-ndjson')
  print(f"Uploaded {os.path.getsize(file_path)} bytes to gs://{GCS_BUCKET}/{object_key}")

def signal_handler(signum, frame):
  """Handle shutdown signals."""
  global collector_process
  if collector_process:
    print("Shutting down collector...")
    collector_process.terminate()
    collector_process.wait(timeout=10)

@functions_framework.cloud_event
def main(cloud_event):
  """
  Cloud Run function to run OpenTelemetry Collector for netflow collection.

  Args:
    cloud_event: CloudEvent object containing Pub/Sub message
  """
  global collector_process

  if not GCS_BUCKET:
    print('Error: GCS_BUCKET environment variable not set')
    return

  try:
    # Set up signal handlers
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

    # Create collector configuration
    config_path, file_path = create_collector_config()
    print(f"Created collector config at {config_path}")
    print(f"Netflow receiver listening on {NETFLOW_SCHEME}://0.0.0.0:{NETFLOW_PORT}")

    # Start OpenTelemetry Collector
    collector_process = subprocess.Popen(
      ['/otelcol-contrib', '--config', config_path],
      stdout=subprocess.PIPE,
      stderr=subprocess.PIPE,
      text=True
    )

    print(f"Started OpenTelemetry Collector (PID: {collector_process.pid})")

    # Run for collection period (e.g., 5 minutes)
    collection_time = 300
    print(f"Collecting netflow data for {collection_time} seconds...")

    start_time = time.time()
    while time.time() - start_time < collection_time:
      if collector_process.poll() is not None:
        stdout, stderr = collector_process.communicate()
        print(f"Collector exited unexpectedly")
        print(f"STDOUT: {stdout}")
        print(f"STDERR: {stderr}")
        break
      time.sleep(10)

    # Stop collector
    if collector_process.poll() is None:
      print("Stopping collector...")
      collector_process.terminate()
      collector_process.wait(timeout=10)

    # Upload collected data
    upload_to_gcs(file_path)

    # Cleanup
    if os.path.exists(file_path):
      os.remove(file_path)
    if os.path.exists(config_path):
      os.remove(config_path)

    print("Netflow collection completed successfully")

  except Exception as e:
    print(f'Error during netflow collection: {str(e)}')
    if collector_process and collector_process.poll() is None:
      collector_process.terminate()
    raise
    • requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*
PyYAML==6.*
    • Dockerfile:
    FROM python:3.12-slim

# Install OpenTelemetry Collector Contrib
RUN apt-get update && apt-get install -y wget && \
  wget https://github.com/open-telemetry/opentelemetry-collector-releases/releases/download/v0.144.0/otelcol-contrib_0.144.0_linux_amd64.deb && \
  dpkg -i otelcol-contrib_0.144.0_linux_amd64.deb && \
  rm otelcol-contrib_0.144.0_linux_amd64.deb && \
  apt-get clean

# Set working directory
WORKDIR /app

# Copy requirements and install
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy function code
COPY main.py .

# Expose netflow port
EXPOSE 2055/udp

# Run function
CMD ["functions-framework", "--target=main", "--port=8080"]

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

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

NetFlow データを送信するようにネットワーク デバイスを構成する

ネットワーク デバイス（ルーター、スイッチ、ファイアウォール）を構成して、NetFlow、sFlow、または IPFIX データを Cloud Run 関数にエクスポートします。

Cloud Run 関数の外部 IP を取得する

  1. GCP コンソールで、[Cloud Run] > [サービス] に移動します。
  2. 関数名 netflow-collector をクリックします。
  3. 上部に表示されている URL（例: https://netflow-collector-xxxxx-uc.a.run.app）をコピーします。
  4. URL からホスト名を抽出します。

  5. nslookup または dig を使用して IP アドレスを解決します。

    nslookup netflow-collector-xxxxx-uc.a.run.app

  • 例: Cisco ルーターの NetFlow 構成

      ! Configure netflow exporter
  flow exporter OTEL-EXPORTER
  destination <CLOUD_RUN_IP>
  transport udp 2055
  source <INTERFACE>
  export-protocol netflow-v9

  ! Configure flow monitor
  flow monitor OTEL-MONITOR
  exporter OTEL-EXPORTER
  record netflow ipv4 original-input

  ! Apply to interface
  interface GigabitEthernet0/0
  ip flow monitor OTEL-MONITOR input
  ip flow monitor OTEL-MONITOR output

  • 例: 一般的な NetFlow 構成

    ほとんどのネットワーク デバイスでは、次の設定を構成します。

    • Netflow バージョン: NetFlow v5、v9、または IPFIX
    • コレクタ IP: Cloud Run 関数の IP アドレス
    • コレクタ ポート: 2055（または構成済みのポート）
    • プロトコル: UDP
    • アクティブ タイムアウト: 60 秒（推奨）
    • 非アクティブ タイムアウト: 15 秒（推奨）

Cloud Scheduler ジョブの作成

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

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

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

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

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

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

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

頻度 CRON 式 ユースケース
5 分おき */5 * * * * 大容量、低レイテンシ
15 分ごと */15 * * * * 検索量が普通
1 時間ごと 0 * * * * 標準（推奨）
6 時間ごと 0 */6 * * * 少量、バッチ処理

統合をテストする

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

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

    Started OpenTelemetry Collector (PID: ...)
Netflow receiver listening on netflow://0.0.0.0:2055
Collecting netflow data for 300 seconds...
Uploaded ... bytes to gs://netflow-logs-bucket/netflow/netflow_YYYYMMDD_HHMMSS.ndjson
Netflow collection completed successfully

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

  9. バケット名 netflow-logs-bucket をクリックします。

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

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

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

  • コレクタが予期せず終了した: Netflow レシーバの構成を確認する
  • アップロードするデータがない: ネットワーク デバイスが正しい IP とポートに Netflow データを送信していることを確認します。
  • 権限が拒否されました: サービス アカウントにストレージ オブジェクト管理者ロールがあることを確認する
  • ポート バインディング エラー: ポート 2055 がすでに使用されていないことを確認します。

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

Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。

サービス アカウントのメールアドレスを取得する

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

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

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

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

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

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

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

      gs://netflow-logs-bucket/netflow/

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

      • なし: 転送後にファイルを削除しません（テストに推奨）。
      • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。

      • 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。

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

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

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

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

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

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

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

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

UDM マッピング テーブル

次の表は、OpenTelemetry Netflow Receiver のログフィールドが Google SecOps UDM フィールドにマッピングされる方法を示しています。

OpenTelemetry フィールド UDM フィールド 説明
source.address principal.ip 送信元 IP アドレス
source.port principal.port 送信元ポート番号
destination.address target.ip 宛先 IP アドレス
destination.port target.port 宛先ポート番号
network.transport network.ip_protocol トランスポート プロトコル（tcp、udp）
network.type network.ip_version IP バージョン（ipv4、ipv6）
flow.io.bytes network.sent_bytes 転送された合計バイト数
flow.io.packets network.sent_packets 転送されたパケットの合計数
flow.type metadata.product_log_id フロータイプ（NETFLOW_V5、NETFLOW_V9、SFLOW_5、IPFIX）
flow.start metadata.event_timestamp フローの開始時間
flow.end network.session_duration フローの終了時間
flow.sampler_address intermediary.ip Netflow エクスポータの IP アドレス
flow.tcp_flags network.tcp_flags TCP フラグ
flow.io.bytes event.idm.read_only_udm.network.sent_bytes 変更履歴からマッピング
source.port event.idm.read_only_udm.principal.port 変更履歴からマッピング
destination.port event.idm.read_only_udm.target.port 変更履歴からマッピング
network.transport event.idm.read_only_udm.network.ip_protocol 変更履歴からマッピング
network.type event.idm.read_only_udm.security_result.detection_fields 変更履歴からマッピング
flow.sequence_num event.idm.read_only_udm.network.session_id 変更履歴からマッピング
flow.sampler_address event.idm.read_only_udm.observer.ip 変更履歴からマッピング
flow.type event.idm.read_only_udm.metadata.product_event_type 変更履歴からマッピング

変更ログ

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

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