1. How Logstash Works

Logstash is a component for data transport and processing.

Through combinations of plugins, Logstash can handle a wide variety of log collection scenarios:

logstash->elasticsearch
filebeat->logstash->elasticsearch
filebeat->kafka->logstash->elasticsearch
filebeat->redis->logstash->elasticsearch

2. Basic Logstash Configuration

Here is the format of a Logstash configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 数据源，例如 Kafka、MySQL
input {

}
# 过滤器，用于处理数据，可选，例如丢掉空值、添加标签等
filter {

}
# 输出，例如 ES、文件
output {

}

The input data sources include:

azure_event_hubs、beats、cloudwatch、couchdb_changes、dead_letter_queue、elastic_agent、elasticsearch、exec、file、ganglia、gelf、generator、github、google_cloud_storage、google_pubsub、graphite、heartbeat、http、http_poller、imap、irc、java_generator、java_stdin、jdbc、jms、jmx、kafka、kinesis、log4j、lumberjack、meetup、pipe、puppet_facter、rabbitmq、redis、relp、rss、s3、s3-sns-sqs、salesforce、snmp、snmptrap、sqlite、sqs、stdin、stomp、syslog、tcp、twitter、udp、unix、varnishlog、websocket、wmi、xmpp。

Reference: ES input plugins

The filter plugins include:

age、aggregate、alter、bytes、cidr、cipher、clone、csv、date、de_dot、dissect、dns、drop、elapsed、elasticsearch、environment、extractnumbers、fingerprint、geoip、grok、http、i18n、java_uuid、jdbc_static、jdbc_streaming、json、json_encode、kv、memcached、metricize、metrics、mutate、prune、range、ruby、sleep、split、syslog_pri、threats_classifier、throttle、tld、translate、truncate、urldecode、useragent、uuid、wurfl_device_detection、xml

Reference: ES filter plugins

The output sinks include:

boundary、circonus、cloudwatch、csv、datadog、datadog_metrics、dynatrace、elastic_app_search、elastic_workplace_search、elasticsearch、email、exec、file、ganglia、gelf、google_bigquery、google_cloud_storage、google_pubsub、graphite、graphtastic、http、influxdb、irc、java_stdout、juggernaut、kafka、librato、loggly、lumberjack、metriccatcher、mongodb、nagios、nagios_nsca、opentsdb、pagerduty、pipe、rabbitmq、redis、redmine、riak、riemann、s3、sink、sns、solr_http、sqs、statsd、stdout、stomp、syslog、tcp、timber、udp、webhdfs、websocket、xmpp、zabbix

Reference: ES output plugins

When configuring, you need to debug against the official documentation according to your use case.

3. Logstash Configuration Examples

3.1 stdin to stdout

1
2
3
4
5
6

input {
    stdin {}
}
output {
    stdout { codec => rubydebug }
}

3.2 File to ES

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

input {
    file {
        path => [ "/data/nginx/logs/nginx_access.log" ]
        start_position => "beginning"
        ignore_older => 0
    }
}
filter {
    # 使用 grok 插件对日志内容进行格式化
    grok {
        match => {
            "message" => "%{COMBINEDAPACHELOG}"
        }
    }
}
output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "nginx-access"
    }
}

3.3 Filebeats to ES

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

input {
    beats {
        port => 8088
    }
}
output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "beats"
    }
}

3.4 Kafka to ES

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

input {
  kafka {
    bootstrap_servers => "${KAFKA_URL:localhost:2187}"
    topics => "${KAFKA_TOPIC:kafka-topic-test}"
    group_id => "${KAFKA_GROUP_ID:logstash-es}"
    consumer_threads => "${KAFKA_CONSUMER_THREADS:6}"
  }
}
filter {
  if ![message] {
    drop { }
  }
  mutate {
    remove_field => [ "headers", "result"]
  }
}
output {
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "kafka"
    }
}