Add support for retreiving from a remote RIB.

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/openconfig/gribi v1.0.0
 	github.com/openconfig/lemming v0.3.2-0.20230914210403-c6484d12af0a
 	github.com/openconfig/testt v0.0.0-20220311054427-efbb1a32ec07
-	github.com/openconfig/ygot v0.29.10
+	github.com/openconfig/ygot v0.29.11-0.20230922181344-6c3af4108a57
 	go.uber.org/atomic v1.10.0
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d
 	google.golang.org/grpc v1.58.0-dev

go.sum

@@ -208,8 +208,8 @@ github.com/openconfig/ygnmi v0.8.7/go.mod h1:7up6qc9l9G4+Cfo37gzO0D7+2g4yqyW+xzh
 github.com/openconfig/ygot v0.6.0/go.mod h1:o30svNf7O0xK+R35tlx95odkDmZWS9JyWWQSmIhqwAs=
 github.com/openconfig/ygot v0.10.4/go.mod h1:oCQNdXnv7dWc8scTDgoFkauv1wwplJn5HspHcjlxSAQ=
 github.com/openconfig/ygot v0.13.2/go.mod h1:kJN0yCXIH07dOXvNBEFm3XxXdnDD5NI6K99tnD5x49c=
-github.com/openconfig/ygot v0.29.10 h1:FRZXxyeCdiJXz6uat5uOm3Hlg+PUu2N0mY+eiva12MI=
-github.com/openconfig/ygot v0.29.10/go.mod h1:RNnn1ytQ8GZV5LPts36l0cyoRjsYYpruiruJEvmU2sg=
+github.com/openconfig/ygot v0.29.11-0.20230922181344-6c3af4108a57 h1:z7bNYTNR1HzxYQNVwBUaf0veA4j4vuTBW7mTLRW4M1w=
+github.com/openconfig/ygot v0.29.11-0.20230922181344-6c3af4108a57/go.mod h1:RNnn1ytQ8GZV5LPts36l0cyoRjsYYpruiruJEvmU2sg=
 github.com/p4lang/p4runtime v1.4.0-rc.5.0.20220728214547-13f0d02a521e h1:AfZKoikDXbZ7zWvO/lvCRzLo7i6lM+gNleYVMxPiWyQ=
 github.com/p4lang/p4runtime v1.4.0-rc.5.0.20220728214547-13f0d02a521e/go.mod h1:m9laObIMXM9N1ElGXijc66/MSM5eheZJLRLxg/TG+fU=
 github.com/pborman/getopt v0.0.0-20190409184431-ee0cd42419d3/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=

rib/helpers.go

@@ -26,7 +26,7 @@ import (
 
 // FromGetResponses returns a RIB from a slice of gRIBI GetResponse messages.
 // The supplied defaultName is used as the default network instance name.
-func FromGetResponses(defaultName string, responses []*spb.GetResponse) (*RIB, error) {
+func FromGetResponses(defaultName string, responses []*spb.GetResponse, opt ...RIBOpt) (*RIB, error) {
 	r := New(defaultName)
 	niAFTs := map[string]*aftpb.Afts{}
 

rib/reconciler/reconcile.go

@@ -24,6 +24,7 @@
 package reconciler
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/openconfig/gribigo/rib"
@@ -50,7 +51,10 @@ type R struct {
 type RIBTarget interface {
 	// Get returns a RIB containing all network-instances and AFTs that are
 	// supported by the RIB.
-	Get() (*rib.RIB, error)
+	Get(context.Context) (*rib.RIB, error)
+	// CleanUp is called to indicate that the RIBTarget should remove any
+	// state or external connections as it is no longer required.
+	CleanUp()
 }
 
 // LocalRIB wraps a RIB that is locally available on the system as a gRIBIgo
@@ -60,10 +64,18 @@ type LocalRIB struct {
 }
 
 // Get returns the contents of the local RIB.
-func (l *LocalRIB) Get() (*rib.RIB, error) {
+func (l *LocalRIB) Get(_ context.Context) (*rib.RIB, error) {
 	return l.r, nil
 }
 
+// CleanUp implements the RIBTarget interface. No local cleanup is required.
+func (l *LocalRIB) CleanUp() {}
+
+var (
+	// Compile time check that LocalRIB implements the RIBTarget interface.
+	_ RIBTarget = &LocalRIB{}
+)
+
 // New returns a new reconciler with the specified intended and target RIBs.
 func New(intended, target RIBTarget) *R {
 	return &R{
@@ -74,14 +86,14 @@ func New(intended, target RIBTarget) *R {
 
 // Reconcile performs a reconciliation operation between the intended and specified
 // remote RIB.
-func (r *R) Reconcile() error {
+func (r *R) Reconcile(ctx context.Context) error {
 	// Get the current contents of intended and target.
-	iRIB, err := r.intended.Get()
+	iRIB, err := r.intended.Get(ctx)
 	if err != nil {
 		return fmt.Errorf("cannot reconcile RIBs, cannot get contents of intended, %v", err)
 	}
 
-	tRIB, err := r.target.Get()
+	tRIB, err := r.target.Get(ctx)
 	if err != nil {
 		return fmt.Errorf("cannot reconcile RIBs, cannot get contents of target, %v", err)
 	}

rib/reconciler/reconcile_test.go

@@ -1,6 +1,7 @@
 package reconciler
 
 import (
+	"context"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -16,7 +17,7 @@ func TestLocalRIB(t *testing.T) {
 		}
 
 		want := rib.New("DEFAULT")
-		got, err := l.Get()
+		got, err := l.Get(context.Background())
 		if err != nil {
 			t.Fatalf("(*LocalRIB).Get(): did not get expected error, got: %v, want: nil", err)
 		}

rib/reconciler/remote.go

@@ -0,0 +1,67 @@
+package reconciler
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/openconfig/gribigo/client"
+	"github.com/openconfig/gribigo/rib"
+
+	spb "github.com/openconfig/gribi/v1/proto/service"
+)
+
+// RemoteRIB implements the RIBTarget interface and wraps a remote gRIBI RIB.
+// The contents are accessed via the gRIBI gRPC API.
+type RemoteRIB struct {
+	c *client.Client
+
+	defaultName string
+}
+
+// NewRemoteRIB returns a new remote gRIBI RIB. The context supplied is used to
+// dial the remote gRIBI server at the address 'addr'. the 'defName' argument
+// is used to identify the name of the default network instance on the server.
+func NewRemoteRIB(ctx context.Context, defName, addr string) (*RemoteRIB, error) {
+	gc, err := client.New()
+	if err != nil {
+		return nil, fmt.Errorf("cannot create gRIBI client, %v", err)
+	}
+
+	r := &RemoteRIB{
+		c:           gc,
+		defaultName: defName,
+	}
+
+	if err := r.c.Dial(ctx, addr); err != nil {
+		return nil, fmt.Errorf("cannot dial remote server, %v", err)
+	}
+	return r, nil
+}
+
+// CleanUp closes the remote connection to the gRIBI server.
+func (r *RemoteRIB) CleanUp() {
+	r.c.Close()
+}
+
+// Get retrieves the contents of the remote gRIBI server's RIB and returns it as a
+// gRIBIgo RIB struct. The context is used for a Get RPC call to the remote server.
+func (r *RemoteRIB) Get(ctx context.Context) (*rib.RIB, error) {
+	resp, err := r.c.Get(ctx, &spb.GetRequest{
+		NetworkInstance: &spb.GetRequest_All{
+			All: &spb.Empty{},
+		},
+		Aft: spb.AFTType_ALL,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("cannot get remote RIB, %v", err)
+	}
+
+	// We always disable the RIB checking function because we want to see entries that have
+	// not got valid references so that we can reconcile them.
+	remRIB, err := rib.FromGetResponses(r.defaultName, []*spb.GetResponse{resp}, rib.DisableRIBCheckFn())
+	if err != nil {
+		return nil, fmt.Errorf("cannot build remote RIB from responses, %v", err)
+	}
+
+	return remRIB, nil
+}

rib/reconciler/remote_test.go

@@ -0,0 +1,184 @@
+package reconciler
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/openconfig/gribigo/aft"
+	"github.com/openconfig/gribigo/rib"
+	"github.com/openconfig/gribigo/server"
+	"github.com/openconfig/gribigo/testcommon"
+	"github.com/openconfig/ygot/ygot"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/status"
+
+	spb "github.com/openconfig/gribi/v1/proto/service"
+)
+
+func newServer(t *testing.T, r *rib.RIB) (string, func()) {
+	creds, err := credentials.NewServerTLSFromFile(testcommon.TLSCreds())
+	if err != nil {
+		t.Fatalf("cannot load TLS credentials, got err: %v", err)
+	}
+	srv := grpc.NewServer(grpc.Creds(creds))
+	s, err := server.NewFake()
+	if err != nil {
+		t.Fatalf("cannot create server, got err: %v", err)
+	}
+	s.InjectRIB(r)
+
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("cannot create listener, got err: %v", err)
+	}
+	spb.RegisterGRIBIServer(srv, s)
+
+	go srv.Serve(l)
+	return l.Addr().String(), srv.Stop
+}
+
+func TestNewRemoteRIB(t *testing.T) {
+	tests := []struct {
+		desc           string
+		inDefName      string
+		inAddrOverride string
+		wantErr        bool
+	}{{
+		desc:      "successful dial",
+		inDefName: "DEFAULT",
+	}, {
+		desc:           "unsuccessful dial",
+		inDefName:      "DEFAULT",
+		inAddrOverride: "invalid.addr:999999",
+		wantErr:        true,
+	}}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			addr, stop := newServer(t, nil)
+			defer stop()
+
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+			defer cancel()
+
+			if tt.inAddrOverride != "" {
+				addr = tt.inAddrOverride
+			}
+
+			if _, err := NewRemoteRIB(ctx, tt.inDefName, addr); (err != nil) != tt.wantErr {
+				t.Fatalf("NewRemoteRIB(ctx, %s, %s): did not get expected error, got: %v, wantErr? %v", tt.inDefName, addr, err, tt.wantErr)
+			}
+		})
+	}
+}
+
+type badGRIBI struct {
+	*spb.UnimplementedGRIBIServer
+}
+
+func (b *badGRIBI) Get(_ *spb.GetRequest, _ spb.GRIBI_GetServer) error {
+	return status.Errorf(codes.Unimplemented, "RPC unimplemented")
+}
+
+func newBadServer(t *testing.T, r *rib.RIB) (string, func()) {
+	creds, err := credentials.NewServerTLSFromFile(testcommon.TLSCreds())
+	if err != nil {
+		t.Fatalf("cannot load TLS credentials, got err: %v", err)
+	}
+	srv := grpc.NewServer(grpc.Creds(creds))
+	s := &badGRIBI{}
+
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("cannot create listener, got err: %v", err)
+	}
+	spb.RegisterGRIBIServer(srv, s)
+
+	go srv.Serve(l)
+	return l.Addr().String(), srv.Stop
+
+}
+
+func TestGet(t *testing.T) {
+	dn := "DEFAULT"
+	tests := []struct {
+		desc            string
+		inDefName       string
+		inServer        func(*testing.T, *rib.RIB) (string, func())
+		inInjectedRIB   *rib.RIB
+		wantRIBContents map[string]*aft.RIB
+		wantErr         bool
+	}{{
+		desc:      "cannot get RIB",
+		inDefName: "DEFAULT",
+		inServer:  newBadServer,
+		wantErr:   true,
+	}, {
+		desc:      "successfully got RIB",
+		inDefName: dn,
+		inServer:  newServer,
+		inInjectedRIB: func() *rib.RIB {
+			r := rib.NewFake(dn)
+			if err := r.InjectNH(dn, 1); err != nil {
+				t.Fatalf("cannot add NH, %v", err)
+			}
+			if err := r.InjectNHG(dn, 1, map[uint64]uint64{1: 1}); err != nil {
+				t.Fatalf("cannot add NHG, %v", err)
+			}
+			if err := r.InjectIPv4(dn, "1.0.0.0/24", 1); err != nil {
+				t.Fatalf("cannot add IPv4, %v", err)
+			}
+			return r.RIB()
+		}(),
+		wantRIBContents: map[string]*aft.RIB{
+			dn: func() *aft.RIB {
+				r := &aft.RIB{}
+				a := r.GetOrCreateAfts()
+				a.GetOrCreateIpv4Entry("1.0.0.0/24").NextHopGroup = ygot.Uint64(1)
+				a.GetOrCreateNextHopGroup(1).GetOrCreateNextHop(1).Weight = ygot.Uint64(1)
+				a.GetOrCreateNextHop(1)
+				return r
+			}(),
+		},
+	}}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			addr, stop := tt.inServer(t, tt.inInjectedRIB)
+			defer stop()
+
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+			defer cancel()
+
+			rr, err := NewRemoteRIB(ctx, tt.inDefName, addr)
+			if err != nil {
+				t.Fatalf("NewRemoteRIB(_, %s, %s): cannot connect, got err: %v", tt.inDefName, addr, err)
+			}
+
+			got, err := rr.Get(ctx)
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("(*RemoteRIB).Get(ctx): did not get expected err, got: %v, wantErr? %v", err, tt.wantErr)
+			}
+
+			if err != nil {
+				return
+			}
+
+			// We can't introspect the private RIB contents, so make a copy to introspect.
+			gotContents, err := got.RIBContents()
+			if err != nil {
+				t.Fatalf("(*RemoteRIB).Get(ctx).RIBContents(): can't introspect RIB, err: %v", err)
+			}
+
+			if diff := cmp.Diff(gotContents, tt.wantRIBContents); diff != "" {
+				t.Fatalf("(*RemoteRIB).Get(ctx): did not get expected contents, diff(-got,+want):\n%s", diff)
+			}
+
+		})
+	}
+}

rib/rib.go

@@ -324,6 +324,15 @@ func (r *RIB) KnownNetworkInstances() []string {
 	return names
 }
 
+// RIBContents returns the contents of the RIB in a manner that an external
+// caller can interact with. It returns a map, keyed by network instance name,
+// with a deep copy of the RIB contents. Since copying large RIBs may be expensive
+// care should be taken with when it is used. A copy is used since the RIB continues
+// to handle concurrent changes to the contents from multiple sources.
+func (r *RIB) RIBContents() (map[string]*aft.RIB, error) {
+	return r.copyRIBs()
+}
+
 // String returns a string representation of the RIB.
 func (r *RIB) String() string {
 	r.nrMu.RLock()
@@ -542,6 +551,11 @@ func (r *RIB) callResolvedEntryHook(optype constants.OpType, netinst string, aft
 // AFT struct, of the set of RIBs stored by the instance r. A DeepCopy of the RIBs is returned,
 // along with an error that indicates whether the entries could be copied.
 func (r *RIB) copyRIBs() (map[string]*aft.RIB, error) {
+	// TODO(robjs): Consider whether we need finer grained locking for each network
+	// instance RIB rather than holding the lock whilst we clone the contents.
+	r.nrMu.RLock()
+	defer r.nrMu.RUnlock()
+
 	rib := map[string]*aft.RIB{}
 	for name, niR := range r.niRIB {
 		// this is likely expensive on very large RIBs, but with today's implementatiom
@@ -1836,7 +1850,8 @@ func protoFromGoStruct(s ygot.ValidatedGoStruct, prefix *gpb.Path, pb proto.Mess
 	if err := protomap.ProtoFromPaths(pb, vals,
 		protomap.ProtobufMessagePrefix(prefix),
 		protomap.ValuePathPrefix(prefix),
-		protomap.IgnoreExtraPaths()); err != nil {
+		protomap.IgnoreExtraPaths(),
+	); err != nil {
 		return fmt.Errorf("cannot unmarshal gNMI paths, %v", err)
 	}