Update Memcache protocol to use ECS fields (#10189)

Here's a summary of what fields changed.

Part of #7968

Changed

- bytes_in -> source.bytes
- bytes_out -> destination.bytes
- responsetime -> event.duration (unit are now nanoseconds)
- transport -> network.transport = udp or tcp

Added

- destination
- event.dataset = memcache
- event.end
- event.start
- network.bytes
- network.community_id
- network.protocol = memcache
- network.type
- source

Unchanged Packetbeat Fields

- status
- type = memcache (we might remove this since we have event.dataset)

CHANGELOG.next.asciidoc

@@ -81,6 +81,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Changed NFS protocol fields to align with ECS. {pull}10153[10153]
 - Changed Thrift protocol fields to align with ECS. {pull}10125[10125]
 - Changed Cassandra protocol fields to align with ECS. {pull}10093[10093]
+- Changed Memcache protocol fields to align with ECS. {pull}10189[10189]
 
 *Winlogbeat*
 

packetbeat/protos/applayer/applayer.go

@@ -26,6 +26,8 @@ import (
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/streambuf"
+
+	"github.com/elastic/beats/packetbeat/pb"
 )
 
 // A Message its direction indicator
@@ -91,9 +93,8 @@ type Transaction struct {
 	// Ts sets the transaction its initial timestamp
 	Ts TransactionTimestamp
 
-	// ResponseTime is the transaction duration in milliseconds. Should be set
-	// to -1 if duration is unknown
-	ResponseTime int32
+	// EndTime is the time the transaction ended.
+	EndTime time.Time
 
 	// Status of final transaction
 	Status string // see libbeat/common/statuses.go
@@ -222,17 +223,25 @@ func (t *Transaction) InitWithMsg(
 func (t *Transaction) Event(event *beat.Event) error {
 	event.Timestamp = t.Ts.Ts
 
+	pbf := &pb.Fields{}
+	pbf.SetSource(&t.Src)
+	pbf.SetDestination(&t.Dst)
+	pbf.Source.Bytes = int64(t.BytesIn)
+	pbf.Destination.Bytes = int64(t.BytesOut)
+	pbf.Event.Dataset = t.Type
+	pbf.Event.Start = t.Ts.Ts
+	pbf.Event.End = t.EndTime
+	pbf.Network.Transport = t.Transport.String()
+	pbf.Network.Protocol = pbf.Event.Dataset
+
 	fields := event.Fields
-	fields["type"] = t.Type
-	fields["responsetime"] = t.ResponseTime
-	fields["src"] = &t.Src
-	fields["dst"] = &t.Dst
-	fields["transport"] = t.Transport.String()
-	fields["bytes_out"] = t.BytesOut
-	fields["bytes_in"] = t.BytesIn
+	fields[pb.FieldsKey] = pbf
+	fields["type"] = pbf.Event.Dataset
 	fields["status"] = t.Status
-	if len(t.Notes) > 0 {
-		fields["notes"] = t.Notes
+	if len(t.Notes) == 1 {
+		event.PutValue("error.message", t.Notes[0])
+	} else if len(t.Notes) > 1 {
+		event.PutValue("error.message", t.Notes)
 	}
 	return nil
 }

packetbeat/protos/memcache/memcache.go

@@ -351,18 +351,17 @@ func newTransaction(requ, resp *message) *transaction {
 		t.Init(requ)
 		t.BytesOut = requ.Size
 		t.BytesIn = resp.Size
-		t.ResponseTime = int32(resp.Ts.Sub(requ.Ts).Nanoseconds() / 1e6) // [ms]
+		t.EndTime = resp.Ts
 		t.Notes = append(t.Notes, requ.Notes...)
 		t.Notes = append(t.Notes, resp.Notes...)
 	case requ != nil && resp == nil:
 		t.Init(requ)
 		t.BytesOut = requ.Size
-		t.ResponseTime = -1
 		t.Notes = append(t.Notes, requ.Notes...)
 	case requ == nil && resp != nil:
 		t.Init(resp)
 		t.BytesIn = resp.Size
-		t.ResponseTime = -1
+		t.EndTime = resp.Ts
 		t.Notes = append(t.Notes, resp.Notes...)
 	}
 

packetbeat/tests/system/test_0040_memcache_tcp_bin_basic.py

@@ -29,7 +29,7 @@ def assert_common(self, objs):
 
         # check transport layer always tcp
         assert all(o['type'] == 'memcache' for o in objs)
-        assert all(o['transport'] == 'tcp' for o in objs)
+        assert all(o['network.transport'] == 'tcp' for o in objs)
         assert all(o['memcache.protocol_type'] == 'binary' for o in objs)
 
     def test_store_load(self):

packetbeat/tests/system/test_0040_memcache_tcp_text_basic.py

@@ -29,7 +29,7 @@ def assert_common(self, objs):
 
         # check transport layer always tcp
         assert all(o['type'] == 'memcache' for o in objs)
-        assert all(o['transport'] == 'tcp' for o in objs)
+        assert all(o['network.transport'] == 'tcp' for o in objs)
         assert all(o['memcache.protocol_type'] == 'text' for o in objs)
 
     def test_store_load(self):

packetbeat/tests/system/test_0041_memcache_udp_bin_basic.py

@@ -30,7 +30,7 @@ def assert_common(self, objs):
 
         # check transport layer always udp
         assert all(o['type'] == 'memcache' for o in objs)
-        assert all(o['transport'] == 'udp' for o in objs)
+        assert all(o['network.transport'] == 'udp' for o in objs)
         assert all(o['memcache.protocol_type'] == 'binary' for o in objs)
 
     def test_store(self):

packetbeat/tests/system/test_0041_memcache_udp_text_basic.py

@@ -31,7 +31,7 @@ def assert_common(self, objs):
 
         # check transport layer always tcp
         assert all(o['type'] == 'memcache' for o in objs)
-        assert all(o['transport'] == 'udp' for o in objs)
+        assert all(o['network.transport'] == 'udp' for o in objs)
         assert all(o['memcache.protocol_type'] == 'text' for o in objs)
 
     def test_store(self):