Add binary for gNMI+gRIBI server, add TLS support.

.github/workflows/go.yml

@@ -36,9 +36,7 @@ jobs:
 
     - name: Race Test
       run: |
-          for i in `find . -type d -mindepth 1 -maxdepth 1 | egrep -v ".git|.github|fluent|doc"`; do
-            go test -race ./$i
-          done
+           go test -race ./...
 
     - name: Coveralls
       if: ${{ matrix.go == '1.14' }}

client/client.go

@@ -4,6 +4,7 @@ package client
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"errors"
 	"fmt"
 	"io"
@@ -14,6 +15,7 @@ import (
 	log "github.com/golang/glog"
 	"go.uber.org/atomic"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 	"lukechampine.com/uint128"
 
 	spb "github.com/openconfig/gribi/v1/proto/service"
@@ -151,7 +153,11 @@ type DialOpt interface {
 func (c *Client) Dial(ctx context.Context, addr string, opts ...DialOpt) error {
 	// TODO(robjs): translate any options within the dial options here, we may
 	// want to consider just accepting some gRPC dialoptions directly.
-	conn, err := grpc.DialContext(ctx, addr, grpc.WithInsecure(), grpc.WithBlock())
+
+	conn, err := grpc.DialContext(ctx, addr,
+		grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{
+			InsecureSkipVerify: true,
+		})), grpc.WithBlock())
 	if err != nil {
 		return fmt.Errorf("cannot dial remote system, %v", err)
 	}
@@ -529,7 +535,7 @@ func (c *Client) StartSending() {
 	defer c.qs.sendMu.Unlock()
 	for _, m := range c.qs.sendq {
 		log.V(2).Infof("sending %s to modify channel", m)
-		c.Q(m)
+		c.q(m)
 	}
 	c.qs.sendq = []*spb.ModifyRequest{}
 }
@@ -598,6 +604,13 @@ func (c *Client) handleModifyResponse(m *spb.ModifyResponse) error {
 	}
 
 	// TODO(robjs): add handling of received operations.
+	for _, r := range m.Result {
+		res, err := c.clearPendingOp(r)
+		c.qs.resultq = append(c.qs.resultq, res)
+		if err != nil {
+			return fmt.Errorf("cannot remove pending operation %d, %v", r.Id, err)
+		}
+	}
 
 	return nil
 }
@@ -623,8 +636,8 @@ func (c *Client) isConverged() bool {
 func (c *Client) addPendingOp(op *spb.AFTOperation) error {
 	c.qs.pendMu.Lock()
 	defer c.qs.pendMu.Unlock()
-	if c.qs.pendq.Ops[op.Id] != nil {
-		return fmt.Errorf("could not enqueue operation %d, duplicate pending ID", op.Id)
+	if v := c.qs.pendq.Ops[op.Id]; v != nil {
+		return fmt.Errorf("could not enqueue operation %d, duplicate pending ID (pending: %v)", op.Id, v)
 	}
 	c.qs.pendq.Ops[op.Id] = &PendingOp{
 		Timestamp: unixTS(),
@@ -633,6 +646,23 @@ func (c *Client) addPendingOp(op *spb.AFTOperation) error {
 	return nil
 }
 
+// clearPendingOp removes the operation with the ID in the specified result from the
+// pending queue and returns the result.
+func (c *Client) clearPendingOp(op *spb.AFTResult) (*OpResult, error) {
+	c.qs.pendMu.Lock()
+	defer c.qs.pendMu.Unlock()
+	v := c.qs.pendq.Ops[op.Id]
+	if v == nil {
+		return nil, fmt.Errorf("could not dequeue operation %d, unknown operation", op.Id)
+	}
+	delete(c.qs.pendq.Ops, op.Id)
+	n := unixTS()
+	return &OpResult{
+		Timestamp: n,
+		Latency:   n - v.Timestamp,
+	}, nil
+}
+
 // updatePendingElection adds the election ID specified by id to the pending transaction
 // queue on the client. There can only be a single pending election, and hence the
 // election ID is updated in place.

cmd/rtr/main.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	"context"
+	"flag"
+
+	log "github.com/golang/glog"
+	"github.com/openconfig/gribigo/device"
+)
+
+var (
+	certFile = flag.String("cert", "", "cert is the path to the server TLS certificate file")
+	keyFile  = flag.String("key", "", "key is the path to the server TLS key file")
+)
+
+func main() {
+	flag.Parse()
+
+	if *certFile == "" || *keyFile == "" {
+		log.Exitf("must specify a TLS certificate and key file")
+	}
+
+	ctx := context.Background()
+	creds, err := device.TLSCredsFromFile(*certFile, *keyFile)
+	if err != nil {
+		log.Exitf("cannot initialise TLS, got: %v", err)
+	}
+	d, cancel, err := device.New(context.Background(), creds)
+	defer cancel()
+	if err != nil {
+		log.Exitf("cannot start device, %v", err)
+	}
+	log.Infof("listening on:\n\tgRIBI: %s\n\tgNMI: %s", d.GRIBIAddr(), d.GNMIAddr())
+	<-ctx.Done()
+}

compliance/compliance.go

@@ -6,11 +6,13 @@ package compliance
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/openconfig/gribigo/chk"
 	"github.com/openconfig/gribigo/client"
 	"github.com/openconfig/gribigo/fluent"
+	"github.com/openconfig/gribigo/server"
 	"google.golang.org/grpc/codes"
 )
 
@@ -87,3 +89,19 @@ func ModifyConnectionSinglePrimaryPreserve(addr string, t testing.TB) {
 		t.Fatalf("did not get expected error type, got: %v, want: %v", ce, want)
 	}
 }
+
+// AddIPv4EntrySuccess adds a simple IPv4 Entry which references a next-hop-group to the gRIBI target.
+func AddIPv4EntrySuccess(addr string, t testing.TB) {
+	c := fluent.NewClient()
+	c.Connection().WithTarget(addr).WithRedundancyMode(fluent.ElectedPrimaryClient).WithInitialElectionID(0, 1).WithPersistence()
+	c.Start(context.Background(), t)
+	c.Modify().AddEntry(t, fluent.IPv4Entry().WithPrefix("1.1.1.1/32").WithNetworkInstance(server.DefaultNIName).WithNextHopGroup(42))
+	c.StartSending(context.Background(), t)
+	err := c.Await(context.Background(), t)
+	if err != nil {
+		t.Fatalf("got unexpected error from server, got: %v, want: nil", err)
+	}
+	// TODO(robjs): add verification of the received gRIBI results.
+	// TODO(robjs): add gNMI subscription using generated telemetry library.
+	fmt.Printf("%v\n", c.Results(t))
+}

compliance/compliance_test.go

@@ -1,9 +1,11 @@
 package compliance
 
 import (
+	"context"
 	"strings"
 	"testing"
 
+	"github.com/openconfig/gribigo/device"
 	"github.com/openconfig/gribigo/negtest"
 	"github.com/openconfig/gribigo/testcommon"
 )
@@ -32,12 +34,19 @@ func TestModifyConnectionParameters(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.in.ShortName, func(t *testing.T) {
-			startServer, addr := testcommon.Server()
-			go startServer()
+			creds, err := device.TLSCredsFromFile(testcommon.TLSCreds())
+			if err != nil {
+				t.Fatalf("cannot load credentials, got err: %v", err)
+			}
+			d, cancel, err := device.New(context.Background(), creds)
+			defer cancel()
+			if err != nil {
+				t.Fatalf("cannot start server, %v", err)
+			}
 
 			if tt.wantFatalMsg != "" {
 				if got := negtest.ExpectFatal(t, func(t testing.TB) {
-					tt.in.Fn(addr, t)
+					tt.in.Fn(d.GRIBIAddr(), t)
 				}); !strings.Contains(got, tt.wantFatalMsg) {
 					t.Fatalf("did not get expected fatal error, got: %s, want: %s", got, tt.wantFatalMsg)
 				}
@@ -46,14 +55,14 @@ func TestModifyConnectionParameters(t *testing.T) {
 
 			if tt.wantErrorMsg != "" {
 				if got := negtest.ExpectError(t, func(t testing.TB) {
-					tt.in.Fn(addr, t)
+					tt.in.Fn(d.GRIBIAddr(), t)
 				}); !strings.Contains(got, tt.wantErrorMsg) {
 					t.Fatalf("did not get expected error, got: %s, want: %s", got, tt.wantErrorMsg)
 				}
 			}
 
 			// Any unexpected error will be caught by being called directly on t from the fluent library.
-			tt.in.Fn(addr, t)
+			tt.in.Fn(d.GRIBIAddr(), t)
 		})
 	}
 }

constants/const.go

@@ -0,0 +1,16 @@
+// package constants defines constants that are shared amongst multiple gRIBIgo packages.
+package constants
+
+// OpType indicates the type of operation that was performed in contexts where it
+// is not available, such as callbacks to user-provided functions.
+type OpType int64
+
+const (
+	_ OpType = iota
+	// ADD indicates that the operation called was an Add.
+	ADD
+	// DELETE indicates that the operation called was a Delete.
+	DELETE
+	// MODIFY indicates that the operation called was a Modify.
+	MODIFY
+)

demo/ipv4/ipv4.go

@@ -0,0 +1,4 @@
+// Package ipv4 is a very simple (test) demo package that shows
+// an IPv4 entry being sent to the gRIBI server using the fluent
+// API and client packages defined in this package.
+package ipv4

demo/ipv4/ipv4_test.go

@@ -0,0 +1,26 @@
+package ipv4
+
+import (
+	"flag"
+	"testing"
+	"time"
+
+	"github.com/openconfig/gribigo/compliance"
+)
+
+var (
+	addr = flag.String("addr", "", "address of the gRIBI target")
+	run  = flag.Bool("run", false, "whether to run the tests, this stops this file causing failures in CI")
+)
+
+func TestDemo(t *testing.T) {
+	flag.Parse()
+	if !*run {
+		return
+	}
+	if *addr == "" {
+		t.Fatalf("must specify an address")
+	}
+	compliance.AddIPv4EntrySuccess(*addr, t)
+	time.Sleep(2 * time.Second)
+}