安科网

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

jiaomrswang

jiaomrswang

2019-06-26

关注 关注

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

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

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

方案

业务层 >> kafka队列 >> logstash 消费 >> elasticsearch

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

优点

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

缺点

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

搭建

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

  1. logstash内置一套模板
  2. elasticsearch 本身可以自定义一套模板

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

Logstash 配置文件说明

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

input {
    kafka {
        // 需要读取的kafka队列集群配置
        bootstrap_servers => "xxx.xxx.xxx.xxx:9092"
        // 配置的消费者的group名称，因为同一个组内的消费消息不会重复
        group_id => "logstash-group"
        // 主题配置
        topics => "kibana_log"
        // 从未消费过的偏移量开始
        auto_offset_reset =>"earliest"
    }
}

// 重要，下面单独讲
filter {
  json {
    source => "message"
  }
}

output {
    // stdout可以省略，这个是为了命令行模式下方便调试
    stdout{ codec => rubydebug }
    elasticsearch {
        // es 集群
        hosts=>"xxx.xxx.xxx.xxx:9200"
        // 重要：取消logstash自定义模板功能，进而强制使用es的内置模板
        manage_template=>false
        // 需要匹配的模板名称
        index=>"logstash-dev-%{+YYYY.MM.dd}"
    }
}

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

filter {
  json {
    source => "message"
  }
}

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

Php 实现写入

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

$position = YnUtil::getPosition();

$urlData = parse_url(\Wii::app()->request->url);
$path    = $urlData['path'] ?? '';

$params = [
    'category'   => strval($category),
    'appType'    => strval(YnUtil::getAppType()),
    'appVersion' => strval(YnUtil::getAppVersion()),
    'host'       => strval(\Wii::app()->request->hostInfo),
    'uri'        => strval(\Wii::app()->request->url),
    'uid'        => strval(\Wii::app()->user->getUid()),
    'path'       => strval($path),
    'server'     => strval(gethostname()),
    'geoip'      => [
        'ip'       => strval(\Yii::$app->request->userIP),
        'location' => [
            'lat' => floatval($position['latitude']),
            'lon' => floatval($position['longitude']),
        ],
    ],
    'userAgent'  => strval(\Wii::app()->request->userAgent),
    'message'    => is_array($message) ? Json::encode($message) : strval($message),
    // '@timestamp' => intval(microtime(true) * 1000),
];

索引模板

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

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

PUT _template/logstash-dev
{
    "index_patterns": "logstash-dev*",
    "aliases": {},
    "order":1,
    "mappings": {
      // 这里使用logs是因为logstash默认的type类型
      "logs": {
        // 动态模板
        "dynamic_templates": [
          {
            "string_fields": {
              "match": "*",
              "match_mapping_type": "string",
              "mapping": {
                "fields": {
                  "keyword": {
                    "ignore_above": 256,
                    "type": "keyword"
                  }
                },
                "norms": false,
                "type": "text"
              }
            }
          }
        ],
        // 这里对属性进行了类型设置
        "properties": {
          "@timestamp": {
            "type": "date"
          },
          "@version": {
            "type": "keyword"
          },
          "appType": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "appVersion": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "category": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "geoip": {
            "dynamic": "true",
            "properties": {
              "ip": {
                "type": "ip"
              },
              "latitude": {
                "type": "half_float"
              },
              "location": {
                "type": "geo_point"
              },
              "longitude": {
                "type": "half_float"
              }
            }
          },
          "host": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "message": {
            "type": "text",
            "norms": false
          },
          "server": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "uid": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "uri": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "path": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          },
          "userAgent": {
            "type": "text",
            "norms": false,
            "fields": {
              "keyword": {
                "type": "keyword",
                "ignore_above": 256
              }
            }
          }
        }
      }
    },
    "settings": {
      "index": {
        "number_of_shards": "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/...

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

全新索引模板

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

PUT _template/logstash-dev
{
    "index_patterns": "logstash-dev*",
    "aliases": {},
    "order":1,
    "mappings": {
      "logs": {
        "dynamic_templates": [
          {
            "string_fields": {
              "match": "*",
              "match_mapping_type": "string",
              "mapping": {
                "fields": {
                  "keyword": {
                    "ignore_above": 256,
                    "type": "keyword"
                  }
                },
                "norms": false,
                "type": "text"
              }
            }
          }
        ],
        "properties": {
          "@timestamp": {
            "type": "date"
          },
          "@version": {
            "type": "keyword"
          },
          "geoip": {
            "dynamic": "true",
            "properties": {
              "ip": {
                "type": "ip"
              },
              "latitude": {
                "type": "half_float"
              },
              "location": {
                "type": "geo_point"
              },
              "longitude": {
                "type": "half_float"
              }
            }
          }
        }
      }
    },
    "settings": {
      "index": {
        "number_of_shards": "1"
      }
    }
}

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

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

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

快速搭建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，执行搜索命令

$ find ./  -name 'elasticsearch-template*'
./vendor/bundle/jruby/2.3.0/gems/logstash-output-elasticsearch-9.0.2-java/lib/logstash/outputs/elasticsearch/elasticsearch-template-es2x.json
./vendor/bundle/jruby/2.3.0/gems/logstash-output-elasticsearch-9.0.2-java/lib/logstash/outputs/elasticsearch/elasticsearch-template-es5x.json
./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。

# ---------------------------------- Network -----------------------------------
#
# Set the bind address to a specific IP (IPv4 or IPv6):
# 修改此处的host配置为0.0.0.0，这样所有的请求都可以接入进来
network.host: 0.0.0.0
#
# Set a custom port for HTTP:
#
#http.port: 9200

使用client sdk

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

try {
    $hosts = [
        // 这个地方要填写x-pack分配的密码和用户名
        'http://{用户名}:{密码}@192.168.1.11:9200',       // HTTP Basic Authentication
        // 'http://user2:[email protected]:9200' // Different credentials on different host
    ];

    $client = ClientBuilder::create()->setHosts($hosts)->build();

    $position = YnUtil::getPosition();

    $params = [
        'index' => 'logstash-yn-' . date('Ymd'),
        // elastic6版本有个bug，每个索引只能有一个type类型
        'type'  => 'xxxx',
        'id'    => md5(rand(0, 999999999)) . (microtime(true) * 1000),
        'body'  => [
            // 索引创建好后，写入数据一定要注意类型，不然会报错，我这里都会进行格式化一遍
            // 类别作为一个主要字段用于区分日志
            'category'   => strval($category),
            'appType'    => strval(YnUtil::getAppType()),
            'appVersion' => strval(YnUtil::getAppVersion()),
            'host'       => strval($hostInfo),
            'uri'        => strval($url),
            'uid'        => strval($user->getUid()),
            'server'     => strval(gethostname()),
            'geoip'      => [
                'ip'       => strval($ip),
                // 这个很重要，可以实现geo可视化图形
                'location' => [
                    'lat' => floatval($position['latitude']),
                    'lon' => floatval($position['longitude']),
                ],
            ],
            'userAgent'  => strval($userAgent),
            'message'    => Json::encode($message),
            // 这里一定要写入毫秒时间戳
            '@timestamp' => intval(microtime(true) * 1000),
        ],
    ];

    $client->index($params);

} catch (\Exception $e) {

}
  • 安装好x-pack后，sdk也要配置用户名和密码
  • 日志的索引，按照日期来构建名称
  • elastic6版本，每个索引只能有一个type类型，这是一个bug
  • geoip是logstash的模板中定义的字段，可以实现geo可视化（稍后解释模板）
  • @timestamp 这里一定要毫秒时间戳
  • 要捕获报错日志，不要影响业务逻辑

模板说明

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

我们这里以6x作为模板

{
    // 将这个模板应用于所有以 logstash- 为起始的索引。
    "template": "logstash-*",
    "version": 60001,
    "settings": {
        "index.refresh_interval": "5s"
    },
    "mappings": {
        "_default_": {
            //  动态模板说明，很重要，配置了动态模板后，我们可以添加任意字段
            // https://www.elastic.co/guide/en/elasticsearch/reference/current/dynamic-templates.html
            "dynamic_templates": [{
                // 信息字段 官方说这里可以自定义 The template name can be any string value.
                "message_field": {
                    "path_match": "message",
                    "match_mapping_type": "string",
                    "mapping": {
                        "type": "text",
                        "norms": false
                    }
                }
            }, {
                // 字符串字段说明
                "string_fields": {
                    // 匹配所有
                    "match": "*",
                    // 并且字段类型是string的
                    "match_mapping_type": "string",
                    // 
                    "mapping": {
                        "type": "text",
                        // 这里应该是和受欢迎程度评分相关
                        "norms": false,
                        "fields": {
                            "keyword": {
                                "type": "keyword",
                                "ignore_above": 256
                            }
                        }
                    }
                }
            }],
            // 定义好的属性说明
            "properties": {
                // 时间字段，这个很重要，不然kibana不会出现时间相关的查询控件
                "@timestamp": {
                    "type": "date"
                },
                "@version": {
                    "type": "keyword"
                },
                // 这个可以只写入properies里面的任意一个字段
                "geoip": {
                    "dynamic": true,
                    "properties": {
                        "ip": {
                            "type": "ip"
                        },
                        // 我只是用了这个location
                        "location": {
                            "type": "geo_point"
                        },
                        "latitude": {
                            "type": "half_float"
                        },
                        "longitude": {
                            "type": "half_float"
                        }
                    }
                }
            }
        }
    }
}

遇到的问题

索引创建太复杂怎么办？

使用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中可以管理

结束

不详细的地方，可以留言

logstash kafka elk搭建

jiaomrswang

jiaomrswang

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

logstash_output_kafka:Mysql同步Kafka深入详解

实际业务场景中，会遇到基础数据存在Mysql中，实时写入数据量比较大的情景。迁移至kafka是一种比较好的业务选型方案。如果需要同步历史全量数据+实时更新数据，建议使用logstash。实际上，核心logstash的同步原理的掌握，有助于大家理解类似的各种

weikaixxxxxx 2020-08-01

浅尝 Elastic Stack (五) Logstash + Beats + Kafka

在 Elasticsearch、Kibana、Beats 安装 中讲到推荐架构：。本文基于 Logstash + Beats 读取 Spring Boot 日志 将其改为上述架构

goodstudy 2020-06-05

WINDOW 搭建 ELK 2.4.0

文本是在window下搭建单机版的ELK,ES用到的版本是2.4.0，logstash用的版本是7.6.2，kibana用4.6.2. hosts => "127.0.0.1" #注意此为es的机器地址。如果es的端口不是9200

狐狸小七 2020-05-10

SpringBoot+ELK日志系统搭建

“ELK”是三个开源项目的首字母缩写，这三个项目分别是：Elasticsearch、Logstash 和 Kibana。Elasticsearch 是一个搜索和分析引擎。Logstash 是服务器端数据处理管道，能够同时从多个来源采集数据，转换数据，然后将

技术与更多 2020-04-15

Filebeat简介

Filebeat附带预构建的模块，这些模块包含收集、解析、充实和可视化各种日志文件格式数据所需的配置，每个Filebeat模块由一个或多个文件集组成，这些文件集包含摄取节点管道、Elasticsearch模板、Filebeat勘探者配置和Kibana仪表盘

goodstudy 2020-08-19

ES坑之logstash配置文件

　　parameters => { "sql_last_value" => "UpdateTime" }. 　　statement => "SELECT * FROM t WHERE t

liulin0 2020-07-26

Elasticsearch 的安装和使用（1）

说起ELK相信大家都不陌生，他可不是一个软件，是三个软件 logstash，elasticsearch，Kibana 组成的可以打造大规模日志实时处理的系统。Logstash是一个日志收集、过滤、转发的中间件，主要负责将各条业务线的各类日志统一收集、过滤后

AFei00 2020-07-26

Logstash Oracle同步设置

# 是否清除sql_last_value的记录，需要增量同步时此字段必须为false；# 同步频率，默认每分钟同步一次；where updated_at>:sql_last_value and updated_at<sysdate order

娜娜 2020-07-20

filebeat配置文件

#input设置，支持Docker,Container,HTTP JSON,Log,Kafka,MQTT,NetFlow,Redis,TCP,DCP,Syslog,Stdin. #output设置，可以output到kafka,logstash,elast

偏头痛杨 2020-07-18

logstash配置文件

# How many events to retrieve from inputs before sending to filters+workers. #启用持久队列后，在强制检查点之前的最大书面事件数，默认1024. # If using queue.

winxcoder 2020-07-18

logstash

　　logstash是一个日志转化系统，用户通过定义一个input，filter，和一个output配置来完成日志的收集和存储工作。　　array match =>[ "/var/log/messages", "/var

superviser000 2020-06-28

linux安装logstash7.6.1及搭建简单ELK--logstash收集nginx日志

进入解压后的logstash的bin目录，执行启动命令：./bin/logstash -e ‘input { stdin{} } output{ stdout { codec => rubydebug }}‘。[2020-06-25T11:17:36

liulin0 2020-06-25

ELK组件

　　Filebeat 是用于转发和集中日志数据的轻量级传输程序。Filebeat 可以监听指定的日志文件或位置，从中收集日志事件并将其转发到 Elasticsearch 或 Logstash 进行索引。Filebeat 为每个找到的日志启动 Harvest

liulin0 2020-06-16

超强干货！通过filebeat、logstash、rsyslog 几种方式采集 nginx 日志

由于nginx功能强大，性能突出，越来越多的web应用采用nginx作为http和反向代理的web服务器。而nginx的访问日志不管是做用户行为分析还是安全分析都是非常重要的数据源之一。如何有效便捷的采集nginx的日志进行有效的分析成为大家关注的问题。本

wangfengqingyang 2020-06-11

NSSM-将服务变为window service

nssm install logstash， 弹出如下界面。Startup directory : 填写启动文件目录。Dependencies : 填写此服务启动需要依赖哪个服务

cullinans 2020-06-09

logstash的output插件

和之前 inputs/stdin 插件一样，outputs/stdout 插件也是最基础和简单的输出插件。同样在这里简单介绍一下，作为输出插件的一个共性了解。Logstash 为输出做了多线程的准备。codec 的作用在之前已经讲过。可能除了 codecs

winxcoder 2020-06-08

ElasticSearch stack 6.5.4下载地址

ElasticSearch：windows：https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.5.4.ziphttps://artifacts.elastic.co/d

citic 2020-06-06

ELK-概念

ELK是Elasticsearch、Logstash、Kibana三大开源框架首字母大写简称。市面上也被成为Elastic Stack。其中Elasticsearch是一个基于Lucene、分布式、通过Restful方式进行交互的近实时搜索平台框架。像类似

superviser000 2020-05-30

logstash 命令行配置选项

Logstash有以下标记，你可以使用--help标志来显示此信息。指定此Logstash实例的名称，如果没有赋值，它将默认为当前主机名。从特定的文件或目录加载Logstash配置，如果给定一个目录，则该目录中的所有文件将以字典顺序连接，然后作为一个配置文

yuzhongdelei0 2020-05-15

基于阿里云Elasticsearch的日志采集分析展示

因业务需求，需要将阿里云入口nginx访问日志进行收集统计，最原始的办法当然是将各个nginx服务器每个都统计一遍然后进行累加，这样做很麻烦不可取，鉴于我们之前已经购买过阿里云Elasticsearch服务，就想着把nginx日志采集到阿里云elastic

wzhwangzhuohui 2020-05-12
jiaomrswang

jiaomrswang

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号