ELK Stack高级实战培训

ELK Stack高级应用,饶琛琳,网名：三斗室 日志易产品总监，曾任职微博系统架构师，人人网技术专家 网站运维技术与实践作者 ELK Stack权威指南作者 Puppet实战手册译者,Elasticsearch索引原理 高可用索引集群维护 Logstash原理与维护 高吞吐日志系统实现 数据可视化的秘密 时间序列数据分析 监控系统设计,Elasticsearch索引原理,数据映射 内存策略 模板设置 磁盘控制 性能测试,ELK Stack高级实战培训第二期,数据映射mapping,es实现实时索引的原理 mapping在新增字段时，节点和集群之间的差异可能性 主要数据类型介绍 动态mapping设置 _type的field mapping冲突可能性 multi-fields介绍(.raw的运用) meta fields介绍,实时索引原理,Lucene 把每次生成的倒排索引，叫做一个段(segment)。然后另外使用一个 commit 文件，记录索引内所有的 segment。而生成 segment 的数据来源，则是内存中的 buffer。 新收到的数据写到新的索引文件里。,准实时查询原理,segment刷新到文件系统缓存 默认1s间隔 主动调用/_refresh接口刷到缓存 refresh_interval设置,磁盘同步的translog控制,segment从文件系统缓存刷新到磁盘，更新commit 清空translog 默认30分钟间隔，或512MB阈值 主动调用/_flush接口 index.translog.flush_threshold_size参数index.translog.flush_threshold_ops参数,translog的一致性控制,每 5 秒会强制刷新 translog 日志到磁盘上 index.gateway.local.sync参数 无副本的时候，故障可能导致5s的数据丢失,mapping的更新流程,node1收到新字段； node1确定mapping，通知master； master分发全集群更新成新的mapping； 其他节点按照新mapping处理后来数据中这个字段。,mapping的更新问题,node1收到新字段； node2收到新字段； node1确定mapping，通知master； node2确定mapping，通知master； master分发全集群更新成(node1的)新的mapping； 其他节点按照新mapping处理后来数据中这个字段。 node2的segment format已经定了,主要数据类型,string number(integer,long,double) date ipv4 geo_point nested object binary attachment,string的analyze,默认analyzer是lowercase+english，可以自定义stopwords。 但是.不在stopwords里，而且也加不上。 需要通过char_filter替换成空格。,string的not_analyzed,默认单个term最大长度32KB(32766) 为了保护ES，可以配置ignore_above(非英文按char算，也就是要小于32766/3=10922，因为JSON固定是UTF8格式，一个char占用3B) “message“: “type“: “string“, “index“: “not_analyzed“, “ignore_above“: 20 ,date format,默认格式: strict_date_optional_time|epoch_millis 2016-01-07T22:14:00.000Z | 1452176154000 自定义方式: “mappings“:“type1“:“properties“:“timefield“:“type“:“date“,“format“:“yyyy-MM-dd HH:mm:ss“,geo_points format,geo_points接受多种格式： string的话，是lat,lon array的话，是lon,lat object的话，是“lat“:xxx,“lon“:xxx geohash的话，是“drm3btev3e86“,nested和array的区别,ES会平铺array里的字段。 “user“:“first“:“john“,“last“:“smith“, “first“:“alice“,“last“:“white“ 存储成： “user.first“:“john“,“alice“ “user.last“:“smith“,“white“ 当你想搜john smitch的时候，alice white也命中,nested和array的区别,ES不会平铺nested里的字段。 但是需要专门的nested query和nested agg来请求,attachment,需要安装mapper-attachments插件 可以将PPT、XLS、PDF、HTML等各类文档，以base64格式存入，插件会自动生成: date title name author keywords content_type content_length language content,动态mapping,默认有_default_的动态mapping。 “dynamic_templates“: “named_analyzers“: “match_mapping_type“: “string“, “match“: “*“, 或者 “path_match“:“field.subfield.*“, “mapping“: “type“: “string“, “analyzer“: “name“ ,_type之间的mapping,_type的作用 _type是ES的逻辑层设计。 在数据层，index/shard下就是segment。 index1/type1下的field1:“type“:“string“ index1/type2下的field1:“type“:“double“ 在segment里，只有field1。实际mapping怎么选？,_type之间的mapping,先入关中者王。史记 先number后string，string会转换成number； 先string后number，number会写入失败； 先double后long，long会转换成double； 先long后double，double会转换成long，但是： 0.54存成4603039107142836552,_type之间的mapping,更严重的： 一个是object，一个是string/number/date. 必定有一部分写入失败，数据丢失。 “解决“办法： 该字段跳过indexing阶段，即不解析详细mapping。 后遗症： 没有inverted index和fielddata，也就无法搜索统计。,_type之间的mapping, “field1“: “type“:“object“, “enabled“:false, “index“:“no“ 注意：必须写成object，因为string是只有index没有enabled参数的。,multi-fields,在mapping有冲突的情况下，可以用multi-fields解决一定问题； 在需要时，也可以主动使用multi-fields “city“: “type“: “string“, “fields“: “raw“: “type“: “string“, “index“: “not_analyzed“ ,multi-fields,反过来用： “city“: “type“: “string“, “index“: “not_analyzed“ “fields“: “analyzed“: “type“: “string“ ,meta fields,ES有一系列meta fields在每个document里： _index, _type, _id, _version是基础内容； _source, _all, _field_names是数据内容； _timestamp, _ttl, _size是辅助功能内容； _meta是用户自定义内容。,_source和_all的作用,_source的作用在store，_all的作用在index Lucene中，每个field是需要设置store才能获取原文的； ES中，单独使用_source存原文，各field默认“store“:false _all是为了响应不指明field的全文检索请求。 在script中，可以使用_source.field和docfield来获取内容，分别从_source store和inverted index里拿数据。,_ttl的作用和优劣,_timestamp,_ttl,_size都是index setting中默认不开的。 开启_ttl，意味着ES每60秒要标记过期文档，在segment merge时实际删除。会增加资源消耗。 erval:60s,内存策略,segment merge介绍 merge和optimize的区别和差异 fielddata和doc_values indices.fielddata.cache.size indices.breaker.fielddata.limit indices.breaker.request.limit indices.breaker.total.limit,segment merge过程,过多segment影响数据读取的性能，占用文件句柄、内存资源 专门的merge线程池负责,segment merge配置,indices.store.throttle.max_bytes_per_sec默认20MB index.merge.scheduler.max_thread_count默认3个 Policy策略(tiered，2.0以后唯一选择) index.merge.policy.floor_segment:2MB index.merge.policy.max_merge_at_once:10 index.merge.policy.max_merged_segment:5GB index.merge.policy.segments_per_tier:10,optimize,merge默认最大segment为5GB optimize有单独的一个线程运行 2.0后改名叫/_forcemerge接口 max_num_segments=1,fielddata,搜索靠inverted index(倒排索引) 聚合靠uninverted index(field data cache),inverted index,通过term快速检索 至于term和id，稍后再说。,uninverted index,怎么求request_time字段的平均值? inverted index里没法直接从index知道字段内容 SO: 需要一份 request_time:12,12,13,13,14,14. 直接求一个数组的平均值。,Roaring bitmaps,怎么求单个URL的request_time字段的平均值? 也就是：怎么即用上inverted又用上uninverted？ request_time=12,12,13,13,14,14. url=0,1,0,1,0,1. filtered_request_time=0,12,0,13,0,14.,field data,每次请求，会从整个inverted index里读取出来document，生成field data，以便做聚合。 多大内存都耗不起这个读取过程所以必须预先生成。,disk-based field data(doc_values),将uninverted index的数组变成类似cassandra的列存储 views=1,1000,5000,. 变成 1 1000 5000 . 读取过程利用操作系统的VFS cache,跨segment的docID维护(global ordinals),inverted index有docID，uninverted index也要节省内存 但是聚合需要跨segment做合并！需要全局ID 在ES中，叫global ordinals数据结构。 开启doc_values的情况下，fielddata依然会占用一点内存用于global ordinals。,circuit breaker,没有doc_values的内容，如果不加控制，必然OOM！ 为了保护JVM的heap堆，限制 indices.breaker.total.limit: 70% indices.breaker.fielddata.limit: 60% indices.breaker.request.limit: 40%(每个request),模板template设置,template简介 Logstash的默认模板 _template接口 order关系和模板的merge策略 aliases设置,Template简介,index template一般包括三部分： template，settings，mappings template相当于是match/pattern的作用，可以用通配符,Logstash默认template, ./vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-2.2.0-java/lib/logstash/outputs/elasticsearch/elasticsearch-template.json,_template接口,ES同时支持文本存储的template文件和接口提交的template数据。 但是通过_template看不到template文件设置的内容 建议：只使用_template接口,order与merge策略,如果两个template的template匹配规则有重叠。比如： tpla的”template”:”logstash-*” tplb的“template”:”logstash-new-*” ES会自动merge两个template内部的设置。 为了明确merge时候，谁覆盖谁，提供order选项 “template”:”logstash-new-*”,”order”:1 order数值越大，优先级越高，默认是0。,aliases设置,curl -XPOST http:/localhost:9200/_aliases -d “actions” : “add”:“index”:“test*”,“alias”:“test_a”, “add”:“index”:“test*”,“alias”:“test_k”,“filter”:“term”:“user”:“kimchy” ,aliases在template设置, “template” : “test*”, “aliases” : “index_k” : “filter”:“term”:“user”:“kimchy” ,磁盘控制,cluster.routing.allocation.disk.watermark.low:85% 拒绝分配新分片的阈值 cluster.routing.allocation.disk.watermark.high:90% 触发迁移旧分片的阈值,reindex,input elasticsearch hosts = ““ port = “9200“ index = “old_index“ size = 500 scroll = “5m“ docinfo = true output elasticsearch host = ““ port = “9200“ index = “%metadata_index“ index_type = “%metadata_type“ document_id = “%metadata_id“ ,use Search:Elasticsearch; my $es = Search:Elasticsearch-new( nodes = :9200 ); my $bulk = $es-bulk_helper( index = new_index, ); $bulk-reindex( source = index = old_index, size = 500, # default search_type = scan # default );,性能测试,使用和线上集群相同硬件配置的服务器搭建一个单节点集群。 使用和线上集群相同的映射创建一个 0 副本，1 分片的测试索引。 使用和线上集群相同的数据写入进行压测。 观察写入性能，或者运行查询请求观察搜索聚合性能。 持续压测数小时，使用监控系统记录 eps、requesttime、fielddata cache、GC count 等关键数据。,高可用索引集群维护,备份与恢复 版本升级 节点上下线 冷温热分层 cat API介绍 用zabbix进行长期监控,备份恢复,创建备份仓库 curl -XPUT localhost:9200/_snapshot/backup0 -d “type“: “fs“, “settings“: “location“: “/mnt/snaps“, “compress“: “true“ ,备份恢复,开始备份 curl -XPUT localhost:9200/_snapshot/backup0/2014-11-01?pretty&wait_for_completion -d “indices“: “tweets-2014-11-01:10,tweets-2014-11-01:11,tweets-2014-11-01:12,tweets-2014-11-01:13“, “ignore_unavailable“: “true”, “include_global_state”: “false” ,备份恢复,开始恢复 curl -XPOST localhost:9200/_snapshot/backup0/2014-11-01/_restore?pretty curl -XPOST localhost:9200/_snapshot/backup0/2014-11-01/_restore -d “indices“: “index_1,index_2“, “index_settings“: “index.number_of_replicas“: 0 , “ignore_index_settings“: “index.refresh_interval“ ,版本升级,首先适当加大集群的数据恢复和分片均衡并发度以及磁盘限速： curl -XPUT :9200/_cluster/settings -d “persistent“ : “cluster.routing.allocation.disable_allocation“ : “false“, “cluster.routing.allocation.cluster_concurrent_rebalance“ : “5“, “cluster.routing.allocation.node_concurrent_recoveries“ : “5“, “cluster.routing.allocation.enable“ : “all“ “indices.recovery.concurrent_streams“ : “30“, “indices.recovery.max_bytes_per_sec“ : “2gb“ , “transient“ : “cluster.routing.allocation.enable“ : “all“ ,版本升级,暂停分片分配： curl -XPUT :9200/_cluster/settings -d “transient“ : “cluster.routing.allocation.enable“ : “none“ ,版本升级,下发新版本的安装包 停止数据写入进程 执行一次synced flush，同步副本的commit id： curl -XPOST :9200/_flush/synced 更新软件，重启服务,版本升级,等待集群节点数恢复完整 恢复分片分配： curl -XPUT :9200/_cluster/settings -d “transient“ : “cluster.routing.allocation.enable“ : “all“ 等待集群状态变green 恢复数据写入,reroute接口介绍,reroute 接口，手动完成对分片的分配选择的控制。 reroute 接口支持三种指令：allocate，move 和 cancel。常用的一般是 allocate 和 move。 指令默认只控制副本，需要明确指定allow_primary参数才能控制主分片。,reroute allocate,常见于分片因为STORE SET冲突(还记得么？)无法自动分配成功时的操作： curl -XPOST :9200/_cluster/reroute -d “commands“ : “allocate“ : “index“ : “logstash-2015.05.27“, “shard“ : 61, “node“ : “7“, “allow_primary“ : true ,reroute move,常见于机器资源不统一，需要减轻低配机器的分配数量： curl -XPOST :9200/_cluster/reroute -d “commands“ : “allocate“ : “index“ : “logstash-2015.05.22“, “shard“ : 0, “from_node“ : “1“, “to_node“ : “04“ 注意需要修改集群的allocation设置，否则你move完又被relocation回去。,集群allocation控制,集群是否允许分配 cluster.routing.allocation.enable : none,new,new_primary,all 单个节点并发恢复的分片的个数 cluster.routing.allocation.node_concurrent_recoveries : 2 单个节点并发初始化的主分片的个数 cluster.routing.allocation.node_initial_primaries_recoveries : 4 是否检查同一分片的副本可以分配在单个主机上的多个实例里 cluster.routing.allocation.same_shard.host : false 单个节点复制一个副本分片时的并发网络流个数 indices.recovery.concurrent_streams : 3,集群allocation控制,集群是否允许重均衡 cluster.routing.rebalance.enable : all,primaries,replicas,none 集群何时开始重均衡 cluster.routing.allocation.allow_rebalance : always,indices_primaries_active,indices_all_active 集群内同时重均衡的分片数 cluster.routing.allocation.cluster_concurrent_rebalance : 2 触发重均衡的阈值，越高越不容易触发 cluster.routing.allocation.balance.threshold : 1.0f,节点上线,ES 集群的数据均衡策略：以各节点的分片总数(indices_all_active)作为基准。 新加入集群的节点，分片总数远远低于其他节点。 这时候如果有新索引创建，新索引的所有主分片几乎全分配在这台新节点上。 整个集群的写入压力，压在一个节点上，结果很可能是这个节点直接被压死，集群出现异常。,节点上线,预先计算好索引的分片数后，配置好单节点分片的限额。 例：一个 5 节点的集群，索引主分片 10 个，副本 1 份。则平均下来每个节点应该有 4 个分片 curl -s -XPUT :9200/logstash-2015.05.08/_settings -d “index“: “routing.allocation.total_shards_per_node“ : “5“ ,节点下线,curl -XPOST :9200/_cluster/settings -d “transient“ : “cluster.routing.allocation.exclude._ip“ : ““ ,冷温热分层,热：当前正在写入的索引 温：当前无写入，但是还在读取的索引 冷：当前无读写，随时可以恢复的索引(/_close接口),温热分层方案,N 台机器做热数据的存储。这 N 台热数据节点的 elasticsearc.yml 中配置 node.tag: hot M 台机器做温数据的存储。这 M 台温数据节点中配置 node.tag: stale,温热分层方案,index template新增设置： “order“ : 0, “template“ : “*“, “settings“ : “index.routing.allocation.require.tag“ : “hot“ ,温热分层方案,每天0点，对前一天索引发请求： curl -XPUT :9200/logstash-yyyy.mm.dd/_settings -d “index“: “routing.allocation.require.tag“: “stale“ ,cat API,/_cat/allocation /_cat/shards /_cat/shards/index /_cat/master /_cat/nodes /_cat/indices /_cat/indices/index /_cat/segments /_cat/segments/index /_cat/count /_cat/count/index,/_cat/recovery /_cat/recovery/index /_cat/health /_cat/pending_tasks /_cat/aliases /_cat/aliases/alias /_cat/thread_pool /_cat/plugins /_cat/fielddata /_cat/fielddata/fields,_cat请求参数,显示表头 ?v 显示全部可用参数 ?help 指定显示所需参数 ?h=xxx,yyy,常用 _cat 实例,curl -XGET :9200/_cat/health?v curl -XGET :9200/_cat/indices?bytes=b | sort -rnk8 curl -XGET :9200/_cat/nodes?h=i,sm,tm,fm,fcm curl -XGET :9200/_cat/shards/logstash-mweibo-nginx-2015.06.14?v&h=n,iic,sc curl -XGET :9200/_cat/recovery?active_only curl -XGET :9200/_cat/thread_pool?v,即时监控工具,Kopf Bigdesk(不支持2.x) Marvel,zabbix,pip install es_stats_zabbix ,Logstash原理与维护,Pipeline原理 Plugin架构 通用的jmx监控方式 heartbeat监控 性能测试,pipeline,调用链： bin/logstash - lib/logstash/runner.rb - lib/logstash/agent.rb - lib/logstash/pipeline.rb,pipeline,内部队列部分： config = grammar.parse(configstr) code = pile eval(code) input_to_filter = SizedQueue.new(20) filter_to_output = SizedQueue.new(20),pipeline,input_threads = inputs.each do |input| input_threads Thread.new input.run(input_to_filter) end filter_threads = settings“filter-workers“.times.collect do Thread.new begin while true event = input_to_filter.pop events = filter(event) |newevent| events newevent events.each |event| filter_to_output.push(event) end end end,pipeline,output_threads = Thread.new outputs.each(&:worker_setup) while true event = filter_to_output.pop output(event) end input_threads.each(&:join) filter_threads.each(&:join) output_threads.each(&:join),Logstash:Input:File解析,hostname = Socket.gethostname tail.subscribe do |path, line| logger.debug? & logger.debug(“Received line“, :path = path, :text = line) codec.decode(line) do |event| decorate(event) event“host“ = hostname if !event.include?(“host“) event“path“ = path queue event end end,Logstash:Codec:Line解析,def register converter = LogStash:Util:Charset.new(charset) converter.logger = logger end public def decode(data) yield LogStash:Event.new(“message“ = converter.convert(data) end # def decode,Logstash:Event介绍,CHAR_PLUS = “+“ TIMESTAMP = “timestamp“ VERSION = “version“ VERSION_ONE = “1“ TIMESTAMP_FAILURE_TAG = “_timestampparsefailure“ TIMESTAMP_FAILURE_FIELD = “_timestamp“ public def initialize(data = ) logger = Cabin:Channel.get(LogStash) cancelled = false data = data accessors = LogStash:Util:Accessors.new(data) dataVERSION |= VERSION_ONE dataTIMESTAMP = init_timestamp(dataTIMESTAMP),Logstash:Timestamp介绍,def initialize(time = Time.new) time = time.utc end def self.now Timestamp.new(:Time.now) end JODA_ISO8601_PARSER = org.joda.time.format.ISODateTimeFormat.dateTimeParser UTC = org.joda.time.DateTimeZone.forID(“UTC“) def self.parse_iso8601(t) millis = JODA_ISO8601_PARSER.parseMillis(t) LogStash:Timestamp.at(millis / 1000, (millis % 1000) * 1000) def self.coerce(time) case time when String LogStash:Timestamp.parse_iso8601(time),output_workers实现,define_singleton_method(:handle, method(:handle_worker) worker_queue = SizedQueue.new(20) worker_plugins = workers.times.map self.class.new(params.merge(“workers“ = 1, “codec“ = codec.clone) worker_plugins.map.with_index do |plugin, i| Thread.new(original_params, worker_queue) do |params, queue| LogStash:Util:set_thread_name(“#self.class.config_name.#i“) plugin.register while true event = queue.pop plugin.handle(event) end end end def handle_worker(event) worker_queue.push(event) end,jmx配置,./bin/logstash.lib.sh中添加配置： JAVA_OPTS=“$JAVA_OPTS -Dcom.sun.management.jmxremote“ JAVA_OPTS=“$JAVA_OPTS -Dcom.sun.management.jmxremote.port=9010“ JAVA_OPTS=“$JAVA_OPTS -Dcom.sun.management.jmxremote.local.only=false“ JAVA_OPTS=“$JAVA_OPTS -Dcom.sun.management.jmxremote.authenticate=false“ JAVA_OPTS=“$JAVA_OPTS -Dcom.sun.management.jmxremote.ssl=false“,Zabbix的jmx监控配置,部署zabbix-java-gateway代理网关 配置JMX agent方式的item 通过jconsole查看选择需要的监控项,heartbeat监控,input heartbeat message = “epoch“ interval = 60 type = “heartbeat“ output if type = “heartbeat“ file path = “/data1/logstash-log/5160-%+YYYY.MM.dd.log“ else elasticsearch ,heartbeat监控,“clock“:1435191129,“host“:“,“version“:“1“,“timestamp“:“2015-06-25T00:12:09.042Z“,“type“:“heartbeat“,“zbxkey“:“logstash.heartbeat“,“zbxhost“:“logstash_hostname“ 可以针对clock字段的时间来做报警判断,性能测试,input generator count = 10000000 message = “key1“:“value1“,“key2“:1,2,“key3“:“subkey1“:“subvalue1“ codec = json output null ,性能测试,input generator count = 10000000 message = “key1“:“value1“,“key2“:1,2,“key3“:“subkey1“:“subvalue1“ codec = json output stdout codec = dot ./bin/logstash -f generator_dots.conf | pv -abt /dev/null,性能测试,filter metrics meter = “documents“ add_tag = “metric“ flush_interval = 60 output if “metric“ in tags stdout codec = line format = “1m rate: %documents.rate_1m ( %documents.count )“ ,高吞吐日志系统实现,Kafka介绍 多种架构对比 shipper端的最佳实践 indexer端的最佳实践,Kafka搭建,Zookeeper Kafka 注意各节点配置hostname解析,Logstash-input-kafka配置,input kafka zk_connect = “localhost:2181“ group_id = “logstash“ topic_id = “test“ reset_beginning = false # boolean (optional)， default: false consumer_threads = 5 # number (optional)， default: 1 decorate_events = true # boolean (optional)， default: false ,Kafka常见命令行,topic创建 partition修改 offset监控,Kafka和redis对比,简单扩展 数据高可靠性,Rsyslog介绍,rocket-fast System for log processing RHEL/CentOS6默认5.8.0版 最新8.15.0版,Rsyslog的queue设计,main queue * action queues direct disk linkedlist fixedarray disk-assisted,Rsyslog的rainerscript介绍,main_queue() global() module() input() action() template() ruleset(),Rsyslog的rainerscript介绍,template()示例： template(name=“outfmt“ type=“string“ string=“%msg:F,58:2% %$!msg%n“) template( name=“weiboadJson“ type=“list“ ) constant(value=“timestamp“:“) property(name=“timereported“ dateFormat=“rfc3339“) constant(value=“,“host“:“) property(name=“fromhost-ip“) constant(value=“,“) property(name=“msg“ position.from=“3“),Rsyslog的rainerscript介绍,内置变量 $msg,$rawmsg,$rawmsg-after-pri,$hostname,$source,$fromhost,$fromhost-ip,$syslogtag,$programname,$pri,$pri-text,$syslogfacility,$syslogfacility-text,$syslogseverity,$syslogseverity-text,$timegenerated,$timereported,$timestamp,$inputname,$now,$year,$month,$day,$hour,$hhour,$qhour,$minute,$myhostname,Rsyslog的rai