ELK基础学习

Posted 2020-10-15 挖洞的土拨鼠

第壹部分：Elasticsearch

一 、Elasticsearch增删改查
1、数据写入：
HTTP method：POST
URL：http://127.0.0.1:9200/index-docs-name/data-type-name -d "{jsonstr}"
返回：{"_index":"index-name","_type":"typename","_id":"idstr","_version":"1","created":"true"}

2、数据获取:
HTTP method ：GET
URL：http://127.0.0.1:9200/index-docs-name/data-type-name/id
返回：数据的json结构

3、数据删除:
HTTP method ：DELETE
URL：http://127.0.0.1:9200/index-docs-name/data-type-name/id
或者：http://127.0.0.1:9200/index-docs-name（可以使用通配符）删除多个文档

4、数据更新:
（1）全量再次写入
HTTP method：POST
URL：http://127.0.0.1:9200/index-docs-name/data-type-name/id -d "{jsonstr}"
返回：{"_index":"index-name","_type":"typename","_id":"idstr","_version":"1","created":"true"}
（2）局部更新：
HTTP method：POST
URL：http://127.0.0.1:9200/index-docs-name/data-type-name/id/_update -d "{jsonstr}"

5、数据查询
（1）全文搜索
HTTP method : GET
URL : http://127.0.0.1:9200[/index-docs-name/data-type-name]/_search[?q=xxx] -d "{json}"
（2）聚合请求:
在查询的search输入的结构体中写聚合语句即可。

二、Elasticsearch新建模板 

请求方式--PUT
请求地址a.b.c.d:9200/_template/your_temp_name
模板：
{
"template":"whoisinfo",#这里是模板的名字，索引名字一样自动适配。
"order":"7",#这里是模板号
"state":"open",
"settings": {
"index": {
"creation_date": "1491451435658",
"number_of_shards": "5",#自己控制
"number_of_replicas": "1",#自己控制
"uuid": "0GCKTzVTRAaw-z47TfCaZQ",
"version": {
"created": "2030399"
}
}
}
"mapping":{
"domain":{#这里是这个模板下第n类数据的样子
"properties":{
"colunm_name":{
"index":"not_analyzed"
"type":"date/string……"
"format":"strict_date_optional_time||epoch_millis"#这里是时间格式
}
}
}
}
}

三、Elastincsearch索引与映射

（一）、映射
1、创建
Method : PUT
URL : http://127.0.0.1:9200/index-docs-name/_mappind -d "{jsonstr}"
2、新增字段
Method : PUT
URL : http://127.0.0.1:9200/index-docs-name/_mappind/mapping-name -d "{jsonstr}"
3、删除映射会删除数据
Method : DELETE
URL : http://127.0.0.1:9200/index-docs-name/_mappind/mapping-name
4、获取映射
Method : GET
URL : http://127.0.0.1:9200/index-docs-name/_mappind/mapping-name

（二） 、 常见的字段定义
type 数据类型
index 是否分词（not_analyzed）
format 格式
多重索引
"name":{
"type":"xxxxx"
"fields"::{"xxx":"xxxx"}
}

四、Elasticsearch 分页查询

第一步创建Elasticsearch对象
es = Elasticsearch([{‘host‘:‘192.168.1.103‘,‘port‘:9200}])

第二部创建分页机制
resp = es.search(index, body=query, scroll="24h",size=10000)
24h 是scrollid 有效时间
scroll_id = resp[‘_scroll_id‘] id就是标识
total = resp["hits"]["total"] 总量

第三分页获取数据
resp = es.scroll(scroll_id=scroll_id, scroll="24h")
rdoc = resp["hits"]["hits"]
scroll_id = resp[‘_scroll_id‘]
rdoc是每个数据项字典结构体的list

第贰部分：LogStash

一、logstash的安装
#step-one：依赖于java jdk 所以预先安装JDK
1、下载jdk 首选1.7
2、tar xzvf jdk-7u55-linux-x86.tar.gz
3、mkdir /usr/lib/jvm
4、mv jdk1.7.0_55 /usr/lib/jvm/
5、gedit ~/.bashrc
在末尾输入一下内容并保存退出
export JAVA_HOME=/usr/lib/jvm/jdk1.7.0_55
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH
6、source ~/.bashrc（多个账户下）
#step-two：安装logstash
1、方法1--源代码安装
wget https://download.elastic.co/logstash/logstash/logstash-1.5.1.tar.gz
tar zxvf logstash-1.5.1.tar.gz
mv logstash-1.5.1 logstash
cd ./logstash/
测试语句：bin/logstash -e ‘input { stdin { } } output { stdout {} }‘
配置测试语句：
bin/logstash -e ‘input { stdin { } } output { stdout { codec => rubydebug } }‘
2、方法2--elasticsearch官方仓库安装
wget -O http://packages.elasicsearch.org/GPG-KEY-elasticsearch | apt-key add -cat >> /etc/apt/sources.list
deb http//packages.elasticsearch.org/logstash/1.5/debian stable main
apt-get update
apt-get install logstash

二、按照logstash配置文件运行logstash
bin/logstash -f logstash.conf

三、配置文件logstash.conf示例
input {
stdin{ }
}
output{
stdout{ codec => rubydebug }
elasticsearch { embedded => true }
}

四、配置语法：
1、第一个概念：区段，也就是上文中的input 或者output都是一个区段
#区段之内定义键值对#
2、数据类型
（1）布尔型 true 和 false
（2）字符串 "hostname"
（3）数值 514
（4）数组 [] 类似python中的队列
（5）哈希 match => { key1 => "value1" , key2 => "value2" }

3、语法规则
（1）字段读取支持倒序下标 例如a[-1]
（2）条件判断if / else if / else
（3）判断操作符 != , == , < ,> ,<= ,>= ,=~ ,!~ ,in ,not in , and ,or ,nand ,xor, !{}

4、命令行参数
-e 执行
-f 配置文件
-l 错误日志的输出文件
-P 加载插件\
--verbose 输出一定到调试日志
--debug 输出更多调试信息

五、logstash的处理过程
|filtername |xxx
<inputname <logA
>outputname >elasticsearch

logstash处理的是事件，事件的流转过程;
输入 -》 过滤 -》 输出

六、plugin命令：
查看本机有多少插件可用 plugin list
安装插件 plugin install xxxx
升级插件 plugin update xxx

七 logstash运行
1、服务模式
services logstash start
2、nohup模式
nohup logstash -f ./logstash.conf &
3、screen模式(避免用户退出命令行到哦之程序退出
screen -dmS elksc1
screen -r elksc1
screen list

八、标准输出入
input {
stdin{
add_field => {"key" => "value"}
codec => plain
tags => {"add"}标签
type => "std"类型
}
}
九、文件输入
input {
file {
path => ["fiel1","file2",...]
type => "systen"
start_position => "beginning"duqushujuweizhi
#discover_interval 每隔多久去检查下path选的下新文件 默认15s
#execlude 排出文件 list
#sincedb/sincedb_write_interval
#start_interval 每隔多久监听下新文件
}
}
十、TCP 输入：
input {
tcp {
port => 8888
mode => server
ssl_enable => false
}
}
网络导入旧数据时候常用 与nc配合
nc 127.0.0.1 8888 < olddata

十一、syslog输入：
input {
syslog {
port => "514"
}
}

十二、collectd输入：
input {
collectd {
port => 25289
type => colletced
}
}

十三、Codec
1、json
codec => "json"
注意对于nginx日志 可以把-替换成0
2、多行事件编码
codec => multiline {
pattern => "^\["
negate => true
waht => "previous"
}
3、网络流编码：
codec => netflow {
definitions => "/opt/logstash-1.4.2/lib/logstash/codec/netflow/netflow.yaml"
version => [5]
}

十四、时间处理：
filter {
gork {
match => ["message","%{HTTPDATE:logdate}"]
}
date {
match => ["logdate","dd/MMM/yyyy:HH:mm:ss Z"]
}
}
gork 正则捕获
语法
gork {
match => {
"message"=> "%{WORD} {NUMBER:request_time:float} %{WORD}"
}
}

最佳实践
match => {"message"=>"%{SYSLOGBASE} %{DATA:message}"}