logstash 使用

发表于 | 更新于 | 分类于 | 阅读次数：
| 字数： 9.1k | 时长 ≈ 8 分钟
标签

启动logstash

1
$ bin/logstash -f /opt/logstash/conf.d/ -r true

start.sh 启动脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
#!/bin/sh

# @Author: YL
# @Date:   2019-03-07 17:19:33
# @Last Modified by:   YL
# @Last Modified time: 2019-03-08 10:31:06

cd `dirname $0`
LS_HOME=`pwd`

LS_CONF="$LS_HOME/conf.d/"
LS_HOME_CORE_LIB="$LS_HOME/logstash-core/lib"

# 检测Tomcat是否启动（进程ID）
LS_PID=`ps -ef | grep $LS_HOME_CORE_LIB | grep -v grep | awk '{print $2}'`

#==================================================================================
# JVM Configuration
#==================================================================================
export LANG=en_US.UTF-8
export JAVA_HOME=/usr/java/jdk1.8.0_144

function operate(){
    if [[ "$1" = "kill" ]] ; then
        if [[ "$LS_PID" = "" ]] ; then
            echo "logstash not alive"
        else
            kill -9 $LS_PID
        fi
    elif [[ "$1" = "start" ]]; then
        nohup $LS_HOME/bin/logstash -f $LS_CONF -r true >/dev/null 2>&1 &
    else
        echo "not support: $1"
    fi
}
if [[ "$1" = "" || "$1" = "restart" || "$1" = "start" ]] ; then
    operate kill
    operate start
elif [[ "$1" = "kill" ]] ; then
    operate kill
elif [[ "$1" = "check" ]] ; then
    if [[ "$LS_PID" = "" ]] ; then
        operate start
    fi
else
    echo "not support: $1"
fi

crontab

1
* * * * * /opt/logstash/start.sh check

logstash config 命名

思考：如果logstash配置了多个config(多个input或多个filter)，怎样来控制它们被执行的顺序呢？

答案是logstash是按照config file的文件名，按照字母顺序合并了所有的config file，然后才处理的。

  • 最佳命名实践
1
2
3
4
5
6
7
8
00_input.conf

10_filter_global.conf

20_filter_nginx.conf
21_filter_vsftpd.conf

999_output.conf

条件判断

1
2
3
4
5
6
7
8
9
10
11
12
13
output {
    if [index_prefix] {
        elasticsearch {
            hosts => [ "elasticsearch:9200" ]
            index => "%{index_prefix}-%{+YYYY.MM.dd}"
        }
    } else {
        elasticsearch {
            hosts => [ "elasticsearch:9200" ]
            index => "logstash-%{+YYYY.MM.dd}"
        }
    }
}

比较操作有：

  • 相等: ==, !=, <, >, <=, >=
  • 正则: =~(匹配正则), !~(不匹配正则)
  • 包含: in(包含), not in(不包含)

布尔操作：

  • and(与), or(或), nand(非与), xor(非或)

一元运算符：

  • !(取反)
  • ()(复合表达式), !()(对复合表达式结果取反)

remove _field

  • 删除字段

    1
    2
    3
    mutate {
    	remove_field => [ "source" ]
    }

  • 删除嵌套字段（Nested fields）

    1
    2
    3
    mutate {
    	remove_field => [ "[host][name]" ]
    }

    注意：如果没有host字段，[host][name]引用会创建一个空的host字典，为了避免这种情况，可以按以下方式去做：

    1
    2
    3
    4
    5
    if [host] and [host][name] == "null" {
      mutate {
        remove_field => [ "[host][name]" ]
      }
    }

logstash-logback-encoder

logback发送日志到logstash

  • 解决logstash到elasticsearch时区差8小时的问题

    1
    2
    3
    <timestamp>
        <timeZone>GMT+8</timeZone>
    </timestamp>

  • 配置过滤器，过滤掉不需要的日志

    1
    2
    3
    4
    5
    6
    7
    <filter class="ch.qos.logback.core.filter.EvaluatorFilter">
        <evaluator class="ch.qos.logback.classic.boolex.JaninoEventEvaluator">
            <expression>return message.contains("api_monitor");</expression>
        </evaluator>
        <onMatch>ACCEPT</onMatch>
        <onMismatch>DENY</onMismatch>
    </filter>

pom.xml

1
2
3
4
5
6
7
8
9
10
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>5.3</version>
</dependency>
<dependency>
    <groupId>org.codehaus.janino</groupId>
    <artifactId>janino</artifactId>
    <version>3.0.12</version>
</dependency>

logback-spring.xml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
<springProfile name="pro">
    <appender name="logstash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>192.168.56.101:5044</destination>
        <keepAliveDuration>5 minutes</keepAliveDuration>
        <reconnectionDelay>1 second</reconnectionDelay>
        <filter class="ch.qos.logback.core.filter.EvaluatorFilter">
            <evaluator class="ch.qos.logback.classic.boolex.JaninoEventEvaluator">
                <expression>
                    return message.contains("api_monitor");
                </expression>
            </evaluator>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
        <encoder charset="${logger.charset}" class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder">
            <providers>
                <timestamp>
                    <timeZone>GMT+8</timeZone>
                </timestamp>
                <pattern>
                    <pattern>
                        {
                        "time": "%d{yy-MM-dd HH:mm:ss}",
                        "level": "%p",
                        "thread": "%t",
                        "class": "%c{2}.%M\\(%L\\)",
                        "message": "%m"
                        }
                    </pattern>
                </pattern>
                <stackTrace>
                    <throwableConverter class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
                        <maxDepthPerThrowable>full</maxDepthPerThrowable>
                        <maxLength>full</maxLength>
                        <shortenedClassNameLength>20</shortenedClassNameLength>
                        <rootCauseFirst>true</rootCauseFirst>
                        <inlineHash>true</inlineHash>
                        <!-- generated class names -->
                        <exclude>\$\$FastClassByCGLIB\$\$</exclude>
                        <exclude>\$\$EnhancerBySpringCGLIB\$\$</exclude>
                        <exclude>^sun\.reflect\..*\.invoke</exclude>
                        <!-- JDK internals -->
                        <exclude>^com\.sun\.</exclude>
                        <exclude>^sun\.net\.</exclude>
                        <!-- dynamic invocation -->
                        <exclude>^net\.sf\.cglib\.proxy\.MethodProxy\.invoke</exclude>
                        <exclude>^org\.springframework\.cglib\.</exclude>
                        <exclude>^org\.springframework\.transaction\.</exclude>
                        <exclude>^org\.springframework\.validation\.</exclude>
                        <exclude>^org\.springframework\.app\.</exclude>
                        <exclude>^org\.springframework\.aop\.</exclude>
                        <exclude>^java\.lang\.reflect\.Method\.invoke</exclude>
                        <!-- Spring plumbing -->
                        <exclude>^org\.springframework\.ws\..*\.invoke</exclude>
                        <exclude>^org\.springframework\.ws\.transport\.</exclude>
                        <exclude>^org\.springframework\.ws\.soap\.saaj\.SaajSoapMessage\.</exclude>
                        <exclude>^org\.springframework\.ws\.client\.core\.WebServiceTemplate\.</exclude>
                        <exclude>^org\.springframework\.web\.filter\.</exclude>
                        <!-- Tomcat internals -->
                        <exclude>^org\.apache\.tomcat\.</exclude>
                        <exclude>^org\.apache\.catalina\.</exclude>
                        <exclude>^org\.apache\.coyote\.</exclude>
                        <exclude>^java\.util\.concurrent\.ThreadPoolExecutor\.runWorker</exclude>
                        <exclude>^java\.lang\.Thread\.run$</exclude>
                    </throwableConverter>
                </stackTrace>
            </providers>
        </encoder>
    </appender>
    <root level="info">
        <appender-ref ref="logstash"/>
    </root>
</springProfile>

使用中遇到的问题

gsub

有问题

1
2
3
4
5
6
7
8
9
10
mutate {
    gsub => [
        "city", "5", "深圳",
        "city", "50", "东莞",
        "city", "51", "江门"
    ]
}
# 5进来，替换成“深圳”
# 50进来，会被替换成“深圳0”
# 51进来，会被替换成“深圳1”

正常

1
2
3
4
5
6
7
8
9
10
mutate {
    gsub => [
        "city", "^5$", "深圳",
        "city", "^50$", "东莞",
        "city", "^51$", "江门"
    ]
}
# 5进来，替换成“深圳”
# 50进来，会被替换成“东莞”
# 51进来，会被替换成“江门”
1
2
3
4
5
6
7
8
9
10
mutate {
    gsub => [
        "city", "50", "东莞",
        "city", "51", "江门",
        "city", "5", "深圳"
    ]
}
# 5进来，替换成“深圳”
# 50进来，会被替换成“东莞”
# 51进来，会被替换成“江门”
  • 本文作者： forever杨
  • 本文链接： https://blog.yl-online.top/posts/1e58019c.html
  • 版权声明： 本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。如果文章内容对你有用，请记录到你的笔记中。本博客站点随时会停止服务，请不要收藏、转载！