logstash的各个场景应用（配置文件均已实践过）

#控制台输入 input { stdin { } } output {      #codec输出到控制台 stdout { codec=> rubydebug } #输出到elasticsearch elasticsearch {         hosts => "node18:9200"         codec => json         } #输出到文件 file {     path => "/usr/local/logstash-5.6.10/data/log/logstash/all.log" #指定写入文件路径     flush_interval => 0                  # 指定刷新间隔，0代表实时写入     codec => json     } }

################# Filebeat Configuration Example ########################   # This file is an example configuration file highlighting only the most common # options. The filebeat.full.yml file from the same directory contains all the # supported options with more comments. You can use it as a reference. # # You can find the full configuration reference here: # https://www.elastic.co/guide/en/beats/filebeat/index.html   #===================== Filebeat prospectors =====================   filebeat.prospectors:   # Each - is a prospector. Most options can be set at the prospector level, so # you can use different prospectors for various configurations. # Below are the prospector specific configurations.   - input_type: log     # Paths that should be crawled and fetched. Glob based paths.   paths:     - /home/admin/helloworld/logs/*.log     #- c:\programdata\elasticsearch\logs\*     # Exclude lines. A list of regular expressions to match. It drops the lines that are   # matching any regular expression from the list.   #exclude_lines: ["^DBG"]     # Include lines. A list of regular expressions to match. It exports the lines that are   # matching any regular expression from the list.   #include_lines: ["^ERR", "^WARN"]     # Exclude files. A list of regular expressions to match. Filebeat drops the files that   # are matching any regular expression from the list. By default, no files are dropped.   #exclude_files: [".gz$"]     # Optional additional fields. These field can be freely picked   # to add additional information to the crawled log files for filtering   #fields:   #  level: debug   #  review: 1     ### Multiline options     # Mutiline can be used for log messages spanning multiple lines. This is common   # for Java Stack Traces or C-Line Continuation     # The regexp Pattern that has to be matched. The example pattern matches all lines starting with [   #multiline.pattern: ^\[     # Defines if the pattern set under pattern should be negated or not. Default is false.   #multiline.negate: false     # Match can be set to "after" or "before". It is used to define if lines should be append to a pattern   # that was (not) matched before or after or as long as a pattern is not matched based on negate.   # Note: After is the equivalent to previous and before is the equivalent to to next in Logstash   #multiline.match: after     #====================== General =============================   # The name of the shipper that publishes the network data. It can be used to group # all the transactions sent by a single shipper in the web interface. #name:   # The tags of the shipper are included in their own field with each # transaction published. #tags: ["service-X", "web-tier"]   # Optional fields that you can specify to add additional information to the # output. #fields: #  env: staging   #======================= Outputs ===========================   # Configure what outputs to use when sending the data collected by the beat. # Multiple outputs may be used.   #-------------------------- Elasticsearch output ------------------------------ #output.elasticsearch:   # Array of hosts to connect to.   # hosts: ["localhost:9200"]     # Optional protocol and basic auth credentials.   #protocol: "https"   #username: "elastic"   #password: "changeme"   #--------------------------- Logstash output -------------------------------- output.logstash:   # The Logstash hosts   hosts: ["192.168.80.34:5044"]     # Optional SSL. By default is off.   # List of root certificates for HTTPS server verifications   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]     # Certificate for SSL client authentication   #ssl.certificate: "/etc/pki/client/cert.pem"     # Client Certificate Key   #ssl.key: "/etc/pki/client/cert.key"   #=========================== Logging =======================   # Sets log level. The default log level is info. # Available log levels are: critical, error, warning, info, debug #logging.level: debug   # At debug level, you can selectively enable logging only for some components. # To enable all selectors use ["*"]. Examples of other selectors are "beat", # "publish", "service". #logging.selectors: ["*"]

input {     beats {     port => 5044     codec => "json" } } #filters{ #…………(后续进行说明) #}   output {     # 输出到控制台     # stdout { }       # 输出到redis     redis {         host => "192.168.80.32"   # redis主机地址         port => 6379              # redis端口号         password => "123456"          # redis 密码         #db => 8                   # redis数据库编号         data_type => "channel"    # 使用发布/订阅模式         key => "logstash_list_0"  # 发布通道名称 } #输出到kafka     kafka {         bootstrap_servers => "192.168.80.42:9092"         topic_id         => "test"         } #输出到es elasticsearch {         hosts => "node18:9200"         codec => json         } }

input {     beats {     port => 5044     codec => "json"        }     syslog{        } }   #filter{ # #}   output {     # 输出到控制台     # stdout { }       # 输出到redis     redis {         host => "192.168.80.32"   # redis主机地址         port => 6379              # redis端口号         password => "123456"          # redis 密码        #db => 8                   # redis数据库编号         data_type => "channel"    # 使用发布/订阅模式         key => "logstash_list_0"  # 发布通道名称     }     #输出到kafka     kafka {         bootstrap_servers => "192.168.80.42:9092"         topic_id          => "test"         }      }

input{     kafka {         bootstrap_servers => "192.168.80.42:9092"            topics          => ["test"]            #decroate_events   => true         group_id          => "consumer-test"（消费组）            #decroate_events  => true         auto_offset_reset => "earliest"（初始消费，相当于from beginning，不设置，相当于是监控启动后的kafka的消息生产）    } } #filter{ #} output {        elasticsearch {        hosts => "192.168.80.18:9200"           codec => json        } }

input{     kafka {         bootstrap_servers => "192.168.80.42:9092"             topics          => ["test"]          group_id       => "consumer-test"          #decroate_events  => true        auto_offset_reset => "earliest"    }   } #flter{ # #}   output {        elasticsearch {        hosts => "192.168.80.18:9200"        codec => json        }        }

################# Filebeat Configuration Example #########################   # This file is an example configuration file highlighting only the most common # options. The filebeat.full.yml file from the same directory contains all the # supported options with more comments. You can use it as a reference. # # You can find the full configuration reference here: # https://www.elastic.co/guide/en/beats/filebeat/index.html   #================== Filebeat prospectors===========================   filebeat.prospectors:   # Each - is a prospector. Most options can be set at the prospector level, so # you can use different prospectors for various configurations. # Below are the prospector specific configurations.   - input_type: log     # Paths that should be crawled and fetched. Glob based paths.   paths:     - /home/admin/helloworld/logs/*.log     #- c:\programdata\elasticsearch\logs\*     # Exclude lines. A list of regular expressions to match. It drops the lines that are   # matching any regular expression from the list.   #exclude_lines: ["^DBG"]     # Include lines. A list of regular expressions to match. It exports the lines that are   # matching any regular expression from the list.   #include_lines: ["^ERR", "^WARN"]     # Exclude files. A list of regular expressions to match. Filebeat drops the files that   # are matching any regular expression from the list. By default, no files are dropped.   #exclude_files: [".gz$"]     # Optional additional fields. These field can be freely picked   # to add additional information to the crawled log files for filtering   #fields:   #  level: debug   #  review: 1     ### Multiline options     # Mutiline can be used for log messages spanning multiple lines. This is common   # for Java Stack Traces or C-Line Continuation     # The regexp Pattern that has to be matched. The example pattern matches all lines starting with [   #multiline.pattern: ^\[     # Defines if the pattern set under pattern should be negated or not. Default is false.   #multiline.negate: false     # Match can be set to "after" or "before". It is used to define if lines should be append to a pattern   # that was (not) matched before or after or as long as a pattern is not matched based on negate.   # Note: After is the equivalent to previous and before is the equivalent to to next in Logstash   #multiline.match: after   #============================ General=========================   # The name of the shipper that publishes the network data. It can be used to group # all the transactions sent by a single shipper in the web interface. #name:   # The tags of the shipper are included in their own field with each # transaction published. #tags: ["service-X", "web-tier"]   # Optional fields that you can specify to add additional information to the # output. #fields: #  env: staging   #======================== Outputs ============================   # Configure what outputs to use when sending the data collected by the beat. # Multiple outputs may be used.   #-------------------------- Elasticsearch output ------------------------------ #output.elasticsearch:   # Array of hosts to connect to.   # hosts: ["localhost:9200"]     # Optional protocol and basic auth credentials.   #protocol: "https"   #username: "elastic"   #password: "changeme"   #----------------------------- Logstash output -------------------------------- #output.logstash:   # The Logstash hosts #  hosts: ["192.168.80.34:5044"]   #-----------------------------kafka  output----------------------------------- #output.kafka: #  enabled: true #  hosts: ["192.168.80.42:9092,192.168.80.43:9092,192.168.80.44:9092"] #  topics: 'test' output.kafka:   hosts: ["192.168.80.42:9092"]   topic: test   required_acks: 1       # Optional SSL. By default is off.   # List of root certificates for HTTPS server verifications   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]     # Certificate for SSL client authentication   #ssl.certificate: "/etc/pki/client/cert.pem"     # Client Certificate Key   #ssl.key: "/etc/pki/client/cert.key"   #======================== Logging ============================   # Sets log level. The default log level is info. # Available log levels are: critical, error, warning, info, debug #logging.level: debug   # At debug level, you can selectively enable logging only for some components. # To enable all selectors use ["*"]. Examples of other selectors are "beat", # "publish", "service". #logging.selectors: ["*"]

input{     kafka {         bootstrap_servers => "192.168.80.42:9092"             topics          => ["test"]          group_id       => "consumer-test"          #decroate_events  => true        auto_offset_reset => "earliest"    }   } #flter{ # #}   output {        elasticsearch {        hosts => "192.168.80.18:9200"        codec => json        }        }

input {     beats {     port => 5044     codec => "json"     ssl => true    ssl_certificate_authorities => ["/usr/local/logstash-5.6.10/pki/tls/certs/filebeat.crt"]    ssl_certificate => "/usr/local/logstash-5.6.10/pki/tls/certs/logstash.crt"    ssl_key => "/usr/local/logstash-5.6.10/pki/tls/private/logstash.key" ssl_verify_mode => "force_peer"#（需与ssl_certificate_authorities一起使用）        }     syslog{        } }   output {     # 输出到控制台     # stdout { }       # 输出到redis     redis {         host => "192.168.80.32"   # redis主机地址         port => 6379              # redis端口号         password => "123456"          # redis 密码        #db => 8                   # redis数据库编号         data_type => "channel"    # 使用发布/订阅模式         key => "logstash_list_0"  # 发布通道名称     }     #输出到kafka     kafka {         bootstrap_servers => "192.168.80.42:9092"         topic_id          => "test"         }          #输出到es     elasticsearch {        hosts => "node18:9200"        codec => json        }   }

################ #Filebeat Configuration Example #####################   # This file is an example configuration file highlighting only the most common # options. The filebeat.full.yml file from the same directory contains all the # supported options with more comments. You can use it as a reference. # # You can find the full configuration reference here: # https://www.elastic.co/guide/en/beats/filebeat/index.html   #=================== Filebeat prospectors ========================   filebeat.prospectors:   # Each - is a prospector. Most options can be set at the prospector level, so # you can use different prospectors for various configurations. # Below are the prospector specific configurations.   - input_type: log     # Paths that should be crawled and fetched. Glob based paths.   paths:     - /home/admin/helloworld/logs/*.log     #- c:\programdata\elasticsearch\logs\*     # Exclude lines. A list of regular expressions to match. It drops the lines that are   # matching any regular expression from the list.   #exclude_lines: ["^DBG"]     # Include lines. A list of regular expressions to match. It exports the lines that are   # matching any regular expression from the list.   #include_lines: ["^ERR", "^WARN"]     # Exclude files. A list of regular expressions to match. Filebeat drops the files that   # are matching any regular expression from the list. By default, no files are dropped.   #exclude_files: [".gz$"]     # Optional additional fields. These field can be freely picked   # to add additional information to the crawled log files for filtering   #fields:   #  level: debug   #  review: 1     ### Multiline options     # Mutiline can be used for log messages spanning multiple lines. This is common   # for Java Stack Traces or C-Line Continuation     # The regexp Pattern that has to be matched. The example pattern matches all lines starting with [   #multiline.pattern: ^\[     # Defines if the pattern set under pattern should be negated or not. Default is false.   #multiline.negate: false     # Match can be set to "after" or "before". It is used to define if lines should be append to a pattern   # that was (not) matched before or after or as long as a pattern is not matched based on negate.   # Note: After is the equivalent to previous and before is the equivalent to to next in Logstash   #multiline.match: after   #======================== General ============================   # The name of the shipper that publishes the network data. It can be used to group # all the transactions sent by a single shipper in the web interface. #name:   # The tags of the shipper are included in their own field with each # transaction published. #tags: ["service-X", "web-tier"]   # Optional fields that you can specify to add additional information to the # output. #fields: #  env: staging   #========================= Outputs ===========================   # Configure what outputs to use when sending the data collected by the beat. # Multiple outputs may be used.   #----------------------------- Elasticsearch output ------------------------------ #output.elasticsearch:   # Array of hosts to connect to.   # hosts: ["localhost:9200"]     # Optional protocol and basic auth credentials.   #protocol: "https"   #username: "elastic"   #password: "changeme"   #----------------------------- Logstash output -------------------------------- output.logstash: # The Logstash hosts   hosts: ["192.168.80.18:5044"] #加密传输   ssl.certificate_authorities: ["/usr/local/filebeat-5.6.10/pki/tls/certs/logstash.crt"]   ssl.certificate: "/usr/local/filebeat-5.6.10/pki/tls/certs/filebeat.crt"   ssl.key: "/usr/local/filebeat-5.6.10/pki/tls/private/filebeat.key"    #----------------------------- kafka  output----------------------------------- #output.kafka: #  hosts: ["192.168.80.42:9092"] #  topic: test #  required_acks: 1     # Optional SSL. By default is off.   # List of root certificates for HTTPS server verifications   #ssl.certificate_authorities: ["/etc/pki/root/ca.pem"]     # Certificate for SSL client authentication   #ssl.certificate: "/etc/pki/client/cert.pem"     # Client Certificate Key   #ssl.key: "/etc/pki/client/cert.key"   #========================== Logging =========================   # Sets log level. The default log level is info. # Available log levels are: critical, error, warning, info, debug #logging.level: debug   # At debug level, you can selectively enable logging only for some components. # To enable all selectors use ["*"]. Examples of other selectors are "beat", # "publish", "service". #logging.selectors: ["*"]

input {     file {         path => [             # 这里填写需要监控的文件             "/home/admin/helloworld/logs/catalina.out"         ]     } }   output {     kafka {     # 输出到控制台     # stdout { }     # 输出到kafka     bootstrap_servers => "192.168.80.42:9092"     topic_id          => "test"     } }

input{ #从redis读取  redis {         host => "192.168.80.32"   # redis主机地址         port => 6379              # redis端口号        password  => "123456"      # redis 密码         #db => 8                   # redis数据库编号         data_type => "channel"    # 使用发布/订阅模式         key => "logstash_list_0"  # 发布通道名称 } #从kafka读取  kafka {         bootstrap_servers => "192.168.80.42:9092"            topics          => ["test"]         auto_offset_reset => "earliest"        } }   output {     #输出到文件     file {         path => "/usr/local/logstash-5.6.10/data/log/logstash/all1.log" # 指定写入文件路径 #       message_format => "%{host} %{message}"         # 指定写入格式         flush_interval => 0                             # 指定刷新间隔，0代表实时写入      codec => json        }    #输出到es    elasticsearch {        hosts => "node18:9200"        codec => json        } }

input {  stdin { }     jdbc {         jdbc_connection_string => "jdbc:mysql://192.168.80.18:3306/fyyq-mysql"         jdbc_user => "fyyq"         jdbc_password => "fyyq@2017"    jdbc_driver_library => "/usr/local/logstash-5.6.10/mysql-connector-java-5.1.46.jar"         jdbc_driver_class => "com.mysql.jdbc.Driver"         jdbc_paging_enabled => "true"         statement_filepath => "/usr/local/logstash-5.6.10/mysql2es.sql"         #schedule => "* * * * *"     }  }    output {      stdout {         codec => json_lines     }     elasticsearch {         hosts => "node18:9200"         #index => "mainIndex"         #document_type => "user"         #document_id => "%{id}"     } }

select * from sys_log

input {     beats {       port => 5044       #codec => "json"       ssl => true    ssl_certificate_authorities => ["/usr/local/logstash-5.6.10/pki/tls/certs/filebeat.crt"]       ssl_certificate => "/usr/local/logstash-5.6.10/pki/tls/certs/logstash.crt"       ssl_key => "/usr/local/logstash-5.6.10/pki/tls/private/logstash.key"       ssl_verify_mode => "force_peer"                            } }   filter{    grok {        match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"} } }     output {     # 输出到控制台     # stdout { }       # 输出到redis     redis {         host => "192.168.80.32"   # redis主机地址         port => 6379              # redis端口号         password => "123456"          # redis 密码        #db => 8                   # redis数据库编号         data_type => "channel"    # 使用发布/订阅模式         key => "logstash_list_0"  # 发布通道名称     }     #输出到kafka     kafka {         bootstrap_servers => "192.168.80.42:9092"         topic_id          => "test"                             }           #输出到es     elasticsearch {                            hosts => "node18:9200"                            codec => json                            }     #输出到hdfs      webhdfs {      host => "192.168.80.42"      port => 50070      path => "/user/logstash/dt=%{+YYYY-MM-dd}/%{@source_host}-%{+HH}.log"      user => "hadoop"        } }

Setting

Input type

Required

Setting

Input type

Required

Setting

Input type

Required

filter {   grok {match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }   } }

55.3.244.1 GET /index.html 15824 0.043

client: 55.3.244.1（IP） method: GET（方法） request: /index.html（请求文件路径） bytes: 15824（字节数） duration: 0.043（访问时长）

filter {     grok {         match => { "message" => "%{COMBINEDAPACHELOG}"}     } }

192.168.80.183 - - [04/Jan/2018:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"

"clientip" => "192.168.80.183", "timestamp" => "04/Jan/2018:05:13:42 +0000", "verb" => "GET", "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png", "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"", "response" => "200", "bytes" => "203023", "agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",

filter {   grok{ 　　match=>{ 　　　　"message"=>"%{IP:clientip}\s+(?<mypattern>[A-Z]+)"}     } }

12.12.12.12 ABC

"clientip" => "12.12.12.12", "mypattern" => "ABC"

filter {     grok {         #match => { "message" => "%{COMBINEDAPACHELOG}"}          match => { "message" => "%{IP:clientip}\s+%{IP:clientip1}"}     }     mutate {     remove_field => ["message"]     remove_field => ["host"]    } }

1.1.1.1 2.2.2.2

{   "_index": "logstash-2018.07.31",   "_type": "log",   "_id": "AWTuNdzp6Wkp4mVEj3Fh",   "_version": 1,   "_score": null,   "_source": {     "@timestamp": "2018-07-31T02:41:00.014Z",     "offset": 1114,     "clientip": "1.1.1.1",     "@version": "1",     "input_type": "log",     "beat": {       "name": "node183",       "hostname": "node183",       "version": "5.6.10"     },     "source": "/home/usieip/bdp-datashare/logs/a.log",     "type": "log",     "clientip1": "2.2.2.2",     "tags": [       "beats_input_codec_plain_applied"     ]   },   "fields": {     "@timestamp": [       1533004860014     ]   },   "sort": [     1533004860014   ] }

filter {     grok {          match => { "message" =>  "%{DATA:ymd} %{DATA:sfm} %{DATA:http} %{DATA:info}  %{GREEDYDATA:index}"} } }

2018-07-30 17:04:31.317 [http-bio-8080-exec-19] INFO  c.u.i.b.m.s.i.LogInterceptor - ViewName: modules/datashare/front/index

{   "_index": "logstash-2018.07.31",   "_type": "log",   "_id": "AWTvhiPD6Wkp4mVEj3GU",   "_version": 1,   "_score": null,   "_source": {     "offset": 125,     "input_type": "log",     "index": "c.u.i.b.m.s.i.LogInterceptor - ViewName: modules/datashare/front/index",     "source": "/home/usieip/bdp-datashare/logs/b.log",     "type": "log",     "tags": [],     "ymd": "2018-07-30",     "@timestamp": "2018-07-31T08:48:17.948Z",     "@version": "1",     "beat": {       "name": "node183",       "hostname": "node183",       "version": "5.6.10"     },     "http": "[http-bio-8080-exec-19]",     "sfm": "17:04:31.317",     "info": "INFO"   },   "fields": {     "ymd": [       1532908800000     ],     "@timestamp": [       1533026897948     ]   },   "sort": [     1533026897948   ] }

patterns_dir => ["/opt/logstash/patterns","/opt/logstash/extra_patterns"]

remove_field => ["foo _％{somefield}"]

Setting

Input type

Required