文章 - 搜索客，搜索人自己的社区

大家聊一聊使用的什么版本的Elasticsearch，看看Elasticsearch版本变化

我是最近从lucene过渡Elasticsearch的，直接用的最新的2.2.0版本的。发现离线安装插件的方式和以前不一样了，一些配置也有改变，最大的问题是java client api 连接报了如下的异常，我是参照官方api测试的，地址：https://www.elastic.co/guide/e ... .html

org.elasticsearch.transport.RemoteTransportException: Failed to deserialize exception response from stream

谷歌都说是服务和客户端的jvm不一致，我是本机环境测试的，所以，现在这个问题都还没解决，有遇到过的么，还是和版本有关系啊

继续阅读 »

我是最近从lucene过渡Elasticsearch的，直接用的最新的2.2.0版本的。发现离线安装插件的方式和以前不一样了，一些配置也有改变，最大的问题是java client api 连接报了如下的异常，我是参照官方api测试的，地址：https://www.elastic.co/guide/e ... .html

org.elasticsearch.transport.RemoteTransportException: Failed to deserialize exception response from stream

谷歌都说是服务和客户端的jvm不一致，我是本机环境测试的，所以，现在这个问题都还没解决，有遇到过的么，还是和版本有关系啊收起阅读 »

Elastic线下交流活动走起来！

线上交流不过瘾？那就参加线下交流活动吧！
这里是搜罗的最新的线下交流活动预告：

广州：http://www.meetup.com/guangzhou-Elastic-Meetup/
上海：http://www.meetup.com/shanghai-Elastic-Meetup/
北京：http://www.meetup.com/beijing-Elastic-Meetup/

大家分别找到组织报名参加吧，貌似有些还需要场地支持，大家一起出谋划策，把活动办起来吧。

继续阅读 »

es索引模版配置不当导致的aggs聚合查询字段显示错误的问题

今天在es中对http日志的状态码status进行aggs搜索出现字段内容显示不正常的问题，记录过程：

http日志的情况：
1、http日志从logstash写入es时，状态码配置为status，其内容为 200 ，302 ，400 ，404等。
2、使用kibana对该日志的索引进行查询，在discover页面中显示的status内容跟logstash的内容一致，是正常的。

出现问题的场景：
（我这里使用的是kibana的sense插件进行的查询，如果直接使用curl python-ES也是一样的）
查询该索引：
POST http-2016.03.18/_search
{
"fields": ["status"],
"query":{
"bool":{
"must": [
{
"range" : {
"@timestamp" : {"gte" : "now-5m"}
}
}
]
}
},
"_source": "false",
"size": 0,
"aggs": {
"status_type": {
"terms":{"field":"status"}
}
}
}

查询返回的结果中aggregations部分的内容：
"aggregations" : {
"status_type" : {
"doc_count_error_upper_bound" : 0,
"sum_other_doc_count" : 0,
"buckets" : [ {
"key" : -56,
"doc_count" : 376341
}, {
"key" : 46,
"doc_count" : 51439
}, {
"key" : 45,
"doc_count" : 5543
}, {
"key" : 48,
"doc_count" : 1669
}, {
"key" : -108,
"doc_count" : 1068
}, {
"key" : -50,
"doc_count" : 11
}, {
"key" : -109,
"doc_count" : 8
}, {
"key" : -112,
"doc_count" : 4
}

寻找原因：
起先先去掉了查询的aggs部分，单独查询query的内容：
POST http-2016.03.18/_search
{
"fields": ["status"],
"query":{
"bool":{
"must": [
{
"range" : {
"@timestamp" : {"gte" : "now-5m"}
}
}
]
}
}
}

返回的结果中，hits显示的status字段内容是正常的：
"hits": {
"total": 1242104,
"max_score": 1,
"hits": [
{
"_index": "http-2016.03.18",
"_type": "log",
"_id": "AVOI3EiwidwPAhB1e7gQ",
"_score": 1,
"fields": {
"status": [
"200"
]
}
}
......

然后查询了http索引的索引信息和模版配置：
GET /http-2016.03.18/
GET /_template/http
发现其中http的status的属性type类型的内容是byte ：
"properties": {
"@timestamp": {
"type": "date",
"format": "strict_date_optional_time||epoch_millis"
},
......
......
"status": {
"type": "byte"
},
......
......

原因：
在aggs查询中发现了status字段显示错误的情况，status的type类型在es模版中定义成了byte类型，当status的值超过127后将出现溢出的情况，因此修改为short后，恢复了正常。
（对于http的状态码status，其type类型使用short已经足够了，如果使用integer，long或默认的string类型也是可以的，这里影响的是存储空间占用的大小。）

继续阅读 »

今天在es中对http日志的状态码status进行aggs搜索出现字段内容显示不正常的问题，记录过程：

http日志的情况：
1、http日志从logstash写入es时，状态码配置为status，其内容为 200 ，302 ，400 ，404等。
2、使用kibana对该日志的索引进行查询，在discover页面中显示的status内容跟logstash的内容一致，是正常的。

出现问题的场景：
（我这里使用的是kibana的sense插件进行的查询，如果直接使用curl python-ES也是一样的）
查询该索引：
POST http-2016.03.18/_search
{
"fields": ["status"],
"query":{
"bool":{
"must": [
{
"range" : {
"@timestamp" : {"gte" : "now-5m"}
}
}
]
}
},
"_source": "false",
"size": 0,
"aggs": {
"status_type": {
"terms":{"field":"status"}
}
}
}

查询返回的结果中aggregations部分的内容：
"aggregations" : {
"status_type" : {
"doc_count_error_upper_bound" : 0,
"sum_other_doc_count" : 0,
"buckets" : [ {
"key" : -56,
"doc_count" : 376341
}, {
"key" : 46,
"doc_count" : 51439
}, {
"key" : 45,
"doc_count" : 5543
}, {
"key" : 48,
"doc_count" : 1669
}, {
"key" : -108,
"doc_count" : 1068
}, {
"key" : -50,
"doc_count" : 11
}, {
"key" : -109,
"doc_count" : 8
}, {
"key" : -112,
"doc_count" : 4
}

寻找原因：
起先先去掉了查询的aggs部分，单独查询query的内容：
POST http-2016.03.18/_search
{
"fields": ["status"],
"query":{
"bool":{
"must": [
{
"range" : {
"@timestamp" : {"gte" : "now-5m"}
}
}
]
}
}
}

返回的结果中，hits显示的status字段内容是正常的：
"hits": {
"total": 1242104,
"max_score": 1,
"hits": [
{
"_index": "http-2016.03.18",
"_type": "log",
"_id": "AVOI3EiwidwPAhB1e7gQ",
"_score": 1,
"fields": {
"status": [
"200"
]
}
}
......

然后查询了http索引的索引信息和模版配置：
GET /http-2016.03.18/
GET /_template/http
发现其中http的status的属性type类型的内容是byte ：
"properties": {
"@timestamp": {
"type": "date",
"format": "strict_date_optional_time||epoch_millis"
},
......
......
"status": {
"type": "byte"
},
......
......

原因：
在aggs查询中发现了status字段显示错误的情况，status的type类型在es模版中定义成了byte类型，当status的值超过127后将出现溢出的情况，因此修改为short后，恢复了正常。
（对于http的状态码status，其type类型使用short已经足够了，如果使用integer，long或默认的string类型也是可以的，这里影响的是存储空间占用的大小。）

收起阅读 »

Elastic{ON}16的资料已经放出

视频：https://www.elastic.co/elasticon/conf/2016/sf
PPT: https://speakerdeck.com/elastic

继续阅读 »

使用 SQL 查询 Elasticsearch

我新写了一个用 SQL 查询 Elasticsearch 的工具 https://github.com/taowen/es-monitor，欢迎大家使用。详细的文档参见：https://segmentfault.com/a/1190000003502849

在此之前，有这么三个SQL查询Elasticsearch的工具：

Crate.io 的问题是它不是Elasticsearch，它的聚合是自己实现的版本，和Elasticsearch的Aggregation是两套东西。
http://sqltoelasticsearch.fr/ 语法支持很不晚上，同时 WHERE 和 GROUP BY 就翻译错了。
https://github.com/NLPchina/elasticsearch-sql 的问题在于其用Java来翻译SQL太笨拙了，如果要达到同样的SQL语法支持程度还要增加大量的Java代码。
如果只是支持SQL，很多Elasticsearch的功能是无法被充分释放的。比如Elasticsearch支持sub aggregation，每个sub aggregation就是OLAP里的下钻一次的概念。而且每下钻一次都可以有自己的指标计算。简单的SQL是无法表达这样的特性的。所以我扩充了一下SQL的语义，使得其更贴近Elasticsearch聚合的工作方式：

$ cat << EOF | ./es_query.py http://127.0.0.1:9200 

    WITH SELECT MAX(market_cap) AS max_all_times FROM symbol AS all_symbols; 

    WITH SELECT MAX(market_cap) AS max_at_2000 FROM all_symbols WHERE ipo_year=2000 AS year_2000; 

    WITH SELECT MAX(market_cap) AS max_at_2001 FROM all_symbols WHERE ipo_year=2001 AS year_2001;

EOF

希望我的小工具可以帮到你

继续阅读 »

我新写了一个用 SQL 查询 Elasticsearch 的工具 https://github.com/taowen/es-monitor，欢迎大家使用。详细的文档参见：https://segmentfault.com/a/1190000003502849

在此之前，有这么三个SQL查询Elasticsearch的工具：

Crate.io 的问题是它不是Elasticsearch，它的聚合是自己实现的版本，和Elasticsearch的Aggregation是两套东西。
http://sqltoelasticsearch.fr/ 语法支持很不晚上，同时 WHERE 和 GROUP BY 就翻译错了。
https://github.com/NLPchina/elasticsearch-sql 的问题在于其用Java来翻译SQL太笨拙了，如果要达到同样的SQL语法支持程度还要增加大量的Java代码。
如果只是支持SQL，很多Elasticsearch的功能是无法被充分释放的。比如Elasticsearch支持sub aggregation，每个sub aggregation就是OLAP里的下钻一次的概念。而且每下钻一次都可以有自己的指标计算。简单的SQL是无法表达这样的特性的。所以我扩充了一下SQL的语义，使得其更贴近Elasticsearch聚合的工作方式：

$ cat << EOF | ./es_query.py http://127.0.0.1:9200 

    WITH SELECT MAX(market_cap) AS max_all_times FROM symbol AS all_symbols; 

    WITH SELECT MAX(market_cap) AS max_at_2000 FROM all_symbols WHERE ipo_year=2000 AS year_2000; 

    WITH SELECT MAX(market_cap) AS max_at_2001 FROM all_symbols WHERE ipo_year=2001 AS year_2001;

EOF

希望我的小工具可以帮到你
收起阅读 »

ElasticSearch2.1.1安装及简单配置说明

目前最新版ES超级详细的安装、配置流程。
根据自己真实的安装过程以及多篇博客文章的重要提示编写。
按照文档中的说明一步一步操作，分分钟就能开始ES2.1.1的非凡体验！

nest驱动IndexName问题

nest驱动访问ES，按照官网文档，使用如下程序可以正常索引：
static void Main()
        {
            var node = new Uri("http://localhost:9200");

            var settings = new ConnectionSettings(
                node,
                defaultIndex: "my-application"
            );
            var client = new ElasticClient(settings);
            var person = new Person
            {
                Id = "1",
                Firstname = "Martijn",
                Lastname = "Laarman"
            };
            var index = client.Index(person);
        }
调整为手动设置indexName时出错，示例代码如下：
static void Main()
        {
            var node = new Uri("http://localhost:9200");

            var settings = new ConnectionSettings(
                node,
                defaultIndex: "my-application"
            );
            settings.MapDefaultTypeIndices(d => d.Add(typeof(Person), "constIndex"));
            var client = new ElasticClient(settings);
            var person = new Person
            {
                Id = "1",
                Firstname = "Martijn",
                Lastname = "Laarman"
            };
            var index = client.Index(person);
        }
出错提示为：
{StatusCode: 400,
Method: PUT,
Url: http://localhost:9200/constIndex/automobile/1,
Request: {
"firstname": "Martijn",
"lastname": "Laarman",
"id": "1"
},
Response: <Response stream not captured or already read to completion by serializer, set ExposeRawResponse() on connectionsettings to force it to be set on>}

using System;

using Nest;



namespace ConsoleApplication1

{

    class Program

    {

        private static IIndexResponse Index<T>(T person, string indexName) where T : class

        {

            var node = new Uri("http://localhost:9200&quot;);



            var settings = new ConnectionSettings(node,defaultIndex: indexName);

            var client = new ElasticClient(settings);

            return client.Index(person);

        }



        static void Main()

        {

            string indexNameError = typeof(Person).FullName

                .Substring(typeof(Person).FullName.LastIndexOf(".", StringComparison.Ordinal) + 1) + "Indexs";

            const string indexNameOk = "test";

            var person = new Person

            {

                Id = "1",

                Firstname = "Martijn",

                Lastname = "Laarman"

            };

            var ok = Index(person, indexNameOk);

            Console.WriteLine("Result:" + ok.IsValid);

            var error = Index(person, indexNameError);

            Console.WriteLine("Result:" + error.IsValid);

            Console.ReadKey();

        }

       

    }



   [Serializable]

    public class Person

    {

        public string Firstname { get; set; }

        public string Lastname { get; set; }

        public string Id { get; set; }

    }

}

继续阅读 »

nest驱动访问ES，按照官网文档，使用如下程序可以正常索引：
static void Main()
        {
            var node = new Uri("http://localhost:9200");

            var settings = new ConnectionSettings(
                node,
                defaultIndex: "my-application"
            );
            var client = new ElasticClient(settings);
            var person = new Person
            {
                Id = "1",
                Firstname = "Martijn",
                Lastname = "Laarman"
            };
            var index = client.Index(person);
        }
调整为手动设置indexName时出错，示例代码如下：
static void Main()
        {
            var node = new Uri("http://localhost:9200");

            var settings = new ConnectionSettings(
                node,
                defaultIndex: "my-application"
            );
            settings.MapDefaultTypeIndices(d => d.Add(typeof(Person), "constIndex"));
            var client = new ElasticClient(settings);
            var person = new Person
            {
                Id = "1",
                Firstname = "Martijn",
                Lastname = "Laarman"
            };
            var index = client.Index(person);
        }
出错提示为：
{StatusCode: 400,
Method: PUT,
Url: http://localhost:9200/constIndex/automobile/1,
Request: {
"firstname": "Martijn",
"lastname": "Laarman",
"id": "1"
},
Response: <Response stream not captured or already read to completion by serializer, set ExposeRawResponse() on connectionsettings to force it to be set on>}

using System;

using Nest;



namespace ConsoleApplication1

{

    class Program

    {

        private static IIndexResponse Index<T>(T person, string indexName) where T : class

        {

            var node = new Uri("http://localhost:9200&quot;);



            var settings = new ConnectionSettings(node,defaultIndex: indexName);

            var client = new ElasticClient(settings);

            return client.Index(person);

        }



        static void Main()

        {

            string indexNameError = typeof(Person).FullName

                .Substring(typeof(Person).FullName.LastIndexOf(".", StringComparison.Ordinal) + 1) + "Indexs";

            const string indexNameOk = "test";

            var person = new Person

            {

                Id = "1",

                Firstname = "Martijn",

                Lastname = "Laarman"

            };

            var ok = Index(person, indexNameOk);

            Console.WriteLine("Result:" + ok.IsValid);

            var error = Index(person, indexNameError);

            Console.WriteLine("Result:" + error.IsValid);

            Console.ReadKey();

        }

       

    }



   [Serializable]

    public class Person

    {

        public string Firstname { get; set; }

        public string Lastname { get; set; }

        public string Id { get; set; }

    }

}

收起阅读 »

Packetbeat协议扩展开发教程（3）

书接上回：http://elasticsearch.cn/article/53

前面介绍了Packetbeat的项目结构，今天终于要开始写代码了，想想还是有点小激动呢。（你快点吧，拖半天了）
网络传输两大协议TCP和UDP，我们的所有协议都不离这两种，HTTP、MySQL走的是TCP传输协议，DNS走的是UDP协议，在Packetbeat里面，实现一个自己的协议非常简单，继承并实现这两者对应的接口就行了，我们看一下长什么样：
打开一个现有的UDP和HTTP协议接口定义：
/~/go/src/github.com/elastic/beats/packetbeat/protos/protos.go

// Functions to be exported by a protocol plugin

type ProtocolPlugin interface {

	// Called to initialize the Plugin

	Init(test_mode bool, results publisher.Client) error

 

	// Called to return the configured ports

	GetPorts() int

}

 

type TcpProtocolPlugin interface {

	ProtocolPlugin

 

	// Called when TCP payload data is available for parsing.

	Parse(pkt *Packet, tcptuple *common.TcpTuple,

		dir uint8, private ProtocolData) ProtocolData

 

	// Called when the FIN flag is seen in the TCP stream.

	ReceivedFin(tcptuple *common.TcpTuple, dir uint8,

		private ProtocolData) ProtocolData

 

	// Called when a packets are missing from the tcp

	// stream.

	GapInStream(tcptuple *common.TcpTuple, dir uint8, nbytes int,

		private ProtocolData) (priv ProtocolData, drop bool)

 

	// ConnectionTimeout returns the per stream connection timeout.

	// Return <=0 to set default tcp module transaction timeout.

	ConnectionTimeout() time.Duration

}

 

type UdpProtocolPlugin interface {

	ProtocolPlugin

 

	// ParseUdp is invoked when UDP payload data is available for parsing.

	ParseUdp(pkt *Packet)

}

TcpProtocolPlugin：TCP协议插件的接口定义，依次是：Parse() 解析Packet，ReceivedFin()处理TCP断开连接，GapInStream()处理空包丢包，ConnectionTimeout()超时时间；
UdpProtocolPlugin: UDP协议的接口定义，UDP协议是不需要握手和保障数据可靠性的，扔出去就结束，速度快，不保证数据可靠送达，所以只有ParseUdp一个方法需要实现，比较简单；
ProtocolPlugin：TCP和UDP都需要实现ProtocolPlugin的基础接口，其实就定义了获取端口和初始化接口。

请问：
Packetbeat怎么工作的？

回答：
每一个协议都有一个固定的端口用于通信，你要做的事情就是定义协议端口，然后按协议是TCP还是UDP来实现对应的接口，Packetbeat将会截获指定端口的数据包（Packet），然后如果交给你定义的方法来进行解析，TCP是Parse，UDP是ParseUdp，都在上面的接口定义好的，然后将解析出来的结构化数据封装成Json，然后扔给Elasticsearch，后续的就的如何对这些数据做一些有趣的分析和应用了。

貌似很简单嘛！

进入每个端口的数据包，我们假设是一个自来水管，拧开80端口，哗啦啦出来的全是HTTP请求的数据包，Packetbeat里面Http协议监听的是80端口啊，所有这些包统统都交给Packetbeat里面的Http协议模块来进行解析，Http协议会一个个的检查这些数据包，也就是每个数据包都会调用一次Parse接口，到这里提到了传过来一个Packet，我们看看它的数据结构长什么样？

type Packet struct {

	Ts      time.Time

	Tuple   common.IpPortTuple

	Payload byte

}

Packet结构简单，
Ts是收到数据包的时间戳；
Tuple是一个来源IP+来源端口和目的IP+目的端口的元组；
Payload就是这个包里面的传输的有用的数据，应用层的字节数据，不包括IP和TCP/UDP头信息，是不是处理起来简单许多。

首选我们确定SMTP协议的配置，每个协议在packetbeat.yml的protocol下面都应该有一个配置节点，如下：

protocols:

  smtp:

    # Configure the ports where to listen for Smtp traffic. You can disable

    # the Smtp protocol by commenting out the list of ports.

    ports: [25]

还需要在对应的config类文件：packetbeat/config/config.go，增加SMTP的结构体，目前只支持一个端口参数，继承基类ProtocolCommon就行，如下：

git diff config/config.go

@@ -42,6 +42,7 @@ type Protocols struct {

        Pgsql    Pgsql

        Redis    Redis

        Thrift   Thrift

+       Smtp     Smtp

 }

 

 type Dns struct {

@@ -118,5 +119,9 @@ type Redis struct {

        Send_response *bool

 }

 

+type Smtp struct {

+	ProtocolCommon        `yaml:",inline"`

+}

+

 // Config Singleton

 var ConfigSingleton Config

在protos文件夹下面，新增smtp目录，并新增空白文件smtp.go，路径：packetbeat/protos/smtp/smtp.go，
这里就是解析SMTP协议的地方，也是我们扩展协议的主要的工作。

...TODO...

修改protos/protos.go，增加SMTP协议枚举，这里记得保证顺序一致，并且protocol名称必须和配置的节点名称一致，如这里都是smtp。

git diff protos/protos.go

@@ -103,6 +103,7 @@ const (

        MongodbProtocol

        DnsProtocol

        MemcacheProtocol

+       SmtpProtocol

 )

 

 // Protocol names

@@ -116,6 +117,7 @@ var ProtocolNames = string{

        "mongodb",

        "dns",

        "memcache",

+       "smtp",

 }

继续修改packetbeat.go主文件，允许SMTP协议并加载。

git diff packetbeat.go

@@ -27,6 +27,7 @@ import (

        "github.com/elastic/packetbeat/protos/tcp"

        "github.com/elastic/packetbeat/protos/thrift"

        "github.com/elastic/packetbeat/protos/udp"

+       "github.com/elastic/packetbeat/protos/smtp"

        "github.com/elastic/packetbeat/sniffer"

 )

 

@@ -43,6 +44,7 @@ var EnabledProtocolPlugins map[protos.Protocol]protos.ProtocolPlugin = map[proto

        protos.ThriftProtocol:   new(thrift.Thrift),

        protos.MongodbProtocol:  new(mongodb.Mongodb),

        protos.DnsProtocol:      new(dns.Dns),

+       protos.SmtpProtocol:      new(smtp.Smtp),

 }

做完上面一系列修改之后，一个空白的SMTP协议的插件的架子就搭好了，并且插件也注册到了Packetbeat里面了，接下来我们再把packetbeat/protos/smtp/smtp.go按照TCPplugin接口的要求实现一下。

说实话TCP处理起来很难，开始之前，我们先明确几个概念，TCP协议是有状态的，并且是流式的，我们关注的是七层应用层的消息，如HTTP里面的一个HTTP请求和返回，但是TCP底层都是一系列数据包，并且不同的请求的数据包是混杂在一起的，也就是说一个数据包里面可能只是一个HTTP请求的一部分也可能包含多条HTTP请求的一部分，所以Parse()里面需要处理跨数据包的状态信息，我们要把这些数据包和具体的七层的应用层的消息关联起来。

现在我们仔细看看Parse()接口的各个参数定义是做什么用的

Parse(pkt *Packet, tcptuple *common.TcpTuple,

		dir uint8, private ProtocolData) ProtocolData

pkt不用说了，是送进来的数据包，前面已经介绍了其数据结构，tcptuple是该数据包所属的TCP数据流所在的唯一标示（一个未关闭的TCP数据量包含若干数据包，直到TCP链接关闭），使用tcptuple.Hashable()获取唯一值；dir参数标示数据包在TCP数据流中的流向，和第一个TCP数据包方向一致是TcpDirectionOriginal，否则是TcpDirectionReverse；private参数可用来在TCP流中存储状态信息，可在运行时转换成具体的强类型，任意修改和传递给下一个Parse方法，简单来说就是进行中间数据的共享。

下面看段MySQL模块里面的例子

 priv := mysqlPrivateData{}

        if private != nil {

                var ok bool

                priv, ok = private.(mysqlPrivateData)

                if !ok {

                        priv = mysqlPrivateData{}

                }

        }

 

        [ ... ]

 

        return priv

上面的代码就是将private强制转换成mysqlPrivateData结构，然后再使用。
我们再继续看后续怎么处理这些包的一个逻辑例子

ok, complete := mysqlMessageParser(priv.Data[dir])

                if !ok {

                        // drop this tcp stream. Will retry parsing with the next

                        // segment in it

                        priv.Data[dir] = nil

                        logp.Debug("mysql", "Ignore MySQL message. Drop tcp stream.")

                        return priv

                }

 

                if complete {

                        mysql.messageComplete(tcptuple, dir, stream)

                } else {

                        // wait for more data

                        break

                }

mysqlMessageParser是一个解析mysql消息的方法，细节我们忽略，我们只需要关心它的返回，ok标示成功或者失败，true则继续处理，false表示数据包不能用，那就直接忽略；第二个参数complete表示判断这一个MySQL消息是否已经完整了，如果完整了，我们就可以扔出去了，否则继续等待剩下的消息内容。

好的，我们看看SMTP协议怎么折腾吧，先看看一个邮件交互的流程图，来自RFC5321：

由上图可见，发送端和邮件服务器通过一系列命令来执行邮件的发送，下面看看一个具体的命令操作流程(来源：简单邮件传输协议)[/url]

S: 220 www.example.com ESMTP Postfix

C: HELO mydomain.com

S: 250 Hello mydomain.com

C: MAIL FROM: 

S: 250 Ok

C: RCPT TO: 

S: 250 Ok

C: DATA

S: 354 End data with .

C: Subject: test message

C: From:""< sender@mydomain.com>

C: To:""< friend@example.com>

C:

C: Hello,

C: This is a test.

C: Goodbye.

C: .

S: 250 Ok: queued as 12345

C: quit

S: 221 Bye

上面的过程可以看到就几个命令就能将邮件发送出去，但是其实SMTP协议比较复杂，还包括身份认证、附件、多媒体编码等等，我们今天精简一下，我们目前只关心谁给谁发了邮件，发送内容先不管，这样相比完整的SMTP协议(RFC5321)，我们只需要关注以下几个命令：
MAIL：开始一份邮件 mail　from: xxx@xx.com
RCPT: 标识单个的邮件接收人；常在mail命令后面可有多个rcpt　to: xx@xx.com
QUIT：结束SMTP会话，不一定发送了邮件，注意
RESET:重置会话，当前传输被取消

最终希望通过Packetbeat将这些数据解析并处理成我们想要的如下JSON数据，即大功告成：

{

"timestamp":"2016-1-15 12:00:00",

"from":"medcl@example.co",

"to":["lcdem@example.co"]

}

我们还需要一个测试数据，这里有一个下载各种协议测试数据包的地方，由wireshark站点提供：https://wiki.wireshark.org/SampleCaptures/
Ctrl+F找到SMTP的下载地址：smtp.pcap
用wireshark打开我们刚刚下载的smtp.pcap文件，然后再输入过滤条件：tcp.port == 25，只看25端口的数据，如下图：

上图可以看到25端口的跑的数据有很多，不过我们只关心我们需要的那几个命令就好了。

打开/~/go/src/github.com/elastic/beats/packetbeat/protos/smtp/smtp.go
定义smtpPrivateData，里面的Data是一个数组，分别是TCP两个方向的数据，SmtpMessage是解析出来的邮件信息

type smtpPrivateData struct{

	Data [2]*SmtpStream

}



type SmtpStream struct {

	tcptuple *common.TcpTuple



	data byte



	parseOffset int

	isClient    bool

	message *SmtpMessage

}



type SmtpMessage struct {

	Ts   time.Time

	From string

	To string

}

然后参照MySQL协议，定义相应的方法，最终如下：

package smtp



import (

	"github.com/elastic/beats/libbeat/common"

	"github.com/elastic/beats/libbeat/logp"

	"github.com/elastic/beats/libbeat/publisher"

	"github.com/elastic/beats/packetbeat/config"

	"github.com/elastic/beats/packetbeat/protos"

	"github.com/elastic/beats/packetbeat/protos/tcp"

	"bytes"

	"time"

	"strings"

)



type smtpPrivateData struct{

	Data [2]*SmtpStream

}



type SmtpStream struct {

	tcptuple *common.TcpTuple



	data byte



	parseOffset int

	isClient    bool



	message *SmtpMessage

}



type SmtpMessage struct {

	start int

	end   int



	Ts   time.Time

	From string

	To string

	IgnoreMessage bool

}



type Smtp struct {

	SendRequest         bool

	SendResponse        bool

	transactionTimeout time.Duration

	Ports         int

	results publisher.Client

}



func (smtp *Smtp) initDefaults() {

	smtp.SendRequest = false

	smtp.SendResponse = false

	smtp.transactionTimeout = protos.DefaultTransactionExpiration

}



func (smtp *Smtp) setFromConfig(config config.Smtp) error {

	smtp.Ports = config.Ports

	if config.SendRequest != nil {

		smtp.SendRequest = *config.SendRequest

	}

	if config.SendResponse != nil {

		smtp.SendResponse = *config.SendResponse

	}



	if config.TransactionTimeout != nil && *config.TransactionTimeout > 0 {

		smtp.transactionTimeout = time.Duration(*config.TransactionTimeout) * time.Second

	}



	return nil

}



func (smtp *Smtp) GetPorts() int {

	return smtp.Ports

}



func (smtp *Smtp) Init(test_mode bool, results publisher.Client) error {

	smtp.initDefaults()



	if !test_mode {

		err := smtp.setFromConfig(config.ConfigSingleton.Protocols.Smtp)

		if err != nil {

			return err

		}

	}

	smtp.results = results



	return nil

}



func readLine(data byte, offset int) (bool, string, int) {

	q := bytes.Index(data[offset:], byte("\r\n"))

	if q == -1 {

		return false, "", 0

	}

	return true, string(data[offset : offset+q]), offset + q + 2

}



func (smtp *Smtp) Parse(pkt *protos.Packet, tcptuple *common.TcpTuple, dir uint8, private protos.ProtocolData, ) protos.ProtocolData {



	defer logp.Recover("ParseSmtp exception")



	priv := smtpPrivateData{}

	if private != nil {

		var ok bool

		priv, ok = private.(smtpPrivateData)

		if !ok {

			priv = smtpPrivateData{}

		}

	}



	if priv.Data[dir] == nil {

		priv.Data[dir] = &SmtpStream{

			tcptuple: tcptuple,

			data:     pkt.Payload,

			message:  &SmtpMessage{Ts: pkt.Ts},

		}

	} else {

		// concatenate bytes

		priv.Data[dir].data = append(priv.Data[dir].data, pkt.Payload...)

		if len(priv.Data[dir].data) > tcp.TCP_MAX_DATA_IN_STREAM {

			logp.Debug("smtp", "Stream data too large, dropping TCP stream")

			priv.Data[dir] = nil

			return priv

		}

	}



	stream := priv.Data[dir]

	for len(stream.data) > 0 {

		if stream.message == nil {

			stream.message = &SmtpMessage{Ts: pkt.Ts}

		}



		ok, complete := stmpMessageParser(priv.Data[dir])

		if !ok {

			// drop this tcp stream. Will retry parsing with the next

			// segment in it

			priv.Data[dir] = nil

			logp.Debug("smtp", "Ignore SMTP message. Drop tcp stream. Try parsing with the next segment")

			return priv

		}



		if complete {

			smtp.messageComplete(tcptuple, dir, stream)

		} else {

			logp.Debug("smtp","still wait message...")

			// wait for more data

			break

		}

	}



	return priv

}



func (smtp *Smtp) ConnectionTimeout() time.Duration {

	return smtp.transactionTimeout

}



func stmpMessageParser(s *SmtpStream) (bool, bool) {



	var value string=""



	for s.parseOffset < len(s.data) {





		logp.Debug("smtp", "Parse message: %s", string(s.data[s.parseOffset]))





		if strings.HasPrefix(string(s.data[s.parseOffset]),"MAIL" ) {



			logp.Debug("smtp", "Hit MAIL command: %s", string(s.data[s.parseOffset]))



			found, line, off := readLine(s.data, s.parseOffset)

			if !found {

				return true, false

			}



			value = line[1:]

			logp.Debug("smtp", "value  %s", value)



			s.parseOffset = off

		} else {

			logp.Debug("smtp", "Unexpected message starting with %s", s.data[s.parseOffset:])

			return false, false

		}

	}



	return true, false

}



func handleSmtp(stmp *Smtp, m *SmtpMessage, tcptuple *common.TcpTuple,

dir uint8, raw_msg byte) {

	logp.Info("smtp","handle smtp message...")



	//TODO



}



// Called when the parser has identified a full message.

func (smtp *Smtp) messageComplete(tcptuple *common.TcpTuple, dir uint8, stream *SmtpStream) {



	logp.Info("smtp","message completed...")



	// all ok, ship it

	msg := stream.data[stream.message.start:stream.message.end]



	if !stream.message.IgnoreMessage {

		handleSmtp(smtp, stream.message, tcptuple, dir, msg)

	}



	// and reset message

	stream.PrepareForNewMessage()

}



func (stream *SmtpStream) PrepareForNewMessage() {

	logp.Info("smtp","prepare for new message...")



	stream.data = stream.data[stream.parseOffset:]

	stream.parseOffset = 0

	stream.isClient = false

	stream.message = nil

}







func (smtp *Smtp) GapInStream(tcptuple *common.TcpTuple, dir uint8,

nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {



	defer logp.Recover("GapInStream(smtp) exception")



	if private == nil {

		return private, false

	}



	return private, true

}



func (smtp *Smtp) ReceivedFin(tcptuple *common.TcpTuple, dir uint8,

private protos.ProtocolData) protos.ProtocolData {



	logp.Info("smtp","stream closed...")



	// TODO: check if we have data pending and either drop it to free

	// memory or send it up the stack.

	return private

}

现在切换到命令行，编译一下

cd ~/go/src/github.com/elastic/beats/packetbeat

make

编译成功，一个滚烫的packetbeat可执行文件就躺在当前目录下了，运行一下先，参数-I 指定pcap文件（还记得前面下载的那个测试文件吧）

./packetbeat -d "smtp" -c etc/packetbeat.yml -I ~/Downloads/smtp.pcap  -e -N

运行查看控制台输出结果:

➜  packetbeat git:(smtpbeat) ✗ ./packetbeat -d "smtp" -c etc/packetbeat.yml -I ~/Downloads/smtp.pcap  -e -N 

2016/01/15 10:12:19.058535 publish.go:191: INFO Dry run mode. All output types except the file based one are disabled.

2016/01/15 10:12:19.058570 geolite.go:24: INFO GeoIP disabled: No paths were set under output.geoip.paths

2016/01/15 10:12:19.058592 publish.go:262: INFO Publisher name: medcls-MacBook.local

2016/01/15 10:12:19.058724 beat.go:145: INFO Init Beat: packetbeat; Version: 1.0.0

2016/01/15 10:12:19.059758 beat.go:171: INFO packetbeat sucessfully setup. Start running.

2016/01/15 10:12:20.155335 smtp.go:163: DBG  Parse message: 2

2016/01/15 10:12:20.155416 smtp.go:180: DBG  Unexpected message starting with 250-xc90.websitewelcome.com Hello GP [122.162.143.157]

250-SIZE 52428800

250-PIPELINING

250-AUTH PLAIN LOGIN

250-STARTTLS

250 HELP

2016/01/15 10:12:22.310974 smtp.go:163: DBG  Parse message: F

2016/01/15 10:12:22.311025 smtp.go:180: DBG  Unexpected message starting with From: "Gurpartap Singh" 

To: 

Subject: SMTP

Date: Mon, 5 Oct 2009 11:36:07 +0530

Message-ID: <000301ca4581$ef9e57f0$cedb07d0$@in>

MIME-Version: 1.0

...

成功了，邮件内容都在控制台输出了，但这还不是我们要的最终结果，我需要里面的关键信息，我们继续修改smtp.go这个文件。
留待下回分解。

继续阅读 »

书接上回：http://elasticsearch.cn/article/53

前面介绍了Packetbeat的项目结构，今天终于要开始写代码了，想想还是有点小激动呢。（你快点吧，拖半天了）
网络传输两大协议TCP和UDP，我们的所有协议都不离这两种，HTTP、MySQL走的是TCP传输协议，DNS走的是UDP协议，在Packetbeat里面，实现一个自己的协议非常简单，继承并实现这两者对应的接口就行了，我们看一下长什么样：
打开一个现有的UDP和HTTP协议接口定义：
/~/go/src/github.com/elastic/beats/packetbeat/protos/protos.go

// Functions to be exported by a protocol plugin

type ProtocolPlugin interface {

	// Called to initialize the Plugin

	Init(test_mode bool, results publisher.Client) error

 

	// Called to return the configured ports

	GetPorts() int

}

 

type TcpProtocolPlugin interface {

	ProtocolPlugin

 

	// Called when TCP payload data is available for parsing.

	Parse(pkt *Packet, tcptuple *common.TcpTuple,

		dir uint8, private ProtocolData) ProtocolData

 

	// Called when the FIN flag is seen in the TCP stream.

	ReceivedFin(tcptuple *common.TcpTuple, dir uint8,

		private ProtocolData) ProtocolData

 

	// Called when a packets are missing from the tcp

	// stream.

	GapInStream(tcptuple *common.TcpTuple, dir uint8, nbytes int,

		private ProtocolData) (priv ProtocolData, drop bool)

 

	// ConnectionTimeout returns the per stream connection timeout.

	// Return <=0 to set default tcp module transaction timeout.

	ConnectionTimeout() time.Duration

}

 

type UdpProtocolPlugin interface {

	ProtocolPlugin

 

	// ParseUdp is invoked when UDP payload data is available for parsing.

	ParseUdp(pkt *Packet)

}

TcpProtocolPlugin：TCP协议插件的接口定义，依次是：Parse() 解析Packet，ReceivedFin()处理TCP断开连接，GapInStream()处理空包丢包，ConnectionTimeout()超时时间；
UdpProtocolPlugin: UDP协议的接口定义，UDP协议是不需要握手和保障数据可靠性的，扔出去就结束，速度快，不保证数据可靠送达，所以只有ParseUdp一个方法需要实现，比较简单；
ProtocolPlugin：TCP和UDP都需要实现ProtocolPlugin的基础接口，其实就定义了获取端口和初始化接口。

请问：
Packetbeat怎么工作的？

回答：
每一个协议都有一个固定的端口用于通信，你要做的事情就是定义协议端口，然后按协议是TCP还是UDP来实现对应的接口，Packetbeat将会截获指定端口的数据包（Packet），然后如果交给你定义的方法来进行解析，TCP是Parse，UDP是ParseUdp，都在上面的接口定义好的，然后将解析出来的结构化数据封装成Json，然后扔给Elasticsearch，后续的就的如何对这些数据做一些有趣的分析和应用了。

貌似很简单嘛！

进入每个端口的数据包，我们假设是一个自来水管，拧开80端口，哗啦啦出来的全是HTTP请求的数据包，Packetbeat里面Http协议监听的是80端口啊，所有这些包统统都交给Packetbeat里面的Http协议模块来进行解析，Http协议会一个个的检查这些数据包，也就是每个数据包都会调用一次Parse接口，到这里提到了传过来一个Packet，我们看看它的数据结构长什么样？

type Packet struct {

	Ts      time.Time

	Tuple   common.IpPortTuple

	Payload byte

}

Packet结构简单，
Ts是收到数据包的时间戳；
Tuple是一个来源IP+来源端口和目的IP+目的端口的元组；
Payload就是这个包里面的传输的有用的数据，应用层的字节数据，不包括IP和TCP/UDP头信息，是不是处理起来简单许多。

首选我们确定SMTP协议的配置，每个协议在packetbeat.yml的protocol下面都应该有一个配置节点，如下：

protocols:

  smtp:

    # Configure the ports where to listen for Smtp traffic. You can disable

    # the Smtp protocol by commenting out the list of ports.

    ports: [25]

还需要在对应的config类文件：packetbeat/config/config.go，增加SMTP的结构体，目前只支持一个端口参数，继承基类ProtocolCommon就行，如下：

git diff config/config.go

@@ -42,6 +42,7 @@ type Protocols struct {

        Pgsql    Pgsql

        Redis    Redis

        Thrift   Thrift

+       Smtp     Smtp

 }

 

 type Dns struct {

@@ -118,5 +119,9 @@ type Redis struct {

        Send_response *bool

 }

 

+type Smtp struct {

+	ProtocolCommon        `yaml:",inline"`

+}

+

 // Config Singleton

 var ConfigSingleton Config

在protos文件夹下面，新增smtp目录，并新增空白文件smtp.go，路径：packetbeat/protos/smtp/smtp.go，
这里就是解析SMTP协议的地方，也是我们扩展协议的主要的工作。

...TODO...

修改protos/protos.go，增加SMTP协议枚举，这里记得保证顺序一致，并且protocol名称必须和配置的节点名称一致，如这里都是smtp。

git diff protos/protos.go

@@ -103,6 +103,7 @@ const (

        MongodbProtocol

        DnsProtocol

        MemcacheProtocol

+       SmtpProtocol

 )

 

 // Protocol names

@@ -116,6 +117,7 @@ var ProtocolNames = string{

        "mongodb",

        "dns",

        "memcache",

+       "smtp",

 }

继续修改packetbeat.go主文件，允许SMTP协议并加载。

git diff packetbeat.go

@@ -27,6 +27,7 @@ import (

        "github.com/elastic/packetbeat/protos/tcp"

        "github.com/elastic/packetbeat/protos/thrift"

        "github.com/elastic/packetbeat/protos/udp"

+       "github.com/elastic/packetbeat/protos/smtp"

        "github.com/elastic/packetbeat/sniffer"

 )

 

@@ -43,6 +44,7 @@ var EnabledProtocolPlugins map[protos.Protocol]protos.ProtocolPlugin = map[proto

        protos.ThriftProtocol:   new(thrift.Thrift),

        protos.MongodbProtocol:  new(mongodb.Mongodb),

        protos.DnsProtocol:      new(dns.Dns),

+       protos.SmtpProtocol:      new(smtp.Smtp),

 }

做完上面一系列修改之后，一个空白的SMTP协议的插件的架子就搭好了，并且插件也注册到了Packetbeat里面了，接下来我们再把packetbeat/protos/smtp/smtp.go按照TCPplugin接口的要求实现一下。

说实话TCP处理起来很难，开始之前，我们先明确几个概念，TCP协议是有状态的，并且是流式的，我们关注的是七层应用层的消息，如HTTP里面的一个HTTP请求和返回，但是TCP底层都是一系列数据包，并且不同的请求的数据包是混杂在一起的，也就是说一个数据包里面可能只是一个HTTP请求的一部分也可能包含多条HTTP请求的一部分，所以Parse()里面需要处理跨数据包的状态信息，我们要把这些数据包和具体的七层的应用层的消息关联起来。

现在我们仔细看看Parse()接口的各个参数定义是做什么用的

Parse(pkt *Packet, tcptuple *common.TcpTuple,

		dir uint8, private ProtocolData) ProtocolData

pkt不用说了，是送进来的数据包，前面已经介绍了其数据结构，tcptuple是该数据包所属的TCP数据流所在的唯一标示（一个未关闭的TCP数据量包含若干数据包，直到TCP链接关闭），使用tcptuple.Hashable()获取唯一值；dir参数标示数据包在TCP数据流中的流向，和第一个TCP数据包方向一致是TcpDirectionOriginal，否则是TcpDirectionReverse；private参数可用来在TCP流中存储状态信息，可在运行时转换成具体的强类型，任意修改和传递给下一个Parse方法，简单来说就是进行中间数据的共享。

下面看段MySQL模块里面的例子

 priv := mysqlPrivateData{}

        if private != nil {

                var ok bool

                priv, ok = private.(mysqlPrivateData)

                if !ok {

                        priv = mysqlPrivateData{}

                }

        }

 

        [ ... ]

 

        return priv

上面的代码就是将private强制转换成mysqlPrivateData结构，然后再使用。
我们再继续看后续怎么处理这些包的一个逻辑例子

ok, complete := mysqlMessageParser(priv.Data[dir])

                if !ok {

                        // drop this tcp stream. Will retry parsing with the next

                        // segment in it

                        priv.Data[dir] = nil

                        logp.Debug("mysql", "Ignore MySQL message. Drop tcp stream.")

                        return priv

                }

 

                if complete {

                        mysql.messageComplete(tcptuple, dir, stream)

                } else {

                        // wait for more data

                        break

                }

mysqlMessageParser是一个解析mysql消息的方法，细节我们忽略，我们只需要关心它的返回，ok标示成功或者失败，true则继续处理，false表示数据包不能用，那就直接忽略；第二个参数complete表示判断这一个MySQL消息是否已经完整了，如果完整了，我们就可以扔出去了，否则继续等待剩下的消息内容。

好的，我们看看SMTP协议怎么折腾吧，先看看一个邮件交互的流程图，来自RFC5321：

由上图可见，发送端和邮件服务器通过一系列命令来执行邮件的发送，下面看看一个具体的命令操作流程(来源：简单邮件传输协议)[/url]

S: 220 www.example.com ESMTP Postfix

C: HELO mydomain.com

S: 250 Hello mydomain.com

C: MAIL FROM: 

S: 250 Ok

C: RCPT TO: 

S: 250 Ok

C: DATA

S: 354 End data with .

C: Subject: test message

C: From:""< sender@mydomain.com>

C: To:""< friend@example.com>

C:

C: Hello,

C: This is a test.

C: Goodbye.

C: .

S: 250 Ok: queued as 12345

C: quit

S: 221 Bye

上面的过程可以看到就几个命令就能将邮件发送出去，但是其实SMTP协议比较复杂，还包括身份认证、附件、多媒体编码等等，我们今天精简一下，我们目前只关心谁给谁发了邮件，发送内容先不管，这样相比完整的SMTP协议(RFC5321)，我们只需要关注以下几个命令：
MAIL：开始一份邮件 mail　from: xxx@xx.com
RCPT: 标识单个的邮件接收人；常在mail命令后面可有多个rcpt　to: xx@xx.com
QUIT：结束SMTP会话，不一定发送了邮件，注意
RESET:重置会话，当前传输被取消

最终希望通过Packetbeat将这些数据解析并处理成我们想要的如下JSON数据，即大功告成：

{

"timestamp":"2016-1-15 12:00:00",

"from":"medcl@example.co",

"to":["lcdem@example.co"]

}

我们还需要一个测试数据，这里有一个下载各种协议测试数据包的地方，由wireshark站点提供：https://wiki.wireshark.org/SampleCaptures/
Ctrl+F找到SMTP的下载地址：smtp.pcap
用wireshark打开我们刚刚下载的smtp.pcap文件，然后再输入过滤条件：tcp.port == 25，只看25端口的数据，如下图：

上图可以看到25端口的跑的数据有很多，不过我们只关心我们需要的那几个命令就好了。

打开/~/go/src/github.com/elastic/beats/packetbeat/protos/smtp/smtp.go
定义smtpPrivateData，里面的Data是一个数组，分别是TCP两个方向的数据，SmtpMessage是解析出来的邮件信息

type smtpPrivateData struct{

	Data [2]*SmtpStream

}



type SmtpStream struct {

	tcptuple *common.TcpTuple



	data byte



	parseOffset int

	isClient    bool

	message *SmtpMessage

}



type SmtpMessage struct {

	Ts   time.Time

	From string

	To string

}

然后参照MySQL协议，定义相应的方法，最终如下：

package smtp



import (

	"github.com/elastic/beats/libbeat/common"

	"github.com/elastic/beats/libbeat/logp"

	"github.com/elastic/beats/libbeat/publisher"

	"github.com/elastic/beats/packetbeat/config"

	"github.com/elastic/beats/packetbeat/protos"

	"github.com/elastic/beats/packetbeat/protos/tcp"

	"bytes"

	"time"

	"strings"

)



type smtpPrivateData struct{

	Data [2]*SmtpStream

}



type SmtpStream struct {

	tcptuple *common.TcpTuple



	data byte



	parseOffset int

	isClient    bool



	message *SmtpMessage

}



type SmtpMessage struct {

	start int

	end   int



	Ts   time.Time

	From string

	To string

	IgnoreMessage bool

}



type Smtp struct {

	SendRequest         bool

	SendResponse        bool

	transactionTimeout time.Duration

	Ports         int

	results publisher.Client

}



func (smtp *Smtp) initDefaults() {

	smtp.SendRequest = false

	smtp.SendResponse = false

	smtp.transactionTimeout = protos.DefaultTransactionExpiration

}



func (smtp *Smtp) setFromConfig(config config.Smtp) error {

	smtp.Ports = config.Ports

	if config.SendRequest != nil {

		smtp.SendRequest = *config.SendRequest

	}

	if config.SendResponse != nil {

		smtp.SendResponse = *config.SendResponse

	}



	if config.TransactionTimeout != nil && *config.TransactionTimeout > 0 {

		smtp.transactionTimeout = time.Duration(*config.TransactionTimeout) * time.Second

	}



	return nil

}



func (smtp *Smtp) GetPorts() int {

	return smtp.Ports

}



func (smtp *Smtp) Init(test_mode bool, results publisher.Client) error {

	smtp.initDefaults()



	if !test_mode {

		err := smtp.setFromConfig(config.ConfigSingleton.Protocols.Smtp)

		if err != nil {

			return err

		}

	}

	smtp.results = results



	return nil

}



func readLine(data byte, offset int) (bool, string, int) {

	q := bytes.Index(data[offset:], byte("\r\n"))

	if q == -1 {

		return false, "", 0

	}

	return true, string(data[offset : offset+q]), offset + q + 2

}



func (smtp *Smtp) Parse(pkt *protos.Packet, tcptuple *common.TcpTuple, dir uint8, private protos.ProtocolData, ) protos.ProtocolData {



	defer logp.Recover("ParseSmtp exception")



	priv := smtpPrivateData{}

	if private != nil {

		var ok bool

		priv, ok = private.(smtpPrivateData)

		if !ok {

			priv = smtpPrivateData{}

		}

	}



	if priv.Data[dir] == nil {

		priv.Data[dir] = &SmtpStream{

			tcptuple: tcptuple,

			data:     pkt.Payload,

			message:  &SmtpMessage{Ts: pkt.Ts},

		}

	} else {

		// concatenate bytes

		priv.Data[dir].data = append(priv.Data[dir].data, pkt.Payload...)

		if len(priv.Data[dir].data) > tcp.TCP_MAX_DATA_IN_STREAM {

			logp.Debug("smtp", "Stream data too large, dropping TCP stream")

			priv.Data[dir] = nil

			return priv

		}

	}



	stream := priv.Data[dir]

	for len(stream.data) > 0 {

		if stream.message == nil {

			stream.message = &SmtpMessage{Ts: pkt.Ts}

		}



		ok, complete := stmpMessageParser(priv.Data[dir])

		if !ok {

			// drop this tcp stream. Will retry parsing with the next

			// segment in it

			priv.Data[dir] = nil

			logp.Debug("smtp", "Ignore SMTP message. Drop tcp stream. Try parsing with the next segment")

			return priv

		}



		if complete {

			smtp.messageComplete(tcptuple, dir, stream)

		} else {

			logp.Debug("smtp","still wait message...")

			// wait for more data

			break

		}

	}



	return priv

}



func (smtp *Smtp) ConnectionTimeout() time.Duration {

	return smtp.transactionTimeout

}



func stmpMessageParser(s *SmtpStream) (bool, bool) {



	var value string=""



	for s.parseOffset < len(s.data) {





		logp.Debug("smtp", "Parse message: %s", string(s.data[s.parseOffset]))





		if strings.HasPrefix(string(s.data[s.parseOffset]),"MAIL" ) {



			logp.Debug("smtp", "Hit MAIL command: %s", string(s.data[s.parseOffset]))



			found, line, off := readLine(s.data, s.parseOffset)

			if !found {

				return true, false

			}



			value = line[1:]

			logp.Debug("smtp", "value  %s", value)



			s.parseOffset = off

		} else {

			logp.Debug("smtp", "Unexpected message starting with %s", s.data[s.parseOffset:])

			return false, false

		}

	}



	return true, false

}



func handleSmtp(stmp *Smtp, m *SmtpMessage, tcptuple *common.TcpTuple,

dir uint8, raw_msg byte) {

	logp.Info("smtp","handle smtp message...")



	//TODO



}



// Called when the parser has identified a full message.

func (smtp *Smtp) messageComplete(tcptuple *common.TcpTuple, dir uint8, stream *SmtpStream) {



	logp.Info("smtp","message completed...")



	// all ok, ship it

	msg := stream.data[stream.message.start:stream.message.end]



	if !stream.message.IgnoreMessage {

		handleSmtp(smtp, stream.message, tcptuple, dir, msg)

	}



	// and reset message

	stream.PrepareForNewMessage()

}



func (stream *SmtpStream) PrepareForNewMessage() {

	logp.Info("smtp","prepare for new message...")



	stream.data = stream.data[stream.parseOffset:]

	stream.parseOffset = 0

	stream.isClient = false

	stream.message = nil

}







func (smtp *Smtp) GapInStream(tcptuple *common.TcpTuple, dir uint8,

nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {



	defer logp.Recover("GapInStream(smtp) exception")



	if private == nil {

		return private, false

	}



	return private, true

}



func (smtp *Smtp) ReceivedFin(tcptuple *common.TcpTuple, dir uint8,

private protos.ProtocolData) protos.ProtocolData {



	logp.Info("smtp","stream closed...")



	// TODO: check if we have data pending and either drop it to free

	// memory or send it up the stack.

	return private

}

现在切换到命令行，编译一下

cd ~/go/src/github.com/elastic/beats/packetbeat

make

编译成功，一个滚烫的packetbeat可执行文件就躺在当前目录下了，运行一下先，参数-I 指定pcap文件（还记得前面下载的那个测试文件吧）

./packetbeat -d "smtp" -c etc/packetbeat.yml -I ~/Downloads/smtp.pcap  -e -N

运行查看控制台输出结果:

➜  packetbeat git:(smtpbeat) ✗ ./packetbeat -d "smtp" -c etc/packetbeat.yml -I ~/Downloads/smtp.pcap  -e -N 

2016/01/15 10:12:19.058535 publish.go:191: INFO Dry run mode. All output types except the file based one are disabled.

2016/01/15 10:12:19.058570 geolite.go:24: INFO GeoIP disabled: No paths were set under output.geoip.paths

2016/01/15 10:12:19.058592 publish.go:262: INFO Publisher name: medcls-MacBook.local

2016/01/15 10:12:19.058724 beat.go:145: INFO Init Beat: packetbeat; Version: 1.0.0

2016/01/15 10:12:19.059758 beat.go:171: INFO packetbeat sucessfully setup. Start running.

2016/01/15 10:12:20.155335 smtp.go:163: DBG  Parse message: 2

2016/01/15 10:12:20.155416 smtp.go:180: DBG  Unexpected message starting with 250-xc90.websitewelcome.com Hello GP [122.162.143.157]

250-SIZE 52428800

250-PIPELINING

250-AUTH PLAIN LOGIN

250-STARTTLS

250 HELP

2016/01/15 10:12:22.310974 smtp.go:163: DBG  Parse message: F

2016/01/15 10:12:22.311025 smtp.go:180: DBG  Unexpected message starting with From: "Gurpartap Singh" 

To: 

Subject: SMTP

Date: Mon, 5 Oct 2009 11:36:07 +0530

Message-ID: <000301ca4581$ef9e57f0$cedb07d0$@in>

MIME-Version: 1.0

...

成功了，邮件内容都在控制台输出了，但这还不是我们要的最终结果，我需要里面的关键信息，我们继续修改smtp.go这个文件。
留待下回分解。收起阅读 »

Packetbeat协议扩展开发教程（2）

书接上回:http://elasticsearch.cn/article/48

我们打开Packetbeat项目，看看里面长什么样：

现在beats项目都合并在一起了，第一级可以看到各个子项目：
/libbeat: 公共依赖；
/filebeat: 替代Logstash-forwarder，处理日志类型数据；
/packetbeat: 本文扩展重点，网络抓包；
/topbeat: 监控系统性能；
/winlogbeat: 监控windows下面的日志信息；
/vender: 依赖的第三方库；
/tests: 用于测试的pcamp抓包文件，非常有用；
/scripts: 一些用于开发和测试的Docker脚本文件；

现在重点看看/packetbeat下面目录都有些什么：
/packetbeat/main.go：启动入口，里面没有什么逻辑；
/packetbeat/beat/：里面就一个packetbeat.go文件，packetbeat主程序，处理配置和命令行参数，协议需要在这里进行注册；
/packetbeat/config/: 里面就一个config.go文件，定义了所有的配置相关的struct结构体，新协议需要在这里定义其配置的结构体；
/packetbeat/debian/: debian打包相关；
/packetbeat/decoder/: 解码类，网络传输层包的解码；
/packetbeat/docs/: 项目的相关文档；
/packetbeat/etc/: 示例配置文件；
/packetbeat/procs/: 获取系统内核运作状态与进程信息的工具类；
/packetbeat/protos/：自定义协议类，每个目录对应一个应用协议，我们需要在此新增我们的协议，如SMTP；
/packetbeat/sniffer/: 三种不同抓包方式的实现：pcap、af_packet、pf_ring，关于这三者的区别，请参照文档：Traffic Capturing Options；
/packetbeat/tests/：测试相关的文件，里面有每一个协议的pcab抓包样板，还有一堆Python测试脚本；

知道项目的大概架构就知道从哪下手了，下节分解。

继续阅读 »

一个把数据从MySQL同步到Elasticsearch的工具

https://github.com/zhongbiaode ... -sync
这个工具用python实现，主要使用了mysqldump输出xml进行初次同步，以及binlog进行增量同步，欢迎试用以及提出修改意见。
最近刚刚更新了中文文档。

继续阅读 »

社区福利：Elastic-playground

为大家准备了一个测试Elasticsearch/Kibana功能的地方 (Found实例)：
Kibana:
https://6e0ccaba29cd55a7f07f83 ... ibana

用户名/密码：elasticsearch-cn

集群名："e064eb"，使用Java客户端的时候需要，如何连接，参考：http://elasticsearch.cn/article/46

HTTP http://e064eb4b0aa993db28ad513 ... :9200

HTTPS https://e064eb4b0aa993db28ad51 ... :9243

 curl -u elasticsearch-cn:elasticsearch-cn http://e064eb4b0aa993db28ad513 ... 9200/

{

  "name" : "instance-0000000009",

  "cluster_name" : "e064eb4b0aa993db28ad513e4d2df5e3",

  "version" : {

    "number" : "2.1.1",

    "build_hash" : "40e2c53a6b6c2972b3d13846e450e66f4375bd71",

    "build_timestamp" : "2015-12-15T13:05:55Z",

    "build_snapshot" : false,

    "lucene_version" : "5.3.1"

  },

  "tagline" : "You Know, for Search"

}

内置常用插件，有其他插件要安装的请留言。

继续阅读 »

为大家准备了一个测试Elasticsearch/Kibana功能的地方 (Found实例)：
Kibana:
https://6e0ccaba29cd55a7f07f83 ... ibana

用户名/密码：elasticsearch-cn

集群名："e064eb"，使用Java客户端的时候需要，如何连接，参考：http://elasticsearch.cn/article/46

HTTP http://e064eb4b0aa993db28ad513 ... :9200

HTTPS https://e064eb4b0aa993db28ad51 ... :9243

 curl -u elasticsearch-cn:elasticsearch-cn http://e064eb4b0aa993db28ad513 ... 9200/

{

  "name" : "instance-0000000009",

  "cluster_name" : "e064eb4b0aa993db28ad513e4d2df5e3",

  "version" : {

    "number" : "2.1.1",

    "build_hash" : "40e2c53a6b6c2972b3d13846e450e66f4375bd71",

    "build_timestamp" : "2015-12-15T13:05:55Z",

    "build_snapshot" : false,

    "lucene_version" : "5.3.1"

  },

  "tagline" : "You Know, for Search"

}

内置常用插件，有其他插件要安装的请留言。收起阅读 »

elasticsearch-analysis-ik和elasticsearch-analysis-mmseg更新至1.7.0

elasticsearch-analysis-ik：
https://github.com/medcl/elasticsearch-analysis-ik

elasticsearch-analysis-mmseg：
https://github.com/medcl/elast ... -mseg

主要更新配置文件存放路径，之前版本的配置文件存放在elasticsearch的config目录，现在都修改为插件的相对目录了，主要是简化部署，现在可在Found（https://found.elastic.co）部署了。

继续阅读 »

通过elasticsearch-mapper attachment插件实现文件建立索引

1.安装elasticsearch-mapper attachment

bin/plugin install elasticsearch/elasticsearch-mapper-attachments/3.1.1
2.按照插件官方文档来测试
3.插件需要手动把文档内容转化为base64编码然后建立索引，代码如下
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.tika.Tika;
import org.apache.tika.config.TikaConfig;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.Parser;
import org.apache.tika.parser.pdf.PDFParser;
import org.apache.tika.sax.BodyContentHandler;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.xml.sax.ContentHandler;

import com.spatial4j.core.io.ParseUtils;

import static org.elasticsearch.common.xcontent.XContentFactory.*;
public class sysfiles {
   public static void main(String[] args) throws Exception{
       sys();
}

   private static void sys() throws IOException {
       // TODO Auto-generated method stub
       String idxName = "test";
       String idxType = "attachments";
       Settings settings =ImmutableSettings.settingsBuilder().put("cluster.name","az_bsms_elasticsearch").build();
       Client client=new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress("127.0.0.1", 9300));
       String data64=org.elasticsearch.common.Base64.encodeFromFile(filepath);
       XContentBuilder source = jsonBuilder().startObject()
   .field("file", data64)
       .field("text", data64)
       .endObject();

       String id = "file"+11;
       IndexResponse idxResp = client.prepareIndex().setIndex(idxName).setType(idxType).setId(id)
       .setSource(source).setRefresh(true).execute().actionGet();
       System.out.println(idxResp);
       client.close();
   }
4.按官方文档正常的搜索就可以了

继续阅读 »

1.安装elasticsearch-mapper attachment

bin/plugin install elasticsearch/elasticsearch-mapper-attachments/3.1.1
2.按照插件官方文档来测试
3.插件需要手动把文档内容转化为base64编码然后建立索引，代码如下
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.tika.Tika;
import org.apache.tika.config.TikaConfig;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.Parser;
import org.apache.tika.parser.pdf.PDFParser;
import org.apache.tika.sax.BodyContentHandler;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.xml.sax.ContentHandler;

import com.spatial4j.core.io.ParseUtils;

import static org.elasticsearch.common.xcontent.XContentFactory.*;
public class sysfiles {
   public static void main(String[] args) throws Exception{
       sys();
}

   private static void sys() throws IOException {
       // TODO Auto-generated method stub
       String idxName = "test";
       String idxType = "attachments";
       Settings settings =ImmutableSettings.settingsBuilder().put("cluster.name","az_bsms_elasticsearch").build();
       Client client=new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress("127.0.0.1", 9300));
       String data64=org.elasticsearch.common.Base64.encodeFromFile(filepath);
       XContentBuilder source = jsonBuilder().startObject()
   .field("file", data64)
       .field("text", data64)
       .endObject();

       String id = "file"+11;
       IndexResponse idxResp = client.prepareIndex().setIndex(idxName).setType(idxType).setId(id)
       .setSource(source).setRefresh(true).execute().actionGet();
       System.out.println(idxResp);
       client.close();
   }
4.按官方文档正常的搜索就可以了收起阅读 »

Packetbeat协议扩展开发教程（1）

Packetbeat（https://www.elastic.co/products/beats/packetbeat）
是一个开源的网络抓包与分析框架，内置了很多常见的协议解析，如HTPP、MySQL、Thrift等。但是网络协议有很多，如何扩展一个自己的协议呢，本文将为您介绍如何在Packetbeat基础上扩展实现您自己的协议。

开发环境：
1.Go语言
Packetbeat是由Go语言编写，具有高性能和易部署的特点，有关Go语言的更多信息请访问：https://golang.org/。
2.Git
源码管理，相信大家都比较熟悉了。
3.Tcpdump
*nix下的抓包分析，可选，用于调试。
4.Mac本一台
Windows太伤，不建议。
5.IDE
推荐idea，其它只要你顺手都行。

这个教程给大家介绍的是编写一个SMTP协议的扩展，SMTP就是我们发邮件使用的协议，加密的比较麻烦，为了方便，本教程使用不加密的名文传输的SMTP协议，默认对应端口是25。

A.源码签出
登陆Github打开https://github.com/elastic/beats

fork后得到你自己的仓库，比如我的：https://github.com/medcl/packetbeat

#创建相应目录

mkdir -p $GOPATH/src/github.com/elastic/ 

cd $GOPATH/src/github.com/elastic



#签出源码

git clone https://github.com/elastic/beats.git

cd beats



#修改官方仓库为upstream源，设置自己的仓库为origin源

git remote rename origin upstream

git remote add origin git@github.com:medcl/packetbeat.git



#获取上游最新的代码，如果是刚fork的话可不用管

git pull upstream master



#签出一个名为smtpbeat的分支，用于开发这个功能

git checkout -b smtpbeat



#切换到packetbeat模块

cd packetbeat



#获取依赖信息

(mkdir -p $GOPATH/src/golang.org/x/&&cd $GOPATH/src/golang.org/x &&git clone https://github.com/golang/tools.git )

go get github.com/tools/godep



#编译

make

编译出来的文件：packetbeat就在根目录
现在我们测试一下
修改etc/packetbeat.yml,在output下面的elasticsearch下面添加enabled: true，默认是不启用的，另外如果你的Elasticsearch安装了Shield，比如我的Elasticsearch的用户名和密码都是tribe_user，哦，忘了说了，我们的Elasticsearch跑在本机。
packetbeat.yml的详细配置可参见：https://www.elastic.co/guide/e ... .html

output:

  elasticsearch:

    enabled: true

    hosts: ["localhost:9200"]

    username: "tribe_user"

    password: "tribe_user"

现在可以运行命令启动packetbeat了，默认会监听所有内置的协议，如HTTP、DNS等。

./packetbeat -e -c etc/packetbeat.yml  -d "publish"

介绍一下常用的参数：
-N dry run模式，不实际output存储日志
-e 控制台输出调试日志
-d 仅显示对应logger的日志

好的，我们打开几个网页，控制台会有相应的输出，如下：

2015/12/29 14:24:39.965037 preprocess.go:37: DBG  Start Preprocessing

2015/12/29 14:24:39.965366 publish.go:98: DBG  Publish: {

  "@timestamp": "2015-12-29T14:24:39.709Z",

  "beat": {

    "hostname": "medcls-MacBook.local",

    "name": "medcls-MacBook.local"

  },

  "bytes_in": 31,

  "bytes_out": 115,

  "client_ip": "192.168.3.10",

  "client_port": 53669,

  "client_proc": "",

  "client_server": "",

  "count": 1,

  "direction": "out",

  "dns": {

    "additionals_count": 0,

    "answers": [

      {

        "class": "IN",

        "data": "www.a.shifen.com",

        "name": "sp2.baidu.com",

        "ttl": 333,

        "type": "CNAME"

      }

    ],

    "answers_count": 1,

    "authorities": [

      {

        "class": "IN",

        "data": "ns1.a.shifen.com",

        "expire": 86400,

        "minimum": 3600,

        "name": "a.shifen.com",

        "refresh": 5,

        "retry": 5,

        "rname": "baidu_dns_master.baidu.com",

        "serial": 1512240003,

        "ttl": 12,

        "type": "SOA"

      }

    ],

    "authorities_count": 1,

    "flags": {

      "authoritative": false,

      "recursion_allowed": true,

      "recursion_desired": true,

      "truncated_response": false

    },

    "id": 7435,

    "op_code": "QUERY",

    "question": {

      "class": "IN",

      "name": "sp2.baidu.com",

      "type": "AAAA"

    },

    "response_code": "NOERROR"

  },

  "ip": "192.168.3.1",

  "method": "QUERY",

  "port": 53,

  "proc": "",

  "query": "class IN, type AAAA, sp2.baidu.com",

  "resource": "sp2.baidu.com",

  "responsetime": 18,

  "server": "",

  "status": "OK",

  "transport": "udp",

  "type": "dns"

}

2015/12/29 14:24:39.965774 preprocess.go:94: DBG  Forward preprocessed events

2015/12/29 14:24:39.965796 async.go:42: DBG  async forward to outputers (1)

2015/12/29 14:24:40.099973 output.go:103: DBG  output worker: publish 2 events

然后Elasticsearch应该就会有数据进去了，我们看看：

curl http://localhost:9200/_cat/indices\?pretty\=true -u tribe_user:tribe_user

yellow open packetbeat-2015.12.29  5 1   135  0 561.2kb 561.2kb

至此，packetbeat源码的build成功，我们整个开发流程已经跑通了，下一节正式开始介绍SMTP协议的扩展。

继续阅读 »

Packetbeat（https://www.elastic.co/products/beats/packetbeat）
是一个开源的网络抓包与分析框架，内置了很多常见的协议解析，如HTPP、MySQL、Thrift等。但是网络协议有很多，如何扩展一个自己的协议呢，本文将为您介绍如何在Packetbeat基础上扩展实现您自己的协议。

开发环境：
1.Go语言
Packetbeat是由Go语言编写，具有高性能和易部署的特点，有关Go语言的更多信息请访问：https://golang.org/。
2.Git
源码管理，相信大家都比较熟悉了。
3.Tcpdump
*nix下的抓包分析，可选，用于调试。
4.Mac本一台
Windows太伤，不建议。
5.IDE
推荐idea，其它只要你顺手都行。

这个教程给大家介绍的是编写一个SMTP协议的扩展，SMTP就是我们发邮件使用的协议，加密的比较麻烦，为了方便，本教程使用不加密的名文传输的SMTP协议，默认对应端口是25。

A.源码签出
登陆Github打开https://github.com/elastic/beats

fork后得到你自己的仓库，比如我的：https://github.com/medcl/packetbeat

#创建相应目录

mkdir -p $GOPATH/src/github.com/elastic/ 

cd $GOPATH/src/github.com/elastic



#签出源码

git clone https://github.com/elastic/beats.git

cd beats



#修改官方仓库为upstream源，设置自己的仓库为origin源

git remote rename origin upstream

git remote add origin git@github.com:medcl/packetbeat.git



#获取上游最新的代码，如果是刚fork的话可不用管

git pull upstream master



#签出一个名为smtpbeat的分支，用于开发这个功能

git checkout -b smtpbeat



#切换到packetbeat模块

cd packetbeat



#获取依赖信息

(mkdir -p $GOPATH/src/golang.org/x/&&cd $GOPATH/src/golang.org/x &&git clone https://github.com/golang/tools.git )

go get github.com/tools/godep



#编译

make

编译出来的文件：packetbeat就在根目录
现在我们测试一下
修改etc/packetbeat.yml,在output下面的elasticsearch下面添加enabled: true，默认是不启用的，另外如果你的Elasticsearch安装了Shield，比如我的Elasticsearch的用户名和密码都是tribe_user，哦，忘了说了，我们的Elasticsearch跑在本机。
packetbeat.yml的详细配置可参见：https://www.elastic.co/guide/e ... .html

output:

  elasticsearch:

    enabled: true

    hosts: ["localhost:9200"]

    username: "tribe_user"

    password: "tribe_user"

现在可以运行命令启动packetbeat了，默认会监听所有内置的协议，如HTTP、DNS等。

./packetbeat -e -c etc/packetbeat.yml  -d "publish"

介绍一下常用的参数：
-N dry run模式，不实际output存储日志
-e 控制台输出调试日志
-d 仅显示对应logger的日志

好的，我们打开几个网页，控制台会有相应的输出，如下：

2015/12/29 14:24:39.965037 preprocess.go:37: DBG  Start Preprocessing

2015/12/29 14:24:39.965366 publish.go:98: DBG  Publish: {

  "@timestamp": "2015-12-29T14:24:39.709Z",

  "beat": {

    "hostname": "medcls-MacBook.local",

    "name": "medcls-MacBook.local"

  },

  "bytes_in": 31,

  "bytes_out": 115,

  "client_ip": "192.168.3.10",

  "client_port": 53669,

  "client_proc": "",

  "client_server": "",

  "count": 1,

  "direction": "out",

  "dns": {

    "additionals_count": 0,

    "answers": [

      {

        "class": "IN",

        "data": "www.a.shifen.com",

        "name": "sp2.baidu.com",

        "ttl": 333,

        "type": "CNAME"

      }

    ],

    "answers_count": 1,

    "authorities": [

      {

        "class": "IN",

        "data": "ns1.a.shifen.com",

        "expire": 86400,

        "minimum": 3600,

        "name": "a.shifen.com",

        "refresh": 5,

        "retry": 5,

        "rname": "baidu_dns_master.baidu.com",

        "serial": 1512240003,

        "ttl": 12,

        "type": "SOA"

      }

    ],

    "authorities_count": 1,

    "flags": {

      "authoritative": false,

      "recursion_allowed": true,

      "recursion_desired": true,

      "truncated_response": false

    },

    "id": 7435,

    "op_code": "QUERY",

    "question": {

      "class": "IN",

      "name": "sp2.baidu.com",

      "type": "AAAA"

    },

    "response_code": "NOERROR"

  },

  "ip": "192.168.3.1",

  "method": "QUERY",

  "port": 53,

  "proc": "",

  "query": "class IN, type AAAA, sp2.baidu.com",

  "resource": "sp2.baidu.com",

  "responsetime": 18,

  "server": "",

  "status": "OK",

  "transport": "udp",

  "type": "dns"

}

2015/12/29 14:24:39.965774 preprocess.go:94: DBG  Forward preprocessed events

2015/12/29 14:24:39.965796 async.go:42: DBG  async forward to outputers (1)

2015/12/29 14:24:40.099973 output.go:103: DBG  output worker: publish 2 events

然后Elasticsearch应该就会有数据进去了，我们看看：

curl http://localhost:9200/_cat/indices\?pretty\=true -u tribe_user:tribe_user

yellow open packetbeat-2015.12.29  5 1   135  0 561.2kb 561.2kb

至此，packetbeat源码的build成功，我们整个开发流程已经跑通了，下一节正式开始介绍SMTP协议的扩展。收起阅读 »

大家聊一聊使用的什么版本的Elasticsearch，看看Elasticsearch版本变化

Elastic线下交流活动走起来！

es索引模版配置不当导致的aggs聚合查询字段显示错误的问题

Elastic{ON}16的资料已经放出

使用 SQL 查询 Elasticsearch

ElasticSearch2.1.1安装及简单配置说明

推荐一个ES做机器学习的blog，我觉得非常的酷

nest驱动IndexName问题

Packetbeat协议扩展开发教程（3）

Packetbeat协议扩展开发教程（2）

一个把数据从MySQL同步到Elasticsearch的工具

社区福利：Elastic-playground

elasticsearch-analysis-ik和elasticsearch-analysis-mmseg更新至1.7.0

通过elasticsearch-mapper attachment插件实现文件建立索引

Packetbeat协议扩展开发教程（1）

活动推荐

热门文章

热门话题