Protocol Modules

To configure your plugin, you need to add a configuration struct to config/config.go Protocols struct. This struct will be filled by goyaml on startup.

type Protocols struct {
	Dns      Dns
	Http     Http
	Memcache Memcache
	Mysql    Mysql
	Mongodb  Mongodb
	Pgsql    Pgsql
	Redis    Redis
	Thrift   Thrift
}

Next create an ID for the new plugin in protos/protos.go:

// Protocol constants.
const (
	UnknownProtocol Protocol = iota
	HttpProtocol
	MysqlProtocol
	RedisProtocol
	PgsqlProtocol
	ThriftProtocol
	MongodbProtocol
	DnsProtocol
	MemcacheProtocol
)

// Protocol names
var ProtocolNames = []string{
	"unknown",
	"http",
	"mysql",
	"redis",
	"pgsql",
	"thrift",
	"mongodb",
	"dns",
	"memcache",
}

The protocol names must be in the same order as their corresponding protocol IDs. Additionally the protocol name must match the configuration name.

Finally register your new protocol plugin in packetbeat.go EnabledProtocolPlugins:

var EnabledProtocolPlugins map[protos.Protocol]protos.ProtocolPlugin = map[protos.Protocol]protos.ProtocolPlugin{
	protos.HttpProtocol:     new(http.Http),
	protos.MemcacheProtocol: new(memcache.Memcache),
	protos.MysqlProtocol:    new(mysql.Mysql),
	protos.PgsqlProtocol:    new(pgsql.Pgsql),
	protos.RedisProtocol:    new(redis.Redis),
	protos.ThriftProtocol:   new(thrift.Thrift),
	protos.MongodbProtocol:  new(mongodb.Mongodb),
	protos.DnsProtocol:      new(dns.Dns),
}

Once the module is registered, it can be configured, and packets will be processed.

Before implementing all the logic for your new protocol module, it can be helpful to first register the module and implement the minimal plugin interface for printing a debug message on received packets. This way you can test the plugin registration to ensure that it’s working correctly.

For TCP protocols, the Parse() function is the heart of the module. As mentioned earlier, this function is called for every TCP packet that contains data on the configured ports.

It is important to understand that because TCP is a stream-based protocol, the packet boundaries don’t necessarily match the application layer message boundaries. For example, a packet can contain only a part of the message, it can contain a complete message, or it can contain multiple messages.

If you see a packet in the middle of the stream, you have no guaranties that it’s first byte is the beginning of a message. However, if the packet is the first seen in a given TCP stream, then you can assume it is the beginning of the message.

The Parse() function needs to deal with these facts, which generally means that it needs to keep state across multiple packets.

Let’s have a look again at its signature:

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

We’ve already talked about the first parameter, which contains the packet data. The rest of the parameters and the return value are used for maintaining state inside the TCP stream.

The tcptuple is a unique identifier for the TCP stream that the packet is part of. You can use the tcptuple.Hashable() function to get a value that you can store in a map. The dir flag gives you the direction in which the packet is flowing inside the TCP stream. The two possible values are TcpDirectionOriginal if the packet goes in the same direction as the first packet from the stream and TcpDirectionReverse if the packet goes in the other direction.

The private parameter can be used by the module to store state in the TCP stream. The module would typically cast this at runtime to a type of its choice, modify it as needed, and then return the modified value. The next time the TCP layer calls Parse() or another function from the TcpProtocolPlugin interface, it will call the function with the modified private value.

Here is an example of how the MySQL module handles the private data:

	priv := mysqlPrivateData{}
	if private != nil {
		var ok bool
		priv, ok = private.(mysqlPrivateData)
		if !ok {
			priv = mysqlPrivateData{}
		}
	}

	[ ... ]

	return priv

Most modules then use a logic similar to the following to deal with incomplete data (this example is also from MySQL):

		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
		}

The mysqlMessageParser() is the function that tries to parse a single MySQL message. Its implementation is MySQL-specific, so it’s not interesting to us for this guide. It returns two values: ok, which is false if there was a parsing error from which we cannot recover, and complete, which indicates whether a complete and valid message was separated from the stream. These two values are used for deciding what to do next. In case of errors, we drop the stream. If there are no errors, but the message is not yet complete, we do nothing and wait for more data. Finally, if the message is complete, we go to the next level.

This block of code is called in a loop so that it can separate multiple messages found in the same packet.

If the protocol you are working on is running on top of UDP, then all the complexities that TCP parser/decoders need to deal with around extracting messages from packets are no longer relevant.

For an example, see the ParseUdp() function from the DNS module.

Most protocols that Packetbeat supports today are request-response oriented. Packetbeat indexes into Elasticsearch a document for each request-response pair (called a transaction). This way we can have data from the request and the response in the same document and measure the response time.

But this can be different for your protocol. For example for an asynchronous protocol like AMPQ, it makes more sense to index a document for every message, and then no correlation is necessary. On the other hand, for a session-based protocol like SIP, it might make sense to index a document for a SIP transaction or for a full SIP dialog, which can have more than two messages.

The TCP stream or UDP ports are usually good indicators that two messages belong to the same transactions. Therefore most protocol implementations in Packetbeat use a map with tcptuple maps for correlating the requests with the responses. One thing you should be careful about is to expire and remove from this map incomplete transactions. For example, we might see the request that has created an entry in the map, but if we never see the reply, we need to remove the request from memory on a timer, otherwise we risk leaking memory.

After the correlation step, you should have an JSON-like object that can be sent to Elasticsearch for indexing. The way you do that is by publishing it through the results publisher client, which is received by the Init function. The publisher client accepts structures of type common.MapStr, which is essentially a map[string][interface{} with a few more convenience methods added.

As an example, here is the relevant code from the REDIS module:

	event := common.MapStr{}
	event["type"] = "redis"
	if !t.IsError {
		event["status"] = common.OK_STATUS
	} else {
		event["status"] = common.ERROR_STATUS
	}
	event["responsetime"] = t.ResponseTime
	if redis.Send_request {
		event["request"] = t.Request_raw
	}
	if redis.Send_response {
		event["response"] = t.Response_raw
	}
	event["redis"] = common.MapStr(t.Redis)
	event["method"] = strings.ToUpper(t.Method)
	event["resource"] = t.Path
	event["query"] = t.Query
	event["bytes_in"] = uint64(t.BytesIn)
	event["bytes_out"] = uint64(t.BytesOut)

	event["@timestamp"] = common.Time(t.ts)
	event["src"] = &t.Src
	event["dst"] = &t.Dst

	redis.results.PublishEvent(event)

The following fields are required and their presence will be checked by system tests: