0x01 前言
最近在kibana中添加可视化视图时需要增加一个流量统计的图标，可是遇到了很严重的问题。
我们都知道数据是具有类别之分的，例如：
- 数字：long, integer, short, byte, double, float
- 字符串
- 日期格式
- … …
而保存在elasticsearch中的数据一样需要指明格式，要不然就无法生成特定的图表。例如计算总和、均值这类运算则需要数字的支持，这也是合乎常理的，如果值的类型为字符串，这要怎样进行数学运算？
0x02 问题
因为我在配置elasticsearch的时候还没意识到格式的问题，当导入将近20G的数据后，在添加可视化试图时却发现无法添加某些图表。
最显而易见的就是流量统计，我的nginx日志包含bytes这个字段，可是在elasticsearch中是以默认的string类型来存储。这就是导致无法进行数学运算的根本原因。
0x03 思路
因为elasticsearch中已经处理存储着20G的数据，我不希望通过删除索引重新处理数据这种方式解决这个问题，而且日后再遇到这个问题也不可能通过这种粗暴的方式来解决问题。
0x03.1方法一
—————
修正：
2016.12.24：应该先添加模版再调整logstash
—————
在经过一天翻看文档后，我找到了解决办法。大概思路如下：
- 新建索引hk1-v2
- 将logstash输出索引从hk1指向hk1-v2并重启logstash服务
- 为elasticsearch添加”hk1*“模版
- 将hk1重新索引至hk1-v2
- 删除hk1
0x03.2 方法二
为什么要那么麻烦？通过模版重新定义字段的类型不就好了？
其实这也是可以的，如果你的索引是按天或按月生成，通过模版重新定义字段的类型这种方式很适合你。不过要注意的是，模版只对下一次生成的索引有效。例如:
如果你在24号这天添加模版，而且模版适用于索引格式：filebeat-*；
那么你目前正处于filebeat-2016.12.24这个索引的使用期，模版是不会对这个索引有任何作用的；
模版只对24号以后的索引（如：filebeat-2016.12.25）生效。
这样的话问题又来了：我不是使用这种方式来管理我的索引，我只有一个或者按月度生成，这该怎么办？那只好使用方法一来解决问题。
在这里还需要注意的是，如果你的logstash在实时接收日志，例如syslog的UDP 514端口的日志，那么在重启logstash的时候可能会有几十秒处于无法接收的状态，这段时间的日志可能会丢失！
0x04 模版
首先要向elasticsearch添加模版，在这里我使用kibana中的Dev Tools而不使用curl：
上面的Dev Tools中，左侧作为输入框，右侧为输出框。
我们不需要从头编写模版，因为kibana内置filebeat的模版，我们可以直接进行修改，在输入框中输入以下命令：
GET _template
并点击运行按钮：
左侧即为logstash中所有的模版，但我们只需要filebeat的模版，请通过以下命令导出：
GET _template/filebeat*
将输出的所有内容复制到文本编辑器中：
{
"filebeat": {
"order": 0,
"template": "filebeat-*",
"settings": {
"index": {
"refresh_interval": "5s"
}
},
"mappings": {
"_default_": {
"dynamic_templates": [
{
"template1": {
"mapping": {
"ignore_above": 1024,
"index": "not_analyzed",
"type": "{dynamic_type}",
"doc_values": true
},
"match": "*"
}
}
],
"_all": {
"norms": {
"enabled": false
},
"enabled": true
},
"properties": {
"@timestamp": {
"type": "date"
},
"geoip": {
"dynamic": true,
"type": "object",
"properties": {
"location": {
"type": "geo_point"
}
}
},
"offset": {
"type": "long",
"doc_values": "true"
},
"client_ip": {
"type": "ip"
},
"message": {
"index": "analyzed",
"type": "string"
}
}
}
},
"aliases": {}
}
}
在properties字段下添加以下内容：
"bytes": {
"type": "long"
},
要注意这个模版中含有两处properties字段，其中一处是位于geoip下的，请不要修改geoip下的内容：
"geoip": {
"dynamic": true,
"type": "object",
"properties": {
"location": {
"type": "geo_point"
}
}
},
最后要修改模版名称和匹配的索引，修改2、4行：
#模版名称，可自定义
"hk1": {
"order": 0,
#匹配索引名称，以下的意思是所有hk1开头（包含hk1）的索引都使用这个模版
"template": "hk1*",
最后删除第一行与最后一行的”[“与”]”，删除第二行的 “hk1”:
完成后即可通过以下命令向elasticsearch中导入模版：
PUT _template/hk1
{
"order": 0,
"template": "hk1*",
"settings": {
"index": {
"refresh_interval": "5s"
}
},
"mappings": {
"_default_": {
"dynamic_templates": [
{
"template1": {
"mapping": {
"ignore_above": 1024,
"index": "not_analyzed",
"type": "{dynamic_type}",
"doc_values": true
},
"match": "*"
}
}
],
"_all": {
"norms": {
"enabled": false
},
"enabled": true
},
"properties": {
"@timestamp": {
"type": "date"
},
"bytes": {
"type": "long"
},
"geoip": {
"dynamic": true,
"type": "object",
"properties": {
"location": {
"type": "geo_point"
}
}
},
"offset": {
"type": "long",
"doc_values": "true"
},
"client_ip": {
"type": "ip"
},
"message": {
"index": "analyzed",
"type": "string"
}
}
}
},
"aliases": {}
}
0x05 logstash
如果你按照我之前写的文章进行安装（安装配置Elastic Stack 5），那么你可能在elasticsearch的配置文件中允许自动创建”hk1*”索引：
#打开文件 [root@web ~]# vim /etc/elasticsearch/elasticsearch.yml #允许自动创建索引 action.auto_create_index: +pfsense*,+hk1*,+syslog-*
如果没有，请手动创建：
PUT hk1-v2
然后修改output并重启logstash：
[root@web ~]# systemctl restart logstash.service
0x06 索引
这里要注意的是，如果你的elasticsearch不是以集群模式运行并且数据即为庞大，例如数据量为千万级的，那么重新索引的时间会极为漫长。在重新索引的时候，CPU将长时间处于高负荷，磁盘则处于高IO的环境中。
另一个需要注意的点是，重新索引并不是仅处理修改过的部分，而是所有字段都进行重新索引处理，如果你又多个字段都出现错误，那么可以一并解决。
在重新索引的时候，如果你的服务没有配置集群，那么很有可能通过kibana提取数据时会出现超时的情况。
一切准备就绪后就可以通过以下命令进行重新索引：
POST _reindex
{
"source": {
"index": "hk1"
},
"dest": {
"index": "hk1-v2"
}
}
重新索引的操作是以task的形式在后台运行，如果你使用kibana提交这个操作，那么在经过预设的超时时间后，kibana会返还超时的提示；如果是使用curl提交操作，在经过一段时间后也会返还超时。
但这个问题不必担心，因为重新索引的操作是以任务（task）的形式在后台工作的，我们可以通过以下命令查看相关任务，在kibana中输入并运行：
GET _tasks
在输出框中搜索：reindex 即可看到相关任务信息。以下是类似的返还内容（并不是reindex的任务信息，仅用于说明情况）：
"tasks": {
"IPkWEFyNRWC3Gbm7hsoB-A:1167097": {
"node": "IPkWEFyNRWC3Gbm7hsoB-A",
"id": 1167097,
"type": "direct",
"action": "cluster:monitor/tasks/lists[n]",
"start_time_in_millis": 1482558534346,
"running_time_in_nanos": 3590645,
"cancellable": false,
"parent_task_id": "IPkWEFyNRWC3Gbm7hsoB-A:1167096"
},
请注意第二行（高亮行），这是任务ID，如果遇到需要终止任务的情况，请通过以下命令进行终止：
POST _tasks/_cancel/IPkWEFyNRWC3Gbm7hsoB-A:1167097
还有更粗暴的方法：
POST _tasks/_cancel?actions=*reindex
在重新索引的过程，推荐通过elasticsearch-head插件进行监控，安装方法如下：
在Elasticsearch5中使用Elasticsearch-head插件
0x07 结语
在使用ELK时，最好是先配置elasticsearch，添加模版，确定字段与相关的类型，然后再通过logstash等软件向内导入内容。
使用上面介绍的方法会有几十秒的空档期会导致实时内容丢失，其实还有另一种方法在0停摆时间的情况下进行重新索引操作，这个日后再介绍。
不过重启logstash不会导致通过 *beat（如filebeat） 传入的内容丢失，因为相关的软件之间会有检验机制，这个有机会再作介绍。
修改完成后：
有疑问欢迎给我留言！
