[ELK] Cấu hình gửi Logst từ Cisco sang Elasticsearch

1. Cấu hình trên Switch.

Để cấu hình switch Cisco 3750 để gửi logs đến Elasticsearch (ELK) với địa chỉ IP là 192.168.100.111 và cổng 9200, bạn có thể làm theo các bước sau:

Đảm bảo rằng switch Cisco 3750 có kết nối mạng đến ELK server và địa chỉ IP của nó là đúng.

Đăng nhập vào switch bằng cách sử dụng giao diện dòng lệnh (CLI) thông qua một công cụ như PuTTY hoặc sử dụng terminal trực tiếp nếu có. Cài đặt thời gian cho chính xác, ví dụ:

configure terminal
clock timezone UTC 7 0
end
clock set 20:18:00 14 Jun 2023

Cấu hình giao thức Syslog để gửi log tới cổng UDP 5044 bằng lệnh sau:

configure terminal
service timestamps log datetime msec show-timezone
service timestamps debug datetime msec show-timezone
logging source-interface vlan 100
logging host 192.168.100.101 transport udp port 5044
logging trap 6

Lưu ý: Thay thế <tên giao diện mạng> bằng tên giao diện mạng trên switch Cisco 3750 mà bạn muốn sử dụng để gửi logs (ví dụ: VLAN interface, Loopback interface, etc.).

Kiểm tra lại cấu hình và lưu lại:

show logging
write

Nội dung logging như sau:

$ show logging 
Syslog logging: enabled (0 messages dropped, 2 messages rate-limited, 0 flushes, 0 overruns, xml disabled, filtering disabled)

No Active Message Discriminator.



No Inactive Message Discriminator.


    Console logging: disabled
    Monitor logging: level debugging, 0 messages logged, xml disabled,
                     filtering disabled
    Buffer logging:  level debugging, 1181 messages logged, xml disabled,
                    filtering disabled
    Exception Logging: size (4096 bytes)
    Count and timestamp logging messages: disabled
    File logging: disabled
    Persistent logging: disabled

No active filter modules.

    Trap logging: level informational, 1183 message lines logged
        Logging to 192.168.100.111  (udp port 5044, audit disabled,
              link up),
              42 message lines logged, 
              0 message lines rate-limited, 
              0 message lines dropped-by-MD, 
              xml disabled, sequence number disabled
              filtering disabled
        Logging Source-Interface:       VRF Name:
        Vlan100

Trong đoạn output trên là một cấu hình logging của thiết bị Cisco. Dưới đây là một giải thích cho các phần quan trọng của cấu hình đó:

Syslog logging: đã được kích hoạt và hiện không có bất kỳ Message Discriminator nào được áp dụng.
Console logging: đã tắt (disabled).
Monitor logging: cấp độ (level) là “debugging”, không có thông điệp nào được ghi lại (0 messages logged), và không áp dụng bộ lọc (filtering).
Buffer logging: cấp độ (level) là “debugging”, có 1181 thông điệp được ghi lại (1181 messages logged), không áp dụng bộ lọc (filtering).
Exception Logging: kích thước (size) của bộ nhớ được sử dụng để ghi lại thông điệp lỗi là 4096 byte.
Count and timestamp logging messages: đã tắt (disabled).
File logging: đã tắt (disabled).
Persistent logging: đã tắt (disabled).
No active filter modules: không có mô-đun bộ lọc hoạt động.
Trap logging: cấp độ (level) là “informational”, có 1183 dòng thông điệp được ghi lại (1183 message lines logged), gửi đến địa chỉ IP 192.168.100.111 trên cổng UDP 5044, không sử dụng kiểm tra log (audit), đường link đang hoạt động (link up), không có thông điệp bị hạn chế tốc độ (rate-limited), không có thông điệp bị xóa bởi Message Discriminator (dropped-by-MD), không sử dụng định dạng XML, không sử dụng số thứ tự cho thông điệp (sequence number), và không áp dụng bộ lọc (filtering).
Logging Source-Interface: không có thông tin về giao diện nguồn (source interface).
VRF Name: không có thông tin về tên VRF.

Như vậy, cấu hình logging hiện tại của bạn cho phép ghi lại các thông điệp syslog với cấp độ (level) cụ thể và chuyển đến một máy chủ syslog có địa chỉ IP 192.168.100.111 trên cổng UDP 5044.

Sau khi hoàn tất các bước trên, switch Cisco 3750 sẽ gửi logs đến ELK server theo địa chỉ IP và cổng được cấu hình (192.168.100.111:9200). Hãy đảm bảo rằng ELK server của bạn đã được cấu hình đúng để nhận và xử lý logs từ switch.

2. Cấu hình trên Server Logs.

2.1. Lựa chọn nơi để bạn chuyển logs tới.

Gửi log qua Logstash và Elasticsearch có một số khác biệt quan trọng:

Logstash:
- Logstash là một công cụ xử lý và chuyển đổi log được phát triển bởi Elastic. Nó có thể nhận log từ nhiều nguồn khác nhau, xử lý chúng và chuyển tiếp đến các hệ thống khác như Elasticsearch để lưu trữ và tìm kiếm. Logstash cung cấp nhiều plugin đa dạng để xử lý log, chẳng hạn như parsing, phân tích, tiêu chuẩn hóa dữ liệu, và enriching.
- Logstash thường được sử dụng khi bạn cần xử lý log trước khi lưu trữ hoặc thực hiện các phân tích phức tạp trên log. Bạn có thể sử dụng các plugin của Logstash để chuẩn hóa, phân tích và biến đổi log trước khi gửi đến Elasticsearch.
Elasticsearch:
- Elasticsearch là một hệ thống lưu trữ và tìm kiếm dựa trên Apache Lucene, được phát triển bởi Elastic. Nó được thiết kế để lưu trữ và tìm kiếm dữ liệu phân tán nhanh chóng và hiệu quả. Elasticsearch cho phép bạn lưu trữ, truy vấn, và phân tích log một cách mạnh mẽ. Bạn có thể gửi log trực tiếp đến Elasticsearch và sử dụng nó làm hệ thống lưu trữ chính cho log của bạn.
- Elasticsearch thích hợp khi bạn muốn tận dụng tính năng lưu trữ và tìm kiếm của Elasticsearch mà không cần xử lý phức tạp trên log. Elasticsearch cung cấp các khả năng tìm kiếm, lọc và phân tích mạnh mẽ trên log và hỗ trợ việc lưu trữ dữ liệu phân tán và hiệu suất cao.

Tùy thuộc vào yêu cầu và kiến trúc của hệ thống của bạn, bạn có thể chọn sử dụng Logstash và Elasticsearch riêng lẻ hoặc kết hợp cả hai để xử lý và lưu trữ log một cách tối ưu.

2.2. Chuẩn bị server logs.

Để cấu hình Logstash hoặc Elasticsearch để lắng nghe và nhận log từ switch Cisco trên cổng 5044, bạn cần chỉnh sửa file cấu hình tương ứng. Dưới đây là ví dụ nội dung file cấu hình Logstash hoặc Elasticsearch:

Logstash:

Mở file cấu hình của Logstash (thông thường là logstash.conf) để chỉnh sửa. Thêm các cấu hình input để lắng nghe và nhận log từ switch Cisco trên cổng 5044:

input {
  udp {
    port => 5044
    type => "syslog-cisco"
  }
  
  tcp {
    port => 5044
    type => "syslog-cisco"
  }
}

Tiếp theo là phần cấu hình filter, nếu bạn để trống nội dung của filter như trong ví dụ của mình thì không có lỗi xảy ra vì bạn đang sử dụng một câu lệnh if mà không có hành động được xác định bên trong nó. Điều này có nghĩa là nếu trường type trong sự kiện là “syslog-cisco”, không có bất kỳ xử lý hay biến đổi nào sẽ xảy ra.

Tuy nhiên, hãy lưu ý rằng nếu bạn không có bất kỳ xử lý nào trong block if, có thể không cần thiết phải sử dụng câu lệnh if trong trường hợp này. Bạn có thể loại bỏ câu lệnh if và chỉ sử dụng grok mà không có điều kiện, vì các sự kiện không phù hợp với mẫu grok sẽ không được xử lý bởi block grok đó. Trong trường hợp này, tất cả các sự kiện syslog-cisco sẽ được xử lý bởi block grok.

filter {
  if [type] == "syslog-cisco" {
    grok {
      # Nội dung của đoạn filter
    }
  }
}

Cuối cùng là phần cấu hình output.

output {
  if "_grokparsefailure" in [tags] {
    file {
      path => "/tmp/fail-%{type}-%{+YYYY.MM.dd}.log"
    }
  }
  elasticsearch {
    hosts => ["http://es01:9200", "http://es02:9200", "http://es03:9200"]
    sniffing => true
    ssl => false
    ssl_certificate_verification => false
    index => "cisco-logs-%{+YYYY.MM.dd}"
  }
}

Dòng if "_grokparsefailure" in [tags] kiểm tra xem sự kiện có tag _grokparsefailure hay không. Nếu có, nghĩa là sự kiện không khớp với mẫu grok và có lỗi xảy ra trong quá trình phân tích. Trong trường hợp này, sự kiện sẽ được ghi vào một tập tin log để kiểm tra và xử lý sau này.
Trong khối file, path xác định đường dẫn và tên tập tin cho các sự kiện không khớp. Trong ví dụ này, tập tin sẽ được lưu trong thư mục /tmp với tên dạng fail-%{type}-%{+YYYY.MM.dd}.log. %{type} là giá trị của trường type trong sự kiện, %{+YYYY.MM.dd} là ngày hiện tại theo định dạng YYYY.MM.dd.
Khối elasticsearch xác định đầu ra của Logstash là Elasticsearch, nơi các sự kiện sẽ được gửi đến lưu trữ và truy vấn.
hosts chỉ định danh sách các máy chủ Elasticsearch mà Logstash sẽ gửi dữ liệu đến. Trong ví dụ này, có ba máy chủ: es01:9200, es02:9200, và es03:9200.
sniffing => true cho phép Logstash tự động phát hiện và giao tiếp với các nút Elasticsearch trong cụm.
ssl => false và ssl_certificate_verification => false tắt việc sử dụng SSL cho giao tiếp giữa Logstash và Elasticsearch. Đây là cấu hình không an toàn và chỉ nên sử dụng trong môi trường phát triển hoặc kiểm tra.
index => "cisco-logs-%{+YYYY.MM.dd}" xác định mẫu cho tên chỉ mục (index) trong Elasticsearch. Trong ví dụ này, chỉ mục sẽ có tên dạng cisco-logs-YYYY.MM.dd, trong đó %{+YYYY.MM.dd} là ngày hiện tại theo định dạng YYYY.MM.dd.

Sau khi bạn chuẩn bị xong file cấu hình, hãy tiến hành khởi động lại Logstash để cấu hình mới được áp dụng.

systemctl restart logstash

Hoặc nếu bạn sử dụng docker container thì sử dụng lệnh docker restart <container_name>.

docker restart logstash

Đối với Docker có thể sử dụng lệnh docker logs -f <container_name> để xem logs nhé.

docker logs -f logstash

Khi bạn nhìn thấy logs như vậy thì container Logstash đã khởi động xong và có thể sử dụng.

Sending Logstash logs to /usr/share/logstash/logs which is now configured via log4j2.properties
[2023-06-14T22:14:21,507][INFO ][logstash.runner          ] Log4j configuration path used is: /usr/share/logstash/config/log4j2.properties
[2023-06-14T22:14:21,515][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"7.13.2", "jruby.version"=>"jruby 9.2.16.0 (2.5.7) 2021-03-03 f82228dc32 OpenJDK 64-Bit Server VM 11.0.11+9 on 11.0.11+9 +indy +jit [linux-x86_64]"}
[2023-06-14T22:14:22,377][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2023-06-14T22:14:22,970][INFO ][org.reflections.Reflections] Reflections took 30 ms to scan 1 urls, producing 24 keys and 48 values 
[2023-06-14T22:14:23,763][WARN ][deprecation.logstash.inputs.udp] Relying on default value of `pipeline.ecs_compatibility`, which may change in a future major release of Logstash. To avoid unexpected changes when upgrading Logstash, please explicitly declare your desired ECS Compatibility mode.
[2023-06-14T22:14:23,967][WARN ][deprecation.logstash.outputs.elasticsearch] Relying on default value of `pipeline.ecs_compatibility`, which may change in a future major release of Logstash. To avoid unexpected changes when upgrading Logstash, please explicitly declare your desired ECS Compatibility mode.
[2023-06-14T22:14:24,028][INFO ][logstash.outputs.elasticsearch][main] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["http://es01:9200", "http://es02:9200", "http://es03:9200"]}
[2023-06-14T22:14:24,293][INFO ][logstash.outputs.elasticsearch][main] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://es01:9200/, http://es02:9200/, http://es03:9200/]}}
[2023-06-14T22:14:24,413][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://es01:9200/"}
[2023-06-14T22:14:24,466][INFO ][logstash.outputs.elasticsearch][main] Elasticsearch version determined (7.13.2) {:es_version=>7}
[2023-06-14T22:14:24,468][WARN ][logstash.outputs.elasticsearch][main] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>7}
[2023-06-14T22:14:24,523][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://es02:9200/"}
[2023-06-14T22:14:24,569][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://es03:9200/"}
[2023-06-14T22:14:24,647][WARN ][deprecation.logstash.filters.grok][main] Relying on default value of `pipeline.ecs_compatibility`, which may change in a future major release of Logstash. To avoid unexpected changes when upgrading Logstash, please explicitly declare your desired ECS Compatibility mode.
[2023-06-14T22:14:24,691][INFO ][logstash.outputs.elasticsearch][main] Using a default mapping template {:es_version=>7, :ecs_compatibility=>:disabled}
[2023-06-14T22:14:24,721][INFO ][logstash.javapipeline    ][main] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50, "pipeline.max_inflight"=>1000, "pipeline.sources"=>["/usr/share/logstash/pipeline/logstash.conf"], :thread=>"#<Thread:0x1ad5653e run>"}
[2023-06-14T22:14:25,620][INFO ][logstash.javapipeline    ][main] Pipeline Java execution initialization time {"seconds"=>0.9}
[2023-06-14T22:14:25,749][INFO ][logstash.javapipeline    ][main] Pipeline started {"pipeline.id"=>"main"}
[2023-06-14T22:14:25,758][INFO ][logstash.inputs.tcp      ][main][1ea53ce4a1436ab156905017b69a8e5abf3a37fc9fbba66e490a69be25eaba19] Starting tcp input listener {:address=>"0.0.0.0:5044", :ssl_enable=>false}
[2023-06-14T22:14:25,849][INFO ][logstash.inputs.udp      ][main][55b1a52e99cfddb98bca50f48966ad338ae46af27a34ecabdc2d1dc6a11f4d1e] Starting UDP listener {:address=>"0.0.0.0:5044"}
[2023-06-14T22:14:25,866][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2023-06-14T22:14:25,889][INFO ][logstash.inputs.udp      ][main][55b1a52e99cfddb98bca50f48966ad338ae46af27a34ecabdc2d1dc6a11f4d1e] UDP listener started {:address=>"0.0.0.0:5044", :receive_buffer_bytes=>"106496", :queue_size=>"2000"}
[2023-06-14T22:14:29,647][INFO ][logstash.outputs.elasticsearch][main] Elasticsearch pool URLs updated {:changes=>{:removed=>[http://es01:9200/, http://es02:9200/, http://es03:9200/], :added=>[http://192.168.0.4:9200/, http://192.168.0.6:9200/, http://192.168.0.5:9200/]}}
[2023-06-14T22:14:29,653][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://192.168.0.4:9200/"}
[2023-06-14T22:14:29,703][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://192.168.0.6:9200/"}
[2023-06-14T22:14:29,746][WARN ][logstash.outputs.elasticsearch][main] Restored connection to ES instance {:url=>"http://192.168.0.5:9200/"}

2.3. Đẩy logs.

Mình sử dụng script sau để đẩy logs test nhé.

# !/bin/bash
build_logs(){
  destination="192.168.100.111"
  port=5044

  values=(
    'Jun 14 02:43:32.092 UTC: %LINK-3-UPDOWN: Interface Loopback0, changed state to up'
    '*Sep 15 18:20:59.372: %LINK-3-UPDOWN: Interface TenGigabitEthernet2/1/4, changed state to down'
    '*Jun 12 21:57:02.449: %LINEPROTO-5-UPDOWN: Line protocol on Interface Vlan1, changed state to up'
    '*Sep 15 18:20:47.712: %STACKMGR-6-STACK_LINK_CHANGE: Switch 1 R0/0: stack_mgr: Stack port 1 on Switch 1 is up'
    '*Sep 15 18:20:48.227: %DHCPD-4-PING_CONFLICT: DHCP address conflict:  server pinged 192.168.113.2.'
    '*Sep 15 18:20:47.747: %STACKMGR-4-SWITCH_ADDED: Switch 1 R0/0: stack_mgr: Switch 2 has been added to the stack.'
    '*Sep 29 01:29:15.865: %WEBSERVER-5-CONNECTION_FAILED: Switch 1 R0/0: nginx: connection failed from host 192.168.125.18 - Cipher Mismatch/No shared cipher'
    '*Sep 15 18:21:29.069: %PLATFORM_PM-6-FRULINK_INSERTED: 4x1G uplink module inserted in the switch 2 slot 1'
    '*Sep 15 18:21:29.262: %PLATFORM_PM-6-MODULE_INSERTED: SFP module inserted with interface name Gi2/1/1'
    '*Sep 15 18:21:29.327: %PLATFORM_PM-6-MODULE_INSERTED: SFP module inserted with interface name Gi2/1/2'
    'Jun 14 02:49:58.406 UTC: %LINK-5-CHANGED: Interface Loopback0, changed state to administratively down'
    'Jun 13 08:29:23.480 UTC: %DHCPD-4-DECLINE_CONFLICT: DHCP address conflict:  client 01ee.2dd3.ac76.66 declined 192.168.106.252'
    '*Oct  7 21:12:19.538: %SW_MATM-4-MACFLAP_NOTIF: Host 0009.0f09.0004 in vlan 503 is flapping between port Gi1/0/20 and port Gi1/0/19'
    'Jun 12 22:54:06.345 UTC: %SYS-5-CONFIG_I: Configured from console by admin on vty0 (192.168.100.111)'
    '*Sep 15 18:21:45.792: %REDUNDANCY-5-PEER_MONITOR_EVENT: Active detected a standby insertion (raw-event=PEER_FOUND(4))'
  )

  for val in "${values[@]}"
    do
      echo "$val" | nc -w1 -u "$destination" "$port"
    done
}

Sau khi chạy script cong bạn vào Stack Management.

Vào Index Management (1) -> Reload indices (2) -> bạn sẽ nhìn thấy Index đầu tiên (3), đó chính là toàn bộ logs bạn đã đẩy sang bằng script trên hoặc là đẩy từ switch qua.