TrumanDu's Blog

发表于 更新于 分类于 本文字数： 331 阅读时长 ≈ 1 分钟

kafka patition计算

kafka生产信息的时候，可以指定该信息存储在具体的patition上，如果不指定会使用默认的算法计算要落的patition.以下是针对此做一个探讨和理解

消息路由

  1. 指定了 patition，则直接使用；
  2. 未指定 patition 但指定 key，通过对 key 的 value 进行hash 选出一个 patition
  3. patition 和 key 都未指定，使用轮询选出一个 patition。

Kafka Client计算patition

详细代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
//创建消息实例
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {
     if (topic == null)
          throw new IllegalArgumentException("Topic cannot be null");
     if (timestamp != null && timestamp < 0)
          throw new IllegalArgumentException("Invalid timestamp " + timestamp);
     this.topic = topic;
     this.partition = partition;
     this.key = key;
     this.value = value;
     this.timestamp = timestamp;
}

//计算 patition，如果指定了 patition 则直接使用，否则使用 key 计算
private int partition(ProducerRecord<K, V> record, byte[] serializedKey , byte[] serializedValue, Cluster cluster) {
     Integer partition = record.partition();
     if (partition != null) {
          List<PartitionInfo> partitions = cluster.partitionsForTopic(record.topic());
          int lastPartition = partitions.size() - 1;
          if (partition < 0 || partition > lastPartition) {
               throw new IllegalArgumentException(String.format("Invalid partition given with record: %d is not in the range [0...%d].", partition, lastPartition));
          }
          return partition;
     }
     return this.partitioner.partition(record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
}

// 使用 key 选取 patition
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
     List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
     int numPartitions = partitions.size();
     if (keyBytes == null) {
          int nextValue = counter.getAndIncrement();
          List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
          if (availablePartitions.size() > 0) {
               int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
               return availablePartitions.get(part).partition();
          } else {
               return DefaultPartitioner.toPositive(nextValue) % numPartitions;
          }
     } else {
          //对 keyBytes 进行 hash 选出一个 patition
          return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
     }
}

发表于 更新于 分类于 本文字数： 453 阅读时长 ≈ 2 分钟

one step

创建topic-move-test.json,迁移可以指定多个topic

1
2
3
4
5
6
7
{"topics": [{
    "topic": "upgrade-kafka"
  },{
  "topic":"upgrade-kafka1"
  }],
  "version": 1
}

two step

在kafka工作目录，执行以下生成迁移配置，其中91为需要迁移到的broker id

1
bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --topics-to-move-json-file topic-move-test.json --broker-list "91" --generate

生成以下内容：

1
2
3
4
5
6
7
Current partition replica assignment

{"version":1,"partitions":[{"topic":"upgrade-kafka1","partition":1,"replicas":[96]},{"topic":"upgrade-kafka","partition":5,"replicas":[96]},{"topic":"upgrade-kafka1","partition":2,"replicas":[91]},{"topic":"upgrade-kafka","partition":1,"replicas":[92]},{"topic":"upgrade-kafka","partition":4,"replicas":[95]},{"topic":"upgrade-kafka1","partition":0,"replicas":[95]},{"topic":"upgrade-kafka","partition":3,"replicas":[94]},{"topic":"upgrade-kafka","partition":0,"replicas":[91]},{"topic":"upgrade-kafka","partition":2,"replicas":[93]}]}
Proposed partition reassignment configuration

{"version":1,"partitions":[{"topic":"upgrade-kafka","partition":5,"replicas":[91]},{"topic":"upgrade-kafka1","partition":1,"replicas":[91]},{"topic":"upgrade-kafka","partition":1,"replicas":[91]},{"topic":"upgrade-kafka","partition":4,"replicas":[91]},{"topic":"upgrade-kafka1","partition":2,"replicas":[91]},{"topic":"upgrade-kafka","partition":3,"replicas":[91]},{"topic":"upgrade-kafka1","partition":0,"replicas":[91]},{"topic":"upgrade-kafka","partition":0,"replicas":[91]},{"topic":"upgrade-kafka","partition":2,"replicas":[91]}]}

three step

新建expand-cluster-reassignment.json文件，将上一步输出放入，内容如下：

1
{"version":1,"partitions":[{"topic":"upgrade-kafka","partition":5,"replicas":[91]},{"topic":"upgrade-kafka1","partition":1,"replicas":[91]},{"topic":"upgrade-kafka","partition":1,"replicas":[91]},{"topic":"upgrade-kafka","partition":4,"replicas":[91]},{"topic":"upgrade-kafka1","partition":2,"replicas":[91]},{"topic":"upgrade-kafka","partition":3,"replicas":[91]},{"topic":"upgrade-kafka1","partition":0,"replicas":[91]},{"topic":"upgrade-kafka","partition":0,"replicas":[91]},{"topic":"upgrade-kafka","partition":2,"replicas":[91]}]}

four step

按自己需求可以重新编辑expand-cluster-reassignment.json内容，修改broker id.
执行迁移命令

1
bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --reassignment-json-file expand-cluster-reassignment.json --execute

检查执行结果

1
bin/kafka-reassign-partitions.sh --zookeeper 127.0.0.1:2181 --reassignment-json-file expand-cluster-reassignment.json --verify

以下即为迁移成功：

1
2
3
4
5
6
7
8
9
10
11
Status of partition reassignment:
Reassignment of partition [upgrade-kafka,4] completed successfully
Reassignment of partition [upgrade-kafka,5] completed successfully
Reassignment of partition [upgrade-kafka1,0] completed successfully
Reassignment of partition [upgrade-kafka,2] completed successfully
Reassignment of partition [upgrade-kafka1,1] completed successfully
Reassignment of partition [upgrade-kafka,1] completed successfully
Reassignment of partition [upgrade-kafka,3] completed successfully
Reassignment of partition [upgrade-kafka,0] completed successfully
Reassignment of partition [upgrade-kafka1,2] completed successfully

Reference

  1. kafka.apache.org

发表于 更新于 分类于 本文字数： 102 阅读时长 ≈ 1 分钟

zookeeper运维经验

参数配置

在默认zoo.cfg配置中，zookeeper生成的历史镜像,log不会删除，生成的频率也比较快，因此生产环境需要配置以下两个参数

  • autopurge.purgeInterval

解释：清楚间隔，单位小时，默认值为0，修改次值，表示启用清楚

  • autopurge.snapRetainCount

解释：保留数量，默认为3，最小值为3

引用

  1. zookeeper官网

发表于 更新于 分类于 本文字数： 113 阅读时长 ≈ 1 分钟

kafka使用教程

shell操作

1.创建topic

–replication-factor 2：备份因子为2

–partitions 10：partitions数目为10

1
bin/kafka-topics.sh --create --zookeeper 192.168.0.101:2181 --replication-factor 2 --partitions 10 --topic upgrade-kafka_test

2.查询topic

列出所有的topic

1
bin/kafka-topics.sh --list --zookeeper localhost:2181

3.删除topic

首先需要确认集群是否配置delete.topic.enable=true，配置后即可删除，确保topic没有被使用。

1
2
bin/kafka-topics.sh --delete --zookeeper 192.168.0.101:2181 --topic upgrade-kafka_test

程序操作

参考

  1. kafka official website

发表于 更新于 分类于 本文字数： 499 阅读时长 ≈ 2 分钟

Kafka原理及设计理论总结

一、数据可靠性保证

当Producer向Leader发送数据时,可以通过request.required.acks参数设置数据可靠性的级别

  1. 0: 不论写入是否成功,server不需要给Producer发送Response,如果发生异常,server会终止连接,触发Producer更新meta数据;
  2. 1: Leader写入成功后即发送Response,此种情况如果Leader fail,会丢失数据
  3. -1: 等待所有ISR接收到消息后再给Producer发送Response,这是最强保证
    仅设置acks=-1也不能保证数据不丢失,当Isr列表中只有Leader时,同样有可能造成数据丢失。要保证数据不丢除了设置acks=-1, 还要保 证ISR的大小大于等于2,具体参数设置:

(1).request.required.acks:设置为-1 等待所有ISR列表中的Replica接收到消息后采算写成功;
(2).min.insync.replicas: 设置为大于等于2,保证ISR中至少有两个Replica
Producer要在吞吐率和数据可靠性之间做一个权衡

二、数据一致性保证

一致性定义:若某条消息对Consumer可见,那么即使Leader宕机了,在新Leader上数据依然可以被读到

  1. HighWaterMark

简称HW: Partition的高水位，取一个partition对应的ISR中最小的LEO作为HW，消费者最多只能消费到HW所在的位置，另外每个replica都有highWatermark，leader和follower各自负责更新自己的highWatermark状态，highWatermark <= leader. LogEndOffset
2. 对于Leader新写入的msg，

Consumer不能立刻消费，Leader会等待该消息被所有ISR中的replica同步后,更新HW,此时该消息才能被Consumer消费，即Consumer最多只能消费到HW位置

这样就保证了如果Leader Broker失效,该消息仍然可以从新选举的Leader中获取。对于来自内部Broker的读取请求,没有HW的限制。同时,Follower也会维护一份自己的HW,Folloer.HW = min(Leader.HW, Follower.offset)

发表于 更新于 分类于 本文字数： 312 阅读时长 ≈ 1 分钟

一、项目介绍

为了更好监控RedisCluster集群状态信息，提升性能，抛弃之前通过java api获取info信息。此次项目分为两个方面：

  1. 通过redis自带info信息监控集群
  2. 监控网络流量获取对集群的使用情况的监控。

总之，通过不同的粒度监控RedisCluster集群运行状况，提供良好的管理运维平台。

二、技术架构

1.info信息

1
2
3
graph LR
Logstash-->ElasticSearch
ElasticSearch-->Kibana

Logstash良好的插件结构设计，我们可以根据不同场景选择合适的input,filter,output插件。为了高效配置监控集群，input插件我们基于exec自定义了自己的插件redis-exec插件。output插件直接选择elasticsearch插件。

2.网络流量

1
2
3
graph LR
Packagebeat-->ElasticSearch
ElasticSearch-->Kibana

Packagebeat是一个分布式网络数据抓包软件，可以直接监控redis协议信息。

Kibana是一个es可视化的作图工具，根据不同的搜索条件可制定监控不同指标信息的动态图，让使用人员可以直观监控集群运行状况。

三、难点

1.redis-exec插件开发

详见logstash插件开发
2.监控指标

详见国外案例

发表于 更新于 分类于 本文字数： 188 阅读时长 ≈ 1 分钟

centos6.X版本，docker version1.7.1版本创建过程如下：

  1. 运行容器
    1
    $ docker run -d -p 5000:5000 registry
  2. 修改配置

在docker1.3.X版本以后，与docker registry交互默认使用的是https，此处需要修改为http。在/etc/sysconfig/docker文件中添加以下内容即可：

1
other_args="$other_args --insecure-registry myregistry.example.com:5000 "

在1.12最新版本中可以使用以下方式修改，原理是一致的。

1
2
 Create or modify /etc/docker/daemon.json
{ "insecure-registries":["myregistry.example.com:5000"] }

然后重新启动

1
$ service docker restart
  1. 测试新的registry
  • 镜像打标签
    1
    $ docker tag <img_id> myregistry.example.com:5000/truman/opentsdb
  • 提交镜像
    1
    $ docker push  myregistry.example.com:5000/truman/opentsdb
    然后通过docker images即可查看到push 的镜像

在别的机器中就可以拉取镜像了命令如下：

1
docker pull myregistry.example.com:5000/truman/opentsdb

发表于 更新于 分类于 本文字数： 528 阅读时长 ≈ 2 分钟

简介

logstash是一个实时流水式开源数据收集引擎。具有强大的plugin。可以根据自己的业务场景选择不同的input filter output。绝大多数情况下都是结合ElasticSearch Kibana一起使用的，俗称ELK。

模块介绍

Logstash使用管道方式进行日志的搜集处理和输出。有点类似*NIX系统的管道命令 xxx | ccc | ddd，xxx执行完了会执行ccc，然后执行ddd。

在logstash中，包括了三个阶段:

输入input –> 处理filter（不是必须的） –> 输出output

配置文件说明

前面介绍过logstash基本上由三部分组成，input、output以及用户需要才添加的filter，因此标准的配置文件格式如下：

1
2
3
4
5
6
7
8
9
input {
   
}
filter {
    
}
output {
    
}

执行说明

1
bin/logstash -f demo.conf

使用Demo

Output plugins ElasticSearch

案例使用如下：

1
2
3
4
5
6
7
8
9
output {
#stdout {  codec => rubydebug }
elasticsearch {
hosts => ["127.0.0.1:9200"]
template_overwrite => true
index => "rediscluster-%{+YYYY.MM.dd}"
workers => 5
}
}

Output plugins opentsdb

使用logstash收集数据，并发送到opentsdb中。分为三部分：Input,Filter,Output

输入数据时，输入一条数据，回车。以下为三条测试数据：

1
2
3
threads.ThreadCount 1352279077 67 host=server1 port=1006
gc.PSScavenge.CollectionTime 1352279137 1360 host=server2 port=1010
memorypool.CodeCache.Usage_used 1352279137 11625472 host=server1 port=1009

  • Input采用命令行输入数据

    1
    2
    3
    4
    input {
       stdin{
       }
    }

  • Filter过滤组织数据

采用的是grok插件，可以使用其他插件完成相同的目的

1
2
3
4
5
6
7
filter {
    grok { 
        match => { "message" => "%{DATA:metricName} %{NUMBER:unixtime} %{NUMBER:data} host=%{DATA:metricHost} port=%{NUMBER:port}" }  
        remove_field => [ "host" ]
    }
    
}

备忘：

  1. logstash输入数据自带host,@timestamp等自带，为了避免干扰存入opentsdb数据，此处特将隐含的host字段去掉。
  2. DATA/NUMBER等实为grok自带的正则规则。
  • Output输出数据

此处输出数据到opentsdb中，官方文档有误，详见源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
output {
    stdout {  codec => rubydebug }#此处是为了将filter结果输出到控制台中
    opentsdb {
        host => '***.***.***.***'
        port => 4242
        metrics => [
            "%{metricName}",
            "%{data}",
            "host",
            "%{metricHost}",
            "port",
            "%{port}"
        ]
    }
}

备忘：

opentsdb输入信息格式为：put metric timestamp value tagname=tagvalue tag2=value2，在logstash-output-opentsdb插件metrics配置中默认已经输入timestamp，因此metrics需要配置的第一个参数为metricName，第二个参数为 value 之后依次为tagname,tagValue。

参考

  1. https://www.elastic.co/guide/en/logstash/current/index.html

发表于 更新于 分类于 本文字数： 98 阅读时长 ≈ 1 分钟

Filter plugins

  1. grok
    1
    2
    3
    4
    grok {
    match => {"command" => "redis-cli -c -h %{IP:node:} -p %{NUMBER:port}%{DATA:data}" }
    remove_field => [ "host" ]
    }
  2. ruby

功能描述：将redis info 信息格式化按字段输出

1
2
3
4
5
6
7
8
9
10
11
12
ruby {
       code => "fields = event['message'].split(/\r\n|\n/)
       length = fields.length-1
       for i in 1..length do 
         if fields[i].include?':' then
           field = fields[i].split(':')
           event[field[0]] = field[1].to_f
         end
       end
       "
       remove_field => [ "message" ]
   }
  1. mutate

功能描述：字段类型指定

1
2
3
4
5
6
7
8
filter {
        mutate {
            convert => {"latestResponse" => "integer"}
            convert => {"cacheHit" => "string"}
            convert => {"cacheRate" => "float"}
        }
        
}

Output plugins

发表于 更新于 分类于 本文字数： 1.4k 阅读时长 ≈ 5 分钟

背景

logstash强大魅力在于它的插件体系，虽然官方插件很多，但不可能满足所有的要求，因此就需要定制化个性化插件，本次结合Logstash Monitor Redis需求开发专用插件，以实现动态化获取master 实例中info 信息。

logstash插件介绍

体系结构

1
2
3
4
5
6
7
8
9
10
11
12
13
$ tree logstash-input-example
├── Gemfile
├── LICENSE
├── README.md
├── Rakefile
├── lib
│   └── logstash
│       └── inputs
│           └── example.rb
├── logstash-input-example.gemspec
└── spec
    └── inputs
        └── example_spec.rb

其实只需要这logstash-input-example.gemspec,example.rb两个文件即可。
mypluginname_spec.rb 是测试类。

先看看logstash-input-example.gemspec都做了什么吧！

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
Gem::Specification.new do |s|
  s.name = 'logstash-input-example'
  s.version         = '2.0.4'
  s.licenses = ['Apache License (2.0)']
  s.summary = "This example input streams a string at a definable interval."
  s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
  s.authors = ["Elastic"]
  s.email = 'info@elastic.co'
  s.homepage = "http://www.elastic.co/guide/en/logstash/current/index.html"
  s.require_paths = ["lib"]

  # Files
  s.files = Dir['lib/**/*','spec/**/*','vendor/**/*','*.gemspec','*.md','CONTRIBUTORS','Gemfile','LICENSE','NOTICE.TXT']
   # Tests
  s.test_files = s.files.grep(%r{^(test|spec|features)/})

  # Special flag to let us know this is actually a logstash plugin
  s.metadata = { "logstash_plugin" => "true", "logstash_group" => "input" }

  # Gem dependencies
  s.add_runtime_dependency "logstash-core", ">= 2.0.0", "< 3.0.0"
  s.add_runtime_dependency 'logstash-codec-plain'
  s.add_runtime_dependency 'stud', '>= 0.0.22'
  s.add_development_dependency 'logstash-devutils', '>= 0.0.16'
end

上面的信息，只要改改版本和名字，其他的信息基本不需要动。

关键的信息还有：

  • s.require_paths定义了插件核心文件的位置
  • s.add_runtime_dependency 定义了插件运行的环境
    然后再看看example.rb
    这个文件就需要详细说说了，基本的框架如下，
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    # encoding: utf-8
    require "logstash/inputs/base"
    require "logstash/namespace"
    require "stud/interval"
    require "socket" # for Socket.gethostname

    # Generate a repeating message.
    #
    # This plugin is intented only as an example.

    class LogStash::Inputs::Example < LogStash::Inputs::Base
      config_name "example"

      # If undefined, Logstash will complain, even if codec is unused.
      default :codec, "plain"

      # The message string to use in the event.
      config :message, :validate => :string, :default => "Hello World!"

      # Set how frequently messages should be sent.
      #
      # The default, `1`, means send a message every second.
      config :interval, :validate => :number, :default => 1

      public
      def register
        @host = Socket.gethostname
      end # def register

      def run(queue)
        # we can abort the loop if stop? becomes true
        while !stop?
          event = LogStash::Event.new("message" => @message, "host" => @host)
          decorate(event)
          queue << event
          # because the sleep interval can be big, when shutdown happens
          # we want to be able to abort the sleep
          # Stud.stoppable_sleep will frequently evaluate the given block
          # and abort the sleep(@interval) if the return value is true
          Stud.stoppable_sleep(@interval) { stop? }
        end # loop
      end # def run

      def stop
        # nothing to do in this case so it is not necessary to define stop
        # examples of common "stop" tasks:
        #  * close sockets (unblocking blocking reads/accepts)
        #  * cleanup temporary files
        #  * terminate spawned threads
      end
    end # class LogStash::Inputs::Example
    挨行看看！

首先第一行的# encoding: utf-8,不要以为是注释就没什么作用。它定义了插件的编码方式。

下面两行：

require “logstash/inputs/base”
require “logstash/namespace”
引入了插件必备的包。

1
2
class LogStash::Inputs::Example < LogStash::Inputs::Base
  config_name "example"

插件继承自Base基类，并配置插件的使用名称。

下面的一行对参数做了配置，参数有很多的配置属性，完整的如下：

1
config :variable_name,:validate =>:variable_type,:default =>"Default value",:required => boolean,:deprecated => boolean

其中

variable_name就是参数的名称了。
validate 定义是否进行校验，如果不是指定的类型，在logstash -f xxx –configtest的时候就会报错。它支持多种数据类型，比如:string, :password, :boolean, :number, :array, :hash, :path (a file-system path), :codec (since 1.2.0), :bytes.
default 定义参数的默认值
required 定义参数是否是必须值
deprecated 定义参数的额外信息，比如一个参数不再推荐使用了，就可以通过它给出提示！典型的就是es-output里面的Index_type，当使用这个参数时，就会给出提示

插件安装

  1. 便捷安装方式

第一步，首先把这个插件文件夹拷贝到下面的目录中

1
logstash-2.1.0\vendor\bundle\jruby\1.9\gems

第二步，修改logstash根目录下的Gemfile,添加如下的内容：

1
gem "logstash-filter-example", :path => "vendor/bundle/jruby/1.9/gems/logstash-filter-example-1.0.0"

第三步，编写配置文件，test.conf：

1
2
3
4
5
6
7
8
9
10
11
input{
    example{} 
}
filter{
    
}
output{
    stdout{
        codec => rubydebug
    }
}

第四步，输入logstash -f test.conf时，输入任意字符，回车~~~大功告成！

1
2
3
4
5
6
{
       "message" => "Hello World!",
      "@version" => "1",
    "@timestamp" => "2016-01-27T19:17:18.932Z",
          "host" => "cadenza"
}
  1. 官方指导方式

第一步，build

1
gem build logstash-input-example.gemspec

会在当前路径下生成logstash-input-example-2.0.4.gem
第二步，install

1
bin/logstash-plugin install /logstash-input-example/logstash-input-example-2.0.4.gem

验证

1
2
3
validating /logstash-input-example/logstash-input-example-2.0.4.gem >= 0
Valid logstash plugin. Continuing...
Successfully installed 'logstash-input-example' with version '2.0.4'

第三步，查看plugin：

1
bin/logstash-plugin list

第四步，使用

略

开发案例

开发插件实现根据cluster nodes信息获取redis cluster 中master节点 info信息。使用该插件只用输入一条命令，即可动态获取相关信息。

插件开发

此插件是基于exec基础上封装的，主要修改内容为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def execute(command, queue)
  @logger.debug? && @logger.debug("Running exec", :command => command)
  begin
 @io = IO.popen(command)
 fields = (@io.read).split(/\r\n|\n/)
 puts fields
 length = fields.length-1
    	for i in 0..length do 
  if fields[i].include?':' then
	field = fields[i].split(':')
	newcommand = "redis-cli -c -h #{field[0]} -p #{field[1]} info"
	@io = IO.popen(newcommand)
	@codec.decode(@io.read) do |event|
	decorate(event)
	event.set("host", @hostname)
	event.set("command", newcommand)
	queue << event
  end
end

    end
  rescue StandardError => e
    @logger.error("Error while running command",
      :command => command, :e => e, :backtrace => e.backtrace)
  rescue Exception => e
    @logger.error("Exception while running command",
      :command => command, :e => e, :backtrace => e.backtrace)
  ensure
    stop
  end
end

使用Demo

使用方式

1
2
3
4
5
redisexec {
command => "redis-cli -h 127.0.0.1 -p 6379 cluster nodes|grep master|awk '{print $2}'"
interval => 20
type => "info"
}

完整使用案例

将info 信息存储到 ElasticSerach中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
input {
redisexec {
command => "redis-cli -h 127.0.0.1 -p 6379 cluster nodes|grep master|awk '{print $2}'"
interval => 20
type => "info"
}
}
filter {
        grok {
            match => {"command" => "redis-cli -c -h %{IP:node:} -p %{NUMBER:port}%{DATA:data}" }
      remove_field => [ "host" ]
        }
    ruby {
        code => "fields = event['message'].split(/\r\n|\n/)
        length = fields.length-1
        for i in 1..length do
          if fields[i].include?':' then
            field = fields[i].split(':')
            event[field[0]] = field[1].to_f
          end
        end
        "
        remove_field => [ "message" ]
    }
}
output {
#stdout {  codec => rubydebug }
elasticsearch {
hosts => ["127.0.0.1:9200"]
template_overwrite => true
index => "rediscluster-%{+YYYY.MM.dd}"
workers => 5
}
}

参考

  1. http://www.cnblogs.com/xing901022/p/5259750.html
  2. https://github.com/logstash-plugins?utf8=%E2%9C%93&query=example
  3. https://www.elastic.co/guide/en/logstash/current/_how_to_write_a_logstash_input_plugin.html
0%