临窗旋墨

filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)

文章来源：原创作者：临窗旋墨发布时间：2020-09-30 阅读：394 标签：日志

[toc]

### [filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)](http://www.xuqiudong.cn/detail/14)

`Vic.xu`  `20200930`

#### 目标：多台集群机器日志收集到一起，且同名输出，能记录来源机器

1. 多台机器输出的日志名是相同的，比如机器A 输出的日志包括`err.log`，`sql.log`，`common.log`等
2. 希望收集多台机器的日志，输入到同名日志文件，比如把A机器和B机器的`err.log`，`sql.log`，`common.log`收集到C机器的`err.log`，`sql.log`，`common.log`；
3. 在日志中标识来自哪一台机器；
4. 多行日志需要合并，比如一个堆栈信息；

#### 前言：项目有点老了（着急的话，直接调到[最后](#### 五 总结： `linux`下的启动脚本和最终配置文件)看结论吧）

>      一说分布式日志收集分析系统，网上都有一大推的关于ELK的文章。 不过咱的项目比较老了，只是最近才真正做了集群部署，所以才会涉及到分布式的日志收集。
>
>      不过，项目中暂时也不会做日志分析之类的东西。所以就准备先弄一个`filebeat`收集日志到`logstash`，然后通过`logstash`定向输出到文件即可。加上之前因为开发环境和测试环境的隔离，所以写了个实时页面查看日志的小项目，也是勉强够用了。

> **版本**：(因为之前使用过`ElasticSearch`的版本是7.7.1)
>
> - `Filebeat`    `7.7.1`
> - `Logstash`    `7.7.1`

#### 一 为什么不只使用`Logstash`,而是额外引入了`Filebeat`

1. 因为`logstash`是`jvm`跑的，资源消耗比较大，
2. `filebeat`更轻量，占用资源更少
3. 一般结构都是`filebeat`采集日志，然后发送到消息队列，`redis`，`kafaka`。然后`logstash`去获取，利用`filter`功能过滤分析，然后存储到`elasticsearch`中

#### 二 基本思路 `Filebeat` + `Logstash`

1. 在各个需要收集日志的机器上`FileBeat` , 配置输出到远程的`Logstash`中

- [参见Filenbeat官网配置](https://www.elastic.co/guide/en/beats/filebeat/current/configuring-output.html)
   - 日志可以输出到`Es`/`Logstash`/`Kafka`/`Redis`/`File`/`Console`,方便后续扩展

2. 在接收日志文件的机器上安装`Logstash`，收集各个`Filebeat`发送来到日志，输出到文件

- `Logstash`的input方式有很多，包括`Filebeat`，`kafka`等： https://www.elastic.co/guide/en/logstash/current/input-plugins.html

- `Logstash`的output方式有很多,非常方便后续扩展，[参见官网](https://www.elastic.co/guide/en/logstash/current/output-plugins.html)

- 由于本系统无特殊的处理日志的需求，可直接把收集到的日志[输出到文件](https://www.elastic.co/guide/en/logstash/7.x/plugins-outputs-file.html)即

- [输出到文件](https://doc.yonyoucloud.com/doc/logstash-best-practice-cn/output/stdout.html)

```
       output {
           file {
               path => "/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz"
               message_format => "%{message}"
               gzip => true
           }
       }
       ```

#### 三 windows下简单测试`Filebeat`

1. 官网下载windows版本

2. 修改`filebeat.yml`，具体参考官网

```yaml
   filebeat.inputs:
   - type: log
   
     enabled: true
     fields:
       name: vic_first
     paths:
    #   - /var/log/*.log
       - E:/space/idea/blog/logback/*.log
   
   output.file:
     path: d:/filebeat
     filename: filebeat_vic.log
     codec.format:
       string: '%{[fields][name]}%{[@timestamp]} %{[message]}'
     rotate_every_kb: 10240
   
   # 是否开启多行合并
   multiline.type: pattern
   #multiline.pattern: '^\['
   multiline.pattern: '^vic-blog'
   multiline.negate: false
   multiline.match: after
   
   #============================= Filebeat modules ===============================
   filebeat.config.modules:
     path: ${path.config}/modules.d/*.yml
     reload.enabled: false
   
   processors:
     - add_host_metadata: ~
     - add_cloud_metadata: ~
     - add_docker_metadata: ~
     - add_kubernetes_metadata: ~
   
   ```

- 如上：
   - 1. 配置了需要收集的日志文件的位置。
     2. fields 下是携带的额外的数据，一般作为标识使用
     3. 配置了日志的输出的文件
     4. 配置合并多行：因为默认是读取一行就发送，但是其实有的时候多行才算一个日志，比如堆栈信息等

3. 启动`Filebeat`,然后测试写入日志，会被写入到输出位置

> .\filebeat.exe -e -c filebeat.yml
   
4. **其他**

1. 输出到`es`：

```yaml
      output.elasticsearch:
        # Array of hosts to connect to.
        hosts: ["localhost:9200"]
      
        # Protocol - either `http` (default) or `https`.
        #protocol: "https"
      
        # Authentication credentials - either API key or username/password.
        #api_key: "id:api_key"
        #username: "elastic"
        #password: "changeme"
      ```

2. 输出到`logstash`

```yaml
      output.logstash:
        # The Logstash hosts
  hosts: ["localhost:5044"]

```

#### 四  windows下简单测试`Logstash`

1. 官网下载zip解压，

2. cofig下复制logstash.conf，然后配置成控制台输入输出：

```yaml
   input {
     beats {
       port => 5044
     }
   }

filter {
       ruby {
           code => "
               path = event.get('log')['file']['path']
               puts format('path = %<path>s', path: path)
               if (!path.nil?) && (!path.empty?)
                   event.set('filename', path.split('/')[-1])
               end
           "
       }
   }

output {
     file   {
   #    path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{+HH}-vic.log" 
       path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
           codec => line { format => "%{filename} %{[fields][name]}: %{message}"}
     }

stdout {
       codec => rubydebug
     }

}
```

3. 启动Logstash

> bin\logstash -f config\logstash.conf

- 启动成功后可访问localhost:9600，看到版本等信息，则表示启动成功
   - 测试：在启动控制台输入信息， 可看到相关输出信息

#### 一些问题：

1. 在logstash中获取Filebeat中的日志原文件名，然后输出到同名文件:通过filter实现

> 如下，获取输入的日志文件的文件名filename，通过`%{filename}`获取

```
   filter {
       ruby {
           code => "
               path = event.get('log')['file']['path']
               puts format('path = %<path>s', path: path)
               if (!path.nil?) && (!path.empty?)
                   event.set('filename', path.split('/')[-1])
               end
           "
       }
   }
   ```

2. filebeat携带额外的信息到，标志来自哪台机器，通过fields携带(如下，携带的key是name，value是server1)

```yaml
   filebeat.inputs:
   - type: log
     enabled: true
     fields:
       name: server1
   ```

3. logstash获取filebeat携带的额外的信息

```yaml
   %{[fields][name]}
   ```

4. filebeat   不发送最后一行

> Filebeat使用换行符来检测事件的结束。 如果将行逐渐添加到正在采集的文件中，则在最后一行之后需要换行符，否则Filebeat将不会读取文件的最后一行。

#### 五 总结： `linux`下的启动脚本和最终配置文件

> `linux`下的安装和window下没什么区别，`logstash`需要确保安装了`jdk1.8`+;
>
> 启动也是一样的， 只是本人比较懒， 所以写了两个启动脚本`start.sh`  ；运行时带参数的话则输出日志到`catalina.out`，不带参数则不输出日志

#### `linux`下的`filebeat`的启动脚本

```shell
#!/bin/bash

log="/dev/null"
# if input paramter is not null ,then output log to catalina.out. else to /dev/null

if [ $1 != '' ];then
  log="catalina.out"	
fi

nohup ./filebeat -e -c filebeat.yml >$log 2>&1 &

echo "start   filebeat.......... ----> $log"

```

##### `linux`下的`logstash`的启动脚本

```shell
#!/bin/bash

log="/dev/null"
# if input paramter is not null ,then output log to catalina.out. else to /dev/null

if [ $1 != '' ];then
  log="catalina.out"	
fi
nohup bin/./logstash -f config/logstash.conf >$log 2>&1 & 
echo "start logstash.......... ----> $log" 
```

##### `filebeat.yaml`最终配置文件(见注释)

```yaml
# 输入日志
filebeat.inputs:
- type: log
  enabled: true
  # 携带的额外的数据，标识来自哪个机器
  fields:
     name: wsl-ubuntu      
  # 日志的所在位置
  paths:
    - /data/share/logs/input/*.log  
    - /data/share/logs/input2/*.log  
#=========================== 是否开启多行合并================================
multiline.type: pattern
#多行合并的正则，以[ 开头
multiline.pattern: '^\['
# 是否不合并多行合并： 负负得正 表示开启
multiline.negate: false
multiline.match: after
#==========================输出日志到logstash==============================
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]
```

##### `logstash.conf`最终配置文件(见注释)

```yaml
#======从filebeat获取日志==============================
input {
  beats {
    port => 5044
  }
}

#=====解析出来源日志的文件名===================================
filter {
    ruby {
        code => "
            path = event.get('log')['file']['path']
            puts format('path = %<path>s', path: path)
            if (!path.nil?) && (!path.empty?)
                event.set('filename', path.split('/')[-1])
            end
        "
    }
}
#======输出日志==============================
output {
#  输出日志到文件，在日志的开头打印filebeat中携带的name属性，标识来自哪个文件
  file   {
    path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
	codec => line { format => "%{[fields][name]}: %{message}"}
  }
  # 输出日志到控制台  可删除掉
  stdout { 
    codec => rubydebug
  }
  
}

```

文章来源：[临窗旋墨的博客](http://www.xuqiudong.cn/detail/14)，转载注明出处。

时间： 20200930

许秋冬偷偷隐藏起来的markdown原文,测试对蜘蛛是否友好

[toc]

### [filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)](http://www.xuqiudong.cn/detail/14)

`Vic.xu`  `20200930`

#### 目标：多台集群机器日志收集到一起，且同名输出，能记录来源机器

1. 多台机器输出的日志名是相同的，比如机器A 输出的日志包括`err.log`，`sql.log`，`common.log`等
2. 希望收集多台机器的日志，输入到同名日志文件，比如把A机器和B机器的`err.log`，`sql.log`，`common.log`收集到C机器的`err.log`，`sql.log`，`common.log`；
3. 在日志中标识来自哪一台机器；
4. 多行日志需要合并，比如一个堆栈信息；

#### 前言：项目有点老了（着急的话，直接调到[最后](#### 五 总结： `linux`下的启动脚本和最终配置文件)看结论吧）

>      一说分布式日志收集分析系统，网上都有一大推的关于ELK的文章。 不过咱的项目比较老了，只是最近才真正做了集群部署，所以才会涉及到分布式的日志收集。
>
>      不过，项目中暂时也不会做日志分析之类的东西。所以就准备先弄一个`filebeat`收集日志到`logstash`，然后通过`logstash`定向输出到文件即可。加上之前因为开发环境和测试环境的隔离，所以写了个实时页面查看日志的小项目，也是勉强够用了。



> **版本**：(因为之前使用过`ElasticSearch`的版本是7.7.1)
>
> - `Filebeat`    `7.7.1`
> - `Logstash`    `7.7.1` 

#### 一 为什么不只使用`Logstash`,而是额外引入了`Filebeat` 

1. 因为`logstash`是`jvm`跑的，资源消耗比较大，
2. `filebeat`更轻量，占用资源更少
3. 一般结构都是`filebeat`采集日志，然后发送到消息队列，`redis`，`kafaka`。然后`logstash`去获取，利用`filter`功能过滤分析，然后存储到`elasticsearch`中

#### 二 基本思路 `Filebeat` + `Logstash`

1. 在各个需要收集日志的机器上`FileBeat` , 配置输出到远程的`Logstash`中

   - [参见Filenbeat官网配置](https://www.elastic.co/guide/en/beats/filebeat/current/configuring-output.html)
   - 日志可以输出到`Es`/`Logstash`/`Kafka`/`Redis`/`File`/`Console`,方便后续扩展

2. 在接收日志文件的机器上安装`Logstash`，收集各个`Filebeat`发送来到日志，输出到文件

   - `Logstash`的input方式有很多，包括`Filebeat`，`kafka`等： https://www.elastic.co/guide/en/logstash/current/input-plugins.html

   - `Logstash`的output方式有很多,非常方便后续扩展，[参见官网](https://www.elastic.co/guide/en/logstash/current/output-plugins.html)

     - 由于本系统无特殊的处理日志的需求，可直接把收集到的日志[输出到文件](https://www.elastic.co/guide/en/logstash/7.x/plugins-outputs-file.html)即

     - [输出到文件](https://doc.yonyoucloud.com/doc/logstash-best-practice-cn/output/stdout.html)

       ```
       output {
           file {
               path => "/path/to/%{+yyyy/MM/dd/HH}/%{host}.log.gz"
               message_format => "%{message}"
               gzip => true
           }
       }
       ```





#### 三 windows下简单测试`Filebeat`

1. 官网下载windows版本

2. 修改`filebeat.yml`，具体参考官网

   ```yaml
   filebeat.inputs:
   - type: log
   
     enabled: true
     fields:
       name: vic_first
     paths:
    #   - /var/log/*.log
       - E:/space/idea/blog/logback/*.log
   
   output.file:
     path: d:/filebeat
     filename: filebeat_vic.log
     codec.format:
       string: '%{[fields][name]}%{[@timestamp]} %{[message]}'
     rotate_every_kb: 10240
   
   # 是否开启多行合并
   multiline.type: pattern
   #multiline.pattern: '^\['
   multiline.pattern: '^vic-blog'
   multiline.negate: false
   multiline.match: after
   
   #============================= Filebeat modules ===============================
   filebeat.config.modules:
     path: ${path.config}/modules.d/*.yml
     reload.enabled: false
   
   processors:
     - add_host_metadata: ~
     - add_cloud_metadata: ~
     - add_docker_metadata: ~
     - add_kubernetes_metadata: ~
   
   ```

   - 如上：
   - 1. 配置了需要收集的日志文件的位置。
     2. fields 下是携带的额外的数据，一般作为标识使用
     3. 配置了日志的输出的文件
     4. 配置合并多行：因为默认是读取一行就发送，但是其实有的时候多行才算一个日志，比如堆栈信息等

3. 启动`Filebeat`,然后测试写入日志，会被写入到输出位置

   > .\filebeat.exe -e -c filebeat.yml
   
4. **其他**

   1. 输出到`es`：

      ```yaml
      output.elasticsearch:
        # Array of hosts to connect to.
        hosts: ["localhost:9200"]
      
        # Protocol - either `http` (default) or `https`.
        #protocol: "https"
      
        # Authentication credentials - either API key or username/password.
        #api_key: "id:api_key"
        #username: "elastic"
        #password: "changeme"
      ```

   2. 输出到`logstash`

      ```yaml
      output.logstash:
        # The Logstash hosts
  hosts: ["localhost:5044"]

      ```

#### 四  windows下简单测试`Logstash`

1. 官网下载zip解压，

2. cofig下复制logstash.conf，然后配置成控制台输入输出：

     ```yaml
   input {
     beats {
       port => 5044
     }
   }

   filter {
       ruby {
           code => "
               path = event.get('log')['file']['path']
               puts format('path = %<path>s', path: path)
               if (!path.nil?) && (!path.empty?)
                   event.set('filename', path.split('/')[-1])
               end
           "
       }
   }

   output {
     file   {
   #    path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{+HH}-vic.log" 
       path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
           codec => line { format => "%{filename} %{[fields][name]}: %{message}"}
     }

     stdout {
       codec => rubydebug
     }

   }
```

3. 启动Logstash

   > bin\logstash -f config\logstash.conf

   - 启动成功后可访问localhost:9600，看到版本等信息，则表示启动成功
   - 测试：在启动控制台输入信息， 可看到相关输出信息



#### 一些问题：

1. 在logstash中获取Filebeat中的日志原文件名，然后输出到同名文件:通过filter实现

   > 如下，获取输入的日志文件的文件名filename，通过`%{filename}`获取

   ```
   filter {
       ruby {
           code => "
               path = event.get('log')['file']['path']
               puts format('path = %<path>s', path: path)
               if (!path.nil?) && (!path.empty?)
                   event.set('filename', path.split('/')[-1])
               end
           "
       }
   }
   ```

   

2. filebeat携带额外的信息到，标志来自哪台机器，通过fields携带(如下，携带的key是name，value是server1)

   ```yaml
   filebeat.inputs:
   - type: log
     enabled: true
     fields:
       name: server1
   ```

3. logstash获取filebeat携带的额外的信息

   ```yaml
   %{[fields][name]}
   ```

4. filebeat   不发送最后一行

   > Filebeat使用换行符来检测事件的结束。 如果将行逐渐添加到正在采集的文件中，则在最后一行之后需要换行符，否则Filebeat将不会读取文件的最后一行。


#### 五 总结： `linux`下的启动脚本和最终配置文件

> `linux`下的安装和window下没什么区别，`logstash`需要确保安装了`jdk1.8`+;
>
> 启动也是一样的， 只是本人比较懒， 所以写了两个启动脚本`start.sh`  ；运行时带参数的话则输出日志到`catalina.out`，不带参数则不输出日志

#### `linux`下的`filebeat`的启动脚本

```shell
#!/bin/bash

log="/dev/null"
# if input paramter is not null ,then output log to catalina.out. else to /dev/null

if [ $1 != '' ];then
  log="catalina.out"	
fi

nohup ./filebeat -e -c filebeat.yml >$log 2>&1 &

echo "start   filebeat.......... ----> $log" 

```

##### `linux`下的`logstash`的启动脚本

```shell
#!/bin/bash

log="/dev/null"
# if input paramter is not null ,then output log to catalina.out. else to /dev/null

if [ $1 != '' ];then
  log="catalina.out"	
fi
nohup bin/./logstash -f config/logstash.conf >$log 2>&1 & 
echo "start logstash.......... ----> $log" 
```



##### `filebeat.yaml`最终配置文件(见注释)

```yaml
# 输入日志
filebeat.inputs:
- type: log
  enabled: true
  # 携带的额外的数据，标识来自哪个机器
  fields:
     name: wsl-ubuntu      
  # 日志的所在位置
  paths:
    - /data/share/logs/input/*.log  
    - /data/share/logs/input2/*.log  
#=========================== 是否开启多行合并================================
multiline.type: pattern
#多行合并的正则，以[ 开头
multiline.pattern: '^\['
# 是否不合并多行合并： 负负得正 表示开启
multiline.negate: false
multiline.match: after
#==========================输出日志到logstash==============================
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]
```

##### `logstash.conf`最终配置文件(见注释)

```yaml
#======从filebeat获取日志==============================
input {
  beats {
    port => 5044
  }
}

#=====解析出来源日志的文件名===================================
filter {
    ruby {
        code => "
            path = event.get('log')['file']['path']
            puts format('path = %<path>s', path: path)
            if (!path.nil?) && (!path.empty?)
                event.set('filename', path.split('/')[-1])
            end
        "
    }
}
#======输出日志==============================
output {
#  输出日志到文件，在日志的开头打印filebeat中携带的name属性，标识来自哪个文件
  file   {
    path => "/data/share/logs/output/logstash/%{+yyyy-MM-dd}/%{filename}"
	codec => line { format => "%{[fields][name]}: %{message}"}
  }
  # 输出日志到控制台  可删除掉
  stdout { 
    codec => rubydebug
  }
  
}

```



文章来源：[临窗旋墨的博客](http://www.xuqiudong.cn/detail/14)，转载注明出处。

时间： 20200930

filebeat收集日志输出到logstash然后合并输出到同名文件(7.7.1)

许秋冬偷偷隐藏起来的markdown原文,测试对蜘蛛是否友好

发表评论