func FindPidsByCmdlineGrep(prefix string, process string) ([]int, error) {
	defer logp.Recover("FindPidsByCmdlineGrep exception")
	pids := []int{}

	proc, err := os.Open(filepath.Join(prefix, "/proc"))
	if err != nil {
		return pids, fmt.Errorf("Open /proc: %s", err)
	}
	defer proc.Close()

	names, err := proc.Readdirnames(0)
	if err != nil {
		return pids, fmt.Errorf("Readdirnames: %s", err)
	}

	for _, name := range names {
		pid, err := strconv.Atoi(name)
		if err != nil {
			continue
		}

		cmdline, err := ioutil.ReadFile(filepath.Join(prefix, "/proc/", name, "cmdline"))
		if err != nil {
			continue
		}

		if strings.Index(string(cmdline), process) >= 0 {
			pids = append(pids, pid)
		}
	}

	return pids, nil
}

// GapInStream is called when a gap of nbytes bytes is found in the stream (due
// to packet loss).
func (http *httpPlugin) GapInStream(tcptuple *common.TCPTuple, dir uint8,
	nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {

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

	conn := getHTTPConnection(private)
	if conn == nil {
		return private, false
	}

	stream := conn.streams[dir]
	if stream == nil || stream.message == nil {
		// nothing to do
		return private, false
	}

	ok, complete := http.messageGap(stream, nbytes)
	if isDetailed {
		detailedf("messageGap returned ok=%v complete=%v", ok, complete)
	}
	if !ok {
		// on errors, drop stream
		conn.streams[dir] = nil
		return conn, true
	}

	if complete {
		// Current message is complete, we need to publish from here
		http.messageComplete(conn, tcptuple, dir, stream)
	}

	// don't drop the stream, we can ignore the gap
	return private, false
}

func (mysql *Mysql) GapInStream(tcptuple *common.TcpTuple, dir uint8,
	nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {

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

	if private == nil {
		return private, false
	}
	mysqlData, ok := private.(mysqlPrivateData)
	if !ok {
		return private, false
	}
	stream := mysqlData.Data[dir]
	if stream == nil || stream.message == nil {
		// nothing to do
		return private, false
	}

	if mysql.messageGap(stream, nbytes) {
		// we need to publish from here
		mysql.messageComplete(tcptuple, dir, stream)
	}

	// we always drop the TCP stream. Because it's binary and len based,
	// there are too few cases in which we could recover the stream (maybe
	// for very large blobs, leaving that as TODO)
	return private, true
}

// fetch invokes the appropriate Fetch method for the MetricSet and publishes
// the result using the publisher client. This method will recover from panics
// and log a stack track if one occurs.
func (msw *metricSetWrapper) fetch(done <-chan struct{}, out chan<- common.MapStr) error {
	defer logp.Recover(fmt.Sprintf("recovered from panic while fetching "+
		"'%s/%s' for host '%s'", msw.module.Name(), msw.Name(), msw.Host()))

	switch fetcher := msw.MetricSet.(type) {
	case mb.EventFetcher:
		event, err := msw.singleEventFetch(fetcher)
		if err != nil {
			return err
		}
		msw.stats.Add(eventsKey, 1)
		writeEvent(done, out, event)
	case mb.EventsFetcher:
		events, err := msw.multiEventFetch(fetcher)
		if err != nil {
			return err
		}
		for _, event := range events {
			msw.stats.Add(eventsKey, 1)
			if !writeEvent(done, out, event) {
				break
			}
		}
	default:
		return fmt.Errorf("MetricSet '%s/%s' does not implement a Fetcher "+
			"interface", msw.Module().Name(), msw.Name())
	}

	return nil
}

func (thrift *thriftPlugin) GapInStream(tcptuple *common.TCPTuple, dir uint8,
	nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {

	defer logp.Recover("GapInStream(thrift) exception")
	logp.Debug("thriftdetailed", "GapInStream called")

	if private == nil {
		return private, false
	}
	thriftData, ok := private.(thriftPrivateData)
	if !ok {
		return private, false
	}
	stream := thriftData.data[dir]
	if stream == nil || stream.message == nil {
		// nothing to do
		return private, false
	}

	if thrift.messageGap(stream, nbytes) {
		// we need to publish from here
		thrift.messageComplete(tcptuple, dir, stream, &thriftData)
	}

	// we always drop the TCP stream. Because it's binary and len based,
	// there are too few cases in which we could recover the stream (maybe
	// for very large blobs, leaving that as TODO)
	return private, true
}

// Starts the given module
func (m *Module) Start(b *beat.Beat) error {

	defer logp.Recover(fmt.Sprintf("Module %s paniced and stopped running.", m.name))

	if !m.Config.Enabled {
		logp.Debug("helper", "Not starting module %s with metricsets %s as not enabled.", m.name, m.getMetricSetsList())
		return nil
	}

	logp.Info("Setup moduler: %s", m.name)
	err := m.moduler.Setup(m)
	if err != nil {
		return fmt.Errorf("Error setting up module: %s. Not starting metricsets for this module.", err)
	}

	err = m.loadMetricsets()
	if err != nil {
		return fmt.Errorf("Error loading metricsets: %s", err)
	}

	// Setup period
	period, err := time.ParseDuration(m.Config.Period)
	if err != nil {
		return fmt.Errorf("Error in parsing period of module %s: %v", m.name, err)
	}

	// If no period set, set default
	if period == 0 {
		logp.Info("Setting default period for module %s as not set.", m.name)
		period = 1 * time.Second
	}

	var timeout time.Duration

	if m.Config.Timeout != "" {
		// Setup timeout
		timeout, err := time.ParseDuration(m.Config.Timeout)
		if err != nil {
			return fmt.Errorf("Error in parsing timeout of module %s: %v", m.name, err)
		}

		// If no timeout set, set to period as default
		if timeout == 0 {
			logp.Info("Setting default timeout for module %s as not set.", m.name)
			timeout = period
		}
	} else {
		timeout = period
	}
	m.Timeout = timeout

	logp.Info("Start Module %s with metricsets [%s] and period %v", m.name, m.getMetricSetsList(), period)

	m.setupMetricSets()

	m.events = b.Publisher.Connect()
	go m.Run(period, b)

	return nil
}

func createWatchUpdater(monitor *Monitor) func(content []byte) {
	return func(content []byte) {
		defer logp.Recover("Failed applying monitor watch")

		// read multiple json objects from content
		dec := json.NewDecoder(bytes.NewBuffer(content))
		var configs []*common.Config
		for dec.More() {
			var obj map[string]interface{}
			err := dec.Decode(&obj)
			if err != nil {
				logp.Err("Failed parsing json object: %v", err)
				return
			}

			logp.Info("load watch object: %v", obj)

			cfg, err := common.NewConfigFrom(obj)
			if err != nil {
				logp.Err("Failed normalizing json input: %v", err)
				return
			}

			configs = append(configs, cfg)
		}

		// apply read configurations
		if err := monitor.Update(configs); err != nil {
			logp.Err("Failed applying configuration: %v", err)
		}
	}
}

func (dns *Dns) ParseUdp(pkt *protos.Packet) {
	defer logp.Recover("Dns ParseUdp")
	packetSize := len(pkt.Payload)

	debugf("Parsing packet addressed with %s of length %d.",
		pkt.Tuple.String(), packetSize)

	dnsPkt, err := decodeDnsData(TransportUdp, pkt.Payload)
	if err != nil {
		// This means that malformed requests or responses are being sent or
		// that someone is attempting to the DNS port for non-DNS traffic. Both
		// are issues that a monitoring system should report.
		debugf("%s", err.Error())
		return
	}

	dnsTuple := DnsTupleFromIpPort(&pkt.Tuple, TransportUdp, dnsPkt.Id)
	dnsMsg := &DnsMessage{
		Ts:           pkt.Ts,
		Tuple:        pkt.Tuple,
		CmdlineTuple: procs.ProcWatcher.FindProcessesTuple(&pkt.Tuple),
		Data:         dnsPkt,
		Length:       packetSize,
	}

	if dnsMsg.Data.Response {
		dns.receivedDnsResponse(&dnsTuple, dnsMsg)
	} else /* Query */ {
		dns.receivedDnsRequest(&dnsTuple, dnsMsg)
	}
}

// GapInStream is called when a gap of nbytes bytes is found in the stream (due
// to packet loss).
func (http *HTTP) GapInStream(tcptuple *common.TcpTuple, dir uint8,
	nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {

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

	if private == nil {
		return private, false
	}
	httpData, ok := private.(httpConnectionData)
	if !ok {
		return private, false
	}
	stream := httpData.Streams[dir]
	if stream == nil || stream.message == nil {
		// nothing to do
		return private, false
	}

	ok, complete := http.messageGap(stream, nbytes)
	detailedf("messageGap returned ok=%v complete=%v", ok, complete)
	if !ok {
		// on errors, drop stream
		httpData.Streams[dir] = nil
		return httpData, true
	}

	if complete {
		// Current message is complete, we need to publish from here
		http.messageComplete(tcptuple, dir, stream)
	}

	// don't drop the stream, we can ignore the gap
	return private, false
}

func (dns *Dns) ParseUdp(pkt *protos.Packet) {
	defer logp.Recover("Dns ParseUdp")

	logp.Debug("dns", "Parsing packet addressed with %s of length %d.",
		pkt.Tuple.String(), len(pkt.Payload))

	dnsPkt, err := decodeDnsPacket(pkt.Payload)
	if err != nil {
		// This means that malformed requests or responses are being sent or
		// that someone is attempting to the DNS port for non-DNS traffic. Both
		// are issues that a monitoring system should report.
		logp.Debug("dns", NonDnsPacketMsg+" addresses %s, length %d",
			pkt.Tuple.String(), len(pkt.Payload))
		return
	}

	dnsTuple := DnsTupleFromIpPort(&pkt.Tuple, TransportUdp, dnsPkt.ID)
	dnsMsg := &DnsMessage{
		Ts:           pkt.Ts,
		Tuple:        pkt.Tuple,
		CmdlineTuple: procs.ProcWatcher.FindProcessesTuple(&pkt.Tuple),
		Data:         dnsPkt,
		Length:       len(pkt.Payload),
	}

	if dnsMsg.Data.QR == Query {
		dns.receivedDnsRequest(&dnsTuple, dnsMsg)
	} else /* Response */ {
		dns.receivedDnsResponse(&dnsTuple, dnsMsg)
	}
}

func (m *Module) FetchMetricSets(b *beat.Beat, metricSet *MetricSet) {

	m.wg.Add(1)

	// Catches metric in case of panic. Keeps other metricsets running
	defer m.wg.Done()

	// Separate defer call as is has to be called directly
	defer logp.Recover(fmt.Sprintf("Metric %s paniced and stopped running.", m.name))

	events, err := metricSet.Fetch()

	if err != nil {
		// TODO: Also list module?
		logp.Err("Fetching events in MetricSet %s returned error: %s", metricSet.Name, err)
		// TODO: Still publish event with error
		return
	}

	events, err = m.processEvents(events, metricSet)

	// Async publishing of event
	b.Events.PublishEvents(events)

}

func (dns *Dns) Parse(pkt *protos.Packet, tcpTuple *common.TcpTuple, dir uint8, private protos.ProtocolData) protos.ProtocolData {
	defer logp.Recover("DNS ParseTcp")

	logp.Debug("dns", "Parsing packet addressed with %s of length %d.",
		pkt.Tuple.String(), len(pkt.Payload))

	priv := dnsPrivateData{}

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

	payload := pkt.Payload

	stream := &priv.Data[dir]

	if *stream == nil {
		*stream = &DnsStream{
			tcpTuple: tcpTuple,
			data:     payload,
			message:  &DnsMessage{Ts: pkt.Ts, Tuple: pkt.Tuple},
		}
		if len(payload) <= DecodeOffset {
			logp.Debug("dns", EmptyMsg+" addresses %s",
				tcpTuple.String())

			return priv
		}
	} else {
		(*stream).data = append((*stream).data, payload...)
		dataLength := len((*stream).data)
		if dataLength > tcp.TCP_MAX_DATA_IN_STREAM {
			logp.Debug("dns", "Stream data too large, dropping DNS stream")
			return priv
		}
		if dataLength <= DecodeOffset {
			logp.Debug("dns", EmptyMsg+" addresses %s",
				tcpTuple.String())
			return priv
		}
	}

	data, err := decodeDnsData(TransportTcp, (*stream).data)

	if err != nil {
		logp.Debug("dns", NonDnsCompleteMsg+" addresses %s, length %d",
			tcpTuple.String(), len((*stream).data))

		// wait for decoding with the next segment
		return priv
	}

	dns.messageComplete(tcpTuple, dir, *stream, data)
	return priv
}

// Called when a packets are missing from the tcp
// stream.
func (rpc *Rpc) GapInStream(tcptuple *common.TCPTuple, dir uint8,
	nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {

	defer logp.Recover("GapInRpcStream exception")

	// forced by TCP interface
	return private, false
}

// Called when the FIN flag is seen in the TCP stream.
func (rpc *Rpc) ReceivedFin(tcptuple *common.TCPTuple, dir uint8,
	private protos.ProtocolData) protos.ProtocolData {

	defer logp.Recover("ReceivedFinRpc exception")

	// forced by TCP interface
	return private
}

// GapInStream is called by TCP layer when a packets are missing from the tcp
// stream.
func (mc *Memcache) GapInStream(
	tcptuple *common.TcpTuple,
	dir uint8, nbytes int,
	private protos.ProtocolData,
) (priv protos.ProtocolData, drop bool) {
	debug("memcache(tcp) stream gap detected")

	defer logp.Recover("GapInStream(memcache) exception")
	if !isMemcacheConnection(private) {
		return private, false
	}

	conn := private.(*tcpConnectionData)
	stream := conn.Streams[dir]
	parser := stream.parser
	msg := parser.message

	if msg != nil {
		if msg.IsRequest {
			msg.AddNotes(NoteRequestPacketLoss)
		} else {
			msg.AddNotes(NoteResponsePacketLoss)
		}
	}

	// If we are about to read binary data (length) encoded, but missing gab
	// does fully cover data area, we might be able to continue processing the
	// stream + transactions
	inData := parser.state == parseStateDataBinary ||
		parser.state == parseStateIncompleteDataBinary ||
		parser.state == parseStateData ||
		parser.state == parseStateIncompleteData
	if inData {
		if msg == nil {
			logp.WTF("parser message is nil on data load")
			return private, true
		}

		alreadyRead := stream.Buf.Len() - int(msg.bytesLost)
		dataRequired := int(msg.bytes) - alreadyRead
		if nbytes <= dataRequired {
			// yay, all bytes included in message binary data part.
			// just drop binary data part and recover parsing.
			if msg.isBinary {
				parser.state = parseStateIncompleteDataBinary
			} else {
				parser.state = parseStateIncompleteData
			}
			msg.bytesLost += uint(nbytes)
			return private, false
		}
	}

	// need to drop TCP stream. But try to publish all cached trancsactions first
	mc.pushAllTCPTrans(conn.connection)
	return private, true
}

func (tcp *Tcp) Process(id *flows.FlowID, tcphdr *layers.TCP, pkt *protos.Packet) {
	// This Recover should catch all exceptions in
	// protocol modules.
	defer logp.Recover("Process tcp exception")

	debugf("tcp flow id: %p", id)

	stream, created := tcp.getStream(pkt)
	if stream.conn == nil {
		return
	}

	if id != nil {
		id.AddConnectionID(uint64(stream.conn.id))
	}
	conn := stream.conn

	tcp_start_seq := tcphdr.Seq
	tcp_seq := tcp_start_seq + uint32(len(pkt.Payload))
	lastSeq := conn.lastSeq[stream.dir]
	if isDebug {
		debugf("pkt.start_seq=%v pkt.last_seq=%v stream.last_seq=%v (len=%d)",
			tcp_start_seq, tcp_seq, lastSeq, len(pkt.Payload))
	}

	if len(pkt.Payload) > 0 && lastSeq != 0 {
		if tcpSeqBeforeEq(tcp_seq, lastSeq) {
			if isDebug {
				debugf("Ignoring retransmitted segment. pkt.seq=%v len=%v stream.seq=%v",
					tcphdr.Seq, len(pkt.Payload), lastSeq)
			}
			return
		}

		if tcpSeqBefore(lastSeq, tcp_start_seq) {
			if !created {
				gap := int(tcp_start_seq - lastSeq)
				logp.Warn("Gap in tcp stream. last_seq: %d, seq: %d, gap: %d", lastSeq, tcp_start_seq, gap)
				drop := stream.gapInStream(gap)
				if drop {
					if isDebug {
						debugf("Dropping connection state because of gap")
					}

					// drop application layer connection state and
					// update stream_id for app layer analysers using stream_id for lookups
					conn.id = tcp.getId()
					conn.data = nil
				}
			}
		}
	}

	conn.lastSeq[stream.dir] = tcp_seq
	stream.addPacket(pkt, tcphdr)
}

func (m *Module) FetchMetricSets(metricSet *MetricSet) {

	// Separate defer call as is has to be called directly
	defer logp.Recover(fmt.Sprintf("Metric %s paniced and stopped running.", m.name))

	err := metricSet.Fetch()
	if err != nil {
		logp.Err("Fetching events for MetricSet %s in Module %s returned error: %s", metricSet.Name, m.name, err)
	}
}

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

	defer logp.Recover("ParseMysql exception")

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

	if priv.Data[dir] == nil {
		priv.Data[dir] = &MysqlStream{
			tcptuple: tcptuple,
			data:     pkt.Payload,
			message:  &MysqlMessage{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("mysql", "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 = &MysqlMessage{Ts: pkt.Ts}
		}

		ok, complete := mysqlMessageParser(priv.Data[dir])
		//logp.Debug("mysqldetailed", "mysqlMessageParser returned ok=%b complete=%b", ok, complete)
		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. Try parsing with the next segment")
			return priv
		}

		if complete {
			mysql.messageComplete(tcptuple, dir, stream)
		} else {
			// wait for more data
			break
		}
	}
	return priv
}

// Parse function is used to process TCP payloads.
func (http *HTTP) Parse(pkt *protos.Packet, tcptuple *common.TcpTuple,
	dir uint8, private protos.ProtocolData) protos.ProtocolData {

	defer logp.Recover("ParseHttp exception")

	detailedf("Payload received: [%s]", pkt.Payload)

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

	if priv.Streams[dir] == nil {
		priv.Streams[dir] = &stream{
			tcptuple: tcptuple,
			data:     pkt.Payload,
			message:  &message{Ts: pkt.Ts},
		}

	} else {
		// concatenate bytes
		priv.Streams[dir].data = append(priv.Streams[dir].data, pkt.Payload...)
		if len(priv.Streams[dir].data) > tcp.TCP_MAX_DATA_IN_STREAM {
			debugf("Stream data too large, dropping TCP stream")
			priv.Streams[dir] = nil
			return priv
		}
	}
	stream := priv.Streams[dir]
	if stream.message == nil {
		stream.message = &message{Ts: pkt.Ts}
	}

	parser := newParser(&http.parserConfig)
	ok, complete := parser.parse(stream)
	if !ok {
		// drop this tcp stream. Will retry parsing with the next
		// segment in it
		priv.Streams[dir] = nil
		return priv
	}

	if complete {
		// all ok, ship it
		http.messageComplete(tcptuple, dir, stream)
	}

	return priv
}

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

	defer logp.Recover("ParseAmqp exception")
	detailedf("Parse method triggered")

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

	if priv.Data[dir] == nil {
		priv.Data[dir] = &AmqpStream{
			tcptuple: tcptuple,
			data:     pkt.Payload,
			message:  &AmqpMessage{Ts: pkt.Ts},
		}
	} else {
		// concatenate databytes
		priv.Data[dir].data = append(priv.Data[dir].data, pkt.Payload...)
		if len(priv.Data[dir].data) > tcp.TCP_MAX_DATA_IN_STREAM {
			debugf("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 = &AmqpMessage{Ts: pkt.Ts}
		}

		ok, complete := amqp.amqpMessageParser(stream)
		if !ok {
			// drop this tcp stream. Will retry parsing with the next
			// segment in it
			priv.Data[dir] = nil
			return priv
		}
		if !complete {
			break
		}
		amqp.handleAmqp(stream.message, tcptuple, dir)
	}
	return priv
}