Mengumpulkan log Nagios XI

Didukung di:

Dokumen ini menjelaskan cara menyerap log Nagios XI ke Google Security Operations menggunakan Google Cloud Storage V2.

Nagios XI adalah solusi pemantauan infrastruktur komprehensif yang menyediakan pemantauan server, jaringan, aplikasi, dan layanan. Cloud Monitoring melacak status host dan layanan, metrik performa, serta membuat pemberitahuan untuk masalah infrastruktur IT melalui REST API-nya.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

  • Instance Google SecOps
  • Project GCP dengan Cloud Storage, Cloud Run, Pub/Sub, dan Cloud Scheduler API yang diaktifkan
  • Izin untuk membuat dan mengelola bucket GCS
  • Izin untuk mengelola kebijakan IAM di bucket GCS
  • Izin untuk membuat layanan Cloud Run, topik Pub/Sub, dan tugas Cloud Scheduler
  • Akses istimewa ke Nagios XI dengan izin pengelolaan pengguna
  • Nagios XI versi 5 atau yang lebih baru (dukungan REST API)

Mengonfigurasi akses API Nagios XI

Agar Google SecOps dapat mengambil data pemantauan, Anda harus membuat akun pengguna dengan akses API dan izin hanya baca.

Membuat pengguna hanya baca dengan akses API

  1. Login ke antarmuka web Nagios XI dengan hak istimewa administrator.
  2. Buka Admin > Kelola Pengguna.
  3. Klik Tambahkan Pengguna Baru.
  4. Berikan detail konfigurasi berikut:
    • Nama pengguna: Masukkan nama deskriptif (misalnya, chronicle-integration)
    • Sandi: Masukkan sandi yang aman
    • Nama: Masukkan Chronicle Integration User
    • Alamat Email: Masukkan alamat email yang valid
  5. Di bagian Security Settings, konfigurasikan hal berikut:
    • Tingkat Otorisasi: Pilih Pengguna
    • Dapat melihat semua host dan layanan: Centang opsi ini
    • Akses hanya baca: Centang opsi ini
    • Akses API: Centang opsi ini
  6. Hapus centang pada opsi berikut:
    • Memaksa Perubahan Sandi saat Login Berikutnya
    • Mengirim Informasi Akun Pengguna melalui Email
    • Buat sebagai Kontak Pemantauan
  7. Pastikan Account Enabled dicentang.
  8. Klik Add user.

Mengambil kunci API

  1. Di halaman Manage Users, klik akun pengguna (chronicle-integration).
  2. Di halaman setelan akun pengguna, temukan kolom Kunci API.

  3. Salin nilai kunci API.

Memverifikasi izin

Untuk memverifikasi bahwa akun memiliki izin yang diperlukan:

  1. Login ke Nagios XI dengan akun pengguna chronicle-integration.
  2. Buka Beranda > Status Host.
  3. Jika Anda dapat melihat semua host dan layanan yang dipantau, berarti Anda memiliki izin yang diperlukan.
  4. Jika Anda tidak dapat melihat host atau layanan, hubungi administrator Anda untuk memberikan izin Dapat melihat semua host dan layanan.

Menguji akses API

  • Uji kredensial Anda sebelum melanjutkan integrasi:

    # 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"

    Respons yang berhasil akan menampilkan objek JSON yang berisi informasi status host.

Buat bucket Google Cloud Storage

  1. Buka Konsol Google Cloud.
  2. Pilih project Anda atau buat project baru.
  3. Di menu navigasi, buka Cloud Storage > Buckets.
  4. Klik Create bucket.

  5. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Beri nama bucket Anda Masukkan nama yang unik secara global (misalnya, nagios-xi-logs)
    Location type Pilih berdasarkan kebutuhan Anda (Region, Dual-region, Multi-region)
    Location Pilih lokasi (misalnya, us-central1)
    Kelas penyimpanan Standard (direkomendasikan untuk log yang sering diakses)
    Kontrol akses Seragam (direkomendasikan)
    Alat perlindungan Opsional: Aktifkan pembuatan versi objek atau kebijakan retensi

  6. Klik Create.

Membuat akun layanan untuk fungsi Cloud Run

  1. Di Konsol GCP, buka IAM & Admin > Service Accounts.
  2. Klik Create Service Account.
  3. Berikan detail konfigurasi berikut:
    • Nama akun layanan: Masukkan nagios-logs-collector-sa
    • Service account description: Masukkan Service account for Cloud Run function to collect Nagios XI logs
  4. Klik Create and Continue.
  5. Di bagian Grant this service account access to project, tambahkan peran berikut:
    1. Klik Pilih peran.
    2. Telusuri dan pilih Storage Object Admin.
    3. Klik + Add another role.
    4. Telusuri dan pilih Cloud Run Invoker.
    5. Klik + Add another role.
    6. Telusuri dan pilih Cloud Functions Invoker.
  6. Klik Lanjutkan.
  7. Klik Done.

Memberikan izin IAM pada bucket GCS

  1. Buka Cloud Storage > Buckets.
  2. Klik nama bucket Anda (nagios-xi-logs).
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Tambahkan prinsipal: Masukkan email akun layanan (nagios-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
    • Assign roles: Pilih Storage Object Admin
  6. Klik Simpan.

Membuat topik Pub/Sub

  1. Di Konsol GCP, buka Pub/Sub > Topics.
  2. Klik Create topic.
  3. Berikan detail konfigurasi berikut:
    • ID Topik: Masukkan nagios-logs-trigger
    • Biarkan setelan lainnya sesuai default
  4. Klik Create.

Membuat fungsi Cloud Run untuk mengumpulkan log

Fungsi Cloud Run akan dipicu oleh pesan Pub/Sub dari Cloud Scheduler untuk mengambil log dari REST API Nagios XI dan menulisnya ke GCS.

  1. Di Konsol GCP, buka Cloud Run.
  2. Klik Create service.
  3. Pilih Function (gunakan editor inline untuk membuat fungsi).

  4. Di bagian Configure, berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama layanan nagios-logs-collector
    Region Pilih region yang cocok dengan bucket GCS Anda (misalnya, us-central1)
    Runtime Pilih Python 3.12 atau yang lebih baru

  5. Di bagian Pemicu (opsional):

    1. Klik + Tambahkan pemicu.
    2. Pilih Cloud Pub/Sub.
    3. Di Select a Cloud Pub/Sub topic, pilih nagios-logs-trigger.
    4. Klik Simpan.

  6. Di bagian Authentication:

    1. Pilih Wajibkan autentikasi.
    2. Periksa Identity and Access Management (IAM).

  7. Scroll ke bawah dan luaskan Containers, Networking, Security.

  8. Buka tab Security:

    • Service account: Pilih nagios-logs-collector-sa

  9. Buka tab Containers:

    1. Klik Variables & Secrets.
    2. Klik + Add variable untuk setiap variabel lingkungan:
    Nama Variabel Nilai Contoh Deskripsi
    GCS_BUCKET nagios-xi-logs Nama bucket GCS
    GCS_PREFIX nagios-xi Awalan untuk file log
    STATE_KEY nagios-xi/state.json Jalur file status
    NAGIOS_BASE_URL https://your-nagios-server.example.com URL dasar Nagios XI
    NAGIOS_API_KEY your-api-key Kunci API Nagios XI
    MAX_RECORDS 1000 Jumlah maksimum data per endpoint per eksekusi
    PAGE_SIZE 200 Halaman Rekaman per API
    LOOKBACK_HOURS 24 Periode lihat balik awal

  10. Di bagian Variables & Secrets, scroll ke bawah ke Requests:

    • Waktu tunggu permintaan: Masukkan 600 detik (10 menit)

  11. Buka tab Setelan:

    • Di bagian Resources:
      • Memory: Pilih 512 MiB atau lebih tinggi
      • CPU: Pilih 1

  12. Di bagian Revision scaling:

    • Minimum number of instances: Masukkan 0
    • Maximum number of instances: Masukkan 100

  13. Klik Create.

  14. Tunggu hingga layanan dibuat (1-2 menit).

  15. Setelah layanan dibuat, editor kode inline akan terbuka secara otomatis.

Menambahkan kode fungsi

  1. Masukkan main di kolom Entry point.

  2. Di editor kode inline, buat dua file:

    • 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. Klik Deploy untuk menyimpan dan men-deploy fungsi.

  4. Tunggu hingga deployment selesai (2-3 menit).

Buat tugas Cloud Scheduler

  1. Di Konsol GCP, buka Cloud Scheduler.
  2. Klik Create Job.

  3. Berikan detail konfigurasi berikut:

    Setelan Nilai
    Nama nagios-logs-collector-hourly
    Region Pilih region yang sama dengan fungsi Cloud Run
    Frekuensi 0 * * * * (setiap jam, tepat pada waktunya)
    Zona Waktu Pilih zona waktu (UTC direkomendasikan)
    Jenis target Pub/Sub
    Topik Pilih nagios-logs-trigger
    Isi pesan {} (objek JSON kosong)

  4. Klik Create.

Opsi frekuensi jadwal

Pilih frekuensi berdasarkan volume log dan persyaratan latensi:

Frekuensi Ekspresi Cron Kasus Penggunaan
Setiap 5 menit */5 * * * * Volume tinggi, latensi rendah
Setiap 15 menit */15 * * * * Volume sedang
Setiap jam 0 * * * * Standar (direkomendasikan)
Setiap 6 jam 0 */6 * * * Volume rendah, pemrosesan batch
Harian 0 0 * * * Pengumpulan data historis

Menguji integrasi

  1. Di konsol Cloud Scheduler, temukan tugas Anda (nagios-logs-collector-hourly).
  2. Klik Force run untuk memicu tugas secara manual.
  3. Tunggu beberapa detik.
  4. Buka Cloud Run > Services.
  5. Klik nagios-logs-collector.
  6. Klik tab Logs.

  7. Pastikan fungsi berhasil dieksekusi. Cari:

    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. Buka Cloud Storage > Buckets.

  9. Klik nagios-xi-logs.

  10. Buka folder nagios-xi/.

  11. Pastikan file .ndjson baru telah dibuat dengan stempel waktu saat ini.

Jika Anda melihat error dalam log:

  • HTTP 401: Pastikan variabel lingkungan NAGIOS_API_KEY sudah benar dan pengguna telah mengaktifkan akses API
  • HTTP 403: Pastikan akun pengguna memiliki izin Dapat melihat semua host dan layanan
  • HTTP 429: Pembatasan kecepatan -- fungsi akan otomatis mencoba lagi dengan penundaan
  • Variabel lingkungan tidak ada: Pastikan semua variabel yang diperlukan ditetapkan dalam konfigurasi fungsi Cloud Run

Mengambil akun layanan Google SecOps

  1. Buka Setelan SIEM > Feed.
  2. Klik Tambahkan Feed Berita Baru.
  3. Klik Konfigurasi satu feed.
  4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, Nagios XI Logs GCS).
  5. Pilih Google Cloud Storage V2 sebagai Source type.
  6. Pilih Nagios sebagai Jenis log.
  7. Klik Get Service Account.

  8. Email akun layanan yang unik akan ditampilkan. Contoh:

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

  9. Salin alamat email ini untuk digunakan di langkah berikutnya.

  10. Klik Berikutnya.

  11. Tentukan nilai untuk parameter input berikut:

    • URL bucket penyimpanan: Masukkan URI bucket GCS dengan jalur awalan:

      gs://nagios-xi-logs/nagios-xi/

    • Opsi penghapusan sumber: Pilih opsi penghapusan sesuai preferensi Anda:

      • Never: Tidak pernah menghapus file apa pun setelah transfer (direkomendasikan untuk pengujian).
      • Hapus file yang ditransfer: Menghapus file setelah transfer berhasil.

      • Hapus file yang ditransfer dan direktori kosong: Menghapus file dan direktori kosong setelah transfer berhasil.

    • Usia File Maksimum: Menyertakan file yang diubah dalam beberapa hari terakhir (default adalah 180 hari)

    • Namespace aset: Namespace aset

    • Label penyerapan: Label yang akan diterapkan ke peristiwa dari feed ini

  12. Klik Berikutnya.

  13. Tinjau konfigurasi feed baru Anda di layar Selesaikan, lalu klik Kirim.

Memberikan izin IAM ke akun layanan Google SecOps

  1. Buka Cloud Storage > Buckets.
  2. Klik nagios-xi-logs.
  3. Buka tab Izin.
  4. Klik Grant access.
  5. Berikan detail konfigurasi berikut:
    • Add principals: Tempel email akun layanan Google SecOps
    • Tetapkan peran: Pilih Storage Object Viewer

  6. Klik Simpan.

Tabel pemetaan UDM

Kolom Log Pemetaan UDM Logika
src_ip has_principal Ditetapkan ke "true" jika penggabungan src_ip ke principal.ip dan principal.asset.ip berhasil
src_ip principal.asset.ip Nilai disalin secara langsung
src_ip principal.ip Nilai disalin langsung
jobid jobid_label.key Disetel ke "Jobid"
jobid jobid_label.value Nilai disalin secara langsung
jobid_label principal.resource.attribute.labels Digabungkan dari jobid_label
pid principal.process.pid Nilai disalin langsung
ent metadata.product_event_type Nilai disalin secara langsung
deskripsi metadata.description Nilai disalin secara langsung
msg_ip has_target Ditetapkan ke "true" jika penggabungan msg_ip ke target.ip dan target.asset.ip berhasil
msg_ip target.asset.ip Nilai disalin langsung
msg_ip target.ip Nilai disalin secara langsung
port target.port Nilai disalin langsung dan dikonversi menjadi bilangan bulat
kolom1 principal.asset.hostname Nilai dari column1 jika ent == "SERVICE ALERT", atau dari column2 jika ent == "SERVICE NOTIFICATION"
kolom1 principal.hostname Nilai dari column1 jika ent == "SERVICE ALERT", atau dari column2 jika ent == "SERVICE NOTIFICATION"
kolom1 principal.user.user_display_name Nilai disalin secara langsung jika ent == "SERVICE NOTIFICATION"
kolom2 security_result.summary Nilai dari column3 jika ent == "SERVICE ALERT" dan column2 tidak kosong, atau dari column3 jika ent == "SERVICE NOTIFICATION"
kolom3 security_result.summary Nilai dari column3 jika ent == "SERVICE ALERT" dan column3 tidak kosong, atau dari column3 jika ent == "SERVICE NOTIFICATION"
column4 security_result.severity Nilai disalin secara langsung jika column4 dalam ["LOW", "MEDIUM", "HIGH", "CRITICAL"] dan ent dalam ["SERVICE NOTIFICATION", "SERVICE ALERT"]
column4 security_result.severity_details Nilai disalin langsung jika column4 tidak ada di ["LOW", "MEDIUM", "HIGH", "CRITICAL"] dan ent ada di ["SERVICE NOTIFICATION", "SERVICE ALERT"]
has_principal metadata.event_type Ditetapkan ke "NETWORK_CONNECTION" jika has_principal dan has_target benar, atau "STATUS_UPDATE" jika has_principal benar, atau "USER_UNCATEGORIZED" jika has_principal_user benar, atau "GENERIC_EVENT"
has_target metadata.event_type
has_principal_user metadata.event_type
security_result event.idm.read_only_udm.security_result Digabungkan dari security_result
metadata event.idm.read_only_udm.metadata Diganti namanya dari metadata
target event.idm.read_only_udm.target Diganti namanya dari target
utama event.idm.read_only_udm.principal Diganti namanya dari principal
jaringan event.idm.read_only_udm.network Diganti namanya dari jaringan
metadata.product_name metadata.product_name Tetapkan ke "NAGIOS"
metadata.vendor_name metadata.vendor_name Tetapkan ke "NAGIOS"

Log Perubahan

Melihat Log Perubahan untuk parser ini

Perlu bantuan lain? Dapatkan jawaban dari anggota Komunitas dan profesional Google SecOps.