Teammate fdb fix

dataplane/forwarding/fwdconfig/action.go

@@ -409,3 +409,25 @@ func (m *MirrorActionBuilder) set(a *fwdpb.ActionDesc) {
 func (m *MirrorActionBuilder) actionType() fwdpb.ActionType {
 	return fwdpb.ActionType_ACTION_TYPE_MIRROR
 }
+
+// BridgeLearnActionBuilder is a builder for a bridge learn action.
+type BridgeLearnActionBuilder struct {
+	tableID string
+}
+
+// BridgeLearnAction returns a new bridge learn action builder.
+func BridgeLearnAction(tableID string) *BridgeLearnActionBuilder {
+	return &BridgeLearnActionBuilder{tableID: tableID}
+}
+
+func (b *BridgeLearnActionBuilder) set(a *fwdpb.ActionDesc) {
+	a.Action = &fwdpb.ActionDesc_Bridge{
+		Bridge: &fwdpb.BridgeLearnActionDesc{
+			TableId: &fwdpb.TableId{ObjectId: &fwdpb.ObjectId{Id: b.tableID}},
+		},
+	}
+}
+
+func (b *BridgeLearnActionBuilder) actionType() fwdpb.ActionType {
+	return fwdpb.ActionType_ACTION_TYPE_BRIDGE_LEARN
+}

dataplane/forwarding/fwdtable/bridge/bridge.go

@@ -69,9 +69,17 @@ func (req *learnRequest) DebugString(port fwdport.Port) string {
 // is buffered.
 type Table struct {
 	*exact.Table                     // exact table containing mac entries
-	learn        *queue.Queue        // unbounded queue for learn requests
-	ctx          *fwdcontext.Context // context for finding objects
-	notify       chan bool           // if not nil, a notification is generated when an entry is learned (test only)
+	learn         *queue.Queue        // unbounded queue for learn requests
+	ctx           *fwdcontext.Context // context for finding objects
+	notify        chan bool           // if not nil, a notification is generated when an entry is learned (test only)
+	LearnCallback func(mac []byte, portID string)
+}
+
+// SetLearnCallback sets a callback invoked when a new MAC entry is dynamically learned.
+func (t *Table) SetLearnCallback(cb func(mac []byte, portID string)) {
+	t.ctx.Lock()
+	defer t.ctx.Unlock()
+	t.LearnCallback = cb
 }
 
 // Clear clears the table by deleting all its entries.
@@ -152,6 +160,8 @@ func (t *Table) processLearn(v interface{}) {
 	// we try to learn a mac address that has a static entry.
 	if err := t.AddEntry(desc, []*fwdpb.ActionDesc{&ad}); err != nil {
 		log.Infof("bridge: Skipping learn for %v %v.", req.DebugString(port), err)
+	} else if t.LearnCallback != nil {
+		t.LearnCallback(req.mac, string(fwdport.GetID(port).GetObjectId().GetId()))
 	}
 }
 

dataplane/saiserver/BUILD

@@ -23,6 +23,7 @@ go_library(
     deps = [
         "//dataplane/dplaneopts",
         "//dataplane/forwarding",
+        "//dataplane/forwarding/fwdtable/bridge",
         "//dataplane/forwarding/fwdconfig",
         "//dataplane/forwarding/infra/fwdcontext",
         "//dataplane/proto/packetio",

dataplane/saiserver/attrmgr/attrmgr.go

@@ -18,6 +18,7 @@
 package attrmgr
 
 import (
+	"strconv"
 	"context"
 	"fmt"
 	"log/slog"
@@ -385,3 +386,18 @@ func (mgr *AttrMgr) getID(req, resp proto.Message) (string, error) {
 	}
 	return string(pBytes), nil
 }
+
+// GetOIDsByType returns a list of OIDs that match the given ObjectType.
+func (mgr *AttrMgr) GetOIDsByType(t saipb.ObjectType) []uint64 {
+	mgr.mu.Lock()
+	defer mgr.mu.Unlock()
+	var oids []uint64
+	for id, ty := range mgr.idToType {
+		if ty == t {
+			if oid, err := strconv.ParseUint(id, 10, 64); err == nil {
+				oids = append(oids, oid)
+			}
+		}
+	}
+	return oids
+}

dataplane/saiserver/fdb.go

@@ -16,28 +16,158 @@ package saiserver
 
 import (
 	"context"
+	"fmt"
+	"log/slog"
+	"sync"
 
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 
+	"github.com/openconfig/lemming/dataplane/forwarding/fwdconfig"
 	saipb "github.com/openconfig/lemming/dataplane/proto/sai"
 	"github.com/openconfig/lemming/dataplane/saiserver/attrmgr"
+	fwdpb "github.com/openconfig/lemming/proto/forwarding"
 )
 
 type fdb struct {
 	saipb.UnimplementedFdbServer
-	mgr       *attrmgr.AttrMgr
-	dataplane switchDataplaneAPI
+	mgr         *attrmgr.AttrMgr
+	dataplane   switchDataplaneAPI
+	mu          sync.RWMutex
+	subscribers map[chan *saipb.FdbEventNotificationResponse]struct{}
 }
 
 func newFdb(mgr *attrmgr.AttrMgr, dataplane switchDataplaneAPI, s *grpc.Server) (*fdb, error) {
 	f := &fdb{
-		mgr:       mgr,
-		dataplane: dataplane,
+		mgr:         mgr,
+		dataplane:   dataplane,
+		subscribers: make(map[chan *saipb.FdbEventNotificationResponse]struct{}),
 	}
 	saipb.RegisterFdbServer(s, f)
 	return f, nil
 }
 
+func (f *fdb) subscribe(ch chan *saipb.FdbEventNotificationResponse) func() {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.subscribers[ch] = struct{}{}
+	return func() {
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		delete(f.subscribers, ch)
+	}
+}
+
+func (f *fdb) sendNotification(data *saipb.FdbEventNotificationData) {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+	resp := &saipb.FdbEventNotificationResponse{
+		Data: []*saipb.FdbEventNotificationData{data},
+	}
+	for ch := range f.subscribers {
+		select {
+		case ch <- resp:
+		default:
+			slog.Warn("fdb notification channel full, dropping event")
+		}
+	}
+}
+
 func (f *fdb) FlushFdbEntries(ctx context.Context, req *saipb.FlushFdbEntriesRequest) (*saipb.FlushFdbEntriesResponse, error) {
 	return &saipb.FlushFdbEntriesResponse{}, nil
 }
+
+func (f *fdb) CreateFdbEntry(ctx context.Context, req *saipb.CreateFdbEntryRequest) (*saipb.CreateFdbEntryResponse, error) {
+	slog.InfoContext(ctx, "CreateFdbEntry called", "mac", req.GetEntry().GetMacAddress(), "vlan", req.GetEntry().GetBvId(), "bridge_port", req.GetBridgePortId())
+	entry := req.GetEntry()
+	if entry == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "FDB entry is required")
+	}
+
+	mac := entry.GetMacAddress()
+	if len(mac) == 0 {
+		return nil, status.Errorf(codes.InvalidArgument, "MAC address is required")
+	}
+
+	portOID := req.GetBridgePortId()
+	if portOID == 0 {
+		return nil, status.Errorf(codes.InvalidArgument, "Bridge port ID is required")
+	}
+
+	bpReq := &saipb.GetBridgePortAttributeRequest{
+		Oid:      portOID,
+		AttrType: []saipb.BridgePortAttr{saipb.BridgePortAttr_BRIDGE_PORT_ATTR_PORT_ID},
+	}
+	bpResp := &saipb.GetBridgePortAttributeResponse{}
+	if err := f.mgr.PopulateAttributes(bpReq, bpResp); err != nil {
+		return nil, fmt.Errorf("failed to populate bridge port %d: %v", portOID, err)
+	}
+
+	portID := bpResp.GetAttr().GetPortId()
+	if portID == 0 {
+		return nil, fmt.Errorf("cannot find port ID for bridge port %d", portOID)
+	}
+
+	addReq := fwdconfig.TableEntryAddRequest(f.dataplane.ID(), FDBTable).AppendEntry(
+		fwdconfig.EntryDesc(fwdconfig.ExactEntry(
+			fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_DST).WithBytes(mac),
+		)),
+		fwdconfig.TransmitAction(fmt.Sprint(portID)),
+	).Build()
+
+	if _, err := f.dataplane.TableEntryAdd(ctx, addReq); err != nil {
+		return nil, fmt.Errorf("failed to add FDB entry to dataplane: %v", err)
+	}
+
+	f.sendNotification(&saipb.FdbEventNotificationData{
+		EventType: saipb.FdbEvent_FDB_EVENT_LEARNED,
+		FdbEntry: &saipb.FdbEntry{
+			SwitchId:   entry.GetSwitchId(),
+			MacAddress: mac,
+			BvId:       entry.GetBvId(),
+		},
+		Attrs: []*saipb.FdbEntryAttribute{
+			{
+				BridgePortId: proto.Uint64(portOID),
+			},
+		},
+	})
+
+	return &saipb.CreateFdbEntryResponse{}, nil
+}
+
+func (f *fdb) RemoveFdbEntry(ctx context.Context, req *saipb.RemoveFdbEntryRequest) (*saipb.RemoveFdbEntryResponse, error) {
+	slog.InfoContext(ctx, "RemoveFdbEntry called", "mac", req.GetEntry().GetMacAddress(), "vlan", req.GetEntry().GetBvId())
+	entry := req.GetEntry()
+	if entry == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "FDB entry is required")
+	}
+
+	mac := entry.GetMacAddress()
+	if len(mac) == 0 {
+		return nil, status.Errorf(codes.InvalidArgument, "MAC address is required")
+	}
+
+	delReq := fwdconfig.TableEntryRemoveRequest(f.dataplane.ID(), FDBTable).AppendEntry(
+		fwdconfig.EntryDesc(fwdconfig.ExactEntry(
+			fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_DST).WithBytes(mac),
+		)),
+	).Build()
+
+	if _, err := f.dataplane.TableEntryRemove(ctx, delReq); err != nil {
+		return nil, fmt.Errorf("failed to remove FDB entry from dataplane: %v", err)
+	}
+
+	f.sendNotification(&saipb.FdbEventNotificationData{
+		EventType: saipb.FdbEvent_FDB_EVENT_AGED,
+		FdbEntry: &saipb.FdbEntry{
+			SwitchId:   entry.GetSwitchId(),
+			MacAddress: mac,
+			BvId:       entry.GetBvId(),
+		},
+	})
+
+	return &saipb.RemoveFdbEntryResponse{}, nil
+}

dataplane/saiserver/ports.go

@@ -16,6 +16,7 @@ package saiserver
 
 import (
 	"context"
+	"encoding/binary"
 	"fmt"
 	"log/slog"
 	"net"
@@ -64,6 +65,7 @@ var getInterface = net.InterfaceByName
 func getPreIngressPipeline() []*fwdpb.ActionDesc {
 	return []*fwdpb.ActionDesc{
 		fwdconfig.Action(fwdconfig.LookupAction(tunTermTable)).Build(),          // Decap the packet if we have a tunnel.
+		fwdconfig.Action(fwdconfig.LookupAction(VlanTable)).Build(),             // Classify incoming packets to VLAN.
 		fwdconfig.Action(fwdconfig.LookupAction(inputIfaceTable)).Build(),       // Match packet to interface.
 		fwdconfig.Action(fwdconfig.LookupAction(IngressVRFTable)).Build(),       // Match interface to VRF.
 		fwdconfig.Action(fwdconfig.LookupAction(PreIngressActionTable)).Build(), // Run pre-ingress actions.
@@ -101,7 +103,9 @@ func getL3Pipeline(skipIPValidation bool) []*fwdpb.ActionDesc {
 func getL2Pipeline() []*fwdpb.ActionDesc {
 	return []*fwdpb.ActionDesc{
 		fwdconfig.Action(fwdconfig.LookupAction(IngressActionTable)).Build(), // Run ingress action.
-		fwdconfig.Action(fwdconfig.LookupAction(outputTable)).Build(),        // Take final decision on forward, drop, or trap.
+		fwdconfig.Action(fwdconfig.BridgeLearnAction(FDBTable)).Build(),      // Learn source MAC into FDB.
+		fwdconfig.Action(fwdconfig.LookupAction(FDBTable)).Build(),           // Match dest MAC in FDB.
+		fwdconfig.Action(fwdconfig.LookupAction(FloodTable)).Build(),         // Match VLAN tag if FDB misses.
 		{
 			ActionType: fwdpb.ActionType_ACTION_TYPE_OUTPUT,
 		},
@@ -301,8 +305,10 @@ func (port *port) CreatePort(ctx context.Context, req *saipb.CreatePortRequest)
 		Update: &fwdpb.PortUpdateDesc{
 			Port: &fwdpb.PortUpdateDesc_Kernel{
 				Kernel: &fwdpb.KernelPortUpdateDesc{
-					Inputs:  getPreIngressPipeline(),
-					Outputs: nil,
+					Inputs: getPreIngressPipeline(),
+					Outputs: []*fwdpb.ActionDesc{
+						fwdconfig.Action(fwdconfig.DecapAction(fwdpb.PacketHeaderId_PACKET_HEADER_ID_ETHERNET_VLAN)).Build(),
+					},
 				},
 			},
 		},
@@ -361,7 +367,7 @@ func (port *port) CreatePort(ctx context.Context, req *saipb.CreatePortRequest)
 		fwdconfig.EntryDesc(fwdconfig.ExactEntry(fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_PACKET_PORT_INPUT).WithUint64(nid.GetNid())))).Build()
 	vlanReq.Entries[0].Actions = []*fwdpb.ActionDesc{
 		fwdconfig.Action(fwdconfig.EncapAction(fwdpb.PacketHeaderId_PACKET_HEADER_ID_ETHERNET_VLAN)).Build(),
-		fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET, fwdpb.PacketFieldNum_PACKET_FIELD_NUM_VLAN_TAG).WithUint64Value(uint64(attrs.GetPortVlanId()))).Build(),
+		fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET, fwdpb.PacketFieldNum_PACKET_FIELD_NUM_VLAN_TAG).WithValue(binary.BigEndian.AppendUint16(nil, uint16(attrs.GetPortVlanId())))).Build(),
 	}
 	if _, err := port.dataplane.TableEntryAdd(ctx, vlanReq); err != nil {
 		return nil, err