[2015-12-23 23:20:47,749] ERROR [KafkaApi-0] error when handling request Name: TopicMetadataRequest; Version: 0; CorrelationId: 48; ClientId: ; Topics: test-%{type} (kafka.server.KafkaApis)

kafka.common.InvalidTopicException: topic name test-%{type} is illegal, contains a character other than ASCII alphanumerics, '.', '_' and '-'

        at kafka.common.Topic$.validate(Topic.scala:42)

        at kafka.admin.AdminUtils$.createOrUpdateTopicPartitionAssignmentPathInZK(AdminUtils.scala:181)

        at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:172)

        at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:520)

        at kafka.server.KafkaApis$$anonfun$19.apply(KafkaApis.scala:503)

        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)

        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)

        at scala.collection.immutable.Set$Set1.foreach(Set.scala:74)

        at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)

        at scala.collection.AbstractSet.scala$collection$SetLike$$super$map(Set.scala:47)

        at scala.collection.SetLike$class.map(SetLike.scala:93)

        at scala.collection.AbstractSet.map(Set.scala:47)

        at kafka.server.KafkaApis.getTopicMetadata(KafkaApis.scala:503)

        at kafka.server.KafkaApis.handleTopicMetadataRequest(KafkaApis.scala:542)

        at kafka.server.KafkaApis.handle(KafkaApis.scala:62)

        at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:59)

        at java.lang.Thread.run(Thread.java:744)

tcpdump -nn 'dst port 9092 and tcp[37]==3 and tcp[57]==37'

tcpdump -nn 'dst port 9092 and tcp[(tcp[12]>>2)+5]==3 and tcp[(tcp[12]>>2)+25]==37'

tcpdump -vv -nn -X -s 0 dst port 9092 | grep -C 5 "test-"

1. 倒排词典的索引需要常驻内存，无法GC，需要监控data node上segment memory增长趋势。
2. 各类缓存，field cache, filter cache, indexing cache, bulk queue等等，要设置合理的大小，并且要应该根据最坏的情况来看heap是否够用，也就是各类缓存全部占满的时候，还有heap空间可以分配给其他任务吗？避免采用clear cache等“自欺欺人”的方式来释放内存。
3. 避免返回大量结果集的搜索与聚合。确实需要大量拉取数据的场景，可以采用scan & scroll api来实现。
4. cluster stats驻留内存并无法水平扩展，超大规模集群可以考虑分拆成多个集群通过tribe node连接。
5. 想知道heap够不够，必须结合实际应用场景，并对集群的heap使用情况做持续的监控。
6. 根据监控数据理解内存需求，合理配置各类circuit breaker，将内存溢出风险降低到最低。

{

 “first_name”:”string”,

 “last_name”:”string”

}

{

    "query": {

        "multi_match": {

            "query":       "诸葛瑜",

            "type":        "most_fields",

            "fields":      [ “*_name” ]

        }

    }

}

PUT /my_index/my_type/1

{

    "title": "Quick brown rabbits",

    "body":  "Brown rabbits are commonly seen."

}

PUT /my_index/my_type/2

{

    "title": "Keeping pets healthy",

    "body":  "My quick brown fox eats rabbits on a regular basis."

}

{

    "query": {

        "bool": {

            "should": [

                { "match": { "title": "Brown fox" }},

                { "match": { "body":  "Brown fox" }}

            ]

        }

    }

}

{

    "query": {

        "multi_match": {

            "query":       "peter smith",

            "type":        "most_fields",

            "operator":    "and",

            "fields":      [ "first_name", "last_name" ]

        }

    }

}

{

 “时代”:”三国”,

 “姓名”: [“大司马”，“诸葛亮”]

}

{

    "query": {

        "match_phrase": {

            "姓名": "司马诸葛"

        }

    }

}

{

    "files": [

    {

        "paths": [

            "/var/log/nginx/*.log"

            ],

        "Fields":{

            "type":"nginx"

        },

        "DeadTime": "30m"

    },

    {

        "paths": [

            "/var/log/tomcat/*.log",

            "/var/log/tomcat/*/*.log"

            ],

        "Fields":{

            "type":"tomcat"

        },

        "DeadTime": "30m"

    }

    ],

    "kafka": {

        "broker_list": ["10.0.0.1:9092","10.0.0.2:9092"],

        "topic_id": "topic_name_change_it_{{.type}}",

        "compression_codec": "gzip"

    }

}

• DeadTime:30m 是说超过30分钟没有更新, 就不会再继续跟踪这个文件了(退出goroutine)
• “Fields”:{ “type”:”tomcat” } , 会在每条日志中增加配置的字段
• path目前就是用的golang官方库, 好像是还不支持递归多层目录查找, 反正我翻了一下文档, 没有找到.

"files": [

        {

            "paths": [

                "d:\\target.txt"

                ],

            "FieldNames": ["datetime", "datetime", "s_ip", "cs_method", "cs_uri_stem", "cs_uri_query", "s_port", "time_taken"],

            "Delimiter": "\\s+",

            "QuoteChar": "\""

        }

]

1. PacketBeat，用来嗅探和分析网络流量，如HTTP、MySQL、Redis等
2. TopBeat，用来收集系统的监控信息，功能如其名，类似*nix下的top命令，只不过所有的信息都会发送给后端的集中存储：Elasticsearch，这样你就可以很方便的监控所有的服务器的运行情况了
3. FileBeat，用来收集数据源是文件的数据，比如常见的系统日志、应用日志、网站日志等等，FIleBeat思路来自Logstash-forwarder，Beats团队加入之后重构改写而成，解决的就是Logstash作为Agent采集时占用太多被收集系统资源的问题，Beats家族都是Golang编写，效率高，占用内存和CPU比较少，非常适合作为agent跑着服务器上
4. 。。。

1. 通过 logstash-filter-checksum 自主生成 ES 文档的 _id；
2. 使用上一步生成的 _id 同时发送 logstash-output-elasticsearch 和 logstash-output-percolator
3. Percolator 接口一旦过滤成功，将 _id 发送给 Redis 服务器
4. 其他系统从 Redis 服务器中获取 _id 即可从 ES 里拿到实际数据

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"} }'

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-cli

127.0.0.1:6379> lrange percolator 0 1

1) "{\"matches\":[\"2\"],\"document_id\":\"a5d5c5f69b26ac0597370c9b1e7a8111\"}"

filter {

    ipip {

        source => "clientip"

        target => "ipip"

    }

}

{

    "clientip" : "",

    "ipip" : {

        "country" : "",

        "city" : "",

        "carrier" : "",

        "province" : ""

    }

}

# bin/plugin -i solutions.siren/siren-join/1.0

{"index":{"_index":"index1","_type":"type","_id":"1"}}

{"id":1, "foreign_key":"13"}

{"index":{"_index":"index1","_type":"type","_id":"2"}}

{"id":2}

{"index":{"_index":"index1","_type":"type","_id":"3"}}

{"id":3, "foreign_key": "2"}

{"index":{"_index":"index1","_type":"type","_id":"4"}}

{"id":4, "foreign_key": "14"}

{"index":{"_index":"index1","_type":"type","_id":"5"}}

{"id":5, "foreign_key": "2"}

{"index":{"_index":"index2","_type":"type","_id":"1"}}

{"id":"1", "tag": "aaa"}

{"index":{"_index":"index2","_type":"type","_id":"2"}}

{"id":"2", "tag": "aaa"}

{"index":{"_index":"index2","_type":"type","_id":"3"}}

{"id":"3", "tag": "bbb"}

{"index":{"_index":"index2","_type":"type","_id":"4"}}

{"id":"4", "tag": "ccc"}

# curl -s -XPOST 'http://localhost:9200/index1/_coordinate_search?pretty' -d '

{

    "query":{

        "filtered":{

            "query":{

                "match_all":{}

            },

            "filter":{

                "filterjoin":{

                    "foreign_key":{

                        "index":"index2",

                        "type":"type",

                        "path":"id",

                        "query":{

                            "terms":{

                                "tag":["aaa"]

                            }

                        }

                    }

                }

            }

        }

    },

    "aggs":{

        "avg":{

            "avg":{

                "field":"id"

            }

        }

    }

}'

{

    "coordinate_search" : {

        "actions" : [ {

            "relations" : {

                "from" : {

                    "indices" : [ ],

                    "types" : [ ],

                    "field" : "id"

                },

                "to" : {

                    "indices" : null,

                    "types" : null,

                    "field" : "foreign_key"

                }

            },

            "size" : 2,

            "size_in_bytes" : 20,

            "is_pruned" : false,

            "cache_hit" : true,

            "took" : 0

        } ]

    },

    "took" : 2,

    "timed_out" : false,

    "_shards" : {

        "total" : 5,

        "successful" : 5,

        "failed" : 0

    },

    "hits" : {

        "total" : 2,

        "max_score" : 1.0,

        "hits" : [ {

            "_index" : "index1",

            "_type" : "type",

            "_id" : "5",

            "_score" : 1.0,

            "_source":{"id":5, "foreign_key": "2"}

        }, {

            "_index" : "index1",

            "_type" : "type",

            "_id" : "3",

            "_score" : 1.0,

            "_source":{"id":3, "foreign_key": "2"}

        } ]

    },

    "aggregations" : {

        "avg" : {

            "value" : 4.0

        }

    }

}

    .bars($width): 用柱状图展示数组

    .lines($width, $fill, $show, $steps): 用折线图展示数组

    .points(): 用散点图展示数组

    .color("#c6c6c6"): 改变颜色

    .hide(): 隐藏该数组

    .label("change from %s"): 标签

    .legend($position, $column): 图例位置

    .yaxis($yaxis_number, $min, $max, $position): 设置 Y 轴属性，.yaxis(2) 表示第二根 Y 轴

    .abs(): 对整个数组元素求绝对值

    .precision($number): 浮点数精度

    .testcast($count, $alpha, $beta, $gamma): holt-winters 预测

    .cusum($base): 数组元素之和，再加上 $base

    .derivative(): 对数组求导数

    .divide($divisor): 数组元素除法

    .multiply($multiplier): 数组元素乘法

    .subtract($term): 数组元素减法

    .sum($term): 数组元素加法

    .add(): 同 .sum()

    .plus(): 同 .sum()

    .first(): 返回第一个元素

    .movingaverage($window): 用指定的窗口大小计算移动平均值

    .mvavg(): .movingaverage() 的简写

    .movingstd($window): 用指定的窗口大小计算移动标准差

    .mvstd(): .movingstd() 的简写

    .elasticsearch(): 从 ES 读取数据

    .es(q="querystring", metric="cardinality:uid", index="logstash-*", offset="-1d"): .elasticsearch() 的简写

    .graphite(metric="path.to.*.data", offset="-1d"): 从 graphite 读取数据

    .quandl(): 从 quandl.com 读取 quandl 码

    .worldbank_indicators(): 从 worldbank.org 读取国家数据

    .wbi(): .worldbank_indicators() 的简写

    .worldbank(): 从 worldbank.org 读取数据

    .wb(): .worldbanck() 的简写

1. 预先已经记录成 JSON 了；
2. 日志主体内容是一个巨大无比的数组，数据元素才是实际的单次日志记录；
3. 一次 POST 会有几 MB 到几十 MB 大小。

{"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":"****"}]}

input {

    tcp {

        codec => json

    }

}

filter {

    split {

        field => "logs"

    }

    date {

        match => ["timestamp", "yyyy-MM-dd HH:mm:ss"]

        remove_fields => ["logs", "timestamp"]

    }

}

{"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":"****"}

    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

{"@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":"****"}]}

output {

    elasticsearch {

        host => "127.0.0.1"

        manage_template => true

        template => "/path/to/mytemplate"

        template_name => "myname"

    }

}

{

    "template" : "logstash-*",

    "order" : 1,

    "settings" : { ... },

    "mappings" : { ... }

}

output {

    elasticsearch {

        host => "127.0.0.1"

        manage_template => true

        template_overwrite => true

    }

}