input {

        file {

                type => "mysql-slow"

                path => "/tmp/slows.log"

                codec => multiline {

                        pattern => "^# Time:"

                        negate => true

                        what => "previous"

                }

        }

}

filter {

        if [ type ] == "mysql-slow" {

                grok {

                        match => { "message" => "SELECT SLEEP" }

                        add_tag => [ "sleep_drop" ]

                }

                if "sleep_drop" in [tags] {

                        drop {}

                }

                grok {

                    "message" => "(?m)^#\s+Time\s?.*\s+#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IPV4:clientip})?\]\s+Id:\s+%{NUMBER:row_id:int}\n#\s+Query_time:\s+%{NUMBER:query_time:float}\s+Lock_time:\s+%{NUMBER:lock_time:float}\s+Rows_sent:\s+%{NUMBER:rows_sent:int}\s+Rows_examined:\s+%{NUMBER:rows_examined:int}\n\s*(?:use %{DATA:database};\s*\n)?SET\s+timestamp=%{NUMBER:timestamp};\n\s*(?<sql>(?<action>\w+)([\w.*\W.*])*;)\s*$"    

                }

        }

}

1.将go服务集成到ELK

http://t.cn/A6AsbP9A

2.ES安全功能入门

http://t.cn/A62bW6m4

3.ES动态同义词插件

http://t.cn/A62blCcq

• 编辑：bsll

• 归档：https://ela.st/cn-daily-all

• 订阅：https://ela.st/cn-daily-sub

• 沙龙：https://ela.st/cn-meetup

注意： 此解决方案，短时间内没有复现，还需要长时间验证是否有效。

现象

在使用HTTP方式，Elasticsearch 长时间不查询后，再次查询会出现抛出SocketTimeoutException的问题。

原因

基本逻辑

Elasticsearch 客户端会根据服务器返回的HTTP报文内容，来决定客户端保持HTTP连接Keep-Alive状态的策略。
如果结果如下,那么保持HTTP连接 Keep-Alive状态为120s

Connection: Keep-Alive
Keep-Alive: max=5, timeout = 120

如果不包含上述内容，那么客户端将保持Keep-Alive状态的时间为永久。
事实上，Elasticsearch服务器返回的报文，并没有上述HTTP头内容，所以客户端所有的HTTP连接都为永久保持Keep-Alive。
如果客户端长时间没有发送请求，服务器或者防火墙已经close了HTTP底层的TCP链接，但是此时客户端并不知道，由于Keep Alive是无限期，那么并不会重新建立连接，而是直接发送请求，此时就会得到SocketTimeout异常。

阅读源码

我使用的Elasticsearch的客户端下面的版本

<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>rest</artifactId>
    <version>5.4.1</version>
</dependency>

其HTTP的发送依赖Maven包httpasyncclient.

这个包中的接口ConnectionKeepAliveStrategy，抽象了处理 HTTP Keepalive 的策略，其默认实现为：

@Contract(threading = ThreadingBehavior.IMMUTABLE)
public class DefaultConnectionKeepAliveStrategy implements ConnectionKeepAliveStrategy {

    public static final DefaultConnectionKeepAliveStrategy INSTANCE = new DefaultConnectionKeepAliveStrategy();

    @Override
    public long getKeepAliveDuration(final HttpResponse response, final HttpContext context) {
        Args.notNull(response, "HTTP response");
        final HeaderElementIterator it = new BasicHeaderElementIterator(
                response.headerIterator(HTTP.CONN_KEEP_ALIVE));
        while (it.hasNext()) {
            final HeaderElement he = it.nextElement();
            final String param = he.getName();
            final String value = he.getValue();
            if (value != null && param.equalsIgnoreCase("timeout")) {
                try {
                    return Long.parseLong(value) * 1000;
                } catch(final NumberFormatException ignore) {
                }
            }
        }
        return -1;
    }
}

-1代表多长时间，接口说明不是很清楚。
PoolingNHttpClientConnectionManager 类中的代码，实现了上述对待KeepAlive的逻辑，可以看到-1表示为：无限期

@Override
    public void releaseConnection(
            final NHttpClientConnection managedConn,
            final Object state,
            final long keepalive,
            final TimeUnit tunit) {
        Args.notNull(managedConn, "Managed connection");
        synchronized (managedConn) {
            final CPoolEntry entry = CPoolProxy.detach(managedConn);
            if (entry == null) {
                return;
            }
            if (this.log.isDebugEnabled()) {
                this.log.debug("Releasing connection: " + format(entry) + formatStats(entry.getRoute()));
            }
            final NHttpClientConnection conn = entry.getConnection();
            try {
                if (conn.isOpen()) {
                    entry.setState(state);
                    entry.updateExpiry(keepalive, tunit != null ? tunit : TimeUnit.MILLISECONDS);
                    if (this.log.isDebugEnabled()) {
                        final String s;
                       // keepalive 就是上面接口 ConnectionKeepAliveStrategy.getKeepAliveDuration()的返回值
                        if (keepalive > 0) {
                            s = "for " + (double) keepalive / 1000 + " seconds";
                        } else {
                       // 如果小于0 ，那么策略为indefinitely：无限期。
                            s = "indefinitely";
                        }
                        this.log.debug("Connection " + format(entry) + " can be kept alive " + s);
                    }
                }
            } finally {
                this.pool.release(entry, conn.isOpen() && entry.isRouteComplete());
                if (this.log.isDebugEnabled()) {
                    this.log.debug("Connection released: " + format(entry) + formatStats(entry.getRoute()));
                }
            }
        }
    }

解决方式

自定义类实现ConnectionKeepAliveStrategy接口：

public class CustomConnectionKeepAliveStrategy extends DefaultConnectionKeepAliveStrategy {

    public static final CustomConnectionKeepAliveStrategy INSTANCE = new CustomConnectionKeepAliveStrategy();

    private CustomConnectionKeepAliveStrategy() {
        super();
    }

    /**
     * 最大keep alive的时间（分钟）
     * 这里默认为10分钟，可以根据实际情况设置。可以观察客户端机器状态为TIME_WAIT的TCP连接数，如果太多，可以增大此值。
     */
    private final long MAX_KEEP_ALIVE_MINUTES = 10;

    @Override
    public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
        long keepAliveDuration = super.getKeepAliveDuration(response, context);
        // <0 为无限期keepalive
        // 将无限期替换成一个默认的时间
        if(keepAliveDuration < 0){
            return TimeUnit.MINUTES.toMillis(MAX_KEEP_ALIVE_MINUTES);
        }
        return keepAliveDuration;
    }
}

在创建Elasticserach Client时，配置

RestClientBuilder builder = RestClient.builder(hosts);
builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
    @Override
    public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
        httpClientBuilder.setKeepAliveStrategy(CustomConnectionKeepAliveStrategy.INSTANCE);
        return httpClientBuilder;
    }
});

1.跨集群复制索引工具

http://t.cn/A6ADWeIA

2.通过组合布尔查询提高ES查询结果的相关性

http://t.cn/A6ADlEA5

3.Beats导入RabbitMQ日志至阿里云Elatic Stack实现可视化分析

http://t.cn/A6ADlOtX

• 编辑：bsll

• 归档：https://ela.st/cn-daily-all

• 订阅：https://ela.st/cn-daily-sub

• 沙龙：https://ela.st/cn-meetup