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을 입력합니다. - 이메일 주소: 유효한 이메일 주소를 입력합니다.
- 사용자 이름: 설명이 포함된 이름을 입력합니다 (예:
- 보안 설정 섹션에서 다음을 구성합니다.
- 승인 수준: 사용자를 선택합니다.
- 모든 호스트 및 서비스 보기 가능: 이 옵션을 선택합니다.
- 읽기 전용 액세스: 이 옵션을 선택합니다.
- API 액세스: 이 옵션을 선택합니다.
- 다음 옵션을 선택 해제합니다.
- 다음 로그인 시 비밀번호 변경 강제
- 사용자 계정 정보 이메일 보내기
- 모니터링 연락처로 만들기
- 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 함수의 서비스 계정 만들기
- GCP 콘솔에서 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 콘솔에서 Cloud Run으로 이동합니다.
- 서비스 만들기를 클릭합니다.
- 함수 (인라인 편집기를 사용하여 함수 만들기)를 선택합니다.
구성 섹션에서 다음 구성 세부정보를 입력합니다.
설정 값 서비스 이름 nagios-logs-collector리전 GCS 버킷과 일치하는 리전을 선택합니다 (예: us-central1).런타임 Python 3.12 이상을 선택합니다. 트리거 (선택사항) 섹션에서 다음을 수행합니다.
- + 트리거 추가를 클릭합니다.
- Cloud Pub/Sub을 선택합니다.
- Cloud Pub/Sub 주제 선택에서
nagios-logs-trigger를 선택합니다. - 저장을 클릭합니다.
인증 섹션에서 다음을 구성합니다.
- 인증 필요를 선택합니다.
- ID 및 액세스 관리 (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분)를 입력합니다.
- 요청 제한 시간:
설정 탭으로 이동합니다.
- 리소스 섹션에서 다음을 수행합니다.
- 메모리: 512MiB 이상 선택
- CPU: 1을 선택합니다.
- 리소스 섹션에서 다음을 수행합니다.
버전 확장 섹션에서 다음을 수행합니다.
- 최소 인스턴스 수:
0를 입력합니다. - 최대 인스턴스 수:
100를 입력합니다.
- 최소 인스턴스 수:
만들기를 클릭합니다.
서비스가 생성될 때까지 기다립니다 (1~2분).
서비스가 생성되면 인라인 코드 편집기가 자동으로 열립니다.
함수 코드 추가
- 진입점 필드에 main을 입력합니다.
인라인 코드 편집기에서 다음 두 파일을 만듭니다.
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 함수와 동일한 리전 선택 주파수 0 * * * *(매시간 정각)시간대 시간대 선택 (UTC 권장) 타겟 유형 Pub/Sub 주제 nagios-logs-trigger선택메일 본문 {}(빈 JSON 객체)만들기를 클릭합니다.
일정 빈도 옵션
로그 볼륨 및 지연 시간 요구사항에 따라 빈도를 선택합니다.
| 빈도 | 크론 표현식 | 사용 사례 |
|---|---|---|
| 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 recordsCloud Storage> 버킷으로 이동합니다.
nagios-xi-logs를 클릭합니다.nagios-xi/폴더로 이동합니다.현재 타임스탬프로 새
.ndjson파일이 생성되었는지 확인합니다.
로그에 오류가 표시되는 경우:
- HTTP 401:
NAGIOS_API_KEY환경 변수가 올바르고 사용자에게 API 액세스가 사용 설정되어 있는지 확인합니다. - HTTP 403: 사용자 계정에 모든 호스트 및 서비스 보기 권한이 있는지 확인합니다.
- HTTP 429: 비율 제한 - 함수가 백오프로 자동 재시도됩니다.
- 환경 변수 누락: Cloud Run 함수 구성에 필요한 모든 변수가 설정되어 있는지 확인합니다.
Google SecOps 서비스 계정 가져오기
- SIEM 설정 > 피드로 이동합니다.
- 새 피드 추가를 클릭합니다.
- 단일 피드 구성을 클릭합니다.
- 피드 이름 필드에 피드 이름을 입력합니다(예:
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/
소스 삭제 옵션: 환경설정에 따라 삭제 옵션을 선택합니다.
- 삭제 안함: 전송 후 파일을 삭제하지 않습니다 (테스트에 권장).
- 전송된 파일 삭제: 전송이 완료되면 파일을 삭제합니다.
전송된 파일 및 빈 디렉터리 삭제: 전송이 완료되면 파일과 빈 디렉터리를 삭제합니다.
최대 파일 기간: 지난 일수 동안 수정된 파일을 포함합니다 (기본값은 180일).
애셋 네임스페이스: 애셋 네임스페이스
수집 라벨: 이 피드의 이벤트에 적용할 라벨
다음을 클릭합니다.
확정 화면에서 새 피드 구성을 검토한 다음 제출을 클릭합니다.
Google SecOps 서비스 계정에 IAM 권한 부여
- Cloud Storage> 버킷으로 이동합니다.
nagios-xi-logs를 클릭합니다.- 권한 탭으로 이동합니다.
- 액세스 권한 부여를 클릭합니다.
- 다음 구성 세부정보를 제공합니다.
- 주 구성원 추가: Google SecOps 서비스 계정 이메일을 붙여넣습니다.
- 역할 할당: 스토리지 객체 뷰어를 선택합니다.
저장을 클릭합니다.
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 | 값이 직접 복사되고 정수로 변환됨 |
| 1 열 | principal.asset.hostname | ent == 'SERVICE ALERT'인 경우 column1의 값, ent == 'SERVICE NOTIFICATION'인 경우 column2의 값 |
| 1 열 | principal.hostname | ent == 'SERVICE ALERT'인 경우 column1의 값, ent == 'SERVICE NOTIFICATION'인 경우 column2의 값 |
| 1 열 | 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에서 병합됨 |
| 메타데이터 | event.idm.read_only_udm.metadata | 메타데이터에서 이름이 변경됨 |
| 대상 | event.idm.read_only_udm.target | 타겟에서 이름이 변경됨 |
| 주 구성원 | event.idm.read_only_udm.principal | 주 구성원에서 이름이 변경됨 |
| 네트워크 | event.idm.read_only_udm.network | 네트워크에서 이름 변경됨 |
| metadata.product_name | metadata.product_name | 'NAGIOS'로 설정 |
| metadata.vendor_name | metadata.vendor_name | 'NAGIOS'로 설정 |
변경 로그
도움이 더 필요하신가요? 커뮤니티 회원 및 Google SecOps 전문가에게 문의하여 답변을 받으세요.