精华内容
下载资源
问答
  • Beats

    千次阅读 2018-05-10 17:31:03
    Beats概述 什么是Beats 官方定义是:轻量级数据传送工具。 Beats其实是一组beat工具的统称,它包含有很多的工具。 beat工具 Filebeat:针对日志文件 Metricbeat:度量数据,可以搜集系统性能和软件数据,将...

    Beats概述


    什么是Beats

    官方定义是:轻量级数据传送工具。

    Beats其实是一组beat工具的统称,它包含有很多的工具。

    beat工具

    • Filebeat:针对日志文件
    • Metricbeat:度量数据,可以搜集系统性能和软件数据,将这些数据存在Elasticsearch中就可以进行展示、告警等功能。
    • Packetbeat:网络数据,比如抓包
    • Winlogbeat:主要针对Windows数据
    • Hearbeat:健康检查

    架构

    Beats工具位于数据的起始端,它们获取到数据后可以直接传递到Elasticsearch中进行存储,或者传输给Logstash进行日志处理,最终都会交给Kibana进行展示。

    Filebeat


    Filebeat处理流程

    Filebeat主要是针对日志文件进行处理的,它的处理流程如下:

    1、输入Input:输入就是日志文件
    2、处理 Filter:对获取到的日志文件进行处理
    3、输出 Output:将结果输出

    Fliebeat构成

    这里写图片描述

    从图中可以看出,Filebeat的输出对象可以使Elasticsearch、Logstash、Kafka、Redis等。而Filebeat有两个组成部分:Prospector和Harvester。

    Prospector:负责检测指定的日志文件是否存在变化。
    Harvester:当 Prospector 检测到指定的日志发生变化时就会通知Harvester来收集新的日志,然后发送到output对象。

    在一个Filebeat中可以有多个Prospector对象,每一个Prospector对象针对的日志对象都是不同的。每个Prospector会对自己匹配到的日志文件启动一个Harvester。

    Filebeat配置


    Filebeat配置是通过yaml语法进行配置的。

    Input配置

    示例:

    filebeat.prospectors:
        - input_type:log
          paths:
            - /var/log/apache/httpd-*.log
        - input_type:log
          paths:
            - /var/log/messages
            - /var/log/*.log

    其中,input_type有两种类型:
    - log:日志文件
    - stdin:标准输入

    Output配置

    Output支持的输出对象有:
    - Console:标准输出
    - Elasticsearch
    - Logstash
    - Kafka
    - redis
    - file:输出到另一个文件中

    示例1,输出到elasticsearch:

    output.elasticsearch:
      hosts:["http://10.10.99.229:9200"]
      username:"admin"
      password:"admin"

    如果加了用户权限认证则需要配置username和password

    示例2,输出到标准输出:

    output.console:
      pretty:true

    Filter配置

    Filebeat对读入的内容可以通过Filter进行一定的处理。但是其处理能力相对较弱。

    Input时处理:
    - Include_lines:读入这一行
    - exclude_lines:不读入这一行
    - exclude_files:不读取这个文件

    Output前处理 – Processer:
    - drop_event:如果读入数据满足一定条件就丢弃
    - drop_fields:丢弃满足条件的字段
    - Decode_json_fields:对符合json格式的字段进行json解析
    - Include_fields:加入字段

    示例

    processirs:
      - drop_event:
        when:
          regexp:
            message: "^DBG."

    当匹配到正则表达式的时候丢弃不存储。

    processors:
      - decode_json_fields:
          fields:["inner"]

    将日志中的inner字段的json格式正常化处理

    Filebeat Module


    Filebeat Module对从Field写数据到ES的过程进行了合理的封装,比如nginx,mysql,apache的日志等。

    他具体封装了filebeat.yaml、ingest node pipeline、kibana dashboard,这样就极大简化了日志存储的过程。

    Elasticsearch Igest Node


    前面提到,Filebeat缺乏数据转换的能力,在Elasticsearch5.X后,官方提供了一个 Elasticsearch Igest Node的新的node类型,它可以在写入ES前进行数据转换,它使用pipeline api。

    Filebeat 收集Nginx日志


    通过stdin收集日志,通过console输出结果。

    安装Filebeat

    首先下载Filebeat 5.6.4版本,可以通过命令下载:

    wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-5.6.4-linux-x86_64.tar.gz

    解压并将解压后的目录移动到 /usr/local:

    tar zxf filebeat-5.6.4-linux-x86_64.tar.gz
    mv filebeat-5.6.4-linux-x86_64 /usr/local/filebeat-5.6.4

    修改权限:

    chown elastic:elastic -R filebeat-5.6.4/

    我们要分析的nginx访问日志的信息如下:
    这里写图片描述

    设置Filebeat 配置

    首先设置一个nginx.yaml 的配置文件,内容如下:

    filebeat.prospectors:
    - input_type: stdin
    
    output.console:
      pretty: true

    分析日志

    通过标准输入向filebeat中输入日志信息,同时指定日志分析的配置文件:

    head -2 /var/log/nginx/access.log | ./filebeat -e -c nginx.yaml

    这里写图片描述

    可以看到,从标准输入输入了日志文件的两行,日志分析出了两组数据。

    Packetbeat简介


    packetbeat能够实时抓取网络包,并能自动解析应用层协议。

    packetbeat抓取http包并解析

    这里以packetbeat解析es的http请求为例:

    下载packetbeat:

    wget https://artifacts.elastic.co/downloads/beats/packetbeat/packetbeat-5.6.9-linux-x86_64.tar.gz

    解压并修改属主:

    tar zxf packetbeat-5.6.9-linux-x86_64.tar.gz
    mv packetbeat-5.6.9-linux-x86_64 /usr/local/packetbeat-5.6.9
    chown -R elastic:elastic packetbeat-5.6.9

    创建packetbeat配置文件:

    packetbeat.interface.device: ens192
    packetbeat.protocols.http: 
      ports: [9200]
      send_request: true
      include_body_for: ["application/json","x-www-form-urlencoded"]
    output.console:
      pretty: true

    interface.device指定了监听的网卡,protocols.http指定了抓包的协议为http,并且和端口。send_request: true表示记录了详细的信息。include_body_for设置了记录哪些格式的数据。

    运行抓包:

    sudo ./packetbeat -e -c es.yml -strict.perms=false

    这里之所以用到sudo是应为转包命令普通用户权限可能不足。-strict.perms=false表示不检查文件的权限。

    这里写图片描述

    展开全文
  • beats

    2017-02-27 20:12:44
    3509. 【NOIP2013模拟11.5B组】倒霉的小C(beats)  (File IO): input:beats.in output:beats.out Time Limits: 1000 ms Memory Limits: 262144 KB Detailed Limits  Goto ProblemSet ...

    Description

    小G最近迷上了岛国动漫《Angel Beats》,她为了画出一个更霸气的Angel Beats的logo,想了如下办法:

    从(0,0)开始,画到(n,1),再从(n,1),画到(2*n,-1),再到(3*n,2),再到(4*n,-2),依此类推,即每次画出一个(n,(-1)^(i+1)*i)的向量,一共画出n个这样的向量。现在小G想让小C求出这个图形穿过了多少格点(坐标都是整数)。

    由于小C想要认真地听他的数学课并且想自己在接力赛中因RP暴光而发生接力棒传错这类的糗事,所以这个问题就交给你啦。小G说,如果连你也解决不好,就把你的RP也吸光。

    Input

    输入文件中仅一行为一个整数n。

    Output

    输出文件中仅一行为一个数,表示穿过的格点数。

    Sample Input

    4

    Sample Output

    9

    Data Constraint

    对于20%的数据,满足:n<=100;

    对于40%的数据,满足:n<=10^5;

    对于60%的数据,满足:n<=10^7;

    对于100%的数据,满足:n<=10^10。

    #include<iostream>
    #include<cstdio>
    #include<cmath>
    using namespace std;
    long long a,b,d,ans;
    long long fai(long long);
    int main()
    {
    	freopen("beats.in","r",stdin);
    	freopen("beats.out","w",stdout);
    	scanf("%lld",&a);
    	for (b=1;b<=trunc(sqrt(a));b++)
    	{
    		if (a%b==0)
    		{
    			d=b;
    			ans=ans+d*fai(a/d);
    			if (d!=a/d) 
    			{
    				d=a/b;
    				ans=ans+d*fai(a/d);
    			}
    		}
    	}
    	ans++;
    	printf("%lld",ans);
    	return 0;
    }
    long long fai(long long x)
    {
    	int i;
    	long long t;
    	t=x;
    	i=2;
    	while (x>1) 
    	{
    		if (x%i==0) 
            {
    			t=(t-t/i);
    			while (x%i==0) 
        		{
    				x=x/i;
    			}
    		}
    		i++;
    	}
    	return t;
    }

    展开全文
  • beats-源码

    2021-03-30 02:00:03
    beats
  • Beats-源码

    2021-03-27 15:54:43
    Beats
  • BeatsBeats processors

    千次阅读 2020-12-17 19:47:02
    但是针对每个 beats 来讲,它们也分别有自己的一组 processors 来可以帮我们处理数据。我们可以访问 Elastic 的官方网站来查看针对 filebeat 的所有 processors。 也就是说,我们可以在配置 beats 的时候并同时配置...

    我们通常的做法是使用 Elasticsearch 的 ingest node 或者 Logstash 来对数据进行清洗。这其中包括删除,添加,丰富,转换等等。但是针对每个 beat 来讲,它们也分别有自己的一组 processors 来可以帮我们处理数据。我们可以访问  Elastic 的官方网站来查看针对 filebeat 的所有 processors。 也就是说,我们可以在配置 beats 的时候并同时配置相应的 processors 来对数据进行处理。每个 processor 能够修改经过它的事件。

    如果你想了解 ingest pipeline 是如何清洗这些事件的,请阅读我之前的文章 “Elastic可观测性 - 运用 pipeline 使数据结构化”。在之前文章 “深入理解 Dissect ingest processor” 中,我讲述了 dissect ingest processor 的应用。在今天的文章中,我将使用同样的 beat processor 来说明如何对数据进行格式化。

    使用  Filebeat 来对数据进行处理

    在今天的实验中,我们将使用如下是例子来进行。我们创建一个叫做 sample.log 的文件,其内容如下:

    sample.log

    "321 - App01 - WebServer is starting"
    "321 - App01 - WebServer is up and running"
    "321 - App01 - WebServer is scaling 2 pods"
    "789 - App02 - Database is will be restarted in 5 minutes"
    "789 - App02 - Database is up and running"
    "789 - App02 - Database is refreshing tables"
    
    

    由于 Filebeat 是以换行符来识别每一行的数据的,所以我在文件的最后一行也加上了一个换行符以确保最后一行的数据能被导入。

    我们创建一个叫做 filebeat_processors.yml 的 filebeat 配置文件:

    filebeat_processors.yml

    它的内容如下:

    filebeat.inputs:
    - type: log
      enabled: true
      paths:
        - /Users/liuxg/data/beatsprocessors/sample.log
    
    processors:
     - dissect:
         tokenizer: '"%{pid|integer} - %{service.name} - %{service.status}"'
         field: "message"
         target_prefix: ""
         trim_values: "all"
     - drop_fields:
         fields: ["ecs", "agent", "log", "input", "host"]
     
    setup.template.enabled: false
    setup.ilm.enabled: false
     
    output.elasticsearch:
      hosts: ["localhost:9200"]
      index: "sample"
      bulk_max_size: 1000

    请注意你需要依据自己 sample.log 的位置修改上面的 paths 中的路径。请注意在上面,我们在 tokenizer 中是这样定义的:

    '"%{pid|integer} - %{service.name} - %{service.status}"'

    我们使用了 ' 字符来定义字符串。我们也可以使用 " 符号来定义字符串。在这种情况下,我们必须使用 \" 来进行 escape(转义):

    "\"%{pid|integer} - %{service.name} - %{service.status}\""

    上面的两种选择完全依赖于你自己的喜好。我个人比较喜欢第一种的写法。它显得比较干净,整洁!

    另外,如果我们不想要其中的一个字段,那么我们可以使用 %{},或者更加易懂的方式 %{?field},比如,在上面的例子中,我们不想要 service.name 这个字段,我们可以使用如下的方法:

    '"%{pid|integer} - %{?service.name} - %{service.status}"'

    那么经过 Beats processor 的处理过后,我们将看不到 service.name 这个字段了。

    在实际的使用中,有时我们甚至想丢掉一些满足条件的事件。在这种情况下,我们把整个的事件都可以去掉,比如:

    filebeat_processors.yml

    filebeat.inputs:
    - type: log
      enabled: true
      paths:
        - /Users/liuxg/data/beatsprocessors/sample.log
    
    processors:
     - dissect:
         tokenizer: '"%{pid|integer} - %{service.name} - %{service.status}"'
         field: "message"
         target_prefix: ""
         trim_values: "all"
     - drop_fields:
         fields: ["ecs", "agent", "log", "input", "host"]
     - drop_event:
         when:
           equals:
             pid: 789
     
    setup.template.enabled: false
    setup.ilm.enabled: false
     
    output.elasticsearch:
      hosts: ["localhost:9200"]
      index: "sample"
      bulk_max_size: 1000

    我们使用了 drop_event 这个 processor。当这个事件的 pid 的值为 789 时,那么整个事件将被删除,从而我们在 Elasticsearch 中看不到这个事件。

    在上面,我们使用了 drop_fields , dissect 及 drop_event 三个 processors。当我们使用 dissect 处理器时,有时我们的日志里的空格是不规整的,比如有的是一个,有的是两个空格:

    针对这种情况,我们可以使用 trim_values 来除去多余的空格:

    我们使用如下的命令来运行 filebeat:

    ./filebeat -e -c ~/data/beatsprocessors/filebeat_processors.yml 

    同样地,我们需要根据自己的配置文件路径修改上面的路径。

    运行完上面的命令后,我们可以在 Kibana 中进行查询 sample 索引的内容:

    GET sample/_search
    {
      "took" : 0,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,
        "failed" : 0
      },
      "hits" : {
        "total" : {
          "value" : 6,
          "relation" : "eq"
        },
        "max_score" : 1.0,
        "hits" : [
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "qrBscHYBpymojx8hDWuV",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:18:16.540Z",
              "message" : "\"321 - App01 - WebServer is starting\"",
              "service" : {
                "name" : "App01",
                "status" : "WebServer is starting"
              },
              "pid" : 321
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "q7BscHYBpymojx8hDWuV",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:18:16.541Z",
              "pid" : 321,
              "message" : "\"321 - App01 - WebServer is up and running\"",
              "service" : {
                "name" : "App01",
                "status" : "WebServer is up and running"
              }
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "rLBscHYBpymojx8hDWuV",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:18:16.541Z",
              "message" : "\"321 - App01 - WebServer is scaling 2 pods\"",
              "service" : {
                "name" : "App01",
                "status" : "WebServer is scaling 2 pods"
              },
              "pid" : 321
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "rbBscHYBpymojx8hDWuV",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:18:16.541Z",
              "message" : "\"789 - App02 - Database is will be restarted in 5 minutes\"",
              "pid" : 789,
              "service" : {
                "name" : "App02",
                "status" : "Database is will be restarted in 5 minutes"
              }
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "rrBscHYBpymojx8hDWuV",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:18:16.541Z",
              "service" : {
                "name" : "App02",
                "status" : "Database is up and running"
              },
              "pid" : 789,
              "message" : "\"789 - App02 - Database is up and running\""
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "r7BscHYBpymojx8hDWuV",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:18:16.541Z",
              "service" : {
                "status" : "Database is refreshing tables",
                "name" : "App02"
              },
              "message" : "\"789 - App02 - Database is refreshing tables\"",
              "pid" : 789
            }
          }
        ]
      }
    }

    显然,我们得到了一个结构化的索引。在上面,我们对 pid 还进行了从字符串到整型值的转换。

    我们甚至可以重新对一个字段命名,比如:

    filebeat_processors.yml

    filebeat.inputs:
    - type: log
      enabled: true
      paths:
        - /Users/liuxg/data/beatsprocessors/sample.log
    
    processors:
     - drop_fields:
         fields: ["ecs", "agent", "log", "input", "host"]
     - dissect:
         tokenizer: '"%{pid|integer} - %{service.name} - %{service.status}"'
         field: "message"
         target_prefix: "" 
     - rename:
         fields:
            - from: "pid"
              to: "PID"
         ignore_missing: false
         fail_on_error: true    
     
    setup.template.enabled: false
    setup.ilm.enabled: false
     
    output.elasticsearch:
      hosts: ["localhost:9200"]
      index: "sample"
      bulk_max_size: 1000

    重新运行上面的配置文件,我们发现:

    {
      "took" : 0,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,
        "failed" : 0
      },
      "hits" : {
        "total" : {
          "value" : 6,
          "relation" : "eq"
        },
        "max_score" : 1.0,
        "hits" : [
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "UrB5cHYBpymojx8h7oCK",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:33:26.114Z",
              "service" : {
                "status" : "WebServer is starting",
                "name" : "App01"
              },
              "message" : "\"321 - App01 - WebServer is starting\"",
              "PID" : 321
            }
          },
       ...

    之前的 pid 已经转换为 PID 字段。

    我们还可以通过脚本来实现对事件的处理,比如:

    filebeat_processors.yml

    filebeat.inputs:
    - type: log
      enabled: true
      paths:
        - /Users/liuxg/data/beatsprocessors/sample.log
    
    processors:
     - drop_fields:
         fields: ["ecs", "agent", "log", "input", "host"]
     - dissect:
         tokenizer: '"%{pid|integer} - %{service.name} - %{service.status}"'
         field: "message"
         target_prefix: "" 
     - rename:
         fields:
            - from: "pid"
              to: "PID"
         ignore_missing: false
         fail_on_error: true
     - script:
         lang: javascript
         id: my_filter
         params:
            pid: 789
         source: >
           var params = {pid: 0};
           function register(scriptParams) {
              params = scriptParams;
           }
           function process(event) {
              if (event.Get("PID") == params.pid) {
                  event.Cancel();
              }
           }        
     
    setup.template.enabled: false
    setup.ilm.enabled: false
     
    output.elasticsearch:
      hosts: ["localhost:9200"]
      index: "sample"
      bulk_max_size: 1000

    在上面,当 PID 的值为 789 时,我们将过滤这个事件。重新运行 filebeat:

    {
      "took" : 0,
      "timed_out" : false,
      "_shards" : {
        "total" : 1,
        "successful" : 1,
        "skipped" : 0,
        "failed" : 0
      },
      "hits" : {
        "total" : {
          "value" : 3,
          "relation" : "eq"
        },
        "max_score" : 1.0,
        "hits" : [
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "5bCBcHYBpymojx8hrIup",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:41:53.478Z",
              "PID" : 321,
              "service" : {
                "status" : "WebServer is starting",
                "name" : "App01"
              },
              "message" : "\"321 - App01 - WebServer is starting\""
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "5rCBcHYBpymojx8hrIup",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:41:53.479Z",
              "message" : "\"321 - App01 - WebServer is up and running\"",
              "service" : {
                "status" : "WebServer is up and running",
                "name" : "App01"
              },
              "PID" : 321
            }
          },
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "57CBcHYBpymojx8hrIup",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T11:41:53.479Z",
              "service" : {
                "status" : "WebServer is scaling 2 pods",
                "name" : "App01"
              },
              "message" : "\"321 - App01 - WebServer is scaling 2 pods\"",
              "PID" : 321
            }
          }
        ]
      }
    }

    我们发现所有关于 PID 为789 的事件都被过滤掉了。

    我们设置可以通过 script 的方法为事件添加一个 tag。当然由于这是一种 Javascript 的脚本编程,我们甚至可以依据一些条件对事件添加不同的 tag。

    filebeat_processors.yml

    filebeat.inputs:
    - type: log
      enabled: true
      paths:
        - /Users/liuxg/data/beatsprocessors/sample.log
    
    processors:
     - drop_fields:
         fields: ["ecs", "agent", "log", "input", "host"]
     - dissect:
         tokenizer: '"%{pid|integer} - %{service.name} - %{service.status}"'
         field: "message"
         target_prefix: "" 
     - rename:
         fields:
            - from: "pid"
              to: "PID"
         ignore_missing: false
         fail_on_error: true
     - script:
         lang: javascript
         id: my_filter
         params:
            pid: 789
         source: >
           var params = {pid: 0};
           function register(scriptParams) {
              params = scriptParams;
           }
           function process(event) {
              if (event.Get("PID") == params.pid) {
                  event.Cancel();
              }
    
              event.Tag("myevent")
           }
     
    setup.template.enabled: false
    setup.ilm.enabled: false
     
    output.elasticsearch:
      hosts: ["localhost:9200"]
      index: "sample"
      bulk_max_size: 1000

    在上面,我们添加了 event.Tag("myevent")。重新运行我们可以看到:

        "hits" : [
          {
            "_index" : "sample",
            "_type" : "_doc",
            "_id" : "C7CScHYBpymojx8hkKVy",
            "_score" : 1.0,
            "_source" : {
              "@timestamp" : "2020-12-17T12:00:20.365Z",
              "message" : "\"321 - App01 - WebServer is starting\"",
              "PID" : 321,
              "service" : {
                "name" : "App01",
                "status" : "WebServer is starting"
              },
              "tags" : [
                "myevent"
              ]
            }
          },

    在上面,我们可以看到 tags 字段里有一个叫做 myevent 的值。
    在今天的介绍中,我就当是抛砖引玉。更多关于 Filebeat 的 Beats processors,请参阅链接 https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#processors

    在今天的文章中,我们介绍了一种数据处理的方式。这种数据处理可以在 beats 中进行实现,而不需要在 Elasticsearch 中的 ingest node 中实现。在实际的使用中,你需要依据自己的架构设计来实现不同的设计方案。

    展开全文
  • ziplock-beats:Ziplock Beats App
  • Beats 简介

    千次阅读 2021-02-24 23:44:58
    Beats 简介 Filebeat 架构 用于监控、收集服务器日志文件 部署与运行 下载(版本为:filebeat-6.5.4):https://www.elastic.co/downloads/beats mkdir /itcast/beats tar -xvf filebeat-6.5.4-linux-x86_64....

    Beats 简介

    在这里插入图片描述
    在这里插入图片描述
    在这里插入图片描述

    Filebeat

    在这里插入图片描述

    架构

    用于监控、收集服务器日志文件
    在这里插入图片描述

    部署与运行

    下载(版本为:filebeat-6.5.4):https://www.elastic.co/downloads/beats

    mkdir /itcast/beats
    tar -xvf filebeat-6.5.4-linux-x86_64.tar.gz
    cd filebeat-6.5.4-linux-x86_64
    #创建如下配置文件 itcast.yml
    filebeat.inputs:
    - type: stdin
    enabled: true
    setup.template.settings:
    index.number_of_shards: 3
    output.console:
    pretty: true
    enable: true
    #启动filebeat
    ./filebeat -e -c itcast.yml
    #输入hello运行结果如下:
    hello
    {
     "@timestamp": "2019-01-12T12:50:03.585Z",
     "@metadata": { #元数据信息
      "beat": "filebeat",
      "type": "doc",
      "version": "6.5.4"
    },
     "source": "",
     "offset": 0,
     "message": "hello",  #输入的内容
     "prospector": { #标准输入勘探器
      "type": "stdin"
    },
     "input": {  #控制台标准输入
      "type": "stdin"
    },
     "beat": { #beat版本以及主机信息
      "name": "itcast01",
      "hostname": "itcast01",
      "version": "6.5.4"
    },
     "host": {
      "name": "itcast01"
    }
    }
    

    读取文件

    #配置读取文件项 itcast-log.yml
    filebeat.inputs:
    - type: log
    enabled: true
    paths:
    - /itcast/beats/logs/*.log
    setup.template.settings:
    index.number_of_shards: 3
    output.console:
    pretty: true
    enable: true
    #启动filebeat
    ./filebeat -e -c itcast-log.yml
    #/haoke/beats/logs下创建a.log文件,并输入如下内容
    hello
    world
    #观察filebeat输出
    {
     "@timestamp": "2019-01-12T14:16:10.192Z",
      "@metadata": {
      "beat": "filebeat",
      "type": "doc",
      "version": "6.5.4"
    },
     "host": {
      "name": "itcast01"
    },
     "source": "/haoke/beats/logs/a.log",
     "offset": 0,
     "message": "hello",
     "prospector": {
      "type": "log"
    },
     "input": {
      "type": "log"
    },
     "beat": {
      "version": "6.5.4",
      "name": "itcast01",
      "hostname": "itcast01"
    }
    }
    {
     "@timestamp": "2019-01-12T14:16:10.192Z",
     "@metadata": {
      "beat": "filebeat",
      "type": "doc",
      "version": "6.5.4"
    },
     "prospector": {
      "type": "log"
    },
     "input": {
      "type": "log"
    },
     "beat": {
      "version": "6.5.4",
      "name": "itcast01",
      "hostname": "itcast01"
    },
     "host": {
      "name": "itcast01"
    },
     "source": "/haoke/beats/logs/a.log",
     "offset": 6,
     "message": "world"
    }
    

    可以看出,已经检测到日志文件有更新,立刻就会读取到更新的内容,并且输出到控制台。

    自定义字段

    #配置读取文件项 itcast-log.yml
    filebeat.inputs:
    - type: log
    enabled: true
    paths:
      - /itcast/beats/logs/*.log
    tags: ["web"]  #添加自定义tag,便于后续的处理
    fields:  #添加自定义字段
     from: itcast-im
    fields_under_root: true #true为添加到根节点,false为添加到子节点中
    setup.template.settings:
    index.number_of_shards: 3
    output.console:
    pretty: true
    enable: true
    #启动filebeat
    ./filebeat -e -c itcast-log.yml
    #/haoke/beats/logs下创建a.log文件,并输入如下内容
    123
    #执行效果
    {
     "@timestamp": "2019-01-12T14:37:19.845Z",
     "@metadata": {
      "beat": "filebeat",
      "type": "doc",
      "version": "6.5.4"
    },
     "offset": 0,
     "tags": [
      "haoke-im"
    ],
     "prospector": {
      "type": "log"
    },
     "beat": {
      "name": "itcast01",
      "hostname": "itcast01",
      "version": "6.5.4"
    },
     "host": {
      "name": "itcast01"
    },
     "source": "/itcast/beats/logs/a.log",
     "message": "123",
     "input": {
      "type": "log"
    },
     "from": "haoke-im"
    }
    

    输出到Elasticsearch

    # itcast-log.yml
    filebeat.inputs:
    - type: log
    enabled: true
    paths:
     - /itcast/beats/logs/*.log
    tags: ["haoke-im"]
    fields:
     from: haoke-im
    fields_under_root: false
    setup.template.settings:
    index.number_of_shards: 3 #指定索引的分区数
    output.elasticsearch: #指定ES的配置
    hosts: ["192.168.1.7:9200","192.168.1.7:9201","192.168.1.7:9202"]
    

    在日志文件中输入新的内容进行测试:
    在这里插入图片描述
    查看数据:
    在这里插入图片描述

    Filebeat工作原理

    Filebeat由两个主要组件组成:prospector 和 harvester。

    • harvester:
      • 负责读取单个文件的内容。
      • 如果文件在读取时被删除或重命名,Filebeat将继续读取文件。
    • prospector
      • prospector 负责管理harvester并找到所有要读取的文件来源。
      • 如果输入类型为日志,则查找器将查找路径匹配的所有文件,并为每个文件启动一个harvester。
      • Filebeat目前支持两种prospector类型:log和stdin。
    • Filebeat如何保持文件的状态
      • Filebeat 保存每个文件的状态并经常将状态刷新到磁盘上的注册文件中。
      • 该状态用于记住harvester正在读取的最后偏移量,并确保发送所有日志行。
      • 如果输出(例如Elasticsearch或Logstash)无法访问,Filebeat会跟踪最后发送的行,并在输出再次可用时继续读取文件。
      • 在Filebeat运行时,每个prospector内存中也会保存的文件状态信息,当重新启动Filebeat时,将使用注册文件的数据来重建文件状态,Filebeat将每个harvester在从保存的最后偏移量继续读取。
      • 文件状态记录在data/registry文件中。
        启动命令:
    /filebeat -e -c itcast.yml
    ./filebeat -e -c itcast.yml -d "publish"
    #参数说明
    -e: 输出到标准输出,默认输出到syslog和logs下
    -c: 指定配置文件
    -d: 输出debug信息
    #测试: ./filebeat -e -c itcast-log.yml -d "publish"
    DEBUG  [publish]    pipeline/processor.go:308    Publish event: {
     "@timestamp": "2019-01-12T15:03:50.820Z",
     "@metadata": {
      "beat": "filebeat",
      "type": "doc",
      "version": "6.5.4"
    },
     "offset": 0,
     "tags": [
      "haoke-im"
    ],
     "input": {
      "type": "log"
    },
     "prospector": {
      "type": "log"
    },
     "beat": {
      "name": "itcast01",
      "hostname": "itcast01",
      "version": "6.5.4"
    },
     "source": "/haoke/beats/logs/a.log",
     "fields": {
      "from": "haoke-im"
    },
     "host": {
      "name": "itcast01"
    },
     "message": "456"
    }
    

    读取Nginx日志文件

    # itcast-nginx.yml
    filebeat.inputs:
    - type: log
    enabled: true
    paths:
      - /usr/local/nginx/logs/*.log
    tags: ["nginx"]
    setup.template.settings:
    index.number_of_shards: 3 #指定索引的分区数
    output.elasticsearch: #指定ES的配置
    hosts: ["192.168.40.133:9200","192.168.40.134:9200","192.168.40.135:9200"]
    #启动
    ./filebeat -e -c itcast-nginx.yml
    

    在这里插入图片描述
    启动后,可以在Elasticsearch中看到索引以及查看数据:
    在这里插入图片描述
    在这里插入图片描述
    可以看到,在message中已经获取到了nginx的日志,但是,内容并没有经过处理,只是读取到原数据,那么对于我们后期的操作是不利的,有办法解决吗?

    Module

    前面要想实现日志数据的读取以及处理都是自己手动配置的,其实,在Filebeat中,有大量的Module,可以简化我们的配置,直接就可以使用,如下:

    ./filebeat modules list
    Enabled:
    Disabled:
    apache2
    auditd
    elasticsearch
    haproxy
    icinga
    iis
    kafka
    kibana
    logstash
    mongodb
    mysql
    nginx
    osquery
    postgresql
    redis
    suricata
    system
    traefik
    

    可以看到,内置了很多的module,但是都没有启用,如果需要启用需要进行enable操作:

    ./filebeat modules enable nginx #启动
    ./filebeat modules disable nginx #禁用
    Enabled:
    nginx
    Disabled:
    apache2
    auditd
    elasticsearch
    haproxy
    icinga
    iis
    kafka
    kibana
    logstash
    mongodb
    mysql
    redis
    osquery
    postgresql
    suricata
    system
    traefik
    

    可以发现,nginx的module已经被启用。

    nginx module 配置

    - module: nginx
     # Access logs
    access:
     enabled: true
     var.paths: ["/usr/local/nginx/logs/access.log*"]
      # Set custom paths for the log files. If left empty,
      # Filebeat will choose the paths depending on your OS.
       #var.paths:
     # Error logs
    error:
     enabled: true
     var.paths: ["/usr/local/nginx/logs/error.log*"]
      # Set custom paths for the log files. If left empty,
      # Filebeat will choose the paths depending on your OS.
      #var.paths:
    

    配置filebeat

    #vim itcast-nginx.yml
    filebeat.inputs:
    #- type: log
    # enabled: true
    # paths:
    #  - /usr/local/nginx/logs/*.log
    # tags: ["nginx"]
    setup.template.settings:
    index.number_of_shards: 3
    output.elasticsearch:
    hosts: ["192.168.40.133:9200","192.168.40.134:9200","192.168.40.135:9200"]
    filebeat.config.modules:
    path: ${path.config}/modules.d/*.yml
    reload.enabled: false
    

    测试

    ./filebeat -e -c itcast-nginx.yml
    #启动会出错,如下
    ERROR  fileset/factory.go:142 Error loading pipeline: Error loading pipeline for
    fileset nginx/access: This module requires the following Elasticsearch plugins:
    ingest-user-agent, ingest-geoip. You can install them by running the following
    commands on all the Elasticsearch nodes:
      sudo bin/elasticsearch-plugin install ingest-user-agent
      sudo bin/elasticsearch-plugin install ingest-geoip
     
    #解决:需要在Elasticsearch中安装ingest-user-agent、ingest-geoip插件
    #在资料中可以找到,ingest-user-agent.tar、ingest-geoip.tar、ingest-geoip-conf.tar 3个文件
    #其中,ingest-user-agent.tar、ingest-geoip.tar解压到plugins下
    #ingest-geoip-conf.tar解压到config下
    #问题解决。
    

    在这里插入图片描述
    测试发现,数据已经写入到了Elasticsearch中,并且拿到的数据更加明确了:
    在这里插入图片描述
    当然了,其他的Module的用法参加官方文档:
    https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-modules.html
    在这里插入图片描述

    Metricbeat

    在这里插入图片描述

    • 定期收集操作系统或应用服务的指标数据
    • 存储到Elasticsearch中,进行实时分析

    Metricbeat组成

    Metricbeat有2部分组成,一部分是Module,另一部分为Metricset。

    • Module
      • 收集的对象,如:mysql、redis、nginx、操作系统等;
    • Metricset
      • 收集指标的集合,如:cpu、memory、network等;
        以Redis Module为例:
        在这里插入图片描述

    部署与收集系统指标

    tar -xvf metricbeat-6.5.4-linux-x86_64.tar.gz
    cd metricbeat-6.5.4-linux-x86_64
    vim metricbeat.yml
    metricbeat.config.modules:
    path: ${path.config}/modules.d/*.yml
    reload.enabled: false
    setup.template.settings:
    index.number_of_shards: 1
    index.codec: best_compression
    setup.kibana:
    output.elasticsearch:
    hosts: ["192.168.40.133:9200","192.168.40.134:9200","192.168.40.135:9200"]
    processors:
     - add_host_metadata: ~
     - add_cloud_metadata: ~
    #启动
    ./metricbeat -e
    

    在ELasticsearch中可以看到,系统的一些指标数据已经写入进去了:
    在这里插入图片描述
    system module配置:

    root@itcast01:modules.d# cat system.yml
    # Module: system
    # Docs: https://www.elastic.co/guide/en/beats/metricbeat/6.5/metricbeat-module-
    system.html
    - module: system
    period: 10s
    metricsets:
     - cpu
     - load
     - memory
     - network
     - process
     - process_summary
      #- core
      #- diskio
      #- socket
    process.include_top_n:
     by_cpu: 5    # include top 5 processes by CPU
     by_memory: 5  # include top 5 processes by memory
    - module: system
    period: 1m
    metricsets:
     - filesystem
     - fsstat
    processors:
    - drop_event.when.regexp:
      system.filesystem.mount_point: '^/(sys|cgroup|proc|dev|etc|host|lib)($|/)'
      - module: system
    period: 15m
    metricsets:
     - uptime
    #- module: system
    # period: 5m
    # metricsets:
    #  - raid
    # raid.mount_point: '/'
    

    Module

    ./metricbeat modules list  #查看列表
    Enabled:
    system #默认启用
    Disabled:
    aerospike
    apache
    ceph
    couchbase
    docker
    dropwizard
    elasticsearch
    envoyproxy
    etcd
    golang
    graphite
    haproxy
    http
    jolokia
    kafka
    kibana
    kubernetes
    kvm
    logstash
    memcached
    mongodb
    munin
    mysql
    nginx
    php_fpm
    postgresql
    prometheus
    rabbitmq
    redis
    traefik
    uwsgi
    vsphere
    windows
    zookeeper
    

    Nginx Module

    开启nginx的状态查询

    在nginx中,需要开启状态查询,才能查询到指标数据。

    #重新编译nginx
    ./configure --prefix=/usr/local/nginx --with-http_stub_status_module
    make
    make install
    ./nginx -V  #查询版本信息
    nginx version: nginx/1.11.6
    built by gcc 4.4.7 20120313 (Red Hat 4.4.7-23) (GCC)
    configure arguments: --prefix=/usr/local/nginx --with-http_stub_status_module
    #配置nginx
    vim nginx.conf
    location /nginx-status {
    stub_status on;
    access_log off;
    }
    

    测试:
    在这里插入图片描述
    结果说明:

    • Active connections:正在处理的活动连接数
    • server accepts handled requests
      • 第一个 server 表示Nginx启动到现在共处理了9个连接
      • 第二个 accepts 表示Nginx启动到现在共成功创建 9 次握手
      • 第三个 handled requests 表示总共处理了 21 次请求
      • 请求丢失数 = 握手数 - 连接数 ,可以看出目前为止没有丢失请求
    • Reading: 0 Writing: 1 Waiting: 1
      • Reading:Nginx 读取到客户端的 Header 信息数
      • Writing:Nginx 返回给客户端 Header 信息数
      • Waiting:Nginx 已经处理完正在等候下一次请求指令的驻留链接(开启keep-alive的情况下,这个值等于Active - (Reading+Writing))

    配置Nginx Module

    #启用redis module
    ./metricbeat modules enable nginx
    #修改redis module配置
    vim modules.d/nginx.yml
    # Module: nginx
    # Docs: https://www.elastic.co/guide/en/beats/metricbeat/6.5/metricbeat-module-
    nginx.html
    - module: nginx
     #metricsets:
     # - stubstatus
    period: 10s
     # Nginx hosts
    hosts: ["http://192.168.40.133"]
     # Path to server status. Default server-status
    server_status_path: "nginx-status"
     #username: "user"
     #password: "secret"
    #启动
    ./metricbeat -e
    

    测试:
    在这里插入图片描述
    可以看到,nginx的指标数据已经写入到了Elasticsearch。
    更多的Module使用参见官方文档:
    https://www.elastic.co/guide/en/beats/metricbeat/current/metricbeat-modules.html

    展开全文
  • Beats-登陆页面 欢迎! :waving_hand:
  • Lofi-节拍 Lofi-Beats可将Lofi拍子放松/学习 使用npm安装 npm install lofi-beats -g 用法 node lofi-beats
  • 该存储库不再用于从生成正式的 Docker映像。 要为6.6之前的版本构建Beats docker映像,请将此存储库中的分支切换到匹配的版本。
  • Beats耳机app是Beats系列蓝牙耳机的配套手机app,用户在这里可以查看自己耳机的当前状态信息,还可以对耳机进行见的设置和控制。具体功能请需要的朋友前来西西下载体验!应用简介下载Beats应用解锁各种功能,以...
  • ansible-role-beats 安装和配置Elastic Beats。 描述 该角色将以可配置的方式安装和配置任何Elastic Beats。 一次只能安装和配置一个节拍,但是可以多次包含此角色,每个节拍一个。 配置通过yaml dict( beats_...
  • BeatsBeats 简介

    千次阅读 2019-04-26 16:31:17
    Beats介绍2. 第三方开源 Beats3. 开发 Beats3.1. 创建项目3.2. Beater接口3.2.1. New 函数3.2.2. Run方法3.2.3. Stop 方法3.2.4. 主函数3.3. 事件命名规约3.4. 标准化命名 1. Beats介绍 Beats是 Elastic 公司开发...
  • 青春节拍 Teensy Beats Shield 围绕Teensy平台构建的多功能一体式步进音序器,具有LiPo充电,触摸TFT,编码器,按钮和RGB LED。 项目构建日志和照片在这里: : ...
  • Beats:Elastic Beats 介绍

    千次阅读 2019-09-23 15:48:00
    ElasticStack传统上由三个主要组件(Elasticsearch,Logstash和Kibana)组成,早已脱离了这种组合,现在也可以与名为“ Beats”的第四个元素结合使用--一个针对不同用例的日志运送者系列。 现在网上有一种说法叫做...
  • Beats:如何测试 Beats processors

    千次阅读 2021-03-17 16:51:00
    在之前的 “BeatsBeats processors” 文章中,我详细地描述了如何使用 Beat 的 processors 对数据进行清洗。在很多情况下它是非常有用的一种方法。Beats 的 processors 有很多在 ingest pipeline 的 processors 中...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 18,399
精华内容 7,359
关键字:

beats