Mengumpulkan log OpenTelemetry Netflow Receiver

Didukung di:

Google SecOps SIEM

Catatan: Fitur ini tercakup dalam Persyaratan Penawaran Pra-GA dari Persyaratan Khusus Layanan Google Security Operations. Fitur pra-GA mungkin memiliki dukungan terbatas, dan perubahan pada fitur pra-GA mungkin tidak kompatibel dengan versi pra-GA lainnya. Untuk mengetahui informasi selengkapnya, lihat panduan Layanan Dukungan Teknis Google SecOps dan Persyaratan Khusus Layanan Google SecOps.
Jika fitur ini tidak terlihat di lingkungan Anda, lihat Mengelola fitur pratinjau atau hubungi administrator sistem Anda.

Dokumen ini menjelaskan cara menyerap log OpenTelemetry Netflow Receiver ke Google Security Operations menggunakan Google Cloud Storage V2.

OpenTelemetry Netflow Receiver adalah komponen open source yang memantau traffic UDP netflow, sflow, dan IPFIX, lalu mengonversinya menjadi rekaman log OpenTelemetry. Hal ini memungkinkan pemantauan dan analisis traffic jaringan, termasuk identifikasi protokol, analisis volume traffic, pelacakan penggunaan port, dan statistik byte/paket.

Sebelum memulai

Pastikan Anda memiliki prasyarat berikut:

• Instance Google SecOps
• Project GCP dengan Cloud Storage API 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
• Perangkat jaringan yang dapat mengirim data netflow, sflow, atau IPFIX
• Akses untuk mengonfigurasi setelan ekspor perangkat jaringan

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, netflow-logs-bucket)
   Location type	Pilih berdasarkan kebutuhan Anda (Region, Dual-region, Multi-region)
   Location	Pilih lokasi (misalnya, us-central1)
   Kelas penyimpanan	Standar (direkomendasikan untuk log yang sering diakses)
   Kontrol akses	Seragam (direkomendasikan)
   Alat perlindungan	Opsional: Aktifkan pembuatan versi objek atau kebijakan retensi

6. Klik Create.

Buat akun layanan untuk fungsi Cloud Run

Fungsi Cloud Run memerlukan akun layanan dengan izin untuk menulis ke bucket GCS dan dipanggil oleh Pub/Sub.

Membuat akun layanan

1. Di Konsol GCP, buka IAM & Admin > Service Accounts.
2. Klik Create Service Account.
3. Berikan detail konfigurasi berikut:
   • Nama akun layanan: Masukkan netflow-collector-sa
   • Deskripsi akun layanan: Masukkan Service account for Cloud Run function to collect netflow 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.

Peran ini diperlukan untuk:

• Storage Object Admin: Menulis log ke bucket GCS dan mengelola file status
• Cloud Run Invoker: Izinkan Pub/Sub memanggil fungsi
• Cloud Functions Invoker: Mengizinkan pemanggilan fungsi

Memberikan izin IAM pada bucket GCS

Beri akun layanan izin tulis di bucket GCS:

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

Membuat topik Pub/Sub

Buat topik Pub/Sub yang akan dipublikasikan oleh Cloud Scheduler dan akan dilanggan oleh fungsi Cloud Run.

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

Membuat fungsi Cloud Run untuk mengumpulkan log netflow

Cloud Run Function akan menjalankan OpenTelemetry Collector yang menerima data netflow dan mengekspornya 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 Konfigurasi, berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama layanan	netflow-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 topik Pub/Sub netflow-trigger.
   4. Klik Simpan.

6. Di bagian Authentication:

   1. Pilih Wajibkan autentikasi.
   2. Periksa Identity and Access Management (IAM).
   Catatan: Pub/Sub akan otomatis menangani autentikasi saat memanggil fungsi.

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

8. Buka tab Security:

   • Akun layanan: Pilih akun layanan netflow-collector-sa

9. Buka tab Containers:

   1. Klik Variables & Secrets.
   2. Klik + Tambahkan variabel untuk setiap variabel lingkungan:

   Nama Variabel	Nilai Contoh	Deskripsi
   GCS_BUCKET	netflow-logs-bucket	Nama bucket GCS
   GCS_PREFIX	netflow	Awalan untuk file log
   NETFLOW_PORT	2055	Port untuk penerima netflow
   NETFLOW_SCHEME	netflow	Jenis skema: netflow, sflow, atau ipfix
   NETFLOW_SOCKETS	4	Jumlah soket UDP
   NETFLOW_WORKERS	8	Jumlah pekerja decoder

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

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

11. Buka tab Setelan:

    • Di bagian Referensi:
      • Memori: Pilih 1 GiB atau lebih tinggi
      • CPU: Pilih 2

12. Di bagian Penskalaan revisi:

    • Minimum number of instances: Masukkan 1
    • Maximum number of instances: Masukkan 10

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 tiga file:

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

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

Catatan: Fungsi ini akan menjalankan OpenTelemetry Collector yang memproses data netflow dan mengekspornya ke GCS.

Mengonfigurasi perangkat jaringan untuk mengirim data netflow

Konfigurasi perangkat jaringan Anda (router, switch, firewall) untuk mengekspor data netflow, sflow, atau IPFIX ke fungsi Cloud Run.

Mendapatkan IP eksternal fungsi Cloud Run

1. Di Konsol GCP, buka Cloud Run > Services.
2. Klik nama fungsi netflow-collector.
3. Salin URL yang ditampilkan di bagian atas (misalnya, https://netflow-collector-xxxxx-uc.a.run.app).
4. Ekstrak nama host dari URL.

5. Gunakan nslookup atau dig untuk menyelesaikan alamat IP:

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

Catatan: Layanan Cloud Run menggunakan IP dinamis. Untuk deployment produksi, pertimbangkan untuk menggunakan Cloud Load Balancer dengan IP statis atau men-deploy pengumpul di Compute Engine.

• Contoh: Konfigurasi netflow router Cisco

    ! 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
  

• Contoh: Konfigurasi netflow generik

  Untuk sebagian besar perangkat jaringan, konfigurasi setelan berikut:

  • Netflow version: NetFlow v5, v9, atau IPFIX
  • IP Pengumpul: Alamat IP fungsi Cloud Run
  • Port pengumpul: 2055 (atau port yang dikonfigurasi)
  • Protokol: UDP
  • Waktu tunggu aktif: 60 detik (direkomendasikan)
  • Waktu tunggu tidak aktif: 15 detik (direkomendasikan)

Buat tugas Cloud Scheduler

Cloud Scheduler akan memublikasikan pesan ke topik Pub/Sub secara berkala, sehingga memicu fungsi Cloud Run.

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

3. Berikan detail konfigurasi berikut:

   Setelan	Nilai
   Nama	netflow-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 topik Pub/Sub netflow-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

Menguji integrasi

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

7. Pastikan fungsi berhasil dieksekusi. Cari:

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

9. Klik nama bucket netflow-logs-bucket.

10. Buka folder awalan netflow/.

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

Jika Anda melihat error dalam log:

• Collector keluar secara tiba-tiba: Periksa konfigurasi penerima netflow
• Tidak ada data yang akan diupload: Verifikasi bahwa perangkat jaringan mengirimkan data netflow ke IP dan port yang benar
• Akses ditolak: Periksa apakah akun layanan memiliki peran Storage Object Admin
• Error pengikatan port: Pastikan port 2055 belum digunakan

Mengambil akun layanan Google SecOps

Google SecOps menggunakan akun layanan unik untuk membaca data dari bucket GCS Anda. Anda harus memberi akun layanan ini akses ke bucket Anda.

Dapatkan email akun layanan

1. Buka Setelan SIEM > Feed.
2. Klik Tambahkan Feed Baru.
3. Klik Konfigurasi satu feed.
4. Di kolom Nama feed, masukkan nama untuk feed (misalnya, OpenTelemetry Netflow Logs).
5. Pilih Google Cloud Storage V2 sebagai Source type.
6. Pilih NETFLOW_OTEL sebagai Jenis log.

7. Klik Get Service Account. Email akun layanan yang unik akan ditampilkan. Contoh:

   [e-mail]
   

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

   Catatan: Setiap instance Google SecOps memiliki akun layanan yang unik. Jangan gunakan akun layanan dari dokumentasi atau contoh lain.

9. Klik Berikutnya.

10. Tentukan nilai untuk parameter input berikut:

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

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

    Catatan: Selalu sertakan garis miring (/) di akhir URI.

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

      • Jangan pernah: 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.

        Catatan: Jika Anda memilih opsi penghapusan, akun layanan harus memiliki peran Storage Object Admin, bukan Storage Object Viewer. Perbarui izin IAM yang sesuai.

    • 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

11. Klik Berikutnya.

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

Memberikan izin IAM ke akun layanan Google SecOps

Akun layanan Google SecOps memerlukan peran Storage Object Viewer di bucket GCS Anda.

1. Buka Cloud Storage > Buckets.
2. Klik nama bucket netflow-logs-bucket.
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.

Catatan: Jika Anda berencana menggunakan opsi penghapusan "Hapus file yang ditransfer" atau "Hapus file yang ditransfer dan direktori kosong", berikan peran Storage Object Admin, bukan Storage Object Viewer.

Tabel pemetaan UDM

Tabel berikut menunjukkan cara kolom log OpenTelemetry Netflow Receiver dipetakan ke kolom UDM Google SecOps:

Kolom OpenTelemetry	Kolom UDM	Deskripsi
source.address	principal.ip	Alamat IP sumber
source.port	principal.port	Nomor port sumber
destination.address	target.ip	Alamat IP tujuan
destination.port	target.port	Nomor port tujuan
network.transport	network.ip_protocol	Protokol transport (tcp, udp)
network.type	network.ip_version	Versi IP (ipv4, ipv6)
flow.io.bytes	network.sent_bytes	Total byte yang ditransfer
flow.io.packets	network.sent_packets	Total paket yang ditransfer
flow.type	metadata.product_log_id	Jenis alur (NETFLOW_V5, NETFLOW_V9, SFLOW_5, IPFIX)
flow.start	metadata.event_timestamp	Waktu mulai Flow
flow.end	network.session_duration	Waktu berakhir Flow
flow.sampler_address	intermediary.ip	Alamat IP eksportir Netflow
flow.tcp_flags	network.tcp_flags	Flag TCP
flow.io.bytes	event.idm.read_only_udm.network.sent_bytes	Dipetakan dari log perubahan
source.port	event.idm.read_only_udm.principal.port	Dipetakan dari log perubahan
destination.port	event.idm.read_only_udm.target.port	Dipetakan dari log perubahan
network.transport	event.idm.read_only_udm.network.ip_protocol	Dipetakan dari log perubahan
network.type	event.idm.read_only_udm.security_result.detection_fields	Dipetakan dari log perubahan
flow.sequence_num	event.idm.read_only_udm.network.session_id	Dipetakan dari log perubahan
flow.sampler_address	event.idm.read_only_udm.observer.ip	Dipetakan dari log perubahan
flow.type	event.idm.read_only_udm.metadata.product_event_type	Dipetakan dari log perubahan

Log Perubahan

Melihat Log Perubahan untuk parser ini

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