锥智这套日志收集系统流程是：
各种日志文件 —> FileBeats —> Kafka —> Logstash —> ElasticSearch

FileBeats配置

  1. ###################### Filebeat Configuration Example #########################
  3. #=========================== Filebeat inputs =============================
  5. filebeat.inputs:
  6. - enabled: true
  7. paths:
  8. - /data/app/*/logs/mqtt.log
  9. - /data/app/*/logs/exception.log
  10. - /data/app/*/logs/messages.log
  11. - /data/app/*/logs/requestcontext.log
  12. - /data/app/*/logs/responsecontext.log
  13. - /data/app/*/logs/transaction_tracking.log
  14. - /data/app/*/logs/transaction_tracking_step.log
  15. type: log
  17. - enabled: true
  18. exclude_files: []
  19. paths:
  20. - /data/app/*/logs/run.log
  21. type: log
  23. ### Multiline options
  24. # 针对Java日志报错内容进行多行配置
  25. multiline.pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'
  26. multiline.negate: true
  27. multiline.match: after
  30. #============================= Filebeat modules ===============================
  32. filebeat.config.modules:
  33. # Glob pattern for configuration loading
  34. path: ${path.config}/modules.d/*.yml
  36. # Set to true to enable config reloading
  37. reload.enabled: false
  39. #==================== Elasticsearch template setting ==========================
  41. setup.template.settings:
  42. index.number_of_shards: 3
  44. #================================ General =====================================
  48. #============================== Dashboards =====================================
  51. #============================== Kibana =====================================
  53. setup.kibana:
  55. # ============================== kafaka output ===================================
  56. output.kafka:
  57. #  hosts: ['10.111.212.165:9092', '10.111.212.166:9092','10.111.212.167:9092', '10.111.212.168:9092']
  58. hosts: ["10.11.7.25:9092"]
  59. topic: file_logging
  60. required_acks: 1

配置文件其实还是比较好懂得

  1. 输入配置两拨文件，run.log需要多行处理
  2. 发送到Kafka，并且acks方式为1，那就确保全收到

Logstash

  1. input {
  2.     kafka {
  3.         bootstrap_servers => "10.11.7.25:9092"
  4.         topics => ["file_logging"]
  5.         consumer_threads => 8
  6.         codec => json {
  7.             charset => "UTF-8"
  8.         }
  9.         client_id => "logstash-logging"
  10.         type => "logging"
  11.     }
  12.     kafka {
  13.         bootstrap_servers => "10.11.7.25:9092"
  14.         topics => ["middleware_monitor"]
  15.         consumer_threads => 4
  16.         client_id => "logstash-monitor"
  17.         codec => json {
  18.             charset => "UTF-8"
  19.         }
  20.         type => "monitor"
  21.     }
  22.     kafka {
  23.         bootstrap_servers => "10.11.7.25:9092"
  24.         topics => ["server_monitor_topic"]
  25.         consumer_threads => 4
  26.         codec => json
  27.         client_id => "logstash-services"
  28.         type => "services"
  29.     }
  30. }
  32. filter {
  33.   if[type] == "logging"{
  34.     mutate{
  35.         split => {"source" => "/"}
  36.     }
  38.     mutate{
  39.         add_field => {
  40.            "service" => "%{[source][3]}"
  41.            "is_nginx" => "%{[source][4]}"
  42.            "level" => "%{[source][5]}"
  43.         }
  44.     }
  45.     mutate{
  46.        remove_field => ["beat", "source", "offset", "prospector", "input", "@metadata"]
  47.     }
  48.   }
  49. }
  51. output {
  52.     if[type] == "monitor"{
  53.         elasticsearch{
  54.             hosts => ['10.11.7.24:9200']
  55.             index => "monitor-%{+YYYY.MM.dd}"
  56.             timeout => 300
  57.         }
  58.     } else if[type] == "services"{
  59.         elasticsearch{
  60.             hosts => ['10.11.7.24:9200']
  61.             index => "server-monitor-%{+YYYY.MM.dd}"
  62.             timeout => 300
  63.         }
  64.     }
  65.     if[type] == "logging"{
  66.        if[level] == "exception.log"{
  67.            elasticsearch{
  68.               hosts => ['10.11.7.24:9200']
  69.               index => "exception-logging-%{+YYYY.MM.dd}"
  70.               timeout => 300
  71.            }
  72.        }
  73.        if[level] == "responsecontext.log"{
  74.            elasticsearch{
  75.               hosts => ['10.11.7.24:9200']
  76.               index => "responsecontext-logging-%{+YYYY.MM.dd}"
  77.               timeout => 300
  78.            }
  79.        }
  80.        if[level] == "requestcontext.log"{
  81.            elasticsearch{
  82.               hosts => ['10.11.7.24:9200']
  83.               index => "requestcontext-logging-%{+YYYY.MM.dd}"
  84.               timeout => 300
  85.            }
  86.        }
  87.        if[level] == "run.log"{
  88.            elasticsearch{
  89.               hosts => ['10.11.7.24:9200']
  90.               index => "run-logging-%{+YYYY.MM.dd}"
  91.               timeout => 300
  92.            }
  93.        }
  94.        if[level] == "transaction_tracking_step.log"{
  95.            elasticsearch{
  96.               hosts => ['10.11.7.24:9200']
  97.               index => "transaction-logging-%{+YYYY.MM.dd}"
  98.               timeout => 300
  99.            }
  100.        }
  101.        if[level] == "transaction_tracking.log"{
  102.            elasticsearch{
  103.               hosts => ['10.11.7.24:9200']
  104.               index => "transaction-logging-%{+YYYY.MM.dd}"
  105.               timeout => 300
  106.            }
  107.        }
  108.        if[level] == "messages.log"{
  109.            elasticsearch{
  110.               hosts => ['10.11.7.24:9200']
  111.               index => "messages-logging-%{+YYYY.MM.dd}"
  112.               timeout => 300
  113.            }
  114.        }
  115.        if[level] == "access.log"{
  116.            elasticsearch{
  117.               hosts => ['10.11.7.24:9200']
  118.               index => "nginx-logging-%{+YYYY.MM.dd}"
  119.               timeout => 300
  120.            }
  121.        }
  122.        if[level] == "error.log"{
  123.            elasticsearch{
  124.               hosts => ['10.11.7.24:9200']
  125.               index => "nginx-logging-%{+YYYY.MM.dd}"
  126.               timeout => 300
  127.            }
  128.        }
  129.        if[level] == "mqtt.log"{
  130.            elasticsearch{
  131.               hosts => ['10.11.7.24:9200']
  132.               index => "mqtt-logging-%{+YYYY.MM.dd}"
  133.               timeout => 300
  134.            }
  135.        }
  136.     }
  137. }

Input配置

  • bootstrap_servers => “10.11.7.25:9092”：指定Kafka位置
  • topics => [“file_logging”]：指定Topics
  • consumer_threads => 8：指定消费者线程

  • codec => json {

    1.      charset => "UTF-8"<br />        }

  • client_id => “logstash-logging”：消费者id

  • type => “logging”：给这组数据一个标签

    Filter配置

  • 通过mutate做了字段的分割，添加字段和移除字段

    Output配置

  • 进不同的索引