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 アクセス権を持つ読み取り権限のみのユーザーを作成する
- 管理者権限で Nagios XI ウェブ インターフェースにログインします。
- [管理者> ユーザーの管理] に移動します。
- [新しいユーザーの追加] をクリックします。
- 次の構成情報を指定します。
- ユーザー名: わかりやすい名前を入力します(例:
chronicle-integration)。 - パスワード: 安全なパスワードを入力します。
- 名前: 「
Chronicle Integration User」と入力します。 - メールアドレス: 有効なメールアドレスを入力します。
- ユーザー名: わかりやすい名前を入力します(例:
- [セキュリティ設定] セクションで、次のように構成します。
- Authorization Level: [User] を選択します。
- Can see all hosts and services: このオプションをオンにします。
- 読み取り専用アクセス: このオプションをオンにします。
- API アクセス: このオプションをオンにします。
- 次のオプションのチェックボックスをオフにします。
- 次回のログイン時にパスワードの変更を要求する
- Email User Account Information
- モニタリングの連絡先として作成
- [Account Enabled] がオンになっていることを確認します。
- [ユーザーを追加] をクリックします。
API キーを取得する
- [ユーザーの管理] ページで、ユーザー アカウント(
chronicle-integration)をクリックします。 - ユーザー アカウント設定ページで、[API キー] フィールドを見つけます。
API キーの値をコピーします。
権限を確認する
アカウントに必要な権限があることを確認するには:
chronicle-integrationユーザー アカウントで Nagios XI にログインします。- [ホーム]> [ホストのステータス] に移動します。
- モニタリング対象のすべてのホストとサービスが表示されている場合は、必要な権限が付与されています。
- ホストまたはサービスが表示されない場合は、管理者に連絡して [すべてのホストとサービスを表示できる] 権限を付与してもらいます。
テスト 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 バケットを作成する
- Google Cloud コンソールに移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( nagios-xi-logsなど)を入力します。ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション 場所を選択します(例: us-central1)。ストレージ クラス Standard(アクセス頻度の高いログにおすすめ) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Cloud Run functions のサービス アカウントを作成する
- GCP Console で、[IAM と管理] > [サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成情報を指定します。
- サービス アカウント名: 「
nagios-logs-collector-sa」と入力します。 - サービス アカウントの説明:
Service account for Cloud Run function to collect Nagios XI logsと入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [ストレージ オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
GCS バケットに対する IAM 権限を付与する
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
nagios-xi-logs)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルの追加: サービス アカウントのメールアドレス(
nagios-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [ストレージ オブジェクト管理者] を選択します。
- プリンシパルの追加: サービス アカウントのメールアドレス(
- [保存] をクリックします。
Pub/Sub トピックの作成
- GCP Console で、[Pub/Sub> トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成情報を指定します。
- トピック ID: 「
nagios-logs-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Nagios XI REST API からログを取得して GCS に書き込みます。
- GCP Console で、[Cloud Run] に移動します。
- [サービスを作成] をクリックします。
- [関数] を選択します(インライン エディタを使用して関数を作成します)。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 nagios-logs-collectorリージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、
nagios-logs-triggerを選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント:
nagios-logs-collector-saを選択します。
- サービス アカウント:
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 説明 GCS_BUCKETnagios-xi-logsGCS バケット名 GCS_PREFIXnagios-xiログファイルの接頭辞 STATE_KEYnagios-xi/state.json状態ファイルのパス NAGIOS_BASE_URLhttps://your-nagios-server.example.comNagios XI のベース URL NAGIOS_API_KEYyour-api-keyNagios XI API キー MAX_RECORDS1000実行あたりのエンドポイントあたりの最大レコード数 PAGE_SIZE200API ページあたりのレコード数 LOOKBACK_HOURS24初期ルックバック期間 [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト:
600秒(10 分)を入力します。
- リクエストのタイムアウト:
[設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
- メモリ: 512 MiB 以上を選択します。
- CPU: 1 を選択します。
- [リソース] セクションで次の操作を行います。
[リビジョン スケーリング] セクションで、次の操作を行います。
- インスタンスの最小数:
0と入力します。 - [インスタンスの最大数] に「
100」と入力します。
- インスタンスの最小数:
[作成] をクリックします。
サービスが作成されるまで待ちます(1 ~ 2 分)。
サービスが作成されると、インライン コードエディタが自動的に開きます。
関数コードを追加する
- [エントリ ポイント] フィールドに「main」と入力します。
インライン コードエディタで、次の 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
[デプロイ] をクリックして関数を保存し、デプロイします。
デプロイが完了するまで待ちます(2 ~ 3 分)。
Cloud Scheduler ジョブの作成
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 nagios-logs-collector-hourlyリージョン Cloud Run functions と同じリージョンを選択する 周波数 0 * * * *(1 時間ごと、毎正時)タイムゾーン タイムゾーンを選択(UTC を推奨) ターゲット タイプ Pub/Sub トピック nagios-logs-triggerを選択メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
スケジュールの頻度オプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
| 頻度 | CRON 式 | ユースケース |
|---|---|---|
| 5 分おき | */5 * * * * |
大容量、低レイテンシ |
| 15 分ごと | */15 * * * * |
検索量が普通 |
| 1 時間ごと | 0 * * * * |
標準(推奨) |
| 6 時間ごと | 0 */6 * * * |
少量、バッチ処理 |
| 毎日 | 0 0 * * * |
過去のデータの収集 |
統合をテストする
- Cloud Scheduler コンソールで、ジョブ(
nagios-logs-collector-hourly)を見つけます。 - [強制実行] をクリックして、ジョブを手動でトリガーします。
- 数秒待ちます。
- Cloud Run > [サービス] に移動します。
nagios-logs-collectorをクリックします。- [ログ] タブをクリックします。
関数が正常に実行されたことを確認します。以下のものを探します。
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[Cloud Storage] > [バケット] に移動します。
nagios-xi-logsをクリックします。nagios-xi/フォルダに移動します。現在のタイムスタンプで新しい
.ndjsonファイルが作成されたことを確認します。
ログにエラーが表示された場合:
- HTTP 401:
NAGIOS_API_KEY環境変数が正しく、ユーザーの API アクセスが有効になっていることを確認します。 - HTTP 403: ユーザー アカウントにすべてのホストとサービスを表示できる権限があることを確認します
- HTTP 429: レート制限。関数はバックオフを使用して自動的に再試行します。
- 環境変数が不足している: 必要なすべての変数が Cloud Run functions の関数の構成に設定されていることを確認します。
Google SecOps サービス アカウントを取得する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
Nagios XI Logs GCS)。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [Nagios] を選択します。
- [サービス アカウントを取得する] をクリックします。
一意のサービス アカウントのメールアドレスが表示されます。次に例を示します。
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーして、次のステップで使用します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://nagios-xi-logs/nagios-xi/
Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストにおすすめ)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
転送されたファイルと空のディレクトリを削除する: 転送が成功した後にファイルと空のディレクトリを削除します。
ファイルの最大経過日数: 過去の日数以内に変更されたファイルを含めます(デフォルトは 180 日)。
アセットの名前空間: アセットの名前空間
Ingestion labels: このフィードのイベントに適用されるラベル
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
Google SecOps サービス アカウントに IAM 権限を付与する
- [Cloud Storage] > [バケット] に移動します。
nagios-xi-logsをクリックします。- [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
[保存] をクリックします。
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 のプロフェッショナルから回答を得ることができます。