Implement L2MC via ACL redirect.

dataplane/forwarding/protocol/attr.go

@@ -191,6 +191,9 @@ var FieldAttr = map[fwdpb.PacketFieldNum]struct {
 	fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID: {
 		Sizes: []int{SizeUint64},
 	},
+	fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID: {
+		Sizes: []int{SizeUint64},
+	},
 	fwdpb.PacketFieldNum_PACKET_FIELD_NUM_POLICER_ID: {
 		Sizes: []int{SizeUint64},
 	},
@@ -232,6 +235,7 @@ var GroupAttr = map[fwdpb.PacketHeaderGroup]struct {
 			fwdpb.PacketFieldNum_PACKET_FIELD_NUM_OUTPUT_IFACE,
 			fwdpb.PacketFieldNum_PACKET_FIELD_NUM_TUNNEL_ID,
 			fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID,
+			fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID,
 			fwdpb.PacketFieldNum_PACKET_FIELD_NUM_POLICER_ID,
 		},
 	},

dataplane/forwarding/protocol/metadata/metadata.go

@@ -46,6 +46,7 @@ type Metadata struct {
 	outputIface    []byte         // L3 output interface id.
 	tunnelID       []byte         // Tunnel ID
 	hostPortID     []byte         // Host port id
+	l2mcGroupID    []byte         // L2MC Group ID
 	policer        []byte         // Policer ID
 	desc           *protocol.Desc // Protocol descriptor.
 }
@@ -127,6 +128,8 @@ func (m *Metadata) Field(id fwdpacket.FieldID) ([]byte, error) {
 		return m.tunnelID, nil
 	case fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID:
 		return m.hostPortID, nil
+	case fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID:
+		return m.l2mcGroupID, nil
 	case fwdpb.PacketFieldNum_PACKET_FIELD_NUM_POLICER_ID:
 		return m.policer, nil
 
@@ -235,6 +238,9 @@ func (m *Metadata) updateSet(id fwdpacket.FieldID, arg []byte) (bool, error) {
 	case fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID:
 		m.hostPortID = arg
 		return true, nil
+	case fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID:
+		m.l2mcGroupID = arg
+		return true, nil
 	case fwdpb.PacketFieldNum_PACKET_FIELD_NUM_POLICER_ID:
 		m.policer = arg
 		return true, nil
@@ -293,6 +299,7 @@ func parse(frame *frame.Frame, desc *protocol.Desc) (protocol.Handler, fwdpb.Pac
 		outputIface: make([]byte, protocol.FieldAttr[fwdpb.PacketFieldNum_PACKET_FIELD_NUM_OUTPUT_IFACE].DefaultSize),
 		tunnelID:    make([]byte, protocol.FieldAttr[fwdpb.PacketFieldNum_PACKET_FIELD_NUM_TUNNEL_ID].DefaultSize),
 		hostPortID:  make([]byte, protocol.FieldAttr[fwdpb.PacketFieldNum_PACKET_FIELD_NUM_HOST_PORT_ID].DefaultSize),
+		l2mcGroupID: make([]byte, protocol.FieldAttr[fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID].DefaultSize),
 		policer:     make([]byte, protocol.FieldAttr[fwdpb.PacketFieldNum_PACKET_FIELD_NUM_POLICER_ID].DefaultSize),
 		attribute32: make(map[uint8][]byte),
 		attribute24: make(map[uint8][]byte),

dataplane/saiserver/BUILD

@@ -43,6 +43,7 @@ go_test(
     srcs = [
         "acl_test.go",
         "hostif_test.go",
+        "l2mc_test.go",
         "policer_test.go",
         "ports_test.go",
         "routing_test.go",

dataplane/saiserver/acl.go

@@ -328,6 +328,18 @@ func (a *acl) CreateAclEntry(ctx context.Context, req *saipb.CreateAclEntryReque
 			},
 		})
 	}
+	if req.ActionRedirect != nil {
+		switch typ := a.mgr.GetType(fmt.Sprint(req.GetActionRedirect().GetOid())); typ {
+		case saipb.ObjectType_OBJECT_TYPE_L2MC_GROUP:
+			aReq.Actions = append(aReq.Actions, []*fwdpb.ActionDesc{
+				fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET,
+					fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID).WithUint64Value(req.GetActionRedirect().GetOid())).Build(),
+				fwdconfig.Action(fwdconfig.LookupAction(L2MCGroupTable)).Build(), // Check L2MC group.
+			}...)
+		default:
+			return nil, status.Errorf(codes.InvalidArgument, "type %q is not supported; only support L2MC Group for ACL Redirect for now", typ.String())
+		}
+	}
 	if req.ActionSetPolicer != nil {
 		aReq.Actions = append(aReq.Actions,
 			fwdconfig.Action(fwdconfig.UpdateAction(fwdpb.UpdateType_UPDATE_TYPE_SET, fwdpb.PacketFieldNum_PACKET_FIELD_NUM_POLICER_ID).

dataplane/saiserver/l2.go

@@ -16,48 +16,190 @@ package saiserver
 
 import (
 	"context"
+	"encoding/binary"
+	"fmt"
 
 	"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"
 	"github.com/openconfig/lemming/dataplane/saiserver/attrmgr"
+	fwdpb "github.com/openconfig/lemming/proto/forwarding"
 
 	saipb "github.com/openconfig/lemming/dataplane/proto/sai"
 )
 
+type l2mcGroupMember struct {
+	oid      uint64
+	groupId  uint64
+	outputId uint64
+}
+
 type l2mcGroup struct {
 	saipb.UnimplementedL2McGroupServer
 	mgr       *attrmgr.AttrMgr
 	dataplane switchDataplaneAPI
+	groups    map[uint64]map[uint64]*l2mcGroupMember // map[group id][member oid]
 }
 
 func newL2mcGroup(mgr *attrmgr.AttrMgr, dataplane switchDataplaneAPI, s *grpc.Server) *l2mcGroup {
 	mg := &l2mcGroup{
 		mgr:       mgr,
 		dataplane: dataplane,
+		groups:    map[uint64]map[uint64]*l2mcGroupMember{},
 	}
 	saipb.RegisterL2McGroupServer(s, mg)
 	return mg
 }
 
-// TODO: Implement this.
-func (mg *l2mcGroup) CreateL2McGroup(context.Context, *saipb.CreateL2McGroupRequest) (*saipb.CreateL2McGroupResponse, error) {
+func (mg *l2mcGroup) CreateL2McGroup(ctx context.Context, req *saipb.CreateL2McGroupRequest) (*saipb.CreateL2McGroupResponse, error) {
 	id := mg.mgr.NextID()
+	// Update internal data
+	mg.groups[id] = map[uint64]*l2mcGroupMember{}
+	// Update attributes
+	l2mcAttrs := &saipb.L2McGroupAttribute{
+		L2McOutputCount: proto.Uint32(0),
+		L2McMemberList:  []uint64{},
+	}
+	mg.mgr.StoreAttributes(id, l2mcAttrs)
 	return &saipb.CreateL2McGroupResponse{Oid: id}, nil
 }
 
-// TODO: Implement this.
-func (mg *l2mcGroup) CreateL2McGroupMember(context.Context, *saipb.CreateL2McGroupMemberRequest) (*saipb.CreateL2McGroupMemberResponse, error) {
+func (mg *l2mcGroup) portNidFromBrirdgeId(ctx context.Context, outputId uint64) (uint64, error) {
+	req := &saipb.GetBridgePortAttributeRequest{Oid: outputId, AttrType: []saipb.BridgePortAttr{saipb.BridgePortAttr_BRIDGE_PORT_ATTR_PORT_ID}}
+	resp := &saipb.GetBridgePortAttributeResponse{}
+	if err := mg.mgr.PopulateAttributes(req, resp); err != nil {
+		return 0, fmt.Errorf("failed to populate OutputId (oid=%d): %v", outputId, err)
+	}
+	if resp.GetAttr().PortId == nil {
+		return 0, fmt.Errorf("cannot find portId for bridge port %q", outputId)
+	}
+	return resp.GetAttr().GetPortId(), nil
+}
+
+// updateGroupMember updates the member of a L2MC group.
+// If m is nil, remove mid from the group(key: group id), otherwise add m to groups with mid as the key.
+func (mg *l2mcGroup) updateGroupMember(ctx context.Context, gid, mid uint64, m *l2mcGroupMember) error {
+	if m == nil {
+		// Remove the member.
+		delete(mg.groups[gid], mid)
+	} else {
+		// Add a member.
+		mg.groups[gid][mid] = m
+		attr := &saipb.L2McGroupMemberAttribute{
+			L2McGroupId:  &m.groupId,
+			L2McOutputId: &m.outputId,
+		}
+		mg.mgr.StoreAttributes(mid, attr)
+	}
+	mList := []uint64{}
+	for _, m := range mg.groups[gid] {
+		mList = append(mList, m.oid)
+	}
+	gAttr := &saipb.L2McGroupAttribute{}
+	gAttr.L2McMemberList = mList
+	cnt := uint32(len(mList))
+	gAttr.L2McOutputCount = &cnt
+	mg.mgr.StoreAttributes(gid, gAttr)
+
+	var actions []*fwdpb.ActionDesc
+	for _, member := range mg.groups[gid] {
+		portId, err := mg.portNidFromBrirdgeId(ctx, member.outputId)
+		if err != nil {
+			return err
+		}
+		actions = append(actions, &fwdpb.ActionDesc{
+			ActionType: fwdpb.ActionType_ACTION_TYPE_MIRROR,
+			Action: &fwdpb.ActionDesc_Mirror{
+				Mirror: &fwdpb.MirrorActionDesc{
+					PortId: &fwdpb.PortId{
+						ObjectId: &fwdpb.ObjectId{
+							Id: fmt.Sprint(portId),
+						},
+					},
+				},
+			},
+		})
+	}
+	entries := &fwdpb.TableEntryAddRequest{
+		ContextId: &fwdpb.ContextId{Id: mg.dataplane.ID()},
+		TableId: &fwdpb.TableId{
+			ObjectId: &fwdpb.ObjectId{
+				Id: L2MCGroupTable,
+			},
+		},
+		Entries: []*fwdpb.TableEntryAddRequest_Entry{{
+			EntryDesc: &fwdpb.EntryDesc{
+				Entry: &fwdpb.EntryDesc_Exact{
+					Exact: &fwdpb.ExactEntryDesc{
+						Fields: []*fwdpb.PacketFieldBytes{{
+							Bytes: binary.BigEndian.AppendUint64(nil, gid),
+							FieldId: &fwdpb.PacketFieldId{
+								Field: &fwdpb.PacketField{
+									FieldNum: fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID,
+								},
+							},
+						}},
+					},
+				},
+			},
+			Actions: actions,
+		}},
+	}
+	_, err := mg.dataplane.TableEntryAdd(ctx, entries)
+	return err
+}
+
+func (mg *l2mcGroup) CreateL2McGroupMember(ctx context.Context, req *saipb.CreateL2McGroupMemberRequest) (*saipb.CreateL2McGroupMemberResponse, error) {
+	if mg.groups[req.GetL2McGroupId()] == nil {
+		return nil, status.Errorf(codes.FailedPrecondition, "cannot find L2MC group with group ID=%d", req.GetL2McGroupId())
+	}
+	if _, err := mg.portNidFromBrirdgeId(ctx, req.GetL2McOutputId()); err != nil {
+		return nil, err
+	}
 	id := mg.mgr.NextID()
+	if err := mg.updateGroupMember(ctx, req.GetL2McGroupId(), id, &l2mcGroupMember{
+		oid:      id,
+		outputId: req.GetL2McOutputId(),
+		groupId:  req.GetL2McGroupId(),
+	}); err != nil {
+		return nil, err
+	}
 	return &saipb.CreateL2McGroupMemberResponse{Oid: id}, nil
 }
 
-// TODO: Implement this.
-func (mg *l2mcGroup) RemoveL2McGroup(context.Context, *saipb.RemoveL2McGroupRequest) (*saipb.RemoveL2McGroupResponse, error) {
+func (mg *l2mcGroup) RemoveL2McGroup(ctx context.Context, req *saipb.RemoveL2McGroupRequest) (*saipb.RemoveL2McGroupResponse, error) {
+	if mg.groups[req.GetOid()] == nil {
+		return nil, status.Errorf(codes.FailedPrecondition, "cannot find L2MC group with group ID=%d", req.GetOid())
+	}
+	delete(mg.groups, req.GetOid())
+	if _, err := mg.dataplane.TableEntryRemove(ctx, fwdconfig.TableEntryRemoveRequest(mg.dataplane.ID(), L2MCGroupTable).AppendEntry(
+		fwdconfig.EntryDesc(fwdconfig.ExactEntry(fwdconfig.PacketFieldBytes(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_L2MC_GROUP_ID).WithUint64(req.GetOid())))).Build()); err != nil {
+		return nil, err
+	}
 	return &saipb.RemoveL2McGroupResponse{}, nil
 }
 
-// TODO: Implement this.
-func (mg *l2mcGroup) RemoveL2McGroupMember(context.Context, *saipb.RemoveL2McGroupMemberRequest) (*saipb.RemoveL2McGroupMemberResponse, error) {
+func (mg *l2mcGroup) RemoveL2McGroupMember(ctx context.Context, req *saipb.RemoveL2McGroupMemberRequest) (*saipb.RemoveL2McGroupMemberResponse, error) {
+	locateMember := func(oid uint64) *l2mcGroupMember {
+		for _, members := range mg.groups {
+			for k, v := range members {
+				if k == oid {
+					return v
+				}
+			}
+		}
+		return nil
+	}
+	m := locateMember(req.GetOid())
+	if m == nil {
+		return nil, status.Errorf(codes.FailedPrecondition, "cannot find L2MC group member with OID %d", req.GetOid())
+	}
+	if err := mg.updateGroupMember(ctx, m.groupId, m.oid, nil); err != nil {
+		return nil, err
+	}
 	return &saipb.RemoveL2McGroupMemberResponse{}, nil
 }