Add new fake port type

dataplane/forwarding/fwdport/ports/BUILD

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "cpu.go",
         "doc.go",
+        "fake.go",
         "group.go",
         "kernel.go",
         "tap.go",
@@ -27,6 +28,7 @@ go_library(
         "@com_github_google_gopacket//:gopacket",
         "@com_github_google_gopacket//afpacket",
         "@com_github_google_gopacket//layers",
+        "@com_github_google_gopacket//pcapgo",
         "@com_github_vishvananda_netlink//:netlink",
         "@org_golang_x_sys//unix",
     ],

dataplane/forwarding/fwdport/ports/fake.go

@@ -0,0 +1,203 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ports
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	log "github.com/golang/glog"
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcapgo"
+
+	"github.com/openconfig/lemming/dataplane/forwarding/fwdaction"
+	"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"
+	"github.com/openconfig/lemming/internal/debug"
+	fwdpb "github.com/openconfig/lemming/proto/forwarding"
+)
+
+func init() {
+	fwdport.Register(fwdpb.PortType_PORT_TYPE_FAKE, fakeBuilder{})
+}
+
+// fakePort is a ports that receives from and writes a linux network device.
+type fakePort struct {
+	fwdobject.Base
+	packetSrc *pcapgo.NgReader
+	packetDst *pcapgo.NgWriter
+	input     fwdaction.Actions
+	output    fwdaction.Actions
+	ctx       *fwdcontext.Context // Forwarding context containing the port
+}
+
+func (p *fakePort) String() string {
+	desc := fmt.Sprintf("Type=%v;<Input=%v>;<Output=%v>", fwdpb.PortType_PORT_TYPE_KERNEL, p.input, p.output)
+	if state, err := p.State(nil); err == nil {
+		desc += fmt.Sprintf("<State=%v>;", state)
+	}
+	return desc
+}
+
+func (p *fakePort) Cleanup() {
+	p.input.Cleanup()
+	p.output.Cleanup()
+	p.input = nil
+	p.output = nil
+}
+
+// Update updates the actions of the port.
+func (p *fakePort) Update(upd *fwdpb.PortUpdateDesc) error {
+	var err error
+	defer func() {
+		if err != nil {
+			p.Cleanup()
+		}
+	}()
+	fakeUpd, ok := upd.Port.(*fwdpb.PortUpdateDesc_Kernel)
+	if !ok {
+		return fmt.Errorf("invalid type for port update")
+	}
+
+	// Acquire new actions before releasing the old ones.
+	if p.input, err = fwdaction.NewActions(fakeUpd.Kernel.GetInputs(), p.ctx); err != nil {
+		return fmt.Errorf("ports: input actions for port %v failed, err %v", p, err)
+	}
+	if p.output, err = fwdaction.NewActions(fakeUpd.Kernel.GetOutputs(), p.ctx); err != nil {
+		return fmt.Errorf("ports: output actions for port %v failed, err %v", p, err)
+	}
+	return nil
+}
+
+func (p *fakePort) process() {
+	src := gopacket.NewPacketSource(p.packetSrc, layers.LinkTypeEthernet)
+	go func() {
+		for {
+			select {
+			case pkt, ok := <-src.Packets():
+				if !ok {
+					log.Warning("src chan closed")
+					return
+				}
+				fwdPkt, err := fwdpacket.New(fwdpb.PacketHeaderId_PACKET_HEADER_ID_ETHERNET, pkt.Data())
+				if err != nil {
+					log.Warningf("failed to create new packet: %v", err)
+					log.V(1).Info(pkt.Dump())
+					fwdport.Increment(p, len(pkt.Data()), fwdpb.CounterId_COUNTER_ID_RX_BAD_PACKETS, fwdpb.CounterId_COUNTER_ID_RX_BAD_OCTETS)
+					continue
+				}
+				fwdPkt.Debug(debug.ExternalPortPacketTrace)
+				fwdPkt.Log().V(2).Info("input packet", "port", p.ID(), "frame", fwdpacket.IncludeFrameInLog)
+				fwdport.Process(p, fwdPkt, fwdpb.PortAction_PORT_ACTION_INPUT, p.ctx, "Kernel")
+			}
+		}
+	}()
+}
+
+var (
+	// Stubs for tests
+	createFile = func(filename string) (io.Writer, error) {
+		return os.Create(filename)
+	}
+
+	openFile = func(filename string) (io.Reader, error) {
+		return os.Open(filename)
+	}
+	timeNow = func() time.Time {
+		return time.Now()
+	}
+)
+
+// Write writes a packet out. If successful, the port returns
+// fwdaction.CONSUME.
+func (p *fakePort) Write(packet fwdpacket.Packet) (fwdaction.State, error) {
+	if err := p.packetDst.WritePacket(gopacket.CaptureInfo{
+		Timestamp:     timeNow(),
+		CaptureLength: packet.Length(),
+		Length:        packet.Length(),
+	}, packet.Frame()); err != nil {
+		return fwdaction.DROP, fmt.Errorf("failed to write eth packet: %v", err)
+	}
+	return fwdaction.CONSUME, nil
+}
+
+// Actions returns the port actions of the specified type
+func (p *fakePort) Actions(dir fwdpb.PortAction) fwdaction.Actions {
+	switch dir {
+	case fwdpb.PortAction_PORT_ACTION_INPUT:
+		return p.input
+	case fwdpb.PortAction_PORT_ACTION_OUTPUT:
+		return p.output
+	}
+	return nil
+}
+
+// State returns the state of the port.
+func (p *fakePort) State(pi *fwdpb.PortInfo) (*fwdpb.PortStateReply, error) {
+	return &fwdpb.PortStateReply{
+		Status: &fwdpb.PortInfo{
+			OperStatus:  fwdpb.PortState_PORT_STATE_ENABLED_UP,
+			AdminStatus: fwdpb.PortState_PORT_STATE_ENABLED_UP,
+		},
+	}, nil
+}
+
+type fakeBuilder struct{}
+
+// Build creates a new port.
+func (fakeBuilder) Build(portDesc *fwdpb.PortDesc, ctx *fwdcontext.Context) (fwdport.Port, error) {
+	fp, ok := portDesc.Port.(*fwdpb.PortDesc_Fake)
+	if !ok {
+		return nil, fmt.Errorf("invalid port type in proto")
+	}
+
+	inFile, err := openFile(fp.Fake.InFile)
+	if err != nil {
+		return nil, err
+	}
+
+	r, err := pcapgo.NewNgReader(inFile, pcapgo.DefaultNgReaderOptions)
+	if err != nil {
+		return nil, err
+	}
+
+	outFile, err := createFile(fp.Fake.OutFile)
+	if err != nil {
+		return nil, err
+	}
+
+	w, err := pcapgo.NewNgWriter(outFile, layers.LinkTypeEthernet)
+	if err != nil {
+		return nil, err
+	}
+
+	p := &fakePort{
+		ctx:       ctx,
+		packetSrc: r,
+		packetDst: w,
+	}
+	list := append(fwdport.CounterList, fwdaction.CounterList...)
+	if err := p.InitCounters("", list...); err != nil {
+		return nil, err
+	}
+
+	p.process()
+	return p, nil
+}

dataplane/forwarding/fwdport/ports/fake_test.go

@@ -0,0 +1,147 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ports
+
+import (
+	"bytes"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/gopacket/layers"
+	"github.com/google/gopacket/pcapgo"
+	"github.com/openconfig/gnmi/errdiff"
+
+	fwdpb "github.com/openconfig/lemming/proto/forwarding"
+)
+
+const (
+	fileHeader = "0a0d0d0a440000004d3c2b1a01000000ffffffffffffffff04000800676f7061636b657402000500616d643634000000030005006c696e757800000000000000440000000100000038000000010000000000000002000500696e7466300000000c0005006c696e757800000009000100090000000000000038000000"
+)
+
+func writeHex(t testing.TB, w io.Writer, hexStr string) []byte {
+	h, err := hex.DecodeString(hexStr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := w.Write(h); err != nil {
+		t.Fatal(err)
+	}
+	return h
+}
+
+func TestFakeBuild(t *testing.T) {
+	tests := []struct {
+		desc       string
+		openErr    error
+		createErr  error
+		inDesc     *fwdpb.PortDesc
+		inContents string
+		wantErr    string
+	}{{
+		desc:    "wrong desc type",
+		inDesc:  &fwdpb.PortDesc{Port: &fwdpb.PortDesc_Cpu{}},
+		wantErr: "invalid port type in proto",
+	}, {
+		desc:    "open error",
+		inDesc:  &fwdpb.PortDesc{Port: &fwdpb.PortDesc_Fake{Fake: &fwdpb.FakePortDesc{}}},
+		openErr: fmt.Errorf("open error"),
+		wantErr: "open error",
+	}, {
+		desc:    "new reader err",
+		inDesc:  &fwdpb.PortDesc{Port: &fwdpb.PortDesc_Fake{Fake: &fwdpb.FakePortDesc{}}},
+		wantErr: "EOF",
+	}, {
+		desc:       "create err",
+		inDesc:     &fwdpb.PortDesc{Port: &fwdpb.PortDesc_Fake{Fake: &fwdpb.FakePortDesc{}}},
+		inContents: fileHeader,
+		createErr:  fmt.Errorf("create error"),
+	}, {
+		desc:       "sucess",
+		inDesc:     &fwdpb.PortDesc{Port: &fwdpb.PortDesc_Fake{Fake: &fwdpb.FakePortDesc{}}},
+		inContents: fileHeader,
+	}}
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			in := &bytes.Buffer{}
+			writeHex(t, in, tt.inContents)
+			out := &bytes.Buffer{}
+			openFile = func(filename string) (io.Reader, error) {
+				return in, tt.openErr
+			}
+			createFile = func(filename string) (io.Writer, error) {
+				return out, tt.openErr
+			}
+			_, err := (fakeBuilder{}).Build(tt.inDesc, nil)
+			if d := errdiff.Check(err, tt.wantErr); d != "" {
+				t.Fatalf("Build() unexpected error diff: %s", d)
+			}
+		})
+	}
+}
+
+type fakeWriter struct {
+	bytes.Buffer
+	writeErr error
+}
+
+func (fw *fakeWriter) Write(data []byte) (int, error) {
+	if fw.writeErr == nil {
+		return fw.Buffer.Write(data)
+	}
+	return 0, fw.writeErr
+}
+
+func TestFakeWrite(t *testing.T) {
+	tests := []struct {
+		desc     string
+		writeErr error
+		wantErr  string
+		want     string
+	}{{
+		desc:     "write error",
+		writeErr: fmt.Errorf("write err"),
+		wantErr:  "write err",
+	}, {
+		desc: "success", // packet metadata                                           // packet content
+		want: fileHeader + "060000005c000000000000000000000000ca9a3b3c0000003c000000" + "1111111111111111111111110000686900000000000000000000000000000000000000000000000000000000000000000000000000000000000000005c000000",
+	}}
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			fw := &fakeWriter{
+				writeErr: tt.writeErr,
+			}
+			w, err := pcapgo.NewNgWriter(fw, layers.LinkTypeEthernet)
+			if err != nil {
+				t.Fatal(err)
+			}
+			fp := &fakePort{
+				packetDst: w,
+			}
+			timeNow = func() time.Time { return time.Unix(1, 0) }
+			_, err = fp.Write(createEthPacket(t))
+			err = w.Flush()
+			if d := errdiff.Check(err, tt.wantErr); d != "" {
+				t.Fatalf("Write() unexpected error diff: %s", d)
+			}
+			if d := cmp.Diff(fmt.Sprintf("%x", fw.Bytes()), tt.want); d != "" {
+				t.Fatalf("Write() unexpected diff(-got,+want)\n:%s", d)
+			}
+		})
+	}
+}