Implement sai port stats (#314)

* Implement sai port stats

* fixes

dataplane/forwarding/fwdaction/actions/mirror_test.go

@@ -197,6 +197,7 @@ func TestMirror(t *testing.T) {
 		mirrored.EXPECT().Attributes().Return(nil).AnyTimes()
 		mirrored.EXPECT().Log().Return(testr.New(t)).AnyTimes()
 		mirrored.EXPECT().LogMsgs().Return(nil).AnyTimes()
+		mirrored.EXPECT().Field(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_SRC, 0)).Return([]byte{0}, nil).AnyTimes()
 		mirrored.EXPECT().Update(opFID, fwdpacket.OpSet, gomock.Any()).Return(nil).AnyTimes()
 		mirrored.EXPECT().Update(inFID, fwdpacket.OpSet, gomock.Any()).Return(nil).AnyTimes()
 		mirrored.EXPECT().Update(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_DST, 0),
@@ -210,6 +211,7 @@ func TestMirror(t *testing.T) {
 		original.EXPECT().Mirror(fields).Return(mirrored, nil).AnyTimes()
 		original.EXPECT().Attributes().Return(nil).AnyTimes()
 		original.EXPECT().Log().Return(testr.New(t)).AnyTimes()
+		original.EXPECT().Field(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_SRC, 0)).Return([]byte{0}, nil).AnyTimes()
 		original.EXPECT().Field(opFID).Return(make([]byte, protocol.SizeUint64), nil).AnyTimes()
 		original.EXPECT().Field(inFID).Return(make([]byte, protocol.SizeUint64), nil).AnyTimes()
 

dataplane/forwarding/fwdport/port.go

@@ -63,6 +63,10 @@ var CounterList = []fwdpb.CounterId{
 	fwdpb.CounterId_COUNTER_ID_TX_DROP_OCTETS,
 	fwdpb.CounterId_COUNTER_ID_TX_ADMIN_DROP_PACKETS,
 	fwdpb.CounterId_COUNTER_ID_TX_ADMIN_DROP_OCTETS,
+	fwdpb.CounterId_COUNTER_ID_TX_UCAST_PACKETS,
+	fwdpb.CounterId_COUNTER_ID_TX_NON_UCAST_PACKETS,
+	fwdpb.CounterId_COUNTER_ID_RX_UCAST_PACKETS,
+	fwdpb.CounterId_COUNTER_ID_RX_NON_UCAST_PACKETS,
 }
 
 // A Port is an entry or exit point within the forwarding plane. Each port
@@ -236,6 +240,16 @@ func Input(port Port, packet fwdpacket.Packet, dir fwdpb.PortAction, ctx *fwdcon
 
 	Increment(port, packet.Length(), fwdpb.CounterId_COUNTER_ID_RX_PACKETS, fwdpb.CounterId_COUNTER_ID_RX_OCTETS)
 	SetInputPort(packet, port)
+	mac, err := packet.Field(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_DST, 0))
+	if err != nil {
+		Increment(port, packet.Length(), fwdpb.CounterId_COUNTER_ID_RX_ERROR_PACKETS, fwdpb.CounterId_COUNTER_ID_RX_ERROR_OCTETS)
+		return err
+	}
+	if mac[0]%2 == 0 { // Unicast address is when is least significant bit of the 1st octet is 0.
+		port.Increment(fwdpb.CounterId_COUNTER_ID_RX_UCAST_PACKETS, 1)
+	} else {
+		port.Increment(fwdpb.CounterId_COUNTER_ID_RX_NON_UCAST_PACKETS, 1)
+	}
 
 	packet.Log().V(3).Info("input packet", "port", port.ID(), "frame", fwdpacket.IncludeFrameInLog)
 	state, err := fwdaction.ProcessPacket(packet, port.Actions(dir), port)
@@ -279,6 +293,16 @@ func Output(port Port, packet fwdpacket.Packet, dir fwdpb.PortAction, _ *fwdcont
 	}()
 	Increment(port, packet.Length(), fwdpb.CounterId_COUNTER_ID_TX_PACKETS, fwdpb.CounterId_COUNTER_ID_TX_OCTETS)
 	SetOutputPort(packet, port)
+	mac, err := packet.Field(fwdpacket.NewFieldIDFromNum(fwdpb.PacketFieldNum_PACKET_FIELD_NUM_ETHER_MAC_SRC, 0))
+	if err != nil {
+		Increment(port, packet.Length(), fwdpb.CounterId_COUNTER_ID_TX_ERROR_PACKETS, fwdpb.CounterId_COUNTER_ID_TX_ERROR_OCTETS)
+		return err
+	}
+	if mac[0]%2 == 0 { // Unicast address is when is least significant bit of the 1st octet is 0.
+		port.Increment(fwdpb.CounterId_COUNTER_ID_TX_UCAST_PACKETS, 1)
+	} else {
+		port.Increment(fwdpb.CounterId_COUNTER_ID_TX_NON_UCAST_PACKETS, 1)
+	}
 
 	packet.Log().V(3).Info("output packet", "frame", fwdpacket.IncludeFrameInLog)
 	state, err := fwdaction.ProcessPacket(packet, port.Actions(dir), port)

dataplane/standalone/saiserver/attrmgr/attrmgr.go

@@ -113,7 +113,7 @@ func (mgr *AttrMgr) Interceptor(ctx context.Context, req any, info *grpc.UnarySe
 			return respMsg, nil
 		}
 		mgr.storeAttributes(id, reqMsg)
-	} else if strings.Contains(info.FullMethod, "Get") {
+	} else if strings.Contains(info.FullMethod, "Get") && strings.Contains(info.FullMethod, "Attribute") {
 		if err := mgr.PopulateAttributes(reqMsg, respMsg); err != nil {
 			return nil, err
 		}

dataplane/standalone/saiserver/attrmgr/attrmgr_test.go

@@ -122,6 +122,18 @@ func TestInterceptor(t *testing.T) {
 				PreIngressAcl: proto.Uint64(300),
 			},
 		},
+	}, {
+		desc: "stats request",
+		info: &grpc.UnaryServerInfo{FullMethod: "/lemming.dataplane.sai.Port/GetPortStats"},
+		attrs: map[string]map[int32]*protoreflect.Value{
+			"10": {
+				int32(saipb.SwitchAttr_SWITCH_ATTR_CPU_PORT): ptrToValue(protoreflect.ValueOfUint64(100)),
+			},
+		},
+		req: &saipb.GetPortStatsRequest{
+			Oid: 10,
+		},
+		want: &saipb.GetPortStatsResponse{},
 	}}
 	for _, tt := range tests {
 		t.Run(tt.desc, func(t *testing.T) {

dataplane/standalone/saiserver/ports.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/openconfig/lemming/dataplane/standalone/saiserver/attrmgr"
 
+	log "github.com/golang/glog"
+
 	saipb "github.com/openconfig/lemming/dataplane/standalone/proto"
 	dpb "github.com/openconfig/lemming/proto/dataplane"
 	fwdpb "github.com/openconfig/lemming/proto/forwarding"
@@ -35,6 +37,7 @@ type portDataplaneAPI interface {
 	ID() string
 	CreatePort(ctx context.Context, req *dpb.CreatePortRequest) (*dpb.CreatePortResponse, error)
 	PortState(ctx context.Context, req *fwdpb.PortStateRequest) (*fwdpb.PortStateReply, error)
+	ObjectCounters(context.Context, *fwdpb.ObjectCountersRequest) (*fwdpb.ObjectCountersReply, error)
 }
 
 func newPort(mgr *attrmgr.AttrMgr, dataplane portDataplaneAPI, s *grpc.Server) *port {
@@ -257,6 +260,51 @@ func (port *port) SetPortAttribute(ctx context.Context, req *saipb.SetPortAttrib
 	return &saipb.SetPortAttributeResponse{}, nil
 }
 
+// GetPortStats returns the stats for a port.
+func (port *port) GetPortStats(ctx context.Context, req *saipb.GetPortStatsRequest) (*saipb.GetPortStatsResponse, error) {
+	resp := &saipb.GetPortStatsResponse{}
+	counters, err := port.dataplane.ObjectCounters(ctx, &fwdpb.ObjectCountersRequest{
+		ContextId: &fwdpb.ContextId{Id: port.dataplane.ID()},
+		ObjectId:  &fwdpb.ObjectId{Id: fmt.Sprint(req.GetOid())},
+	})
+	if err != nil {
+		return nil, err
+	}
+	counterMap := map[fwdpb.CounterId]uint64{}
+	for _, c := range counters.GetCounters() {
+		counterMap[c.GetId()] = c.GetValue()
+	}
+
+	for _, id := range req.GetCounterIds() {
+		switch id {
+		case saipb.PortStat_PORT_STAT_IF_IN_UCAST_PKTS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_RX_UCAST_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_IN_NON_UCAST_PKTS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_RX_NON_UCAST_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_IN_ERRORS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_RX_ERROR_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_OUT_UCAST_PKTS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_TX_UCAST_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_OUT_NON_UCAST_PKTS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_TX_NON_UCAST_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_OUT_ERRORS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_TX_ERROR_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_IN_OCTETS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_RX_OCTETS])
+		case saipb.PortStat_PORT_STAT_IF_OUT_OCTETS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_TX_OCTETS])
+		case saipb.PortStat_PORT_STAT_IF_IN_DISCARDS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_RX_DROP_PACKETS])
+		case saipb.PortStat_PORT_STAT_IF_OUT_DISCARDS:
+			resp.Values = append(resp.Values, counterMap[fwdpb.CounterId_COUNTER_ID_TX_DROP_PACKETS])
+		default:
+			resp.Values = append(resp.Values, 0)
+			log.Infof("unknown port stat: %v", id)
+		}
+	}
+	return resp, nil
+}
+
 func (port *port) Reset() {
 	port.portToEth = make(map[uint64]string)
 	port.nextEth = 1

dataplane/standalone/saiserver/ports_test.go

@@ -17,6 +17,7 @@ package saiserver
 import (
 	"context"
 	"fmt"
+	"io"
 	"net"
 	"testing"
 
@@ -251,6 +252,88 @@ func TestSetPortAttribute(t *testing.T) {
 	}
 }
 
+func TestGetPortStats(t *testing.T) {
+	tests := []struct {
+		desc         string
+		req          *saipb.GetPortStatsRequest
+		counterReply *fwdpb.ObjectCountersReply
+		want         *saipb.GetPortStatsResponse
+		wantErr      string
+	}{{
+		desc: "all stats",
+		req: &saipb.GetPortStatsRequest{
+			Oid: 1,
+			CounterIds: []saipb.PortStat{
+				saipb.PortStat_PORT_STAT_IF_IN_UCAST_PKTS,
+				saipb.PortStat_PORT_STAT_IF_IN_NON_UCAST_PKTS,
+				saipb.PortStat_PORT_STAT_IF_IN_ERRORS,
+				saipb.PortStat_PORT_STAT_IF_OUT_UCAST_PKTS,
+				saipb.PortStat_PORT_STAT_IF_OUT_NON_UCAST_PKTS,
+				saipb.PortStat_PORT_STAT_IF_OUT_ERRORS,
+				saipb.PortStat_PORT_STAT_IF_IN_OCTETS,
+				saipb.PortStat_PORT_STAT_IF_OUT_OCTETS,
+				saipb.PortStat_PORT_STAT_IF_IN_DISCARDS,
+				saipb.PortStat_PORT_STAT_IF_OUT_DISCARDS,
+			},
+		},
+		counterReply: &fwdpb.ObjectCountersReply{
+			Counters: []*fwdpb.Counter{{
+				Id:    fwdpb.CounterId_COUNTER_ID_TX_DROP_PACKETS,
+				Value: 1,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_RX_DROP_PACKETS,
+				Value: 2,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_TX_OCTETS,
+				Value: 3,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_RX_OCTETS,
+				Value: 4,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_TX_ERROR_PACKETS,
+				Value: 5,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_TX_NON_UCAST_PACKETS,
+				Value: 6,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_TX_UCAST_PACKETS,
+				Value: 7,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_RX_ERROR_PACKETS,
+				Value: 8,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_RX_NON_UCAST_PACKETS,
+				Value: 9,
+			}, {
+				Id:    fwdpb.CounterId_COUNTER_ID_RX_UCAST_PACKETS,
+				Value: 10,
+			}},
+		},
+		want: &saipb.GetPortStatsResponse{
+			Values: []uint64{10, 9, 8, 7, 6, 5, 4, 3, 2, 1},
+		},
+	}}
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			dplane := &fakePortDataplaneAPI{
+				counterReplies: []*fwdpb.ObjectCountersReply{tt.counterReply},
+			}
+			c, _, stopFn := newTestPort(t, dplane)
+			defer stopFn()
+			got, gotErr := c.GetPortStats(context.TODO(), tt.req)
+			if diff := errdiff.Check(gotErr, tt.wantErr); diff != "" {
+				t.Fatalf("SetPortAttribute() unexpected err: %s", diff)
+			}
+			if gotErr != nil {
+				return
+			}
+			if d := cmp.Diff(got, tt.want, protocmp.Transform()); d != "" {
+				t.Errorf("SetPortAttribute() failed: diff(-got,+want)\n:%s", d)
+			}
+		})
+	}
+}
+
 func TestCreateHostif(t *testing.T) {
 	tests := []struct {
 		desc            string
@@ -380,8 +463,10 @@ func newTestHostif(t testing.TB, api portDataplaneAPI) (saipb.HostifClient, *att
 }
 
 type fakePortDataplaneAPI struct {
-	gotPortStateReq  []*fwdpb.PortStateRequest
-	gotPortCreateReq []*dpb.CreatePortRequest
+	gotPortStateReq   []*fwdpb.PortStateRequest
+	gotPortCreateReq  []*dpb.CreatePortRequest
+	counterReplies    []*fwdpb.ObjectCountersReply
+	counterRepliesIdx int
 }
 
 func (f *fakePortDataplaneAPI) ID() string {
@@ -397,3 +482,12 @@ func (f *fakePortDataplaneAPI) PortState(_ context.Context, req *fwdpb.PortState
 	f.gotPortStateReq = append(f.gotPortStateReq, req)
 	return nil, nil
 }
+
+func (f *fakePortDataplaneAPI) ObjectCounters(context.Context, *fwdpb.ObjectCountersRequest) (*fwdpb.ObjectCountersReply, error) {
+	if f.counterRepliesIdx > len(f.counterReplies) {
+		return nil, io.EOF
+	}
+	r := f.counterReplies[f.counterRepliesIdx]
+	f.counterRepliesIdx++
+	return r, nil
+}

proto/forwarding/forwarding_common.proto

@@ -269,6 +269,10 @@ enum CounterId {
       35;  // Number of packet debugged after receiving.
   COUNTER_ID_RX_DEBUG_OCTETS =
       36;                // Number of octets debugged after receiving.
+  COUNTER_ID_RX_UCAST_PACKETS = 37;
+  COUNTER_ID_RX_NON_UCAST_PACKETS = 38;
+  COUNTER_ID_TX_UCAST_PACKETS = 39;
+  COUNTER_ID_TX_NON_UCAST_PACKETS = 40;
   COUNTER_ID_MAX = 255;  // Maximum counter id.
 }