Elk的理论与实践

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Elk的理论与实践相关的知识,希望对你有一定的参考价值。

搜索引擎:
    索引组件:获取数据-->建立文档-->文档分析-->文档索引(倒排索引)
    搜索组件:用户搜索接口-->建立查询(将用户键入的信息转换为可处理的查询对象)-->搜索查询-->展现结果

    索引组件:Lucene
    搜索组件:Solr, ElasticSearch    
    注意:mysql数据库中的myisam引擎支持全文索引,但是格式比较复杂,不适于作为搜索

引擎的组件;
Lucene Core:
Apache LuceneTM is a high-performance, full-featured text search engine library written entirely in Java. It is a technology suitable for nearly any application that requires full-text search, especially cross-platform.
Solr:
SolrTM is a high performance search server built using Lucene Core, with XML/HTTP and JSON/Python/Ruby APIs, hit highlighting, faceted search, caching, replication, and a web admin interface.
ElasticSearch:
Elasticsearch is a distributed, RESTful search and analytics engine capable of solving a growing number of use cases. As the heart of the Elastic Stack, it centrally stores your data so you can discover the expected and uncover the unexpected.

Elastic Stack:
    ElasticSearch
    Logstash
        Logstash is an open source, server-side data processing pipeline that ingests data from a multitude of sources simultaneously, transforms it, and then sends it to your favorite “stash.” (Ours is Elasticsearch, naturally.)
    Beats:
        Filebeat:Log Files
        Metricbeat:Metrics
        Packetbeat:Network Data
        Winlogbeat:Windows Event Logs
        Heartbeat:Uptime Monitoring
    Kibana:
        Kibana lets you visualize your Elasticsearch data and navigate the Elastic Stack, so you can do anything from learning why you‘re getting paged at 2:00 a.m. to understanding the impact rain might have on your quarterly numbers.

    TF/IDF算法:   
        https://zh.wikipedia.org/wiki/Tf-idf

ES的核心组件:
    物理组件:
        集群:
            状态:green, yellow, red
        节点:
        Shard:

    Lucene的核心组件:
        索引(index):数据库(database)
        类型(type):表(table)
        文档(Document):行(row)
        映射(Mapping):

ElasticSearch 5的程序环境:
    配置文件:
        /etc/elasticsearch/elasticsearch.yml
        /etc/elasticsearch/jvm.options
        /etc/elasticsearch/log4j2.properties
    Unit File:elasticsearch.service
    程序文件:
        /usr/share/elasticsearch/bin/elasticsearch
        /usr/share/elasticsearch/bin/elasticsearch-keystore:
        /usr/share/elasticsearch/bin/elasticsearch-plugin:管理插件程序        

    搜索服务:
        9200/tcp

    集群服务:
        9300/tcp

els集群的工作逻辑:
    多播、单播:9300/tcp
    关键因素:clustername

    所有节点选举一个主节点,负责管理整个集群的状态(green/yellow/red),以及各shards的分布方式;

    插件:
elk实现框图:

注意:elk是由elastic stack search、logstash和kibana组成的,如图中间颜色比较暗的是elastic
stack search实现的部分,而下面的数据收集部分由logstash实现,最后kibana负责上方的图形搜
索界面接口;但是logstash数据收集器是由JRuby语言开发的,是用ruby语言先通过java解释器将
其翻译成java语言,之后进行编译执行,效率很低,故而出现了filebeat轻量级组件来代替它;
logstash是通过在每个要采集的日志服务器植入agent组件,一旦日志有变化就将改变的数据拉取
到logstash服务器进行数据的文档化,之后将文档化的数据交给elastic stack search集群进行相
关处理。由于基于lucene的solr搜索引擎在后期没有支持大数据分布式的存储,被elk所取代;
http://lucene.apache.org/ 将数据文档化之后数据形成索引的lucene网址
https://www.elastic.co/ elk访问地址,可以下载els镜像
https://db-engines.com/en/ 体现数据库地位的网址
elasticsearch集群: elasticsearch是由java开发的
准备工作:关闭防火墙、配置chrony时间同步、用本地文件进行dns解析
https://mirrors.cnnic.cn 清华大学的elastic stack search的镜像网站,下载速度快
yum install java-1.8.0-openjdk-devel -y
rpm -ivh elasticsearch-5.6.8.rpm java编写的
scp elasticsearch-5.6.8.rpm server2:/root/ 复制过去后进行rpm安装
scp elasticsearch-5.6.8.rpm server3:/root/
cd /etc/elasticsearch/
vim elasticsearch.yml
cluster.name: myels
node.name: server1
path.data: /els/data
path.logs: /els/logs 需要在外面创建目录,设置属组和属主为elasticsearch用户
network.host: 192.168.43.60
discovery.zen.ping.unicast.hosts: ["server1","server2","server3"]
discovery.zen.minimum_master_nodes: 1 2个节点正常就可以正常使用

vim jvm.options
-Xms1g 注意初始化值和最大值要相同
-Xmx1g
mkdir /els/{data,logs} -pv && chown -R elasticsearch.elasticsearch /els/*
scp elasticsearch.yml jvm.options server2:/etc/elasticsearch/
vim elasticsearch.yml
network.host: 192.168.43.63
node.name: server2
scp elasticsearch.yml jvm.options server3:/etc/elasticsearch/
vim elasticsearch.yml
network.host: 192.168.43.62
node.name: server3
java -version
systemctl daemon-reload && systemctl start elasticsearch
ss -ntl
curl http://server1:9200/ 看测试是否成功
tail /els/logs/myels.log 可以查看日志找错误
free -m 查看内存的大小,以便定虚拟机的初始化值
curl -XGET ‘http://server1:9200/_cluster/health?pretty=true‘ 发起查询请求

集群配置:
elasticsearch.yml配置文件:
cluster.name: myels
node.name: node1
path.data: /data/els/data
path.logs: /data/els/logs
network.host: 0.0.0.0
http.port: 9200 9200端口是客户端用的,9300是集群内部进行通信的
discovery.zen.ping.unicast.hosts: ["node1", "node2", "node3"]
discovery.zen.minimum_master_nodes: 2
· node.attr.rack: r1 表示可以集群分片到不同的机架,以防止机架中交换机断网

RESTful API:        crud操作(create、read、update、delete)
    curl  -X<VERB> ‘<PROTOCOL>://<HOST>:<PORT>/<PATH>?<QUERY_STRING>‘ -d ‘<BODY>‘
        <BODY>:json格式的请求主体;

    <VERB>  请求方法
        GET,POST,PUT,DELETE;GET 为默认的方法

    特殊PATH:/_cat, /_search, /_cluster

    <PATH>
        /index_name/type/Document_ID/

     curl -XGET ‘http://10.1.0.67:9200/_cluster/health?pretty=true‘

     curl -XGET ‘http://10.1.0.67:9200/_cluster/stats?pretty=true‘

    curl -XGET ‘http://10.1.0.67:9200/_cat/nodes?pretty‘

    curl -XGET ‘http://10.1.0.67:9200/_cat/health?pretty‘

curl http://server1:9200/_cat/indices 查看索引信息

    创建文档:
        curl  -XPUT  

    特殊PATH:/_cat, /_search, /_cluster

    文档:
        {"key1": "value1", "key2": value, ...}

ELS:分布式、开源、RESTful、近乎实时
    集群:一个或多个节点的集合;
    节点:运行的单个els实例;
    索引:切成多个独立的shard;(以Lucene的视角,每个shard即为一个独立而完整的索引)
        primary shard:r/w
        replica shard: r

查询:
    ELS:很多API
        _cluster, _cat, _search

    curl -X GET ‘<SCHEME://<HOST>:<PORT>/[INDEX/TYPE/]_search?q=KEYWORD&sort=DOMAIN:[asc|desc]&from=#&size=#&_source=DOMAIN_LIST‘

        /_search:搜索所有的索引和类型;
        /INDEX_NAME/_search:搜索指定的单个索引;
        /INDEX1,INDEX2/_search:搜索指定的多个索引;
        /s*/_search:搜索所有以s开头的索引;
        /INDEX_NAME/TYPE_NAME/_search:搜索指定的单个索引的指定类型;

    简单字符串的语法格式
        http://lucene.apache.org/core/6_6_0/queryparser/org/apache/lucene/queryparser/classic/package-summary.html#package.description

    查询类型:Query DSL,简单字符串;

        文本匹配的查询条件:
            (1) q=KEYWORD, 相当于q=_all:KEYWORD
            (2) q=DOMAIN:KEYWORD

                    {
                        "name" : "Docker in Action",
                        "publisher" : "wrox",
                        "datatime" : "2015-12-01",
                        "author" : "Blair"
                    }

                    _all: "Docker in Action Wrox 2015-12-01 Blair"

            修改默认查询域:df属性

        查询修饰符:
            https://www.elastic.co/guide/en/elasticsearch/reference/current/search-uri-request.html

        自定义分析器:
            analyzer=

        默认操作符:OR/AND
            default_operator, 默认值为OR

        返回字段:
            fields=

            注:5.X不支持;

        结果排序:
            sort=DOMAIN:[asc|desc]

        搜索超时:
            timeout=

        查询结果窗口:
            from=,默认为0;
            size=, 默认为10;

    Lucene的查询语法:
        q=
            KEYWORD
            DOMAIN:KEYWORD

        +DOMAIN:KEYWORD -DOMAIN:KEYWORD 

    els支持从多类型的查询:
        Full text queries

安装elasticsearch-head插件: 可以极大的简化命令行查询的复杂度
    5.X:
        (1) 设置elasticsearch.yml配置文件:
            http.cors.enabled: true
            http.cors.allow-origin: "*"

        (2) 安装head:
            $ git clone https://github.com/mobz/elasticsearch-head.git
            $ cd elasticsearch-head
            $ npm install   出错了,待解决
            $ npm run start

ELK:
    E: elasticsearch
    L: logstash,日志收集工具;
        ELK Beats Platform:
            PacketBeat:网络报文分析工具,统计收集报文信息;
            Filebeat:是logstash forwarder的替换者,因此是一个日志收集工具;
            Topbeat:用来收集系统基础数据,如cpu、内存、io等相关的统计信息;
            Winlogbeat
            Metricbeat
            用户自定义beat:
    logstash的安装:

wget https://mirrors.tuna.tsinghua.edu.cn/elasticstack/yum/elastic-5.x/5.6.8/logstash-
5.6.8.rpm
yum install java-1.8.0-openjdk-devel
rpm -ivh logstash-5.6.8.rpm
rpm -ql logstash|grep logstash$
vim /etc/logstash/conf.d/example1.conf
input{
stdin{}
}

output{
stdout{
codec => rubydebug
}
}
/usr/share/logstash/bin/logstash -f ./example1.conf -t 测试是否有语法错误
/usr/share/logstash/bin/logstash -f ./example1.conf 进行运行程序

    input {
        ...
    }

    filter{
        ...
    }

    output {
        ...
    }

    简单示例配置:

        input {
            stdin {}
        }

        output {
            stdout {
                codec => rubydebug
            }
        }

    示例2:从文件输入数据,经grok过滤器插件过滤之后输出至标准输出:
        input {
            file {
                path => ["/var/log/httpd/access_log"]
                start_position => "beginning"
            }
        }

        filter {
            grok {
                match => {
                    "message" => "%{COMBINEDAPACHELOG}"
                }
                remove_field: "message"
            }
        }

        output {
            stdout {
                codec => rubydebug
            }
        }

    示例3:date filter插件示例:
            filter {
                    grok {
                            match => {
                                    "message" => "%{HTTPD_COMBINEDLOG}"
                            }
                            remove_field => "message"
                    }
                    date {
                            match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
                            remove_field => "timestamp"
                    }

            }               

    插件:mutate(改变内容)
        The mutate filter allows you to perform general mutations on fields. You can rename, remove, replace, and modify fields in your events.

    示例4:mutate filter插件
        filter {
                grok {
                        match => {
                                "message" => "%{HTTPD_COMBINEDLOG}"
                        }
                }
                date {
                        match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
                }
                mutate {
                        rename => {
                                "agent" => "user_agent"
                        }
                }
        } 

    示例5:geoip插件

        filter {
                grok {
                        match => {
                                "message" => "%{HTTPD_COMBINEDLOG}"
                        }
                }
                date {
                        match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
                }
                mutate {
                        rename => {
                                "agent" => "user_agent"
                        }
                }
                geoip {
                        source => "clientip"
                        target => "geoip"
                        database => "/etc/logstash/maxmind/GeoLite2-City.mmdb"
                }
        }            
      echo ‘47.98.120.224 - - [31/May/2018:16:22:58 +0800] "GET / HTTP/1.1" 200 21 "-"  

"curl/7.29.0"‘ >> /var/log/httpd/access_log 追加httpd日志,看是否可以查询到ip地址信息

    示例3:使用Redis
        (1) 从redis加载数据
            input {
                redis {
                    batch_count => 1
                    data_type => "list"
                    key => "logstash-list"
                    host => "192.168.0.2"
                    port => 6379
                    threads => 5
                }
            } 

        (2) 将数据存入redis
            output {
                redis {
                    #data_type => "channel"
                    #key => "logstash-%{+yyyy.MM.dd}"

host => ["192.168.43.66"]
port => 6379
db => 8
data_type => "list"
key => "logstash-%{+YYYY.MMM.dd}"
}
}
/usr/share/logstash/bin/logstash -f ./example6.conf 运行程序将数据输出到redis中
注意:要刷新数据,在redis中找数据
redis中:
yum install redis
vim /etc/redis.conf
bind 0.0.0.0
systemctl restart redis
help @list 查看list命令
lrange logstash-2018.May.31 0 10 查看一定范围的数据
keys * 查看有无数据
select 8 切换到8号数据库
示例4:将数据写入els cluster

        output {
            elasticsearch {
                hosts => ["http://node1:9200/","http://node2:9200/","http://node3:9200/"]
                user => "ec18487808b6908009d3"
                password => "efcec6a1e0"
                index => "logstash-%{+YYYY.MM.dd}"
                document_type => "apache_logs"
            }
        }        

     示例5:综合示例,启用geoip

        input {
            beats {
                port => 5044
            }
        }

        filter {
            grok {
                match => { 
                "message" => "%{COMBINEDAPACHELOG}"
                }
                remove_field => "message"
            }
            geoip {
                source => "clientip"
                target => "geoip"
                database => "/etc/logstash/GeoLite2-City.mmdb"
            }
        }

        output {
            elasticsearch {
                hosts => ["http://172.16.0.67:9200","http://172.16.0.68:9200","http://172.16.0.69:9200"]
                index => "logstash-%{+YYYY.MM.dd}"
                action => "index"
                document_type => "apache_logs"
            }
        }        

    grok:
        %{SYNTAX:SEMANTIC}
            SYNTAX:预定义的模式名称;
            SEMANTIC:给模式匹配到的文本所定义的键名;

            1.2.3.4 GET /logo.jpg  203 0.12
            %{IP:clientip} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

            { clientip: 1.2.3.4, method: GET, request: /logo.jpg, bytes: 203, duration: 0.12}

            %{IPORHOST:client_ip} %{USER:ident} %{USER:auth} [%{HTTPDATE:timestamp}] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-)" %{HOST:domain} %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QS:referrer} %{QS:agent} "(%{WORD:x_forword}|-)" (%{URIHOST:upstream_host}|-) %{NUMBER:upstream_response} (%{WORD:upstream_cache_status}|-) %{QS:upstream_content_type} (%{BASE16FLOAT:upstream_response_time}) > (%{BASE16FLOAT:request_time})

             "message" => "%{IPORHOST:clientip} [%{HTTPDATE:time}] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:http_status_code} %{NUMBER:bytes} "(?<http_referer>S+)" "(?<http_user_agent>S+)" "(?<http_x_forwarded_for>S+)""

             filter {
                grok {
                    match => {
                        "message" => "%{IPORHOST:clientip} [%{HTTPDATE:time}] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:http_status_code} %{NUMBER:bytes} "(?<http_referer>S+)" "(?<http_user_agent>S+)" "(?<http_x_forwarded_for>S+)""
                    }
                    remote_field: message
                }   
            }

            nginx.remote.ip
            [nginx][remote][ip] 

            filter {
                grok {
                    match => { "message" => ["%{IPORHOST:[nginx][access][remote_ip]} - %{DATA:[nginx][access][user_name]} [%{HTTPDATE:[nginx
                    ][access][time]}] "%{WORD:[nginx][access][method]} %{DATA:[nginx][access][url]} HTTP/%{NUMBER:[nginx][access][http_version]}                    " %{NUMBER:[nginx][access][response_code]} %{NUMBER:[nginx][access][body_sent][bytes]} "%{DATA:[nginx][access][referrer]}" "
                    %{DATA:[nginx][access][agent]}""] }
                    remove_field => "message"
                }  
                date {
                    match => [ "[nginx][access][time]", "dd/MMM/YYYY:H:m:s Z" ]
                    remove_field => "[nginx][access][time]"
                }  
                useragent {
                    source => "[nginx][access][agent]"
                    target => "[nginx][access][user_agent]"
                    remove_field => "[nginx][access][agent]"
                }  
                geoip {
                    source => "[nginx][access][remote_ip]"
                    target => "geoip"
                    database => "/etc/logstash/GeoLite2-City.mmdb"
                }  

            }   

            output {                                                                                                     
                elasticsearch {                                                                                      
                    hosts => ["node1:9200","node2:9200","node3:9200"]                                            
                    index => "logstash-ngxaccesslog-%{+YYYY.MM.dd}"                                              
                }                                                                                                    
            }

            注意:
                1、输出的日志文件名必须以“logstash-”开头,方可将geoip.location的type自动设定为"geo_point";
                2、target => "geoip"

    除了使用grok filter plugin实现日志输出json化之外,还可以直接配置服务输出为json格式;

    示例:使用grok结构化nginx访问日志 
        filter {
                grok {
                        match => {
                                "message" => "%{HTTPD_COMBINEDLOG} "%{DATA:realclient}""
                        }
                        remove_field => "message"
                }
                date {
                        match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
                        remove_field => "timestamp"
                }
        }            

    示例:使用grok结构化tomcat访问日志 
        filter {
                grok {
                        match => {
                                "message" => "%{HTTPD_COMMONLOG}"
                        }
                        remove_field => "message"
                }
                date {
                        match => ["timestamp","dd/MMM/YYYY:H:m:s Z"]
                        remove_field => "timestamp"
                }
        } 

    Nginx日志Json化:
        log_format   json  ‘{"@timestamp":"$time_iso8601",‘
                    ‘"@source":"$server_addr",‘
                    ‘"@nginx_fields":{‘
                        ‘"client":"$remote_addr",‘
                        ‘"size":$body_bytes_sent,‘
                        ‘"responsetime":"$request_time",‘
                        ‘"upstreamtime":"$upstream_response_time",‘
                        ‘"upstreamaddr":"$upstream_addr",‘
                        ‘"request_method":"$request_method",‘
                        ‘"domain":"$host",‘
                        ‘"url":"$uri",‘
                        ‘"http_user_agent":"$http_user_agent",‘
                        ‘"status":$status,‘
                        ‘"x_forwarded_for":"$http_x_forwarded_for"‘
                    ‘}‘
                ‘}‘;

        access_log  logs/access.log  json;                  

Conditionals
Sometimes you only want to filter or output an event under certain conditions. For that, you can use a conditional.

Conditionals in Logstash look and act the same way they do in programming languages. Conditionals support if, else if and else statements and can be nested.

The conditional syntax is:

    if EXPRESSION {
    ...
    } else if EXPRESSION {
    ...
    } else {
    ...
    }    

    What’s an expression? Comparison tests, boolean logic, and so on!

    You can use the following comparison operators:

    equality: ==, !=, <, >, <=, >=
    regexp: =~, !~ (checks a pattern on the right against a string value on the left) inclusion: in, not in

    The supported boolean operators are:

        and, or, nand, xor

    The supported unary operators are:

        !
    Expressions can be long and complex. Expressions can contain other expressions, you can negate expressions with !, and you can group them with parentheses (...).

    filter {

        if [type] == ‘tomcat-accesslog‘ {
            grok {}
        }

        if [type] == ‘httpd-accesslog‘ {
            grok {}
        }

    }

1、lucene索引组件
lucene由3部分组成:
index:对应于db数据,每个索引为一个db
type:对应于table数据,如每个应用的日志都是不同的,放在不同的table中
document:对应于row数据,是键值对组的存放
mapping:映射,对每个字段key的数据类型进行规定
2、es组件

在存储索引时采用节点集群存储,索引进行分片处理,以增加冗余度;分片具有主副。在搜索时
通过总线调度到存储即可,不用在文件中写死;es组件集成了lucene,是中间的一部分实现

3、es搜索组件的集群状态

es的集群状态由3中颜色进行表示:
green:所有shard主副片口可以正常使用
yellow:存在某个或某些分片缺少主或副
red:存在某个或某些分片同时缺少主和副
如果发生网络分区,两个节点之间不能进行通信了,这时候就会造成脑裂,故而需要进行quorum
投票选择哪个作为正常节点正常工作,哪个下线等待。所以此集群需要奇数个节点

4、logstash插件讲解
logstash插件通过输入插件从指定数据源获取数据,根据输出插件将处理过的数据输出到指定目
标,中间是进行数据过滤的插件,对数据进行文档化处理等操作;
https://www.elastic.co/guide/en/logstash/5.6/index.html 参考文档
logstash使用架构:
logstash服务器-->elasticsearch
logstash服务器/filebeat服务器-->redis服务器-->logstash服务器-->elasticsearch

5、logstash架构的实现实验
logstash服务器/filebeat服务器-->redis服务器-->logstash服务器-->elasticsearch
(1)filebeat的配置 主机ip 192.168.43.61
wget https://mirrors.tuna.tsinghua.edu.cn/elasticstack/yum/elastic-5.x/5.6.8/filebeat-5.6.8-
x86_64.rpm 先到清华的镜像站点下载filebeat软件
rpm -ivh filebeat-5.6.8-x86_64.rpm 安装软件
cd /etc/filebeat/
vim filebeat.yml
paths:

  • /var/log/httpd/access_log
    #------------------------------- Redis output ----------------------------------
    output.redis:
    enabled: true
    hosts: ["192.168.43.66:6379"]
    db: 6
    datatype: list
    key: filebeat
    systemclt start filebeat
    以上过程实现了filebeat将收集到的数据存入redis
    yum install httpd 日志来源
    echo ‘<h1>HelloWorld</h1>’ > /var/www/html/index.html
    systemctl start httpd
    echo ‘223.5.5.5 - - [1/June/2018:14:03:58 +0800] "GET / HTTP/1.1" 200 21 "-" "curl/7.29.0"‘

    > /var/log/httpd/access_log 创造公网ip的日志以提供日志来源
    curl http://172.18.62.61 访问页面创造日志
    (2)redis的配置 主机ip 192.168.43.66
    yum install redis
    vim /etc/redis.conf
    bind 0.0.0.0
    systemctl start redis
    redis-cli 连接redis

(3)logstash服务器配置 主机ip 192.168.43.61
wget https://mirrors.tuna.tsinghua.edu.cn/elasticstack/yum/elastic-5.x/5.6.8/logstash-
5.6.8.rpm
yum install java-1.8.0-openjdk-devel
rpm -ivh logstash-5.6.8.rpm
rpm -ql logstash|grep logstash$
vim /etc/logstash/conf.d/example8.conf
input{
redis {
host => "192.168.43.66"
port => 6379
db => 6
key => "filebeat"
data_type => "list"
threads => 6
}
}

filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
remove_field => "message"
}
date {
match => [ "timestamp", "dd/MMM/YYYY:H:m:s Z" ]
remove_field => "timestamp"
}
geoip {
source => "clientip"
target => "geoip"
database => "/etc/logstash/maxmind/GeoLite2-City.mmdb"
}
}

output{
elasticsearch {
hosts => ["http://server1:9200","http://server2:9200","http://server3:9200"]
index => "logstash-%{+YYYY.MM.dd}"
document_type => "apache_logs"
}
}

/usr/share/logstash/bin/logstash -f ./example1.conf -t 测试是否有语法错误
systemctl start logstash
(4)elasticsearch的配置 主机ip 192.168.43.60/62/63
准备工作:关闭防火墙、配置chrony时间同步、用本地文件进行dns解析
https://mirrors.cnnic.cn 清华大学的elastic stack search的镜像网站,下载速度快
yum install java-1.8.0-openjdk-devel -y
rpm -ivh elasticsearch-5.6.8.rpm java编写的
scp elasticsearch-5.6.8.rpm server2:/root/ 复制过去后进行rpm安装
scp elasticsearch-5.6.8.rpm server3:/root/
cd /etc/elasticsearch/
vim elasticsearch.yml
cluster.name: myels
node.name: server1
path.data: /els/data
path.logs: /els/logs 需要在外面创建目录,设置属组和属主为elasticsearch用户
network.host: 192.168.43.60
discovery.zen.ping.unicast.hosts: ["server1","server2","server3"]
discovery.zen.minimum_master_nodes: 1 2个节点正常就可以正常使用

vim jvm.options
-Xms1g 注意初始化值和最大值要相同
-Xmx1g
mkdir /els/{data,logs} -pv && chown -R elasticsearch.elasticsearch /els/*
scp elasticsearch.yml jvm.options server2:/etc/elasticsearch/
vim elasticsearch.yml
network.host: 192.168.43.63
node.name: server2
scp elasticsearch.yml jvm.options server3:/etc/elasticsearch/
vim elasticsearch.yml
network.host: 192.168.43.62
node.name: server3
java -version
systemctl daemon-reload && systemctl start elasticsearch
ss -ntl
curl http://server1:9200/ 看测试是否成功
tail /els/logs/myels.log 可以查看日志找错误
free -m 查看内存的大小,以便定虚拟机的初始化值
curl -XGET ‘http://server1:9200/_cluster/health?pretty=true‘ 发起查询请求
(5)kibana的配置
wget https://mirrors.tuna.tsinghua.edu.cn/elasticstack/yum/elastic-5.x/5.6.8/kibana-5.6.8-
x86_64.rpm 从清华镜像网站下载kibana图形界面工具
rpm -ivh kibana-5.6.8-x86_64.rpm 安装软件
cd /etc/kibana/
vim kibana.yml
server.port: 5601 监听端口
server.host: "0.0.0.0" 允许任意主机访问
server.name: "redis" 主机名
elasticsearch.url: "http://server1:9200" 所连接的数据接口
elasticsearch.preserveHost: true
kibana.index: ".kibana" 一些所以文件的放处
systemclt restart kibana
http://172.18.62.66:5601 在浏览器页面访问

以上是关于Elk的理论与实践的主要内容,如果未能解决你的问题,请参考以下文章

企业级日志系统 ELK 原理与实践详细介绍

震惊全网的ELK日志分析系统(齐全详细理论+搭建步骤图释)

震惊全网的ELK日志分析系统(齐全详细理论+搭建步骤图释)

CV学习资料《卷积神经网络与视觉计算》+《深度学习实践计算机视觉》+《视觉SLAM十四讲从理论到实践》电子资料代码分析

如何将理论效率与实践示例进行比较[关闭]

对比学习资料《深度学习入门:基于Python的理论与实现》+《深度学习原理与实践》+《深度学习理论与实战基础篇》电子资料