logstash

logstash

logstash配置log4j控制台无反应

Logstashtiantian666 回复了问题 • 3 人关注 • 2 个回复 • 1318 次浏览 • 4 小时前 • 来自相关话题

Logstash 读取SQL数据库时分页问题

回复

Logstashcxzy888 发起了问题 • 1 人关注 • 0 个回复 • 83 次浏览 • 2019-11-02 22:47 • 来自相关话题

为什么有些列在mapping里面类型设置为keyword,导入数据之后类型自动变成了text?

Elasticsearchzqc0512 回复了问题 • 2 人关注 • 1 个回复 • 98 次浏览 • 2019-10-31 08:45 • 来自相关话题

logstash同步多张相同表结构的数据表到同一个索引

Logstashzm 回复了问题 • 4 人关注 • 4 个回复 • 708 次浏览 • 2019-10-20 17:09 • 来自相关话题

x-pack监控logstash机器数量只能显示3个被监控的logstash机器

LogstashNoLieNoGain 回复了问题 • 3 人关注 • 6 个回复 • 1099 次浏览 • 2019-10-15 23:15 • 来自相关话题

logstash 如何截取指定字符以前的字符串作为output的文件名来用

Logstashdoom 回复了问题 • 3 人关注 • 2 个回复 • 145 次浏览 • 2019-09-30 07:55 • 来自相关话题

Logstash 能不能把常用的参数或经常变换的参数给提取到某一个文件里面去

Logstashtacsklet 回复了问题 • 4 人关注 • 3 个回复 • 144 次浏览 • 2019-09-26 10:53 • 来自相关话题

logstash 运行一段时间后,自动停掉

Logstashounijiangwang 回复了问题 • 6 人关注 • 5 个回复 • 4023 次浏览 • 2019-09-20 14:33 • 来自相关话题

Logstash能否支持这一需求:监听Nginx日志抽取出企业ID然后对该企业的浏览量加一

回复

Logstashzhugw 发起了问题 • 1 人关注 • 0 个回复 • 232 次浏览 • 2019-09-11 21:46 • 来自相关话题

谁能给我一个template demo啊

ElasticsearchJackshufu 回复了问题 • 5 人关注 • 3 个回复 • 5097 次浏览 • 2019-09-08 00:36 • 来自相关话题

Logstash-output-kafka 导致CPU使用率过高

Logstashlaoyang360 回复了问题 • 2 人关注 • 1 个回复 • 387 次浏览 • 2019-08-30 23:07 • 来自相关话题

logstash索引库如何建立?

回复

Logstashhelinyu 发起了问题 • 1 人关注 • 0 个回复 • 351 次浏览 • 2019-08-15 11:37 • 来自相关话题

Logstash迁移ES数据到ES的疑问,求大神解答

Logstashbellengao 回复了问题 • 2 人关注 • 1 个回复 • 405 次浏览 • 2019-08-15 10:44 • 来自相关话题

docker-compose启动elk失败,elasticsearch启动后仅能成功启动logstash或kibana其中一个,另一个卡启动

Logstashyouku 回复了问题 • 2 人关注 • 2 个回复 • 1037 次浏览 • 2019-07-25 09:23 • 来自相关话题

logstash获取数据的时候,能否指定数据类型?

回复

Logstashsailershen 回复了问题 • 1 人关注 • 1 个回复 • 456 次浏览 • 2019-07-19 13:47 • 来自相关话题

条新动态, 点击查看
kennywu76

kennywu76 回答了问题 • 2014-12-02 18:12 • 2 个回复 不感兴趣

logstash获取时间的问题

赞同来自:

产生这个错误的原因是系统的locale设置不是english,因此logstash用到的joda无法理解Dec这样的月份格式。

解决办法为在date filter里加一个locale=>"en"的设置:
filter {
d... 显示全部 »
产生这个错误的原因是系统的locale设置不是english,因此logstash用到的joda无法理解Dec这样的月份格式。

解决办法为在date filter里加一个locale=>"en"的设置:
filter {
date {
locale => "en"
..............
}
}
can

can 回答了问题 • 2014-12-23 19:42 • 3 个回复 不感兴趣

请教个关于logstash output 到es protocol的问题

赞同来自:

感谢@wood大神的指点,我这里总结下哈!
理论上来说,transport没有http的header带来的额外处理开销,所以速度应该要比http快。
node protocol和transport protocol的区别,主要就是node的话,logstash... 显示全部 »
感谢@wood大神的指点,我这里总结下哈!
理论上来说,transport没有http的header带来的额外处理开销,所以速度应该要比http快。
node protocol和transport protocol的区别,主要就是node的话,logstash会成为一个ES node加入集群,理论上性能比transport protocol更好,因为数据少一层转发。但是集群规模大了后,node protocol有一定副作用。 因为每个node上都有集群的状态信息,集群大了后,任何集群状态信息的改变都会在所有node同步,node数量越多,同步的代价越高
关于ES nodes过多可能导致的集群水平扩展问题可以参照这里:
http://www.elasticsearch.org/guide/en/elasticsearch/guide/current/finite-scale.html
ES的transport protocol支持多个host,只是logstash的插件不支持,官网java client api有demo,依葫芦画瓢,改下logstash的ES插件就可以了:
http://www.elasticsearch.org/guide/en/elasticsearch/client/java-api/current/client.html
下面是http与transport比较的讨论帖,有兴趣的也可以看看,多了解了解:
http://elasticsearch-users.115913.n3.nabble.com/TransportClient-vs-HTTP-TransportClient-advantages-td4023014.html
差不多就这些了,希望对后面的同学有帮助!
hexiaohong

hexiaohong 回答了问题 • 2018-05-16 16:08 • 7 个回复 不感兴趣

logstash导入mysql上亿级别数据的效率问题

赞同来自:

我也遇到你同样问题,可以这么解决:
1.给updated_ts时间字段加上索引。
2.分批处理原则
(1)你的SQL:每批处理100000个
SELECT a.party_id AS id ,a.* FROM PARTY_ALL_1 a WHERE a.upd... 显示全部 »
我也遇到你同样问题,可以这么解决:
1.给updated_ts时间字段加上索引。
2.分批处理原则
(1)你的SQL:每批处理100000个
SELECT a.party_id AS id ,a.* FROM PARTY_ALL_1 a WHERE a.updated_ts > '2011-11-17 13:23:58' order by a.updated_ts asc LIMIT 100000;
 
(2)logstash分页的时候每次处理50000个
SELECT * FROM (SELECT a.party_id AS id ,a.* FROM PARTY_ALL_1 a WHERE a.updated_ts > '2011-11-17 13:23:58' order by a.updated_ts asc LIMIT 100000) AS `t1` LIMIT 50000 OFFSET 0;
 
SELECT * FROM (SELECT a.party_id AS id ,a.* FROM PARTY_ALL_1 a WHERE a.updated_ts > '2011-11-17 13:23:58' order by a.updated_ts asc LIMIT 100000) AS `t1` LIMIT 50000 OFFSET 50000;
 
(3)处理两次就写一个最后一条记录的updated_ts时间到指定文件。下一个定时任务启动的时候进行循环处理就行,因为每批处理updated_ts都会改变
 
延迟不用担心会很小的,建议采集还是用用beat,如果数据量大,就在beat和logstash中间加kafka
延迟不用担心会很小的,建议采集还是用用beat,如果数据量大,就在beat和logstash中间加kafka

logstash配置log4j控制台无反应

回复

Logstashtiantian666 回复了问题 • 3 人关注 • 2 个回复 • 1318 次浏览 • 4 小时前 • 来自相关话题

Logstash 读取SQL数据库时分页问题

回复

Logstashcxzy888 发起了问题 • 1 人关注 • 0 个回复 • 83 次浏览 • 2019-11-02 22:47 • 来自相关话题

为什么有些列在mapping里面类型设置为keyword,导入数据之后类型自动变成了text?

回复

Elasticsearchzqc0512 回复了问题 • 2 人关注 • 1 个回复 • 98 次浏览 • 2019-10-31 08:45 • 来自相关话题

logstash同步多张相同表结构的数据表到同一个索引

回复

Logstashzm 回复了问题 • 4 人关注 • 4 个回复 • 708 次浏览 • 2019-10-20 17:09 • 来自相关话题

x-pack监控logstash机器数量只能显示3个被监控的logstash机器

回复

LogstashNoLieNoGain 回复了问题 • 3 人关注 • 6 个回复 • 1099 次浏览 • 2019-10-15 23:15 • 来自相关话题

logstash 如何截取指定字符以前的字符串作为output的文件名来用

回复

Logstashdoom 回复了问题 • 3 人关注 • 2 个回复 • 145 次浏览 • 2019-09-30 07:55 • 来自相关话题

Logstash 能不能把常用的参数或经常变换的参数给提取到某一个文件里面去

回复

Logstashtacsklet 回复了问题 • 4 人关注 • 3 个回复 • 144 次浏览 • 2019-09-26 10:53 • 来自相关话题

logstash 运行一段时间后,自动停掉

回复

Logstashounijiangwang 回复了问题 • 6 人关注 • 5 个回复 • 4023 次浏览 • 2019-09-20 14:33 • 来自相关话题

Logstash能否支持这一需求:监听Nginx日志抽取出企业ID然后对该企业的浏览量加一

回复

Logstashzhugw 发起了问题 • 1 人关注 • 0 个回复 • 232 次浏览 • 2019-09-11 21:46 • 来自相关话题

谁能给我一个template demo啊

回复

ElasticsearchJackshufu 回复了问题 • 5 人关注 • 3 个回复 • 5097 次浏览 • 2019-09-08 00:36 • 来自相关话题

Logstash-output-kafka 导致CPU使用率过高

回复

Logstashlaoyang360 回复了问题 • 2 人关注 • 1 个回复 • 387 次浏览 • 2019-08-30 23:07 • 来自相关话题

logstash索引库如何建立?

回复

Logstashhelinyu 发起了问题 • 1 人关注 • 0 个回复 • 351 次浏览 • 2019-08-15 11:37 • 来自相关话题

Logstash迁移ES数据到ES的疑问,求大神解答

回复

Logstashbellengao 回复了问题 • 2 人关注 • 1 个回复 • 405 次浏览 • 2019-08-15 10:44 • 来自相关话题

docker-compose启动elk失败,elasticsearch启动后仅能成功启动logstash或kibana其中一个,另一个卡启动

回复

Logstashyouku 回复了问题 • 2 人关注 • 2 个回复 • 1037 次浏览 • 2019-07-25 09:23 • 来自相关话题

logstash获取数据的时候,能否指定数据类型?

回复

Logstashsailershen 回复了问题 • 1 人关注 • 1 个回复 • 456 次浏览 • 2019-07-19 13:47 • 来自相关话题

logstash input插件开发

Logstashbellengao 发表了文章 • 0 个评论 • 924 次浏览 • 2018-12-10 19:56 • 来自相关话题

logstash作为一个数据管道中间件,支持对各种类型数据的采集与转换,并将数据发送到各种类型的存储库,比如实现消费kafka数据并且写入到Elasticsearch, 日志文件同步到对象存储S3等,mysql数据同步到Elasticsearch等。

logstash内部主要包含三个模块:

* input: 从数据源获取数据
* filter: 过滤、转换数据
* output: 输出数据

不同类型的数据都可以通过对应的input-plugin, output-plugin完成数据的输入与输出。如需要消费kafka中的数据并写入到Elasticsearch中,则需要使用logstash的kafka-input-plugin完成数据输入,logstash-output-elasticsearch完成数据输出。如果需要对输入数据进行过滤或者转换,比如根据关键词过滤掉不需要的内容,或者时间字段的格式转换,就需要又filter-plugin完成了。

logstash的input插件目前已经有几十种了,支持大多数比较通用或开源的数据源的输入。但如果公司内部开发的数据库或其它存储类的服务不能和开源产品在接口协议上兼容,比如腾讯自研的消息队列服务CMQ不依赖于其它的开源消息队列产品,所以不能直接使用logstash的logstash-input-kafka或logstash-input-rabbitmq同步CMQ中的数据;腾讯云对象存储服务COS, 在鉴权方式上和AWS的S3存在差异,也不能直接使用logstash-input-s3插件从COS中读取数据,对于这种情况,就需要自己开发logstash的input插件了。

本文以开发logstash的cos input插件为例,介绍如何开发logstash的input插件。

logstash官方提供了有个简单的input plugin example可供参考: https://github.com/logstash-plugins/logstash-input-example/

环境准备

logstash使用jruby开发,首先要配置jruby环境:

  1. 安装rvm:

    rvm是一个ruby管理器,可以安装并管理ruby环境,也可以通过命令行切换到不同的ruby版本。

    gpg --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 7D2BAF1CF37B13E2069D6956105BD0E739499BDB
    \curl -sSL https://get.rvm.io | bash -s stable
    source /etc/profile.d/rvm.sh
  2. 安装jruby

    rvm install jruby
    rvm use jruby
  3. 安装包管理工具bundle和测试工具rspec

    gem install bundle
    gem install rspec

从example开始

  1. clone logstash-input-example

    git clone https://github.com/logstash-plugins/logstash-input-example.git
  2. 将clone出来的logstash-input-example源码copy到logstash-input-cos目录,并删除.git文件夹,目的是以logstash-input-example的源码为参考进行开发,同时把需要改动名称的地方修改一下:

    mv logstash-input-example.gemspec logstash-input-cos.gemspec
    mv lib/logstash/inputs/example.rb lib/logstash/inputs/cos.rb
    mv spec/inputs/example_spec.rb spec/inputs/cos_spec.rb
  3. 建立的源码目录结构如图所示:

其中,重要文件的作用说明如下:

  • cos.rb: 主文件,在该文件中编写logstash配置文件的读写与源数据获取的代码,需要继承LogStash::Inputs::Base基类
  • cos_spec.rb: 单元测试文件,通过rspec可以对cos.rb中的代码进行测试
  • logstash-input-cos.gemspec: 类似于maven中的pom.xml文件,配置工程的版本、名称、licene,包依赖等,通过bundle命令可以下载依赖包

配置并下载依赖

因为腾讯云COS服务没有ruby sdk, 因为只能依赖其Java sdk进行开发,首先添加对cos java sdk的依赖。在logstash-input-cos.gemspec中Gem dependencies配置栏中增加以下内容:

# Gem dependencies
  s.requirements << "jar 'com.qcloud:cos_api', '5.4.4'"
  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"
  s.add_runtime_dependency 'logstash-codec-plain'
  s.add_runtime_dependency 'stud', '>= 0.0.22'
  s.add_runtime_dependency 'jar-dependencies'
  s.add_development_dependency 'logstash-devutils', '1.3.6'

相比logstash-input-example.gemspec,增加了对com.qcloud:cos_api包以及jar-dependencies包的依赖,jar-dependencies用于在ruby环境中管理jar包,并且可以跟踪jar包的加载状态。

然后,在logstash-input-cos.gemspec中增加配置:

s.platform = 'java'

这样可以成功下载java依赖包,并且可以在ruby代码中直接调用java代码。

最后,执行以下命令下载依赖:

bundle install

编写代码

logstash-input-cos的代码逻辑其实比较简单,主要是通过执行定时任务,调用cos java sdk中的listObjects方法,获取到指定bucket里的数据,并在每次定时任务执行结束后设置marker保存在本地,再次执行时从marker位置获取数据,以实现数据的增量同步。

jar包的引用

因为要调用cos java sdk中的代码,先引用该jar包:

require 'cos_api-5.4.4.jar'
java_import com.qcloud.cos.COSClient;
java_import com.qcloud.cos.ClientConfig;
java_import com.qcloud.cos.auth.BasicCOSCredentials;
java_import com.qcloud.cos.auth.COSCredentials;
java_import com.qcloud.cos.exception.CosClientException;
java_import com.qcloud.cos.exception.CosServiceException;
java_import com.qcloud.cos.model.COSObjectSummary;
java_import com.qcloud.cos.model.ListObjectsRequest;
java_import com.qcloud.cos.model.ObjectListing;
java_import com.qcloud.cos.region.Region;

读取配置文件

logstash配置文件读取的代码如图所示:

config_name为cos,其它的配置项读取代码按照ruby的代码规范编写,添加类型校验与默认值,就可以从以下配置文件中读取配置项:

input {
    cos {
        "endpoint" => "cos.ap-guangzhou.myqcloud.com"
        "access_key_id" => "*****"
        "access_key_secret" => "****"
        "bucket" => "******"
        "region" => "ap-guangzhou"
        "appId" => "**********"
        "interval" => 60
    }
}

output {
    stdout {
        codec=>rubydebug
    }
}

实现register方法

logstash input插件必须实现另个方法:register 和run

register方法类似于初始化方法,在该方法中可以直接使用从配置文件读取并赋值的变量,完成cos client的初始化,代码如下:

    # 1 初始化用户身份信息(appid, secretId, secretKey)
    cred = com.qcloud.cos.auth.BasicCOSCredentials.new(@access_key_id, @access_key_secret)
    # 2 设置bucket的区域, COS地域的简称请参照 https://www.qcloud.com/document/product/436/6224
    clientConfig = com.qcloud.cos.ClientConfig.new(com.qcloud.cos.region.Region.new(@region))
    # 3 生成cos客户端
    @cosclient = com.qcloud.cos.COSClient.new(cred, clientConfig)
    # bucket名称, 需包含appid
    bucketName = @bucket + "-"+ @appId
    @bucketName = bucketName

    @listObjectsRequest = com.qcloud.cos.model.ListObjectsRequest.new()
    # 设置bucket名称
    @listObjectsRequest.setBucketName(bucketName)
    # prefix表示列出的object的key以prefix开始
    @listObjectsRequest.setPrefix(@prefix)
    # 设置最大遍历出多少个对象, 一次listobject最大支持1000
    @listObjectsRequest.setMaxKeys(1000)
    @listObjectsRequest.setMarker(@markerConfig.getMarker)

示例代码中设置了@cosclient和@listObjectRequest为全局变量, 因为在run方法中会用到这两个变量。

注意在ruby中调用java代码的方式:没有变量描述符;不能直接new Object(),而只能Object.new().

实现run方法

run方法获取数据并将数据流转换成event事件

最简单的run方法为:

def run(queue)
    Stud.interval(@interval) do
      event = LogStash::Event.new("message" => @message, "host" => @host)
      decorate(event)
      queue << event
    end # loop
  end # def run

代码说明:

  • 通过Stud ruby模块执行定时任务,interval可自定义,从配置文件中读取
  • 生成event, 示例代码生成了一个包含两个字段数据的event
  • 调用decorate()方法, 给该event打上tag,如果配置的话
  • queue<<event, 将event插入到数据管道中,发送给filter处理

logstash-input-cos的run方法实现为:

def run(queue)
    @current_thread = Thread.current
    Stud.interval(@interval) do
      process(queue)
    end
end

def process(queue)
    @logger.info('Marker from: ' + @markerConfig.getMarker)

    objectListing = @cosclient.listObjects(@listObjectsRequest)
    nextMarker = objectListing.getNextMarker()
    cosObjectSummaries = objectListing.getObjectSummaries()
    cosObjectSummaries.each do |obj|
       # 文件的路径key
       key = obj.getKey()

       if stop?
         @logger.info("stop while attempting to read log file")
         break
       end
       # 根据key获取内容
       getObject(key) { |log|
         # 发送消息
         @codec.decode(log) do |event|
           decorate(event)
           queue << event
         end
       }

       #记录 marker
       @markerConfig.setMarker(key)
       @logger.info('Marker end: ' + @markerConfig.getMarker)
    end
  end

  # 获取下载输入流
 def getObject(key, &block)
    getObjectRequest = com.qcloud.cos.model.GetObjectRequest.new(@bucketName, key)
    cosObject = @cosclient.getObject(getObjectRequest)
    cosObjectInput = cosObject.getObjectContent()
    buffered =BufferedReader.new(InputStreamReader.new(cosObjectInput))
    while (line = buffered.readLine())
      block.call(line)
    end
  end

测试代码

在spec/inputs/cos_spec.rb中增加如下测试代码:

# encoding: utf-8
require "logstash/devutils/rspec/spec_helper"
require "logstash/inputs/cos"

describe LogStash::Inputs::Cos do

  it_behaves_like "an interruptible input plugin" do
    let(:config) { {
        "endpoint" => 'cos.ap-guangzhou.myqcloud.com',
        "access_key_id" => '*',
        "access_key_secret" => '*',
        "bucket" => '*',
         "region" => 'ap-guangzhou',
         "appId" => '*',
        "interval" => 60 } }
  end
end

rspec是一个ruby测试库,通过bundle命令执行rspec:

bundle exec rspec

如果cos.rb中的代码没有语法或运行时错误,则会出现如果信息表明测试成功:

Finished in 0.8022 seconds (files took 3.45 seconds to load)
1 example, 0 failures

构建并测试input-plugin-cos

build

使用gem对input-plugin-cos插件源码进行build:

gem build logstash-input-cos.gemspec

构建完成后会生成一个名为logstash-input-cos-0.0.1-java.gem的文件

test

在logstash的解压目录下,执行一下命令安装logstash-input-cos plugin:

./bin/logstash-plugin install /usr/local/githome/logstash-input-cos/logstash-input-cos-0.0.1-java.gem

执行结果为:

Validating /usr/local/githome/logstash-input-cos/logstash-input-cos-0.0.1-java.gem
Installing logstash-input-cos
Installation successful

另外,可以通过./bin/logstash-plugin list命令查看logstash已经安装的所有input/output/filter/codec插件。

生成配置文件cos.logstash.conf,内容为:

input {
    cos {
        "endpoint" => "cos.ap-guangzhou.myqcloud.com"
        "access_key_id" => "*****"
        "access_key_secret" => "****"
        "bucket" => "******"
        "region" => "ap-guangzhou"
        "appId" => "**********"
        "interval" => 60
    }
}

output {
    stdout {
        codec=>rubydebug
    }
}

该配置文件使用腾讯云官网账号的secret_id和secret_key进行权限验证,拉取指定bucket里的数据,为了测试,将output设置为标准输出。

执行logstash:

./bin/logstash -f cos.logstash.conf

输出结果为:

Sending Logstash's logs to /root/logstash-5.6.4/logs which is now configured via log4j2.properties
[2018-07-30T19:26:17,039][WARN ][logstash.runner          ] --config.debug was specified, but log.level was not set to 'debug'! No config info will be logged.
[2018-07-30T19:26:17,048][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.4/modules/netflow/configuration"}
[2018-07-30T19:26:17,049][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.4/modules/fb_apache/configuration"}
[2018-07-30T19:26:17,252][INFO ][logstash.inputs.cos      ] Using version 0.1.x input plugin 'cos'. This plugin isn't well supported by the community and likely has no maintainer.
[2018-07-30T19:26:17,341][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
[2018-07-30T19:26:17,362][INFO ][logstash.inputs.cos      ] Registering cos input {:bucket=>"bellengao", :region=>"ap-guangzhou"}
[2018-07-30T19:26:17,528][INFO ][logstash.pipeline        ] Pipeline main started
[2018-07-30T19:26:17,530][INFO ][logstash.inputs.cos      ] Marker from:
log4j:WARN No appenders could be found for logger (org.apache.http.client.protocol.RequestAddCookies).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[2018-07-30T19:26:17,574][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-07-30T19:26:17,714][INFO ][logstash.inputs.cos      ] Marker end: access.log
{
       "message" => "77.179.66.156 - - [25/Oct/2016:14:49:33 +0200] \"GET / HTTP/1.1\" 200 612 \"-\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36\"",
      "@version" => "1",
    "@timestamp" => 2018-07-30T11:26:17.710Z
}
{
       "message" => "77.179.66.156 - - [25/Oct/2016:14:49:34 +0200] \"GET /favicon.ico HTTP/1.1\" 404 571 \"http://localhost:8080/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2840.59 Safari/537.36\"",
      "@version" => "1",
    "@timestamp" => 2018-07-30T11:26:17.711Z
}

在cos中的bucket里上传了名为access.log的nginx日志,上述输出结果中最后打印出来的每个json结构体构成一个event, 其中message消息即为access.log中每一条日志。

一文快速上手Logstash

Logstashmushao999 发表了文章 • 3 个评论 • 2007 次浏览 • 2018-11-17 11:24 • 来自相关话题

本文同步发布在腾讯云+社区Elasticsearch专栏:https://cloud.tencent.com/developer/column/4008
Elasticsearch是当前主流的分布式大数据存储和搜索引擎,可以为用户提供强大的全文本检索能力,广泛应用于日志检索,全站搜索等领域。Logstash作为Elasicsearch常用的实时数据采集引擎,可以采集来自不同数据源的数据,并对数据进行处理后输出到多种输出源,是Elastic Stack 的重要组成部分。本文从Logstash的工作原理,使用示例,部署方式及性能调优等方面入手,为大家提供一个快速入门Logstash的方式。文章最后也给出了一些深入了解Logstash的的链接,以方便大家根据需要详细了解。

Logstash简介

1 Logstash工作原理

1.1 处理过程

Logstash处理过程

如上图,Logstash的数据处理过程主要包括:Inputs, Filters, Outputs 三部分, 另外在Inputs和Outputs中可以使用Codecs对数据格式进行处理。这四个部分均以插件形式存在,用户通过定义pipeline配置文件,设置需要使用的input,filter,output, codec插件,以实现特定的数据采集,数据处理,数据输出等功能

  • (1)Inputs:用于从数据源获取数据,常见的插件如file, syslog, redis, beats 等[详细参考]
  • (2)Filters:用于处理数据如格式转换,数据派生等,常见的插件如grok, mutate, drop, clone, geoip等[详细参考]
  • (3)Outputs:用于数据输出,常见的插件如elastcisearch,file, graphite, statsd等[详细参考]
  • (4)Codecs:Codecs不是一个单独的流程,而是在输入和输出等插件中用于数据转换的模块,用于对数据进行编码处理,常见的插件如json,multiline[详细参考]

可以点击每个模块后面的_详细参考_链接了解该模块的插件列表及对应功能

1.2 执行模型:

  • (1)每个Input启动一个线程,从对应数据源获取数据
  • (2)Input会将数据写入一个队列:默认为内存中的有界队列(意外停止会导致数据丢失)。为了防止数丢失Logstash提供了两个特性: Persistent Queues:通过磁盘上的queue来防止数据丢失 Dead Letter Queues:保存无法处理的event(仅支持Elasticsearch作为输出源)
  • (3)Logstash会有多个pipeline worker, 每一个pipeline worker会从队列中取一批数据,然后执行filter和output(worker数目及每次处理的数据量均由配置确定)

2 Logstash使用示例

2.1 Logstash Hello world

第一个示例Logstash将采用标准输入和标准输出作为input和output,并且不指定filter

  • (1)下载Logstash并解压(需要预先安装JDK8)
  • (2)cd到Logstash的根目录,并执行启动命令如下:
    cd logstash-6.4.0
    bin/logstash -e 'input { stdin { } } output { stdout {} }'
  • (3)此时Logstash已经启动成功,-e表示在启动时直接指定pipeline配置,当然也可以将该配置写入一个配置文件中,然后通过指定配置文件来启动
  • (4)在控制台输入:hello world,可以看到如下输出:
    {
    "@version" => "1",
    "host" => "localhost",
    "@timestamp" => 2018-09-18T12:39:38.514Z,
    "message" => "hello world"
    }  

Logstash会自动为数据添加@version, host, @timestamp等字段

在这个示例中Logstash从标准输入中获得数据,仅在数据中添加一些简单字段后将其输出到标准输出。

2.2 日志采集

这个示例将采用Filebeat input插件(Elastic Stack中的轻量级数据采集程序)采集本地日志,然后将结果输出到标准输出

  • (1)下载示例使用的日志文件[地址],解压并将日志放在一个确定位置
  • (2)安装filebeat,配置并启动[参考]

filebeat.yml配置如下(paths改为日志实际位置,不同版本beats配置可能略有变化,请根据情况调整)

    filebeat.prospectors:
    - input\_type: log
        paths:
            - /path/to/file/logstash-tutorial.log 
    output.logstash:
        hosts: "localhost:5044"

启动命令:

    ./filebeat -e -c filebeat.yml -d "publish"
  • (3)配置logstash并启动

1)创建first-pipeline.conf文件内容如下(该文件为pipeline配置文件,用于指定input,filter, output等):

    input {
        beats {
            port => "5044"
        }
    }
    #filter {
    #}
    output {
        stdout { codec => rubydebug }
    }

codec => rubydebug用于美化输出[参考]

2)验证配置(注意指定配置文件的路径):

    ./bin/logstash -f first-pipeline.conf --config.test_and_exit

3)启动命令:

    ./bin/logstash -f first-pipeline.conf --config.reload.automatic

--config.reload.automatic选项启用动态重载配置功能

4)预期结果:

可以在Logstash的终端显示中看到,日志文件被读取并处理为如下格式的多条数据

    {
        "@timestamp" => 2018-10-09T12:22:39.742Z,
            "offset" => 24464,
          "@version" => "1",
        "input_type" => "log",
              "beat" => {
                "name" => "VM_136_9_centos",
            "hostname" => "VM_136_9_centos",
             "version" => "5.6.10"
        },
              "host" => "VM_136_9_centos",
            "source" => "/data/home/michelmu/workspace/logstash-tutorial.log",
           "message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"http://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
              "type" => "log",
              "tags" => [
            [0] "beats_input_codec_plain_applied"
        ]
    }

相对于示例2.1,该示例使用了filebeat input插件从日志中获取一行记录,这也是Elastic stack获取日志数据最常见的一种方式。另外该示例还采用了rubydebug codec 对输出的数据进行显示美化。

2.3 日志格式处理

可以看到虽然示例2.2使用filebeat从日志中读取数据,并将数据输出到标准输出,但是日志内容作为一个整体被存放在message字段中,这样对后续存储及查询都极为不便。可以为该pipeline指定一个grok filter来对日志格式进行处理

  • (1)在first-pipeline.conf中增加filter配置如下
    input {
        beats {
            port => "5044"
        }
    }
    filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
    }
    output {
        stdout { codec => rubydebug }
    }
  • (2)到filebeat的根目录下删除之前上报的数据历史(以便重新上报数据),并重启filebeat
    sudo rm data/registry
    sudo ./filebeat -e -c filebeat.yml -d "publish"
  • (3)由于之前启动Logstash设置了自动更新配置,因此Logstash不需要重新启动,这个时候可以获取到的日志数据如下:
    {
            "request" => "/style2.css",
              "agent" => "\"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
             "offset" => 24464,
               "auth" => "-",
              "ident" => "-",
         "input_type" => "log",
               "verb" => "GET",
             "source" => "/data/home/michelmu/workspace/logstash-tutorial.log",
            "message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"http://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
               "type" => "log",
               "tags" => [
            [0] "beats_input_codec_plain_applied"
        ],
           "referrer" => "\"http://www.semicomplete.com/projects/xdotool/\"",
         "@timestamp" => 2018-10-09T12:24:21.276Z,
           "response" => "200",
              "bytes" => "4877",
           "clientip" => "86.1.76.62",
           "@version" => "1",
               "beat" => {
                "name" => "VM_136_9_centos",
            "hostname" => "VM_136_9_centos",
             "version" => "5.6.10"
        },
               "host" => "VM_136_9_centos",
        "httpversion" => "1.1",
          "timestamp" => "04/Jan/2015:05:30:37 +0000"
    }

可以看到message中的数据被详细解析出来了

2.4 数据派生和增强

Logstash中的一些filter可以根据现有数据生成一些新的数据,如geoip可以根据ip生成经纬度信息

  • (1)在first-pipeline.conf中增加geoip配置如下
    input {
        beats {
            port => "5044"
        }
    }
     filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
        geoip {
            source => "clientip"
        }
    }
    output {
        stdout { codec => rubydebug }
    }
  • (2)如2.3一样清空filebeat历史数据,并重启
  • (3)当然Logstash仍然不需要重启,可以看到输出变为如下:
    {
            "request" => "/style2.css",
              "agent" => "\"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
              "geoip" => {
                  "timezone" => "Europe/London",
                        "ip" => "86.1.76.62",
                  "latitude" => 51.5333,
            "continent_code" => "EU",
                 "city_name" => "Willesden",
              "country_name" => "United Kingdom",
             "country_code2" => "GB",
             "country_code3" => "GB",
               "region_name" => "Brent",
                  "location" => {
                "lon" => -0.2333,
                "lat" => 51.5333
            },
               "postal_code" => "NW10",
               "region_code" => "BEN",
                 "longitude" => -0.2333
        },
             "offset" => 24464,
               "auth" => "-",
              "ident" => "-",
         "input_type" => "log",
               "verb" => "GET",
             "source" => "/data/home/michelmu/workspace/logstash-tutorial.log",
            "message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"http://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
               "type" => "log",
               "tags" => [
            [0] "beats_input_codec_plain_applied"
        ],
           "referrer" => "\"http://www.semicomplete.com/projects/xdotool/\"",
         "@timestamp" => 2018-10-09T12:37:46.686Z,
           "response" => "200",
              "bytes" => "4877",
           "clientip" => "86.1.76.62",
           "@version" => "1",
               "beat" => {
                "name" => "VM_136_9_centos",
            "hostname" => "VM_136_9_centos",
             "version" => "5.6.10"
        },
               "host" => "VM_136_9_centos",
        "httpversion" => "1.1",
          "timestamp" => "04/Jan/2015:05:30:37 +0000"
    }

可以看到根据ip派生出了许多地理位置信息数据

2.5 将数据导入Elasticsearch

Logstash作为Elastic stack的重要组成部分,其最常用的功能是将数据导入到Elasticssearch中。将Logstash中的数据导入到Elasticsearch中操作也非常的方便,只需要在pipeline配置文件中增加Elasticsearch的output即可。

  • (1)首先要有一个已经部署好的Logstash,当然可以使用腾讯云快速创建一个Elasticsearch创建地址
  • (2)在first-pipeline.conf中增加Elasticsearch的配置,如下
   input {
        beats {
            port => "5044"
        }
    }
     filter {
        grok {
            match => { "message" => "%{COMBINEDAPACHELOG}"}
        }
        geoip {
            source => "clientip"
        }
    }
    output {
        elasticsearch {
            hosts => [ "localhost:9200" ]
        }
    }
  • (3)清理filebeat历史数据,并重启
  • (4)查询Elasticsearch确认数据是否正常上传(注意替换查询语句中的日期)
    curl -XGET 'http://172.16.16.17:9200/logstash-2018.10.09/_search?pretty&q=response=200'
  • (5)如果Elasticsearch关联了Kibana也可以使用kibana查看数据是否正常上报

kibana图示

Logstash提供了大量的Input, filter, output, codec的插件,用户可以根据自己的需要,使用一个或多个组件实现自己的功能,当然用户也可以自定义插件以实现更为定制化的功能。自定义插件可以参考[logstash input插件开发]

3 部署Logstash

演示过如何快速使用Logstash后,现在详细讲述一下Logstash的部署方式。

3.1 安装

  • 安装JDK:Logstash采用JRuby编写,运行需要JDK环境,因此安装Logstash前需要先安装JDK。(当前6.4仅支持JDK8)
  • 安装Logstash:可以采用直接下载压缩包方式安装,也通过APT或YUM安装,另外Logstash支持安装到Docker中。[Logstash安装参考]
  • 安装X-PACK:在6.3及之后版本X-PACK会随Logstash安装,在此之前需要手动安装[参考链接]

3.2 目录结构

logstash的目录主要包括:根目录bin目录配置目录日志目录插件目录数据目录

不同安装方式各目录的默认位置参考[此处]

3.3 配置文件

  • Pipeline配置文件,名称可以自定义,在启动Logstash时显式指定,编写方式可以参考前面示例,对于具体插件的配置方式参见具体插件的说明(使用Logstash时必须配置): 用于定义一个pipeline,数据处理方式和输出源
  • Settings配置文件(可以使用默认配置): 在使用Logstash时可以不用设置,用于性能调优,日志记录等
    • logstash.yml:用于控制logstash的执行过程[参考链接]
    • pipelines.yml: 如果有多个pipeline时使用该配置来配置多pipeline执行[参考链接]
    • jvm.options:jvm的配置
    • log4j2.properties:log4j 2的配置,用于记录logstash运行日志[参考链接]
    • startup.options: 仅适用于Lniux系统,用于设置系统启动项目!
  • 为了保证敏感配置的安全性,logstash提供了配置加密功能[参考链接]

3.4 启动关闭方式

3.4.1 启动

3.5 扩展Logstash

当单个Logstash无法满足性能需求时,可以采用横向扩展的方式来提高Logstash的处理能力。横向扩展的多个Logstash相互独立,采用相同的pipeline配置,另外可以在这多个Logstash前增加一个LoadBalance,以实现多个Logstash的负载均衡。

4 性能调优

[详细调优参考]

  • (1)Inputs和Outputs的性能:当输入输出源的性能已经达到上限,那么性能瓶颈不在Logstash,应优先对输入输出源的性能进行调优。
  • (2)系统性能指标
    • CPU:确定CPU使用率是否过高,如果CPU过高则先查看JVM堆空间使用率部分,确认是否为GC频繁导致,如果GC正常,则可以通过调节Logstash worker相关配置来解决。
    • 内存:由于Logstash运行在JVM上,因此注意调整JVM堆空间上限,以便其有足够的运行空间。另外注意Logstash所在机器上是否有其他应用占用了大量内存,导致Logstash内存磁盘交换频繁。
    • I/O使用率: 1)磁盘IO: 磁盘IO饱和可能是因为使用了会导致磁盘IO饱和的创建(如file output),另外Logstash中出现错误产生大量错误日志时也会导致磁盘IO饱和。Linux下可以通过iostat, dstat等查看磁盘IO情况 2)网络IO: 网络IO饱和一般发生在使用有大量网络操作的插件时。linux下可以使用dstat或iftop等查看网络IO情况
  • (3)JVM堆检查
    • 如果JVM堆大小设置过小会导致GC频繁,从而导致CPU使用率过高
    • 快速验证这个问题的方法是double堆大小,看性能是否有提升。注意要给系统至少预留1GB的空间。
    • 为了精确查找问题可以使用jmap或VisualVM。[参考]
    • 设置Xms和Xmx为相同值,防止堆大小在运行时调整,这个过程非常消耗性能。
  • (4)Logstash worker设置: worker相关配置在logstash.yml中,主要包括如下三个:
    • pipeline.workers: 该参数用以指定Logstash中执行filter和output的线程数,当如果发现CPU使用率尚未达到上限,可以通过调整该参数,为Logstash提供更高的性能。建议将Worker数设置适当超过CPU核数可以减少IO等待时间对处理过程的影响。实际调优中可以先通过-w指定该参数,当确定好数值后再写入配置文件中。
    • pipeline.batch.size: 该指标用于指定单个worker线程一次性执行flilter和output的event批量数。增大该值可以减少IO次数,提高处理速度,但是也以为这增加内存等资源的消耗。当与Elasticsearch联用时,该值可以用于指定Elasticsearch一次bluck操作的大小。
    • pipeline.batch.delay: 该指标用于指定worker等待时间的超时时间,如果worker在该时间内没有等到pipeline.batch.size个事件,那么将直接开始执行filter和output而不再等待。

结束语

Logstash作为Elastic Stack的重要组成部分,在Elasticsearch数据采集和处理过程中扮演着重要的角色。本文通过简单示例的演示和Logstash基础知识的铺陈,希望可以帮助初次接触Logstash的用户对Logstash有一个整体认识,并能较为快速上手。对于Logstash的高阶使用,仍需要用户在使用过程中结合实际情况查阅相关资源深入研究。当然也欢迎大家积极交流,并对文中的错误提出宝贵意见。

MORE:

通过 metadata 使logstash配置更简洁

LogstashLeon J 发表了文章 • 0 个评论 • 2421 次浏览 • 2018-09-04 13:17 • 来自相关话题

从Logstash 1.5开始,我们可以在logstash配置中使用metadata。metadata不会在output中被序列化输出,这样我们便可以在metadata中添加一些临时的中间数据,而不需要去删除它。

我们可以通过以下方式来访问metadata:

[@metadata][foo]

用例

假设我们有这样一条日志:

[2017-04-01 22:21:21] production.INFO: this is a test log message by leon

我们可以在filter中使用grok来做解析:

grok {
      match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{DATA:env}\.%{DATA:log_level}: %{DATA:content}" }
    }

解析的结果为

{
      "env" => "production",
      "timestamp" => "2017-04-01 22:21:21",
      "log_level" => "INFO",
      "content" => "{\"message\":\"[2017-04-01 22:21:21] production.INFO: this is a test log message by leon\"}"
}

假设我们希望

  1. 能把log_level为INFO的日志丢弃掉,但又不想让该字段出现在最终的输出中
  2. 输出的索引名中能体现出env,但也不想让该字段出现在输出结果里

对于1,一种方案是在输出之前通过mutate插件把不需要的字段删除掉,但是一旦这样的处理多了,会让配置文件变得“不干净”。

通过 metadata,我们可以轻松地处理这些问题:

grok {
    match => { "message" => "\[%{TIMESTAMP_ISO8601:timestamp}\] %{DATA:[@metadata][env]}\.%{DATA:[@metadata][log_level]}: %{DATA:content}" }
}

if [@metadata][log_level] == "INFO"{
    drop{}    
}

output{
    elasticsearch {
        hosts => ["127.0.0.1:9200"]
        index => "%{[@metadata][env]}-log-%{+YYYY.MM}"
        document_type => "_doc"
    }
}

除了简化我们的配置文件、减少冗余字段意外,同时也能提高logstash的处理速度。

Elasticsearch input插件

有些插件会用到metadata这个特性,比如elasticsearch input插件:

input {
  elasticsearch {
    host => "127.0.0.1"
    # 把 ES document metadata (_index, _type, _id) 包存到 @metadata 中
    docinfo_in_metadata => true
  }
}

filter{
    ......
}

output {
  elasticsearch {
    document_id => "%{[@metadata][_id]}"
    index => "transformed-%{[@metadata][_index]}"
    type => "%{[@metadata][_type]}"
  }
}

调试

一般来说metadata是不会出现在输出中的,除非使用 rubydebug codec 的方式输出:

output { 
  stdout { 
    codec  => rubydebug {
      metadata => true
    }
  }
}

日志经过处理后输出中会包含:

{
    ....,
    "@metadata" => {
        "env" => "production",
        "log_level" => "INFO"
    }
}

总结

由上可见,metadata提供了一种简单、方便的方式来保存中间数据。这样一方面减少了logstash配置文件的复杂性:避免调用remove_field,另一方面也减少了输出中的一些不必要的数据。通过这篇对metadata的介绍,希望能对大家有所帮助。

elasticTalk,qrcode

ET001 不可不掌握的 Logstash 使用技巧

Logstashrockybean 发表了文章 • 3 个评论 • 1014 次浏览 • 2018-07-21 11:53 • 来自相关话题

Logstash 是 Elastic Stack 中功能最强大的 ETL 工具,相较于 beats 家族,虽然它略显臃肿,但是强在功能丰富、处理能力强大。大家在使用的过程中肯定也体验过其启动时的慢吞吞,那么有什么办法可以减少等待 Logstash 的启动时间,提高编写其处理配置文件的效率呢?本文给大家推荐一个小技巧,帮助大家解决如下两个问题,让大家更好地与这个笨重的大家伙相处。

  1. 减少 Logstash 重启的次数,也就节省宝贵的时间
  2. 方便快捷地向 Logstash 输入需要处理的内容

1. 打开 reload 配置开关

Logstash 启动的时候可以加上 -r 的参数来做到配置文件热加载,效果是:

  • 当你修改了配置文件后,无需重启 Logstash 即可让新配置文件生效。

它的含义如下:

当你写好配置文件,比如 test.conf ,启动命令如下:

bin/logstash -f test.conf -r

启动完毕,修改 test.conf 的内容并保存后,过 1 秒钟,你会发现 Logstash 端有类似如下日志输出(注意红色框标记的部分),此时说明 reload 的成功。

如果你修改的配置文件有错误,会看到报错的日志,你可以根据错误提示修改。

至此,第一个问题解决!

2. 使用 HTTP INPUT

编写配置文件的另一个痛点是需要针对不同格式的输入内容进行详细的测试,以防解析报错的情况出现。此时大家常用标准输入来解决这个问题(stdin input),但是标准输入对于文字编辑支持不太友好,而且配置文件热更新的功能也不支持标准输入。

在这里向大家推荐使用 http input 插件,配置如下:

input{
    http{
        port => 7474
        codec => "json"
    }
}

然后大家再用自己喜欢的 http 请求工具,比如 POSTMan、Insomnia 等向 http://loclahost:7474发送待测试内容即可,如下是 Insomnia 的截图。

至此,第二个问题也解决了。

3. 总结

相信看到这里,大家一定是跃跃欲试了,赶紧打开电脑,找到 Logstash,然后编辑 test.conf,输入如下内容:

input{
    http{
        port => 7474
        codec => "json"
    }
}

filter{

}

output{
        stdout{
        codec => rubydebug{
            metadata => true
        }
    }
}

然后执行启动命令:

bin/logstash -f test.conf -r

打开 Insomnia ,输入要测试的内容,点击发送,开始舒爽流畅的配置文件编写之旅吧!

logstash-filter-elasticsearch的简易安装

Logstashziyou 发表了文章 • 0 个评论 • 907 次浏览 • 2018-05-29 09:38 • 来自相关话题

不同版本的logstash集成的插件不一样,在5.6版本就未集成logstash-filter-elasticsearch插件,所以需要自己安装。 官方提供的方法因为需要联网,并且需要调整插件管理源,比较麻烦,针对logstash-filter-elasticsearch插件,使用下面这种方式安装。 logstash-filter-elasticsearch插件安装 1、在git上下载logstash-filter-elasticsearch压缩包,logstash-filter-elasticsearch.zip, 2、在logstash的目录下新建plugins目录,解压logstash-filter-elasticsearch.zip到此目录下。 3、在logstash目录下的Gemfile中添加一行:
gem "logstash-filter-elasticsearch", :path => "./plugins/logstash-filter-elasticsearch"
4、重启logstash即可。   此方法适用logstash-filter-elasticsearch,但不适用全部logstash插件。

我想只记录这一行,如果写logstash规则

Logstashsunjj 发表了文章 • 0 个评论 • 1158 次浏览 • 2018-03-15 10:26 • 来自相关话题

2018-03-14 22:23:56,833 ERROR [FrontShopController.java:45] : ==dianchou.app.boss.pageController.FrontShopControl lerjava.lang.NullPointerException       at dianchou.app.boss.pageController.FrontShopController.projectDetail(FrontShopController.java:40)       at sun.reflect.GeneratedMethodAccessor495.invoke(Unknown Source)       at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)       at java.lang.reflect.Method.invoke(Method.java:606)       at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205)

kibana使用geoip插件展示地图热力图

Kibanaziyou 发表了文章 • 8 个评论 • 12318 次浏览 • 2018-02-09 16:56 • 来自相关话题

geoip1.png

上图是我们最终的地图效果。

总体步骤:

一、使用logstash geoip插件解析IP字段;

二、配置geoip.location字段为geo_point类型。

三、使用kibana的Coordinate map作图。

具体步骤:

一、解析IP字段

使用logstash的geoip插件 logstash-filter-geoip 解析IP字段,需要在logstash的配置文件中配置geoip的解析配置,配置如下:

geoip {
       source => "ip" //需要解析的IP地址
      }

解析出的效果如下:

"geoip": {
      "city_name": "Wuhan",
      "timezone": "Asia/Shanghai",
      "ip": "117.136.52.200",
      "latitude": 30.5801,
      "country_name": "China",
      "country_code2": "CN",
      "continent_code": "AS",
      "country_code3": "CN",
      "region_name": "Hubei",
      "location": {
        "lon": 114.2734,
        "lat": 30.5801
      },
      "region_code": "42",
      "longitude": 114.2734
    }

备注:这个只是geoip的配置,解析日志的时候需要先解析出ip字段

二、配置geoip字段类型

IP经过logstash解析后就可以使用IP的所有解析信息了,但是如果想要在kibana中作图,就必须把geoip里面的相应信息配置成相应的字段类型,才能够被kibana识别,然后经过聚合作图。 需要配置的字段:geoip.location 需要配置的类型:geo_point 在mapping中的配置为:

"geoip" : {
          "properties" : {
            "location" : {
              "type" : "geo_point",
              "ignore_malformed": "true"  
            }
          }
        }

备注1: ignore_malformed 如果true,格式错误的地理位置被忽略。如果false(默认),格式错误的地理位置引发异常并拒绝整个文档。 此字段需要配置成true,以防地理格式错误导致文档被拒绝。 也可以在所以级别进行设置: "settings": { "index.mapping.ignore_malformed": true }

备注2:需要先设置mapping,再导入数据mapping才会生效,如果先导入数据,再设置mapping,则第二天八点后才会生效(北京时间)。

三、kibana作图

1、在kibana中打开visualize->coordinate map

geoip2.png

2、选择相应的索引进行画图

geoip3.png

3、选择geoip.location作为聚合字段,然后设置Options,调整地图效果即可。

geoip4.png

logstash date不支持BJT时区

Logstashqvitt 发表了文章 • 2 个评论 • 1232 次浏览 • 2017-12-18 10:29 • 来自相关话题

人生日历截图20171218102703.png
收集网络设备的日志,但是格式不统一,很苦恼,而且仅仅时间戳格式就有快10种不同格式,测试发现date不支持BJT时区
人生日历截图20171218102703.png
收集网络设备的日志,但是格式不统一,很苦恼,而且仅仅时间戳格式就有快10种不同格式,测试发现date不支持BJT时区

ELK学习资料整理

经验分享lsyoung 发表了文章 • 0 个评论 • 6773 次浏览 • 2017-04-14 10:17 • 来自相关话题

刚开始学习使用ELK,整理一个学习资料列表,当做备忘录。   1.第一个当然是官方文档
  • ElasticSearch参考手册,学习 DSL查询语法,包括查找(query)、过滤(filter)和聚合(aggs)等。
  • Logstash参考手册,学习数据导入,包括输入(input)、过滤(filter)和输出( output)等,主要是filter中如何对复杂文本 进行拆分和类型 转化。
  • Kibana参考手册,使用Kibana提供的前端界面对数据进行快速展示,主要是对Visulize 模块的使
2.中文文档   欢迎补充……

logstash多时间(date)字段文档导入

Logstashlsyoung 发表了文章 • 1 个评论 • 3122 次浏览 • 2017-04-13 20:43 • 来自相关话题

我们都知道如果文档中有表示时间的字段时可以用如下方法将字段转化为date(timestamp)格式导入
date {
        match => [ "submission_time", "yyyyMMdd" ]
    }
但是如果一条记录中有多个字段需要使用date类型呢?有人遇到了 同样的问题How to parse multiple date fields?其实多次使用date{}语法就行了,例如:
date {
        match => [ "submission_time", "UNIX_MS" ]
        target => "@timestamp"
        }
date {
        match => [ "submission_time", "UNIX_MS" ]
        target => "submission_time"
        }
date {
        match => [ "start_time", "UNIX_MS" ]
        target => "start_time"
        }
date {
        match => [ "end_time", "UNIX_MS" ]
        target => "end_time"
        }
查看官方文档后, 发现一直使用的date是省去了参数target的,而target默认值为@timestamp。

如何反向设置es mapping template

Elasticsearchrunc 发表了文章 • 1 个评论 • 3129 次浏览 • 2016-08-31 14:26 • 来自相关话题

比如,由logstash打到es中的数据,除了其中一个字段比如message,其余字段都想设置为not analyzed,这种情况如何设置?貌似目前es只支持设置那些具体的字段为not analyzed,而不能反过来设置啊?
比如,由logstash打到es中的数据,除了其中一个字段比如message,其余字段都想设置为not analyzed,这种情况如何设置?貌似目前es只支持设置那些具体的字段为not analyzed,而不能反过来设置啊?

Day14: percolator接口在logstash中的运用

Advent三斗室 发表了文章 • 0 个评论 • 4041 次浏览 • 2015-12-16 23:10 • 来自相关话题

我们都知道 Elasticsearch 除了普通的 search 接口以外,还有另一个 Percolator 接口,天生用来做实时过滤告警的。但是由于接口比较复杂,在目前的 ELK 体系中不是很容易运用。 而单纯从 Logstash 来做实时过滤报警,规则又不是很灵活。toplog.io 公司开发了一个 logstash-output-percolator插件,在有一定既定条件的情况下,成功运用上了 Percolator 方案。 这个插件的设计逻辑是:
  1. 通过 logstash-filter-checksum 自主生成 ES 文档的 _id;
  2. 使用上一步生成的 _id 同时发送 logstash-output-elasticsearch 和 logstash-output-percolator
  3. Percolator 接口一旦过滤成功,将 _id 发送给 Redis 服务器
  4. 其他系统从 Redis 服务器中获取 _id 即可从 ES 里拿到实际数据
Percolator 接口的用法简单说是这样: 创建接口:
curl -XPUT 'localhost:9200/patterns/.percolator/my-pattern-id' -d '{"query" : {"match" : {"message" : "ERROR"} } }'
过滤测试:
curl -XGET 'localhost:9200/my-index/my-type/_percolate' -d '{"doc" : {"message" : "ERROR: Service Apache failed to connect to MySQL"} }'
要点就是把文档放在 doc 属性里发送到 _percolate 里。 对应的 Logstash 配置如下:
filter {
    checksum {
        algorithm => "md5"
        keys => ["message"]
    }
}
output {
    elasticsearch {
        host => "localhost"
        cluster => "my-cluster"
        document_id => "%{logstash_checksum}"
        index => "my-index"
    }
    percolator {
        host => "es-balancer"
        redis_host => ["localhost"]
        document_id => "%{logstash_checksum}"
        pattern_index => "patterns"
    }
}
连接上对应的 Redis,就可以看到报警信息了:
$ redis-cli
127.0.0.1:6379> lrange percolator 0 1
1) "{\"matches\":[\"2\"],\"document_id\":\"a5d5c5f69b26ac0597370c9b1e7a8111\"}"
想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day13: ipip.net介绍

Advent三斗室 发表了文章 • 1 个评论 • 3932 次浏览 • 2015-12-16 23:05 • 来自相关话题

Geo 定位在 ELK 应用中是非常重要和有用的一个环节。不幸的是:GeoIP 本身在国内的准确度实在堪忧。高春辉近年成立了一个项目,专注收集细化 IP 地址在国内的数据:http://www.ipip.net。数据分为免费版和收费版两种。项目提供了不少客户端,有趣的是,有社区贡献了一个 Logstash 插件:https://github.com/bittopaz/logstash-filter-ipip。 用法很简单:
filter {
    ipip {
        source => "clientip"
        target => "ipip"
    }
}
生成的 JSON 数据结构类似下面这样:
{
    "clientip" : "",
    "ipip" : {
        "country" : "",
        "city" : "",
        "carrier" : "",
        "province" : ""
    }
}
不过这个插件只实现了收费版的数据库基础格式。免费版的支持,收费版高级的经纬度、基站位置等,都没有随着更新。事实上,我们可以通过 ipip 官方的 Java 库,实现一个更灵活的 logstash-filter-ipip_java 插件出来,下期见。 想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。

Day10: 如何处理数组形式的JSON日志

Advent三斗室 发表了文章 • 1 个评论 • 6577 次浏览 • 2015-12-16 22:57 • 来自相关话题

ELK 收集业务日志的来源,除了应用服务器以外,还有很大一部分来自客户端。考虑到客户端网络流量的因素,一般实现上都不会要求实时上报数据,而是攒一批,等到手机连上 WIFI 网络了,再统一发送出来。所以,这类客户端日志一般都有几个特点:
  1. 预先已经记录成 JSON 了;
  2. 日志主体内容是一个巨大无比的数组,数据元素才是实际的单次日志记录;
  3. 一次 POST 会有几 MB 到几十 MB 大小。
在处理这类数据的时候,第一关是别让数据超长直接给丢弃了(说的就是你啊,Rsyslog);第二关就是拆分 JSON 数组,把几十 MB 数据扔 ES 字段里,显然是不利于搜索和统计需求的。今天我们就来说说怎么拆分 JSON 数组。 假设收到的是这么一段日志:
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
首先我们知道可以在读取的时候把 JSON 数据解析成 LogStash::Event 对象:
input {
    tcp {
        codec => json
    }
}
但是怎么把解析出来的 logs 字段拆分成多个 event 呢?这里我们可以用一个已有插件:logstash-filter-split。
filter {
    split {
        field => "logs"
    }
    date {
        match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]
        remove_fields => ["logs", "timestamp"]
    }
}
这样,就可以得到两个 event 了:
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","@timestamp":"2015-12-10T09:55:00Z","reason":"****"}
{"uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10T09:56:12Z","tracert":"****"}
看起来可能跟这个插件的文档描述不太一样。文档上写的是通过 terminator 字符,切割 field 字符串成多个 event。但实际上,field 设置是会自动判断的,如果 field 内容是字符串,就切割字符串成为数组再循环;如果内容已经是数组了,直接循环:
    original_value = event[@field]

    if original_value.is_a?(Array)
        splits = original_value
    elsif original_value.is_a?(String)
        splits = original_value.split(@terminator, -1)
    else
        raise LogStash::ConfigurationError, "Only String and Array types are splittable. field:#{@field} is of type = #{original_value.class}"
    end

    return if splits.length == 1

    splits.each do |value|
        next if value.empty?

        event_split = event.clone
        @logger.debug("Split event", :value => value, :field => @field)
        event_split[(@target || @field)] = value
        filter_matched(event_split)

        yield event_split
    end
    event.cancel
顺带提一句:这里 yield 在 Logstash 1.5.0 之前,实现有问题,生成的新事件,不会继续执行后续 filter,直接进入到 output 阶段。也就是说,如果你用 Logstash 1.4.2 来执行上面那段配置,生成的两个事件会是这样的:
{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
{"@timestamp":"2015-12-10T09:38:13Z","uid":123456,"upload_datetime":"2015-12-10 11:38:11","type":"network_error","@timestamp":"2015-12-10 17:56:12","tracert":"****","logs":[{"type":"crash","timestamp":"2015-12-10 17:55:00","reason":"****"},{"type":"network_error","timestamp":"2015-12-10 17:56:12","tracert":"****"}]}
想了解更全面的 ELK Stack 知识和细节,欢迎购买我的《ELK Stack权威指南》,也欢迎加 QQ 群:315428175 哟。
  Logstash是一个开源的日志收集管理工具。