test fixes

.github/workflows/buildtest.yml

@@ -63,7 +63,7 @@ jobs:
             ${{ runner.os }}-bazel-
       - name: Install pcap
         run: |
-          sudo apt-get install libpcap-dev
+          sudo apt-get install libpcap-dev libnl-genl-3-dev libnl-3-dev
       - name: Test
         run: make coverage
       - name: Coveralls

dataplane/forwarding/fwd.go

@@ -97,26 +97,6 @@ func (e *Server) UpdateNotification(contextID *fwdpb.ContextId, notification fwd
 	return nil
 }
 
-// UpdatePacketSink updates the packet sink service for a context. If the
-// service is set to nil, no packets are delivered externally for the context.
-// The address is the address of the packet service (used in queries)
-// in the host:port format.
-func (e *Server) UpdatePacketSink(contextID *fwdpb.ContextId, packet fwdcontext.PacketCallback) error {
-	if contextID == nil {
-		return errors.New("fwd: UpdatePacketSink failed, No context ID")
-	}
-
-	ctx, err := e.FindContext(contextID)
-	if err != nil {
-		return fmt.Errorf("fwd: UpdatePacketSink failed, err %v", err)
-	}
-
-	ctx.Lock()
-	defer ctx.Unlock()
-	ctx.SetPacketSink(packet)
-	return nil
-}
-
 // ContextCreate creates a new context. Note that if the packet sink and/or
 // notification services are specified but not reachable, the context creation
 // fails.

dataplane/forwarding/fwdport/ports/BUILD

@@ -16,7 +16,6 @@ go_library(
     deps = [
         "//dataplane/forwarding/fwdaction",
         "//dataplane/forwarding/fwdport",
-        "//dataplane/forwarding/infra/deadlock",
         "//dataplane/forwarding/infra/fwdcontext",
         "//dataplane/forwarding/infra/fwdobject",
         "//dataplane/forwarding/infra/fwdpacket",
@@ -68,14 +67,14 @@ go_test(
         "//dataplane/forwarding/protocol/ethernet",
         "//dataplane/forwarding/protocol/metadata",
         "//dataplane/forwarding/protocol/opaque",
+        "//dataplane/proto/packetio",
         "//proto/forwarding",
         "@com_github_go_logr_logr//testr",
         "@com_github_google_go_cmp//cmp",
         "@com_github_google_gopacket//:gopacket",
         "@com_github_google_gopacket//layers",
         "@com_github_google_gopacket//pcapgo",
         "@com_github_openconfig_gnmi//errdiff",
-        "@org_golang_google_protobuf//proto",
         "@org_uber_go_mock//gomock",
     ],
 )

dataplane/forwarding/fwdport/ports/cpu.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/openconfig/lemming/dataplane/forwarding/fwdaction"
 	"github.com/openconfig/lemming/dataplane/forwarding/fwdport"
-	"github.com/openconfig/lemming/dataplane/forwarding/infra/deadlock"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdcontext"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdobject"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdpacket"
@@ -114,30 +113,20 @@ func (p *CPUPort) punt(v any) {
 		fwdport.Increment(p, 1, fwdpb.CounterId_COUNTER_ID_TX_ERROR_PACKETS, fwdpb.CounterId_COUNTER_ID_TX_ERROR_OCTETS)
 		return
 	}
-
 	p.ctx.RLock()
 	var ingressPID *fwdpb.PortId
 	if port, err := fwdport.InputPort(packet, p.ctx); err == nil {
 		ingressPID = fwdport.GetID(port)
 	}
-	// egressPID := fwdport.GetID(p)
-	// if port, err := fwdport.OutputPort(packet, p.ctx); err == nil {
-	// 	egressPID = fwdport.GetID(port)
-	// }
 	hostPort, err := packet.Field(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID, 0))
 	if err != nil {
 		fwdport.Increment(p, packet.Length(), fwdpb.CounterId_COUNTER_ID_TX_ERROR_PACKETS, fwdpb.CounterId_COUNTER_ID_TX_ERROR_OCTETS)
 		return
 	}
-
 	ingressID, err := strconv.Atoi(ingressPID.GetObjectId().GetId())
 	if err != nil {
 		return
 	}
-	// egressID, err := strconv.Atoi(egressPID.GetObjectId().GetId())
-	// if err != nil {
-	// 	return
-	// }
 
 	response := &pktiopb.PacketOut{
 		Packet: &pktiopb.Packet{
@@ -155,8 +144,8 @@ func (p *CPUPort) punt(v any) {
 		return
 	}
 
-	timer := deadlock.NewTimer(deadlock.Timeout, fmt.Sprintf("Punting packet from port %v", p))
-	defer timer.Stop()
+	// timer := deadlock.NewTimer(deadlock.Timeout, fmt.Sprintf("Punting packet from port %v", p))
+	// defer timer.Stop()
 	if err := ps(response); err != nil {
 		fwdport.Increment(p, packet.Length(), fwdpb.CounterId_COUNTER_ID_TX_ERROR_PACKETS, fwdpb.CounterId_COUNTER_ID_TX_ERROR_OCTETS)
 		log.Errorf("ports: Unable to punt packet, request %+v, err %v.", response, err)

dataplane/forwarding/fwdport/ports/cpu_test.go

@@ -15,14 +15,16 @@
 package ports
 
 import (
+	"encoding/binary"
 	"testing"
 
-	"google.golang.org/protobuf/proto"
+	"github.com/google/go-cmp/cmp"
 
 	"github.com/openconfig/lemming/dataplane/forwarding/fwdport"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdcontext"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdobject"
 	"github.com/openconfig/lemming/dataplane/forwarding/infra/fwdpacket"
+	pktiopb "github.com/openconfig/lemming/dataplane/proto/packetio"
 	fwdpb "github.com/openconfig/lemming/proto/forwarding"
 
 	_ "github.com/openconfig/lemming/dataplane/forwarding/protocol/arp"
@@ -34,46 +36,34 @@ import (
 // A RecordPacketSink records the last injected packet and generates a
 // notification on its channel.
 type RecordPacketSink struct {
-	notify       chan bool
-	lastResponse *fwdpb.PacketSinkResponse
+	notify chan bool
+	resp   *pktiopb.PacketOut
 }
 
 // PacketSink records the inject packet request and generates a notification.
-func (p *RecordPacketSink) PacketSink(resp *fwdpb.PacketSinkResponse) error {
-	p.lastResponse = resp
+func (p *RecordPacketSink) Send(resp *pktiopb.PacketOut) error {
+	p.resp = resp
 	p.notify <- true
 	return nil
 }
 
 // Tests the writes to the CPU port.
 func TestCpuWrite(t *testing.T) {
-	name := "TestPort"
+	name := "1"
 	ctx := fwdcontext.New("test", "fwd")
 	ps := &RecordPacketSink{
 		notify: make(chan bool),
 	}
-	ctx.SetPacketSink(ps.PacketSink)
+	ctx.SetCPUPortSink(ps.Send, func() {})
 
 	// Create a CPU port that exports the ETHER_TYPE and IP_ADDR_DST
 	desc := &fwdpb.PortDesc{
 		PortType: fwdpb.PortType_PORT_TYPE_CPU_PORT,
 		PortId:   fwdport.MakeID(fwdobject.NewID(name)),
 	}
-	ethertype := &fwdpb.PacketFieldId{
-		Field: &fwdpb.PacketField{
-			FieldNum: fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_TYPE,
-			Instance: 0,
-		},
-	}
-	ipaddress := &fwdpb.PacketFieldId{
-		Field: &fwdpb.PacketField{
-			FieldNum: fwdpb.PacketFieldNum_PACKET_FIELD_NUM_IP_ADDR_DST,
-			Instance: 0,
-		},
-	}
+
 	cpu := &fwdpb.CPUPortDesc{
-		QueueId:        name,
-		ExportFieldIds: []*fwdpb.PacketFieldId{ethertype, ipaddress},
+		QueueId: name,
 	}
 	desc.Port = &fwdpb.PortDesc_Cpu{
 		Cpu: cpu,
@@ -95,6 +85,8 @@ func TestCpuWrite(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Unable to create ARP packet, err %v.", err)
 	}
+	packet.Update(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID, 1), fwdpacket.OpSet, binary.BigEndian.AppendUint64(nil, 1))
+	fwdport.SetInputPort(packet, port)
 	fwdport.SetOutputPort(packet, port)
 
 	// Write the packet out of the cpu port and wait for it to be received
@@ -103,19 +95,12 @@ func TestCpuWrite(t *testing.T) {
 		t.Fatalf("Write failed, err %v.", err)
 	}
 	<-ps.notify
-	t.Logf("Got request %+v", ps.lastResponse)
+	t.Logf("Got request %+v", ps.resp)
 
 	// Verify that the packet was received and the parsed fields only have
 	// the ETHER_TYPE set to ARP.
-	list := ps.lastResponse.GetPacket().GetParsedFields()
-	if len(list) != 1 {
-		t.Fatalf("Write failed to get parsed fields, got %v, want 1.", len(list))
-	}
-	want := &fwdpb.PacketFieldBytes{
-		FieldId: ethertype,
-		Bytes:   []byte{0x08, 0x06},
-	}
-	if !proto.Equal(list[0], want) {
-		t.Fatalf("Write failed to get parsed fields, Got %v, want %v", list[0], want)
+	got := ps.resp.GetPacket()
+	if d := cmp.Diff(got.GetFrame(), arp); d != "" {
+		t.Fatalf("Write failed to get parsed fields, diff: %s", d)
 	}
 }

dataplane/forwarding/fwdport/ports/group_test.go

@@ -127,14 +127,14 @@ func TestPortGroupHash(t *testing.T) {
 	defer ctrl.Finish()
 
 	names := []string{"p1", "p2", "p3", "p4"}
-	ctx := fwdcontext.New("test", "fwd")
 
 	hashFn := []fwdpb.AggregateHashAlgorithm{
 		fwdpb.AggregateHashAlgorithm_AGGREGATE_HASH_ALGORITHM_CRC16,
 		fwdpb.AggregateHashAlgorithm_AGGREGATE_HASH_ALGORITHM_CRC32,
 	}
 
 	for id, fn := range hashFn {
+		ctx := fwdcontext.New("test", "fwd")
 		var ports []fwdport.Port
 		for _, name := range names {
 			ports = append(ports, porttestutil.CreateTestPort(t, ctx, fmt.Sprintf("%v-%v", name, id)))

dataplane/forwarding/infra/fwdcontext/context.go

@@ -174,19 +174,6 @@ func (ctx *Context) Notify(event *fwdpb.EventDesc) error {
 
 type CPUPortSink func(*pktiopb.PacketOut) error
 
-// SetPacketSink sets the packet sink service for the context. If the packet
-// sink service is not set to nil, packets are dropped.
-// TODO: Deprecated remove
-func (ctx *Context) SetPacketSink(call PacketCallback) error {
-	ctx.packets = call
-	return nil
-}
-
-// PacketSink returns a handler to the packet sink service.
-func (ctx *Context) PacketSink() PacketCallback {
-	return ctx.packets
-}
-
 // SetCPUPortSink sets the port control service for the context
 func (ctx *Context) SetCPUPortSink(fn CPUPortSink, doneFn func()) error {
 	ctx.cpuPortSink = fn
@@ -204,7 +191,6 @@ func (ctx *Context) CPUPortSink() CPUPortSink {
 // Then it unblocks the caller by sending a message on the channel.
 // Then it cleans up the rest of the objects.
 func (ctx *Context) Cleanup(ch chan bool, isPort func(*fwdpb.ObjectId) bool) {
-	ctx.SetPacketSink(nil)
 	ctx.SetNotification(nil)
 
 	if ctx.cpuPortSinkDone != nil {

dataplane/saiserver/hostif_test.go

@@ -83,7 +83,7 @@ func TestCreateHostif(t *testing.T) {
 			dplane := &fakeSwitchDataplane{
 				ctx: fwdcontext.New("foo", "foo"),
 			}
-			dplane.ctx.SetPacketSink(func(*fwdpb.PacketSinkResponse) error { return nil })
+			dplane.ctx.SetCPUPortSink(func(po *pktiopb.PacketOut) error { return nil }, func() {})
 			c, mgr, stopFn := newTestHostif(t, dplane)
 			// Create switch and ports
 			mgr.StoreAttributes(mgr.NextID(), &saipb.SwitchAttribute{