企业级日志系统架构——ELK（Elasticsearch、Filebeat、Kafka、Logstash、Kibana）

一、概述

ELK是三个开源软件的缩写，分别表明：Elasticsearch , Logstash, Kibana , 它们都是开源软件。新增了一个FileBeat，它是一个轻量级的日志搜集处理东西(Agent)，Filebeat占用资源少，适合于在各个服务器上搜集日志后传输给Logstash，官方也引荐此东西。

大致流程图如下：

企业级日志系统架构——ELK（Elasticsearch、Filebeat、Kafka、Logstash、Kibana）

1）Elasticsearch 存储

Elasticsearch是个开源分布式搜索引擎，供给搜集、剖析、存储数据三大功用。它的特色有：分布式，零装备，主动发现，索引主动分片，索引副本机制，restful风格接口，多数据源，主动搜索负载等。

2）Filebeat 日志数据搜集

filebeat是Beats中的一员，Beats在是一个轻量级日志搜集器，其实Beats家族有6个成员，早期的ELK架构中运用Logstash搜集、解析日志，可是Logstash对内存、cpu、io等资源消耗比较高。相比Logstash，Beats所占体系的CPU和内存简直能够忽略不计。

Filebeat是用于转发和集中日志数据的轻量级传送东西。Filebeat监视您指定的日志文件或位置，搜集日志事情。

目前Beats包含六种东西：

Packetbeat：网络数据（搜集网络流量数据）
Metricbeat：方针（搜集体系、进程和文件体系级别的CPU和内存运用情况等数据）
Filebeat：日志文件（搜集文件数据）
Winlogbeat：windows事情日志（搜集Windows事情日志数据）
Auditbeat：审计数据（搜集审计日志）
Heartbeat：运转时间监控（搜集体系运转时的数据）

工作的流程图如下：

长处

Filebeat 只是一个二进制文件没有任何依靠。它占用资源很少。

缺陷

Filebeat 的运用规模非常有限，因此在某些场景下咱们会碰到问题。在 5.x 版别中，它还具有过滤的才能。

3）Kafka

kafka能协助咱们削峰。ELK能够运用redis作为音讯行列，但redis作为音讯行列不是强项并且redis集群不如专业的音讯发布体系kafka。kafka装置能够参阅我之前的文章：Kafka原理介绍+装置+基本操作（kafka on k8s）。

4）Logstash 过滤

Logstash 主要是用来日志的搜集、剖析、过滤日志的东西，支持许多的数据获取方法。一般工作方法为c/s架构，client端装置在需求搜集日志的主机上，server端负责将收到的各节点日志进行过滤、修正等操作在同时发往elasticsearch上去。

长处

可伸缩性

节拍应该在一组Logstash节点之间进行负载平衡。建议至少运用两个Logstash节点以完成高可用性。每个Logstash节点只布置一个Beats输入是很常见的，但每个Logstash节点也能够布置多个Beats输入，以便为不同的数据源揭露独立的端点。

弹性

Logstash耐久行列供给跨节点故障的维护。对于Logstash中的磁盘级弹性，确保磁盘冗余非常重要。对于内部布置，建议您装备RAID。在云或容器化环境中运转时，建议您运用具有反映数据SLA的仿制战略的永久磁盘。

可过滤

对事情字段履行惯例转换。您能够重命名，删除，替换和修正事情中的字段。

缺陷

Logstash耗资源较大，运转占用CPU和内存高。别的没有音讯行列缓存，存在数据丢掉危险。

5）Kibana 展现

Kibana 也是一个开源和免费的东西，Kibana能够为 Logstash 和 ElasticSearch 供给的日志剖析友好的 Web 界面，能够协助汇总、剖析和搜索重要数据日志。

filebeat和logstash的关系

由于logstash是jvm跑的，资源消耗比较大，所以后来作者又用golang写了一个功用较少可是资源消耗也小的轻量级的logstash-forwarder。不过作者只是一个人，参加elastic.co公司以后，由于es公司本身还收买了另一个开源项目packetbeat，而这个项目专门便是用golang的，有整个团队，所以es公司干脆把logstash-forwarder的开发工作也合并到同一个golang团队来搞，于是新的项目就叫filebeat了。

二、helm3装置ELK

详细流程图如下：

1）预备条件

1、添加helm仓库

$ helm repo add elastic   https://helm.elastic.co

2）helm3装置elasticsearch

1、自界说values

主要是设置storage Class 耐久化和资源约束，本人电脑资源有限，所以这儿就把资源调小了许多，小伙伴们能够根据自己装备自界说哈。

$ # 集群称号
clusterName: "elasticsearch"
# ElasticSearch 6.8+ 默许装置了 x-pack 插件，部分功用免费，这儿选禁用
esConfig:
 elasticsearch.yml: |
    network.host: 0.0.0.0
    cluster.name: "elasticsearch"
    xpack.security.enabled: false
resources:
  requests:
    memory: 1Gi
volumeClaimTemplate:
  storageClassName: "bigdata-nfs-storage"
  accessModes: [ "ReadWriteOnce" ]
  resources:
    requests:
      storage: 3Gi
service:
  type: NodePort
  port: 9000
  nodePort: 31311

禁用Kibana安全提示（Elasticsearch built-in security features are not enabled） xpack.security.enabled: false

2、开端装置Elasitcsearch

装置过程比较慢，由于官方镜像下载比较慢

$ helm install es elastic/elasticsearch -f my-values.yaml  --namespace bigdata

W1207 23:10:57.980283   21465 warnings.go:70] policy/v1beta1 PodDisruptionBudget is deprecated in v1.21+, unavailable in v1.25+; use policy/v1 PodDisruptionBudget
W1207 23:10:58.015416   21465 warnings.go:70] policy/v1beta1 PodDisruptionBudget is deprecated in v1.21+, unavailable in v1.25+; use policy/v1 PodDisruptionBudget
NAME: es
LAST DEPLOYED: Tue Dec  7 23:10:57 2021
NAMESPACE: bigdata
STATUS: deployed
REVISION: 1
NOTES:
1. Watch all cluster members come up.
  $ kubectl get pods --namespace=bigdata -l app=elasticsearch-master -w2. Test cluster health using Helm test.
  $ helm --namespace=bigdata test es

查看，需求所有pod都正常运转才正常，下载镜像有点慢，需求稍等一段时间再查看

$ kubectl get pod -n bigdata -l app=elasticsearch-master
$ kubectl get pvc -n bigdata
$ watch kubectl get pod -n bigdata -l app=elasticsearch-master

3、验证

$ helm --namespace=bigdata test es
$ kubectl get pod,svc -n bigdata -l app=elasticsearch-master -o wide
$ curl 192.168.0.113:31311/_cat/health
$ curl 192.168.0.113:31311/_cat/nodes

4、整理

$ helm uninstall es -n bigdata
$ kubectl delete pvc elasticsearch-master-elasticsearch-master-0 -n bigdata
$ kubectl delete pvc elasticsearch-master-elasticsearch-master-1 -n bigdata
$ kubectl delete pvc elasticsearch-master-elasticsearch-master-2 -n bigdata

3）helm3装置Kibana

1、自界说values

域名（elasticsearch-master-headless.bigdata.svc.cluster.local）的由来不清楚的，能够参阅我之前的文章：Kubernetes（k8s）DNS（CoreDNS）介绍

$ cat <<EOF> my-values.yaml
#此处修正了kibana的装备文件，默许位置/usr/share/kibana/kibana.yaml
kibanaConfig:
   kibana.yml: |
     server.port: 5601
     server.host: "0.0.0.0"
     elasticsearch.hosts: [ "elasticsearch-master-headless.bigdata.svc.cluster.local:9200" ]
resources:
  requests:
    cpu: "1000m"
    memory: "256Mi"
  limits:
    cpu: "1000m"
    memory: "1Gi"
service:
  #type: ClusterIP
  type: NodePort
  loadBalancerIP: ""
  port: 5601
  nodePort: "30026"
EOF

2、开端装置Kibana

$ helm install kibana elastic/kibana -f my-values.yaml  --namespace bigdata

3、验证

$ kubectl get pod,svc -n bigdata -l app=kibana

浏览器访问：http://192.168.0.113:30026/

4、整理

$ helm uninstall kibana -n bigdata

4）helm3装置Filebeat

filebeat默许搜集宿主机上docker的日志途径：/var/lib/docker/containers。如果咱们修正了docker的装置途径要怎样搜集呢，很简单修正chart里的DaemonSet文件里面的hostPath参数：

- name: varlibdockercontainers
  hostPath:
    path: /var/lib/docker/containers   #改为docker装置途径

当然也能够自界说values修正，这儿引荐自界说values方法修正搜集日志途径

1、自界说values

默许是将数据存储到ES，这儿做修正数据存储到Kafka

$ cat <<EOF> my-values.yaml
daemonset:
  filebeatConfig:
    filebeat.yml: |
      filebeat.inputs:
      - type: container
        paths:
          - /var/log/containers/*.log
      output.elasticsearch:
        enabled: false
        host: '${NODE_NAME}'
        hosts: '${ELASTICSEARCH_HOSTS:elasticsearch-master:9200}'
      output.kafka:
       enabled: true
       hosts: ["kafka-headless.bigdata.svc.cluster.local:9092"]
       topic: test
EOF

2、开端装置Filefeat

$ helm install filebeat elastic/filebeat -f my-values.yaml  --namespace bigdata
$ kubectl get pods --namespace=bigdata -l app=filebeat-filebeat -w

3、验证

# 先登录kafka客户端
$ kubectl exec --tty -i kafka-client --namespace bigdata -- bash
# 再消费数据
$ kafka-console-consumer.sh --bootstrap-server kafka.bigdata.svc.cluster.local:9092 --topic test

看到现已能够消费数据了，阐明数据现已存储到kafka了。

查看kafka数据积压情况

$ kubectl exec --tty -i kafka-client --namespace bigdata -- bash
$ kafka-consumer-groups.sh --bootstrap-server kafka-0.kafka-headless.bigdata.svc.cluster.local:9092 --describe --group mygroup

发现许多数据都是处于积压的状况

接下来便是布置logstash去消费kafka数据，最终存储到ES。

4、整理

$ helm uninstall filebeat -n bigdata

5）helm3装置Logstash

1、自界说values

【注意】记得把ES和kafka的地址换成自己环境的。

$ cat <<EOF> my-values.yaml
logstashConfig:
  logstash.yml: |
    xpack.monitoring.enabled: false
logstashPipeline: 
   logstash.yml: |
    input {
      kafka {
            bootstrap_servers => "kafka-headless.bigdata.svc.cluster.local:9092"
            topics => ["test"]
            group_id => "mygroup"
            #如果运用元数据就不能运用下面的byte字节序列化，不然会报错
            #key_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer"
            #value_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer"
            consumer_threads => 1
            #默许为false，只有为true的时候才会获取到元数据
            decorate_events => true
            auto_offset_reset => "earliest"
         }
    }
    filter {
      mutate {
        #从kafka的key中获取数据并依照逗号切开
        split => ["[@metadata][kafka][key]", ","]
        add_field => {
            #将切开后的第一位数据放入自界说的“index”字段中
            "index" => "%{[@metadata][kafka][key][0]}"
        }
      }
    }
    output { 
      elasticsearch {
          pool_max => 1000
          pool_max_per_route => 200
          hosts => ["elasticsearch-master-headless.bigdata.svc.cluster.local:9200"]
          index => "test-%{+YYYY.MM.dd}"
      }
    }
# 资源约束
resources:
  requests:
    cpu: "100m"
    memory: "256Mi"
  limits:
    cpu: "1000m"
    memory: "1Gi"
volumeClaimTemplate:
  accessModes: ["ReadWriteOnce"]
  resources:
    requests:
      storage: 3Gi
EOF

output plugin 输出插件，将事情发送到特定方针：

stdout { codec => rubydebug } // 开启debug形式，可在控制台输出

stdout ：标准输出。将事情输出到屏幕上

output{
    stdout{
        codec => "rubydebug"
    }
}

file ：将事情写入文件

output{
   file {
       path => "/data/logstash/%{host}/{application}
       codec => line { format => "%{message}"} }
    }
}

kafka ：将事情发送到kafka

output{
   kafka{
        bootstrap_servers => "localhost:9092"
        topic_id => "test_topic"  #必需的设置。生成音讯的主题
    }
}

elasticseach ：在es中存储日志

output{
   elasticsearch {
        #user => elastic
        #password => changeme
        hosts => "localhost:9200"
        index => "nginx-access-log-%{+YYYY.MM.dd}"  
    }
}

2、开端装置Logstash

$ helm install logstash elastic/logstash -f my-values.yaml  --namespace bigdata

$ kubectl get pods --namespace=bigdata -l app=logstash-logstash

3、验证

1、登录kibana查看索引是否创立

2、查看logs

$ kubectl logs -f  logstash-logstash-0 -n bigdata >logs
$ tail -100 logs

3、查看kafka消费情况

$ kubectl exec --tty -i kafka-client --namespace bigdata -- bash
$ kafka-consumer-groups.sh --bootstrap-server kafka-0.kafka-headless.bigdata.svc.cluster.local:9092 --describe --group mygroup

4、经过kibana查看索引数据（Kibana版别：7.15.0）创立索引形式

Management-》Stack Management-》Kibana-》Index patterns

经过上面创立的索引形式查询数据（Discover）

4、整理

$ helm uninstall logstash -n bigdata

三、ELK相关的备份组件和备份方法

Elasticsearch备份两种方法：

将数据导出成文本文件，比方经过 elasticdump、esm 等东西将存储在 Elasticsearch 中的数据导出到文件中。适用数据量小的场景。
备份 elasticsearch data 目录中文件的形式来做快照，凭借 Elasticsearch 中 snapshot 接口完成的功用。适用大数据量的场景。

1）Elasticsearch的snapshot快照备份

长处：经过snapshot拍摄快照，然后界说快照备份战略，能够完成快照主动化存储，能够界说各种战略来满足自己不同的备份
缺陷：复原不行灵活，拍摄快照进行备份很快，可是复原的时候没办法随意进行复原，类似虚拟机快照

1、装备备份目录

在 elasticsearch.yml 的装备文件中注明能够用作备份途径 path.repo ，如下所示：

path.repo: ["/mount/backups", "/mount/longterm_backups"]

装备好后，就能够运用 snapshot api 来创立一个 repository 了，如下咱们创立一个名为 my_backup 的 repository。

PUT /_snapshot/my_backup
{
  "type": "fs",
  "settings": {
    "location": "/mount/backups/my_backup"
  }
}

2、开端经过API接口备份 有了 repostiroy 后，咱们就能够做备份了，也叫快照，也便是记录当下数据的状况。如下所示咱们创立一个名为 snapshot_1 的快照。

PUT /_snapshot/my_backup/snapshot_1?wait_for_completion=true

【温馨提示】wait_for_completion 为 true 是指该 api 在备份履行结束后再回来成果，不然默许是异步履行的，咱们这儿为了立刻看到效果，所以设置了该参数，线上履行时不必设置该参数，让其在后台异步履行即可。

3、增量备份

PUT /_snapshot/my_backup/snapshot_2?wait_for_completion=true

当履行结束后，你会发现 /mount/backups/my_backup 体积变大了。这阐明新数据备份进来了。要阐明的一点是，当你在同一个 repository 中做多次 snapshot 时，elasticsearch 会查看要备份的数据 segment 文件是否有改变，如果没有改变则不处理，不然只会把发生改变的 segment file 备份下来。这其实就完成了增量备份。

4、数据康复 经过调用如下 api 即可快速完成康复功用：

POST /_snapshot/my_backup/snapshot_1/_restore?wait_for_completion=true
{
  "indices": "index_1",
  "rename_replacement": "restored_index_1"
}

2）elasticdump备份搬迁es数据

索引数据导出为文件（备份）

# 导出索引Mapping数据
elasticdump \
  --input=http://es实例IP:9200/index_name/index_type \
  --output=/data/my_index_mapping.json \    # 寄存目录
  --type=mapping 
# 导出索引数据
elasticdump \
  --input=http://es实例IP:9200/index_name/index_type \
  --output=/data/my_index.json \
  --type=data

索引数据文件导入至索引（康复）

# Mapping 数据导入至索引
elasticdump \
  --output=http://es实例IP:9200/index_name \
  --input=/home/indexdata/roll_vote_mapping.json \ # 导入数据目录
  --type=mapping
# ES文档数据导入至索引
elasticdump \
  --output=http:///es实例IP:9200/index_name \
  --input=/home/indexdata/roll_vote.json \ 
  --type=data

可直接将备份数据导入另一个es集群

elasticdump --input=http://127.0.0.1:9200/test_event   --output=http://127.0.0.2:9200/test_event --type=data

type类型

type是ES数据导出导入类型，Elasticdump东西支持以下数据类型：

type类型	阐明
mapping	ES的索引映射结构数据
data	ES的数据
settings	ES的索引库默许装备
analyzer	ES的分词器
template	ES的模板结构数据
alias	ES的索引别号
### 3）esm备份搬迁es数据
备份es数据

esm -s http://10.33.8.103:9201 -x "petition_data" -b 5 --count=5000 --sliced_scroll_size=10 --refresh -o=./es_backup.bin

-w 表明线程数 -b 表明一次bulk恳求数据大小，单位MB默许 5M -c 一次scroll恳求数量导入康复es数据

esm -d http://172.16.20.20:9201 -y "petition_data6" -c 5000 -b 5 --refresh -i=./dump.bin

四、彩蛋

还有个日志体系架构跟ELK架构很相似（Elasticsearch、Flume、Kafka、Flink、Kibana），只是把Filebeat换成了Flume，Logstash换成了Flink。后边也会写篇文章分享出来，请耐心等待……

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

企业级日志系统架构——ELK（Elasticsearch、Filebeat、Kafka、Logstash、Kibana）

一、概述

1）Elasticsearch 存储

2）Filebeat 日志数据搜集

3）Kafka

4）Logstash 过滤

5）Kibana 展现

filebeat和logstash的关系

二、helm3装置ELK

1）预备条件

2）helm3装置elasticsearch

1、自界说values

2、开端装置Elasitcsearch

3、验证

4、整理

3）helm3装置Kibana

1、自界说values

2、开端装置Kibana

3、验证

4、整理

4）helm3装置Filebeat

1、自界说values

2、开端装置Filefeat

3、验证

4、整理

5）helm3装置Logstash

1、自界说values

2、开端装置Logstash

3、验证

4、整理

三、ELK相关的备份组件和备份方法

1）Elasticsearch的snapshot快照备份

2）elasticdump备份搬迁es数据

四、彩蛋

近期文章

近期评论