This week I spent some time learning the EK stack with a small project.
After battling with EK specific mac OS docker issues (I guess I should blog about this) I started with a small project: tracking firewall activity on a consumer router. The solution ramped me up on EK though I'm wondering what "best practices" I should have considered. Any feedback?
Hardware/logging:
router -> rpi -> mac OS/docker EK stack
The router logs are in syslog format. The system logs are stored on the rpi using generic log forwarding. The router.log file is manually pulled from rpi onto the mac for processing. (In a live system, rpi would feed data directly to EK for realtime analysis)
EK Config
Filebeat processes the router.log file, sending it to elasticsearch pipelines which perform additional processing. Docker-compose describes a 2 es node, 1 Kibana node configuration. Visualization in Kibana (geo_point and related fields).
Filebeat Config:
Custom filebeat.yml and filebeat.template.json. Relevant fields are:
filebeat.yml:
prospector: document_type:syslog close_inactive: 2m tags: ["merlin-syslog"] _output.elasticsearch:_ index: "merlin-%{+yyyy.MM.dd}" indices: - index: "merlin-dhcp-%{+yyyy.MM.dd}" when.contains: message: "dnsmasq-dhcp" pipeline: "merlin-syslog" pipelines: - pipeline: "merlin-firewall" when.contains: message: "DROP" - pipeline: "merlin-firewall" when.contains: message: "ACCEPT"
filebeat.template.json
properties: { "@timestamp": { "type": "date" }, "geoip": { "properties": { "city_name": { "ignore_above": 1024, "type": "keyword" }, "continent_name": { "ignore_above": 1024, "type": "keyword" }, "country_iso_code": { "ignore_above": 1024, "type": "keyword" }, "location": { "type": "geo_point" }, "region_name": { "ignore_above": 1024, "type": "keyword" } } } } "template": "merlin-*"
Elasticsearch Pipelines
Two pipelines: one for generic syslog processing and one for firewall log processing.
Generic Pipeline
"processors": [ { "grok": { "field": "message", "patterns": ["%{SYSLOGBASE} %{GREEDYDATA:message}"] }, "date": { "field": "timestamp", "formats": [ "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ], "ignore_failure": true } } ]
Firewall Pipeline
"processors": [ { "grok": { "field": "message", "patterns": ["%{SYSLOGBASE} %{WORD:action} <4>%{WORD} IN=%{WORD:in_device} OUT=%{WORD:out_device}?(?:%{GREEDYDATA})<1>SRC=%{IP:src_ip} DST=%{IP:dest_ip} %{GREEDYDATA:message}"] }, "date": { "field": "timestamp", "formats": [ "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ], "ignore_failure": true }, "geoip": { "field": "src_ip", "ignore_failure": true }, "set": { "field": "program", "value": "firewall" } } ]