资讯专栏INFORMATION COLUMN

上云采购季！| 2核2G4M爆款云服务器低至59元/年，更有多台、长期优惠，快来选购！

立即前往
首页/文章专栏/快速搭建ELK日志收集（kafka队列版）

快速搭建ELK日志收集（kafka队列版）

妤锋シ / 1037人阅读

摘要：快速搭建日志收集版本进行文章的第二次修改，包括了之前的简单方案的升级过程。分割线快速搭建日志收集第一版本新项目短时间来实现日志采集。

快速搭建elk日志收集 kafka版本

进行文章的第二次修改，包括了之前的简单方案的升级过程。

因为业务的不断更新升级，为了保证线上业务也能正常使用elk服务，并且使得elk的服务和线业务流解耦（即避免直接写入es的方式可能会带来的耗时影响）所以我们采用了下面最新的方案，也是常规方案

方案 
</>复制代码 
  1. 业务层 >> kafka队列 >> logstash 消费 >> elasticsearch

业务层将日志写入到kafka队列，同事logstash可以开启多个线程，启用同一个group_id来对kafka进行消费，将读取到的日志进行解析后，写入到elasticsearch中，并且按照索引模板进行解析。

优点

业务层可以直接写入到kafka队列中，不用担心elasticsearch的写入效率问题。

缺点

比起之前的简单版本，需要保证kafka队列、logstash的高可用（虽然logstash挂掉后，可以重启后重新读取队列日志）

搭建

整个搭建过程，写入kafka是非常简单的，这里遇到的问题是logstash和elasticsearch索引模板带来的困扰。

logstash内置一套模板

elasticsearch 本身可以自定义一套模板

如何确定使用谁的模板呢？
网上找到的资料，建议采用将模板配置在elasticsearch侧，这样就不用每个logstash进行一个模板配置文件的维护。

Logstash 配置文件说明

官网的文档kafka的input插件
https://www.elastic.co/guide/...

</>复制代码 
  1. input {
  2.     kafka {
  3.         // 需要读取的kafka队列集群配置
  4.         bootstrap_servers => "xxx.xxx.xxx.xxx:9092"
  5.         // 配置的消费者的group名称，因为同一个组内的消费消息不会重复
  6.         group_id => "logstash-group"
  7.         // 主题配置
  8.         topics => "kibana_log"
  9.         // 从未消费过的偏移量开始
  10.         auto_offset_reset =>"earliest"
  11.     }
  12. }
  13. // 重要，下面多带带讲
  14. filter {
  15.   json {
  16.     source => "message"
  17.   }
  18. }
  19. output {
  20.     // stdout可以省略，这个是为了命令行模式下方便调试
  21.     stdout{ codec => rubydebug }
  22.     elasticsearch {
  23.         // es 集群
  24.         hosts=>"xxx.xxx.xxx.xxx:9200"
  25.         // 重要：取消logstash自定义模板功能，进而强制使用es的内置模板
  26.         manage_template=>false
  27.         // 需要匹配的模板名称
  28.         index=>"logstash-dev-%{+YYYY.MM.dd}"
  29.     }
  30. }

先解释下filter配置，当我们从kafka读取消息的时候，消息体是通过message字段来进行传递的，所以message是一个字符串，但是我们的es索引模板可能会非常复杂，所以我们需要对其进行json解析后，再交给es。否则es收到的之后一个message字段。

</>复制代码 
  1. filter {
  2.   json {
  3.     source => "message"
  4.   }
  5. }

再说下模板配置，首先通过kibana的devtool向es中写入了一个模板，我区分了两套环境dev、prod。

Php 实现写入

这里字段都进行了strval转义，为什么呢？这和下面要讲的动态模板有关联的。往下看

</>复制代码 
  1. $position = YnUtil::getPosition();
  2. $urlData = parse_url(Wii::app()->request->url);
  3. $path    = $urlData["path"] ?? "";
  4. $params = [
  5.     "category"   => strval($category),
  6.     "appType"    => strval(YnUtil::getAppType()),
  7.     "appVersion" => strval(YnUtil::getAppVersion()),
  8.     "host"       => strval(Wii::app()->request->hostInfo),
  9.     "uri"        => strval(Wii::app()->request->url),
  10.     "uid"        => strval(Wii::app()->user->getUid()),
  11.     "path"       => strval($path),
  12.     "server"     => strval(gethostname()),
  13.     "geoip"      => [
  14.         "ip"       => strval(Yii::$app->request->userIP),
  15.         "location" => [
  16.             "lat" => floatval($position["latitude"]),
  17.             "lon" => floatval($position["longitude"]),
  18.         ],
  19.     ],
  20.     "userAgent"  => strval(Wii::app()->request->userAgent),
  21.     "message"    => is_array($message) ? Json::encode($message) : strval($message),
  22.     // "@timestamp" => intval(microtime(true) * 1000),
  23. ];
索引模板

下面的模板是写入到es里面的自定义模板，为了防止索引规则名称冲突，这里将order置为1。

我们先来看下第一个模板（这个是不推荐的，因为很繁琐，但是类型很强制有效）

</>复制代码 
  1. PUT _template/logstash-dev
  2. {
  3.     "index_patterns": "logstash-dev*",
  4.     "aliases": {},
  5.     "order":1,
  6.     "mappings": {
  7.       // 这里使用logs是因为logstash默认的type类型
  8.       "logs": {
  9.         // 动态模板
  10.         "dynamic_templates": [
  11.           {
  12.             "string_fields": {
  13.               "match": "*",
  14.               "match_mapping_type": "string",
  15.               "mapping": {
  16.                 "fields": {
  17.                   "keyword": {
  18.                     "ignore_above": 256,
  19.                     "type": "keyword"
  20.                   }
  21.                 },
  22.                 "norms": false,
  23.                 "type": "text"
  24.               }
  25.             }
  26.           }
  27.         ],
  28.         // 这里对属性进行了类型设置
  29.         "properties": {
  30.           "@timestamp": {
  31.             "type": "date"
  32.           },
  33.           "@version": {
  34.             "type": "keyword"
  35.           },
  36.           "appType": {
  37.             "type": "text",
  38.             "norms": false,
  39.             "fields": {
  40.               "keyword": {
  41.                 "type": "keyword",
  42.                 "ignore_above": 256
  43.               }
  44.             }
  45.           },
  46.           "appVersion": {
  47.             "type": "text",
  48.             "norms": false,
  49.             "fields": {
  50.               "keyword": {
  51.                 "type": "keyword",
  52.                 "ignore_above": 256
  53.               }
  54.             }
  55.           },
  56.           "category": {
  57.             "type": "text",
  58.             "norms": false,
  59.             "fields": {
  60.               "keyword": {
  61.                 "type": "keyword",
  62.                 "ignore_above": 256
  63.               }
  64.             }
  65.           },
  66.           "geoip": {
  67.             "dynamic": "true",
  68.             "properties": {
  69.               "ip": {
  70.                 "type": "ip"
  71.               },
  72.               "latitude": {
  73.                 "type": "half_float"
  74.               },
  75.               "location": {
  76.                 "type": "geo_point"
  77.               },
  78.               "longitude": {
  79.                 "type": "half_float"
  80.               }
  81.             }
  82.           },
  83.           "host": {
  84.             "type": "text",
  85.             "norms": false,
  86.             "fields": {
  87.               "keyword": {
  88.                 "type": "keyword",
  89.                 "ignore_above": 256
  90.               }
  91.             }
  92.           },
  93.           "message": {
  94.             "type": "text",
  95.             "norms": false
  96.           },
  97.           "server": {
  98.             "type": "text",
  99.             "norms": false,
  100.             "fields": {
  101.               "keyword": {
  102.                 "type": "keyword",
  103.                 "ignore_above": 256
  104.               }
  105.             }
  106.           },
  107.           "uid": {
  108.             "type": "text",
  109.             "norms": false,
  110.             "fields": {
  111.               "keyword": {
  112.                 "type": "keyword",
  113.                 "ignore_above": 256
  114.               }
  115.             }
  116.           },
  117.           "uri": {
  118.             "type": "text",
  119.             "norms": false,
  120.             "fields": {
  121.               "keyword": {
  122.                 "type": "keyword",
  123.                 "ignore_above": 256
  124.               }
  125.             }
  126.           },
  127.           "path": {
  128.             "type": "text",
  129.             "norms": false,
  130.             "fields": {
  131.               "keyword": {
  132.                 "type": "keyword",
  133.                 "ignore_above": 256
  134.               }
  135.             }
  136.           },
  137.           "userAgent": {
  138.             "type": "text",
  139.             "norms": false,
  140.             "fields": {
  141.               "keyword": {
  142.                 "type": "keyword",
  143.                 "ignore_above": 256
  144.               }
  145.             }
  146.           }
  147.         }
  148.       }
  149.     },
  150.     "settings": {
  151.       "index": {
  152.         "number_of_shards": "1"
  153.       }
  154.     }
  155. }
动态模板概念

上面的模板不是最优的，但是却是一种尝试，模板里面针对每个属性做了设置，这样客户端只需要写入对应的属性就好了。但是如何动态配置呢？如果想加入一个字段，难道还要修改模板么？ 这里引入了动态模板的概念

</>复制代码 
  1. Only the following datatypes can be automatically detected: boolean, date, double, long, object, string. It also accepts * to match all datatypes.

动态映射
https://www.elastic.co/guide/...

动态模板（注意看match和match_mapping_type）
https://www.elastic.co/guide/...

映射属性
https://www.elastic.co/guide/...

然后我们给出了一个全新的模板

全新索引模板

这个模板里面只针对特殊的属性进行了设置，其他的都是通过动态模板扩展的，下面看下效果。

</>复制代码 
  1. PUT _template/logstash-dev
  2. {
  3.     "index_patterns": "logstash-dev*",
  4.     "aliases": {},
  5.     "order":1,
  6.     "mappings": {
  7.       "logs": {
  8.         "dynamic_templates": [
  9.           {
  10.             "string_fields": {
  11.               "match": "*",
  12.               "match_mapping_type": "string",
  13.               "mapping": {
  14.                 "fields": {
  15.                   "keyword": {
  16.                     "ignore_above": 256,
  17.                     "type": "keyword"
  18.                   }
  19.                 },
  20.                 "norms": false,
  21.                 "type": "text"
  22.               }
  23.             }
  24.           }
  25.         ],
  26.         "properties": {
  27.           "@timestamp": {
  28.             "type": "date"
  29.           },
  30.           "@version": {
  31.             "type": "keyword"
  32.           },
  33.           "geoip": {
  34.             "dynamic": "true",
  35.             "properties": {
  36.               "ip": {
  37.                 "type": "ip"
  38.               },
  39.               "latitude": {
  40.                 "type": "half_float"
  41.               },
  42.               "location": {
  43.                 "type": "geo_point"
  44.               },
  45.               "longitude": {
  46.                 "type": "half_float"
  47.               }
  48.             }
  49.           }
  50.         }
  51.       }
  52.     },
  53.     "settings": {
  54.       "index": {
  55.         "number_of_shards": "1"
  56.       }
  57.     }
  58. }

我们上面说了，全都进行了strval转义，为什么呢？因为动态模板里面，匹配的是string类型，如果我们写入的是一个int类型，那么就不会进行自动扩展了。试验后表明，会生成一个int类型的message字段，这样是不合理的。最终生成的效果是如下图的。

分割线 =============================

快速搭建elk日志收集 第一版本

新项目短时间来实现日志采集。

资源

一台8G 4核 500G硬盘 服务器

部署方案

项目部署在4台服务器，每台服务器通过phpsdk直接写入一台es服务器中。
（在本次部署中，没有使用logstash的功能）

缺点

没有使用异步队列，导致直接写入es可能会影响业务逻辑，但是目前只会在开发和测试环境使用。

组件

主要利用elasticsearch 和 kibana
要使用x-pack做安全校验，包括给kibana加入登录授权功能

Elasticsearch
https://www.elastic.co/cn/pro...

Elasticsearch-clients
这里包含的多种语言的sdk包
https://www.elastic.co/guide/...

Kibana
https://www.elastic.co/cn/pro...

Logstash
https://www.elastic.co/cn/pro...

X-pack
安装流程说明很详细，秘钥生成后记得保存下，并且加入x-pack后，kibana和elasticsearch的通讯，需要修改配置文件。另外phpsdk也需要加入秘钥，后面说明。
https://www.elastic.co/cn/pro...

配置 创建索引

es的模板超级复杂的，所以我们要利用标准的现有的模板，需要从logstash中提取一个。
解压下载的logstash-6.1.2.tar，执行搜索命令

</>复制代码 
  1. $ find ./  -name "elasticsearch-template*"
  2. ./vendor/bundle/jruby/2.3.0/gems/logstash-output-elasticsearch-9.0.2-java/lib/logstash/outputs/elasticsearch/elasticsearch-template-es2x.json
  3. ./vendor/bundle/jruby/2.3.0/gems/logstash-output-elasticsearch-9.0.2-java/lib/logstash/outputs/elasticsearch/elasticsearch-template-es5x.json
  4. ./vendor/bundle/jruby/2.3.0/gems/logstash-output-elasticsearch-9.0.2-java/lib/logstash/outputs/elasticsearch/elasticsearch-template-es6x.json

会发现有2x 5x 6x三个模板，这里我们选择6x，要根据你的es版本来选择。
然后创建索引，索引要在数据写入之前创建，因为要给每个字段设置类型。

创建索引模板

https://www.elastic.co/guide/...

按照文档的方式，将获取到的6x通过curl的方式写入到es

Es开启外网访问

因为默认的es配置是开启了localhost:9200端口，用于执行RESTFUL，但是本次我们采用php-sdk的方式，直接写入es，就要求每台业务服务器，都能访问到es。

</>复制代码 
  1. # ---------------------------------- Network -----------------------------------
  2. #
  3. # Set the bind address to a specific IP (IPv4 or IPv6):
  4. # 修改此处的host配置为0.0.0.0，这样所有的请求都可以接入进来
  5. network.host: 0.0.0.0
  6. #
  7. # Set a custom port for HTTP:
  8. #
  9. #http.port: 9200
使用client sdk

下载地址
https://www.elastic.co/guide/...

</>复制代码 
  1. try {
  2.     $hosts = [
  3.         // 这个地方要填写x-pack分配的密码和用户名
  4.         "http://{用户名}:{密码}@192.168.1.11:9200",       // HTTP Basic Authentication
  5.         // "http://user2:pass2@other-host.com:9200" // Different credentials on different host
  6.     ];
  7.     $client = ClientBuilder::create()->setHosts($hosts)->build();
  8.     $position = YnUtil::getPosition();
  9.     $params = [
  10.         "index" => "logstash-yn-" . date("Ymd"),
  11.         // elastic6版本有个bug，每个索引只能有一个type类型
  12.         "type"  => "xxxx",
  13.         "id"    => md5(rand(0, 999999999)) . (microtime(true) * 1000),
  14.         "body"  => [
  15.             // 索引创建好后，写入数据一定要注意类型，不然会报错，我这里都会进行格式化一遍
  16.             // 类别作为一个主要字段用于区分日志
  17.             "category"   => strval($category),
  18.             "appType"    => strval(YnUtil::getAppType()),
  19.             "appVersion" => strval(YnUtil::getAppVersion()),
  20.             "host"       => strval($hostInfo),
  21.             "uri"        => strval($url),
  22.             "uid"        => strval($user->getUid()),
  23.             "server"     => strval(gethostname()),
  24.             "geoip"      => [
  25.                 "ip"       => strval($ip),
  26.                 // 这个很重要，可以实现geo可视化图形
  27.                 "location" => [
  28.                     "lat" => floatval($position["latitude"]),
  29.                     "lon" => floatval($position["longitude"]),
  30.                 ],
  31.             ],
  32.             "userAgent"  => strval($userAgent),
  33.             "message"    => Json::encode($message),
  34.             // 这里一定要写入毫秒时间戳
  35.             "@timestamp" => intval(microtime(true) * 1000),
  36.         ],
  37.     ];
  38.     $client->index($params);
  39. } catch (Exception $e) {
  40. }

安装好x-pack后，sdk也要配置用户名和密码

日志的索引，按照日期来构建名称

elastic6版本，每个索引只能有一个type类型，这是一个bug

geoip是logstash的模板中定义的字段，可以实现geo可视化（稍后解释模板）

@timestamp 这里一定要毫秒时间戳

要捕获报错日志，不要影响业务逻辑

模板说明

官方说明
https://www.elastic.co/guide/...

我们这里以6x作为模板

</>复制代码 
  1. {
  2.     // 将这个模板应用于所有以 logstash- 为起始的索引。
  3.     "template": "logstash-*",
  4.     "version": 60001,
  5.     "settings": {
  6.         "index.refresh_interval": "5s"
  7.     },
  8.     "mappings": {
  9.         "_default_": {
  10.             //  动态模板说明，很重要，配置了动态模板后，我们可以添加任意字段
  11.             // https://www.elastic.co/guide/en/elasticsearch/reference/current/dynamic-templates.html
  12.             "dynamic_templates": [{
  13.                 // 信息字段 官方说这里可以自定义 The template name can be any string value.
  14.                 "message_field": {
  15.                     "path_match": "message",
  16.                     "match_mapping_type": "string",
  17.                     "mapping": {
  18.                         "type": "text",
  19.                         "norms": false
  20.                     }
  21.                 }
  22.             }, {
  23.                 // 字符串字段说明
  24.                 "string_fields": {
  25.                     // 匹配所有
  26.                     "match": "*",
  27.                     // 并且字段类型是string的
  28.                     "match_mapping_type": "string",
  29.                     // 
  30.                     "mapping": {
  31.                         "type": "text",
  32.                         // 这里应该是和受欢迎程度评分相关
  33.                         "norms": false,
  34.                         "fields": {
  35.                             "keyword": {
  36.                                 "type": "keyword",
  37.                                 "ignore_above": 256
  38.                             }
  39.                         }
  40.                     }
  41.                 }
  42.             }],
  43.             // 定义好的属性说明
  44.             "properties": {
  45.                 // 时间字段，这个很重要，不然kibana不会出现时间相关的查询控件
  46.                 "@timestamp": {
  47.                     "type": "date"
  48.                 },
  49.                 "@version": {
  50.                     "type": "keyword"
  51.                 },
  52.                 // 这个可以只写入properies里面的任意一个字段
  53.                 "geoip": {
  54.                     "dynamic": true,
  55.                     "properties": {
  56.                         "ip": {
  57.                             "type": "ip"
  58.                         },
  59.                         // 我只是用了这个location
  60.                         "location": {
  61.                             "type": "geo_point"
  62.                         },
  63.                         "latitude": {
  64.                             "type": "half_float"
  65.                         },
  66.                         "longitude": {
  67.                             "type": "half_float"
  68.                         }
  69.                     }
  70.                 }
  71.             }
  72.         }
  73.     }
  74. }
遇到的问题 索引创建太复杂怎么办？

使用logstash内置的模板

添加了时间字段后，创建数据不显示？

检查是否是毫秒时间戳

安装x-pack后，clientsdk向es写入数据报错，提示校验失败？

需要在链接中添加用户名和密码

类型和模板不匹配导致错误？

写入类型，一定要和索引模板中定义的一致，不然肯定报错！

同一个索引，添加多个type报错？

elasticsearch6的 bug，官方承诺在7进行修复

kibana加入登录验证？

安装x-pack吧

如何保证高可用，无人值守？

使用supervisor

如何清理elasticsearch老数据？

我的方案是：定时脚本来清理7天之前的索引DELETE logstash-xxxx

如何快速执行curl查询？

在kibana中有一个Dev Tools 可以执行curl，并且看到结果

如何格式化kibana的日期格式？

在kibana菜单的Management->Index Patterns中可以管理

结束

不详细的地方，可以留言

GPU云服务器 云服务器 elk日志收集 kafka 日志收集 搭建elk日志服务器 elk日志服务器搭建

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/28196.html

相关文章

  • ELK结合logback搭建日志中心

    摘要：日志监控和分析在保障业务稳定运行时，起到了很重要的作用。本文搭建的的是一个分布式的日志收集和分析系统。对于队列上的这些未处理的日志，有不同的几台进行接收和分析。再由统一的进行日志界面的展示。如等配置文件可以配置，等日志报表可视化熟练 ELK简介ELKStack即Elasticsearch + Logstash + Kibana。日志监控和分析在保障业务稳定运行时，起到了很重要的作用。比...

    tracy 评论0 收藏0

发表评论

登陆后可评论

0条评论

妤锋シ

男|高级讲师
我要私信

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<