Nagios XI ログを収集する

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

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

Nagios XI は、サーバー、ネットワーク、アプリケーション、サービスのモニタリングを提供する包括的なインフラストラクチャ モニタリング ソリューションです。ホストとサービスのステータス、パフォーマンス指標を追跡し、REST API を介して IT インフラストラクチャの問題に関するアラートを生成します。

始める前に

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

  • Google SecOps インスタンス
  • Cloud Storage、Cloud Run、Pub/Sub、Cloud Scheduler API が有効になっている GCP プロジェクト
  • GCS バケットを作成および管理する権限
  • GCS バケットの IAM ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • ユーザー管理権限を持つ Nagios XI への特権アクセス
  • Nagios XI バージョン 5 以降(REST API のサポート)

Nagios XI API アクセスを構成する

Google SecOps がモニタリング データを取得できるようにするには、API アクセス権と読み取り専用権限を持つユーザー アカウントを作成する必要があります。

API アクセス権を持つ読み取り権限のみのユーザーを作成する

  1. 管理者権限で Nagios XI ウェブ インターフェースにログインします。
  2. [管理者> ユーザーの管理] に移動します。
  3. [新しいユーザーの追加] をクリックします。
  4. 次の構成情報を指定します。
    • ユーザー名: わかりやすい名前を入力します(例: chronicle-integration)。
    • パスワード: 安全なパスワードを入力します。
    • 名前: 「Chronicle Integration User」と入力します。
    • メールアドレス: 有効なメールアドレスを入力します。
  5. [セキュリティ設定] セクションで、次のように構成します。
    • Authorization Level: [User] を選択します。
    • Can see all hosts and services: このオプションをオンにします。
    • 読み取り専用アクセス: このオプションをオンにします。
    • API アクセス: このオプションをオンにします。
  6. 次のオプションのチェックボックスをオフにします。
    • 次回のログイン時にパスワードの変更を要求する
    • Email User Account Information
    • モニタリングの連絡先として作成
  7. [Account Enabled] がオンになっていることを確認します。
  8. [ユーザーを追加] をクリックします。

API キーを取得する

  1. [ユーザーの管理] ページで、ユーザー アカウント(chronicle-integration)をクリックします。
  2. ユーザー アカウント設定ページで、[API キー] フィールドを見つけます。
  3. API キーの値をコピーします。

権限を確認する

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

  1. chronicle-integration ユーザー アカウントで Nagios XI にログインします。
  2. [ホーム]> [ホストのステータス] に移動します。
  3. モニタリング対象のすべてのホストとサービスが表示されている場合は、必要な権限が付与されています。
  4. ホストまたはサービスが表示されない場合は、管理者に連絡して [すべてのホストとサービスを表示できる] 権限を付与してもらいます。

テスト API へのアクセス

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

    # Replace with your actual values
    NAGIOS_HOST="https://your-nagios-server.example.com"
    API_KEY="your-api-key"
    
    # Test API access - query host status
    curl -k "${NAGIOS_HOST}/nagiosxi/api/v1/objects/hoststatus?apikey=${API_KEY}&pretty=1"
    

    リクエストが成功すると、ホストのステータス情報を含む JSON オブジェクトが返されます。

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

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

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

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

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

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

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

Pub/Sub トピックの作成

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

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

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

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

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

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

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

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

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

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    GCS_BUCKET nagios-xi-logs GCS バケット名
    GCS_PREFIX nagios-xi ログファイルの接頭辞
    STATE_KEY nagios-xi/state.json 状態ファイルのパス
    NAGIOS_BASE_URL https://your-nagios-server.example.com Nagios XI のベース URL
    NAGIOS_API_KEY your-api-key Nagios XI API キー
    MAX_RECORDS 1000 実行あたりのエンドポイントあたりの最大レコード数
    PAGE_SIZE 200 API ページあたりのレコード数
    LOOKBACK_HOURS 24 初期ルックバック期間
  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
      import time
      
      http = urllib3.PoolManager(
        timeout=urllib3.Timeout(connect=5.0, read=30.0),
        retries=False,
      )
      
      storage_client = storage.Client()
      
      GCS_BUCKET = os.environ.get('GCS_BUCKET')
      GCS_PREFIX = os.environ.get('GCS_PREFIX', 'nagios-xi')
      STATE_KEY = os.environ.get('STATE_KEY', 'nagios-xi/state.json')
      NAGIOS_BASE_URL = os.environ.get('NAGIOS_BASE_URL', '').rstrip('/')
      NAGIOS_API_KEY = os.environ.get('NAGIOS_API_KEY')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '200'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
      ENDPOINTS = [
        'hoststatus',
        'servicestatus',
        'statehistory',
        'logentries',
      ]
      
      @functions_framework.cloud_event
      def main(cloud_event):
        if not all([GCS_BUCKET, NAGIOS_BASE_URL, NAGIOS_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()}")
      
          start_ts = int(last_time.timestamp())
          end_ts = int(now.timestamp())
      
          all_records = []
          newest_time = None
      
          for endpoint in ENDPOINTS:
            records, endpoint_newest = fetch_nagios_endpoint(
              endpoint, start_ts, end_ts
            )
            for record in records:
              record['_nagios_endpoint'] = endpoint
            all_records.extend(records)
      
            if endpoint_newest:
              if newest_time is None or endpoint_newest > newest_time:
                newest_time = endpoint_newest
      
          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}/nagios_logs_{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}")
      
          if newest_time:
            save_state(bucket, newest_time)
          else:
            save_state(bucket, now.isoformat())
      
          print(f"Successfully processed {len(all_records)} records")
      
        except Exception as e:
          print(f'Error processing logs: {str(e)}')
          raise
      
      def fetch_nagios_endpoint(endpoint, start_ts, end_ts):
        base_url = f"{NAGIOS_BASE_URL}/nagiosxi/api/v1/objects/{endpoint}"
      
        records = []
        newest_time = None
        start_index = 0
        page_num = 0
        backoff = 1.0
      
        while True:
          page_num += 1
      
          if len(records) >= MAX_RECORDS:
            print(f"{endpoint}: Reached max_records limit ({MAX_RECORDS})")
            break
      
          remaining = min(PAGE_SIZE, MAX_RECORDS - len(records))
          params = [
            f"apikey={NAGIOS_API_KEY}",
            f"starttime={start_ts}",
            f"endtime={end_ts}",
            f"records={remaining}",
            f"start={start_index}",
          ]
          url = f"{base_url}?{'&'.join(params)}"
      
          try:
            response = http.request('GET', url)
      
            if response.status == 429:
              retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
              print(f"{endpoint}: 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"{endpoint}: HTTP Error {response.status}")
              response_text = response.data.decode('utf-8')
              print(f"Response body: {response_text}")
              break
      
            data = json.loads(response.data.decode('utf-8'))
      
            record_count = data.get('recordcount', 0)
            result_key = get_result_key(endpoint)
            page_results = data.get(result_key, [])
      
            if not page_results:
              print(f"{endpoint}: No more results at offset {start_index}")
              break
      
            print(f"{endpoint} page {page_num}: Retrieved {len(page_results)} events")
            records.extend(page_results)
      
            for event in page_results:
              try:
                event_time = event.get('state_time') or event.get('status_update_time') or event.get('entry_time')
                if event_time:
                  if newest_time is None or event_time > newest_time:
                    newest_time = event_time
              except Exception as e:
                print(f"Warning: Could not parse event time: {e}")
      
            if len(page_results) < remaining:
              print(f"{endpoint}: Reached last page (returned {len(page_results)} < {remaining})")
              break
      
            start_index += len(page_results)
      
          except Exception as e:
            print(f"{endpoint}: Error fetching logs: {e}")
            break
      
        print(f"{endpoint}: Retrieved {len(records)} total records from {page_num} pages")
        return records, newest_time
      
      def get_result_key(endpoint):
        key_map = {
          'hoststatus': 'hoststatus',
          'servicestatus': 'servicestatus',
          'statehistory': 'statehistory',
          'logentries': 'logentry',
        }
        return key_map.get(endpoint, endpoint)
      
      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. 次の構成情報を提供してください。

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

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

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

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

統合をテストする

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

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    hoststatus page 1: Retrieved X events
    servicestatus page 1: Retrieved X events
    statehistory page 1: Retrieved X events
    logentries page 1: Retrieved X events
    Wrote X records to gs://nagios-xi-logs/nagios-xi/nagios_logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. [Cloud Storage] > [バケット] に移動します。

  9. nagios-xi-logs をクリックします。

  10. nagios-xi/ フォルダに移動します。

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

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

  • HTTP 401: NAGIOS_API_KEY 環境変数が正しく、ユーザーの API アクセスが有効になっていることを確認します。
  • HTTP 403: ユーザー アカウントにすべてのホストとサービスを表示できる権限があることを確認します
  • HTTP 429: レート制限。関数はバックオフを使用して自動的に再試行します。
  • 環境変数が不足している: 必要なすべての変数が Cloud Run functions の関数の構成に設定されていることを確認します。

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

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Nagios XI Logs GCS)。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Nagios] を選択します。
  7. [サービス アカウントを取得する] をクリックします。
  8. 一意のサービス アカウントのメールアドレスが表示されます。次に例を示します。

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

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

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

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

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

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

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

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

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

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

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

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

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

UDM マッピング テーブル

ログフィールド UDM マッピング ロジック
src_ip has_principal src_ip を principal.ip と principal.asset.ip にマージできた場合は「true」に設定します。
src_ip principal.asset.ip 値を直接コピーする
src_ip principal.ip 値を直接コピーする
jobid jobid_label.key 「Jobid」に設定
jobid jobid_label.value 値を直接コピーする
jobid_label principal.resource.attribute.labels jobid_label から統合
pid principal.process.pid 値を直接コピーする
ent metadata.product_event_type 値を直接コピーする
説明 metadata.description 値が直接コピーされる
msg_ip has_target msg_ip を target.ip と target.asset.ip にマージできた場合は「true」に設定します
msg_ip target.asset.ip 値を直接コピーする
msg_ip target.ip 値が直接コピーされる
ポート target.port 値が直接コピーされ、整数に変換されます。
column1 principal.asset.hostname ent == "SERVICE ALERT" の場合は column1 の値、ent == "SERVICE NOTIFICATION" の場合は column2 の値
column1 principal.hostname ent == "SERVICE ALERT" の場合は column1 の値、ent == "SERVICE NOTIFICATION" の場合は column2 の値
column1 principal.user.user_display_name ent == "SERVICE NOTIFICATION" の場合、値は直接コピーされます。
column2 security_result.summary ent == "SERVICE ALERT" で column2 が空でない場合は column3 の値、それ以外で ent == "SERVICE NOTIFICATION" の場合は column3 の値
column3 security_result.summary ent == "SERVICE ALERT" かつ column3 が空でない場合は column3 の値、それ以外の場合は ent == "SERVICE NOTIFICATION" の場合は column3 の値
column4 security_result.severity column4 が ["LOW", "MEDIUM", "HIGH", "CRITICAL"] で、ent が ["SERVICE NOTIFICATION", "SERVICE ALERT"] の場合、値は直接コピーされます。
column4 security_result.severity_details column4 が ["LOW", "MEDIUM", "HIGH", "CRITICAL"] に含まれておらず、ent が ["SERVICE NOTIFICATION", "SERVICE ALERT"] に含まれている場合、値は直接コピーされます。
has_principal metadata.event_type has_principal と has_target が true の場合は「NETWORK_CONNECTION」、has_principal が true の場合は「STATUS_UPDATE」、has_principal_user が true の場合は「USER_UNCATEGORIZED」、それ以外の場合は「GENERIC_EVENT」に設定します。
has_target metadata.event_type
has_principal_user metadata.event_type
security_result event.idm.read_only_udm.security_result security_result から統合
metadata event.idm.read_only_udm.metadata metadata から名前変更
ターゲット event.idm.read_only_udm.target target から名前が変更されました
プリンシパル event.idm.read_only_udm.principal プリンシパルから名前を変更
network event.idm.read_only_udm.network ネットワークから名前を変更
metadata.product_name metadata.product_name 「NAGIOS」に設定します。
metadata.vendor_name metadata.vendor_name 「NAGIOS」に設定します。

変更ログ

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

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