ELFK 添加过滤条件以及ela集群

作者: zxhChex | 来源:发表于2019-11-21 21:25 被阅读0次

ELFK 添加过滤条件以及ela集群
ELK+kafka+filebeat搭建生产ELFK集群
SparkJoin
CentOS7安装Redis集群
搭建Ela集群
List之Where方法
IDEA使用.ignore插件
0901_VS_DB_Modify
金蝶云苍穹如何设置列表界面？
Elasticsearch 简介

之前我们已经可以把kibana把日志展示出来了现在我们可以增加一些新的功能比如增加过滤条件，在上面这个文章所构建的环境的基础上。

基本做法就是

在日志挖掘的logstash的配置文件中添加filter条件
这里是在 `logstash/logstash_stdout.conf`中添加

使用`Grok Filter Plugin` 编辑解析Web日志

现在，您有一个工作流程，从Filebeat读取日志行。但是您会注意到日志消息的格式不理想。您要解析日志消息以从日志中创建特定的命名字段。为此，您将使用grok过滤器插件。

因为grok过滤器插件在传入的日志数据中查找模式，所以配置插件需要您决定如何识别用例感兴趣的模式。来自Web服务器日志示例的代表行如下所示：

信息	字段名
IP Address	clientip
User ID	ident
User Authentication	auth
timestamp	timestamp
HTTP Verb	verb
Request body	request
HTTP Version	httpversion
HTTP Status Code	response
Bytes served	bytes
Referrer URL	referrer
User agent	agent

使用方法，编辑 logstash/logstash_stdout.conf
写入下面的内容：

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

假如你在这之前已经运行了logstash 和 filebeat。要想生效现在的过滤配置，您需要强制Filebeat从头开始读取日志文件。
不必重新启动Logstash来接收更改，但是需要删除 filebeat下的注册表文件 registry，此文件一般在安装目录下的 data目录下。

由于Filebeat存储在注册表中收集的每个文件的状态，因此删除注册表文件会强制Filebeat读取从头开始捕获的所有文件。

接下来，使用以下命令重新启动Filebeat即可

使用Geoip Filter插件修改数据
除了解析日志数据以获得更好的搜索之外，过滤插件也可以从现有数据中导出补充信息。例如，geoip插件会查找IP地址，从地址中导出地理位置信息，并将该位置信息添加到日志中。

将Logstash实例配置为使用geoip过滤器插件，将以下行添加到文件的filter部分

geoip {
        source => "clientip"
    }

完整的示例：

input {
  beats {
    port => 5044
    host => "0.0.0.0"
  }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}

output {
  elasticsearch {             -----输出到ela中
    hosts => ["elasticsearch:9200"]
    manage_template => false
    index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
  }
  # stdout { codec => rubydebug }  # 假如有问题，可以打开此行进行调试
}

修改好保存，生效的话，同样先删除Filebeat 注册文件，之后重启 filebeat

kibana截图展示

image

测试您的管道编辑
现在，Logstash管道配置为将数据索引到Elasticsearch集群中，您可以查询Elasticsearch。

过滤nginx的logstash/logstash_stdout.conf

input {
  beats {
    port => 5044
    host => "0.0.0.0"
  }
}
filter {
  if ([fileset][module] == "nginx") {
    if ([fileset][name] == "access") {
      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"
      }
      mutate {
        add_field => { "read_timestamp" => "%{@timestamp}" }
      }
      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 => "[nginx][access][geoip]"
      }
    }
    else if [fileset][name] == "error" {
      grok {
        match => { "message" => ["%{DATA:[nginx][error][time]} \[%{DATA:[nginx][error][level]}\] %{NUMBER:[nginx][error][pid]}#%{NUMBER:[nginx][error][tid]}: (\*%{NUMBER:[nginx][error][connection_id]} )?%{GREEDYDATA:[nginx][error][message]}"] }
        remove_field => "message"
      }
      mutate {
        rename => { "@timestamp" => "read_timestamp" }
      }
      date {
        match => [ "[nginx][error][time]", "YYYY/MM/dd H:m:s" ]
        remove_field => "[nginx][error][time]"
      }
    }
  }
}
output {
  elasticsearch {
    hosts => ["elasticsearch:9200"] 
    manage_template => false
    index => "ddd---%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
  }
  # stdout { codec => rubydebug }  # 假如有问题，可以打开此行进行调试
}

ELA集群

文件 docker-compose.yml 内容

version: "3.2"
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.4.2
    container_name: elasticsearch
    networks:
      - elk-net
    ports:
      - "9200:9200"
    environment:
      - discovery.zen.minimum_master_nodes=2
      - node.name=elasticsearch
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es02,es03
      - cluster.initial_master_nodes=elasticsearch,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1

  es02:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.4.2
    container_name: es02
    networks:
      - elk-net
    environment:
      - discovery.zen.minimum_master_nodes=2
      - node.name=es02
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=elasticsearch,es03
      - cluster.initial_master_nodes=elasticsearch,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1

  es03:
    image: docker.elastic.co/elasticsearch/elasticsearch:7.4.2
    container_name: es03
    networks:
       - elk-net
    environment:
      - discovery.zen.minimum_master_nodes=2
      - node.name=es03
      - cluster.name=es-docker-cluster
      - discovery.seed_hosts=es02,elasticsearch
      - cluster.initial_master_nodes=elasticsearch,es02,es03
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
 
 
  filebeat:
    image: docker.elastic.co/beats/filebeat:7.4.2
    container_name: filebeat
    volumes:
      - type: bind
        source: "./filebeat/2018.log"
        target: "/2018.log"
      - type: bind
        source: "./filebeat/filebeat.yml"
        target: "/usr/share/filebeat/filebeat.yml"
    networks:
      - elk-net
    depends_on:
      - logstash
  logstash:
    image: docker.elastic.co/logstash/logstash:7.4.2
    container_name: logstash
    volumes:
      - type: bind
        source: "./logstash/logstash_stdout.conf"
        target: "/usr/share/logstash/pipeline/logstash.conf"
    networks:
       - elk-net
    depends_on:
      - elasticsearch
    
  kibana:
    image: docker.elastic.co/kibana/kibana:7.4.2
    container_name: kibana
    networks:
      - elk-net
    ports:
      - "5601:5601"
    depends_on:
      - elasticsearch

networks:
   elk-net:
     driver: bridge

ELA服务里的环境变量说明

discovery.zen.minimum_master_nodes=2 #设置最新主节点数，有助于防止脑裂

node.name=elasticsearch #指定节点名字

cluster.name=es-docker-cluster #指定集群名字

discovery.seed_hosts=es02,es03 #写入除自己以外的候选主节点的设备地址，来开启服务时就可以被选为主节点

cluster.initial_master_nodes=elasticsearch,es02,es03 #写入全部的候选主节点的设备地址，来开启服务时就可以被选为主节点

bootstrap.memory_lock=true #在ES运行起来后锁定ES所能使用的堆内存大小，锁定内存大小一般为可用内存的一半左右；锁定内存后就不会使用交换分区。如果不打开此项，当系统物理内存空间不足，ES将使用交换分区，ES如果使用交换分区，那么ES的性能将会变得很差

docker-compose up后访问http://localhost:9200/使用谷歌的Elasticsearch Head插件,可观察到主节点和从节点信息

image

粗框是主节点，细框是从节点
image

对于节点说明

设置最新主节点数
minimum_master_nodes 设定对你的集群的稳定极其重要

当你的集群中有两个 masters（注：主节点）的时候，这个配置有助于防止脑裂。
如果你的集群发生了脑裂，那么你的集群就会处在丢失数据的危险中，因为主节点被认为是这个集群的最高统治者，它决定了什么时候新的索引可以创建，分片是如何移动的等等。如果你有两个 masters 节点，你的数据的完整性将得不到保证，因为你有两个节点认为他们有集群的控制权，就会导致冲突。

此设置应该始终被配置为 master 候选节点的法定个数（大多数个）。法定个数就是 ( master 候选节点个数 / 2) + 1 。这里有几个例子：
如果你有 10 个节点（能保存数据，同时能成为 master），法定数就是 6 。
如果你有 3 个候选 master 节点，和 100 个 data 节点，法定数就是 2 ，你只要数数那些可以做 master 的节点数就可以了。
如果你有两个节点，你遇到难题了。法定数当然是 2 ，但是这意味着如果有一个节点挂掉，你整个集群就不可用了。设置成 1 可以保证集群的功能，但是就无法保证集群脑裂了，像这样的情况，你最好至少保证有 3 个节点。

建议这样配置：
discovery.zen.minimum_master_nodes: 2

但是由于 ELasticsearch 是动态的，你可以很容易的添加和删除节点，但是这会改变这个法定个数。你不得不修改每一个索引节点的配置并且重启你的整个集群只是为了让配置生效。

解决办法是同时添加如下的配置项：
PUT /_cluster/settings
{
    "persistent" : {
        "discovery.zen.minimum_master_nodes" : 2
    }
}

这个配置允许通过 API 调用的方式动态进行配置。
这将成为一个永久的配置，并且无论你配置项里配置的如何，这个将优先生效。
当你添加和删除 master 节点的时候，你需要通过 API 的方式更改这个配置。

ELFK 添加过滤条件以及ela集群
ELFK 容器化之前我们已经可以把kibana把日志展示出来了现在我们可以增加一些新的功能比如增加过滤条件，...
ELK+kafka+filebeat搭建生产ELFK集群
ELK+kafka+filebeat搭建生产ELFK集群 ELK 架构介绍集群服务版本服务版本java1.8....
SparkJoin
left join where onJoin大致包括三个要素：Join方式、Join条件以及过滤条件。其中过滤条件...
CentOS7安装Redis集群
安装依赖安装官方脚本下载Redis源码搭建Redis集群创建集群添加防火墙过滤如有错误，欢迎指出。
搭建Ela集群
参考官方文档重要ela配置参数祥解 node.name=es03节点名字 cluster.name=es-do...
List之Where方法
Where这个方法主要在于为List添加过滤条件，使用委托的方式将过滤条件方法传入。委托中传入参数，返回bool类...
IDEA使用.ignore插件
1.安装.ignore插件 2.创建新过滤文件 3.添加过滤条件 4.最终效果
0901_VS_DB_Modify
一、VS修改（一）维护记录添加查询条件，搜索框（按床位搜索），起始时间修改过滤条件，操作、分区使用一个过滤控...
金蝶云苍穹如何设置列表界面？
点击设计器列表页签进入列表设计界面设置过滤面板配置快速过滤添加单据编号和供应商（名称）常用条件添加单据状态和...
Elasticsearch 简介
1. 基本概念近实时（NRT）建索引，到可搜索，一般仅需要1秒钟；集群（Cluster）集群默认名称: ela...