Add Virtual Database Client + Unit Tests

README.md

@@ -27,13 +27,6 @@ There is also a test program dialout_server_cli for collecting data from dial-ou
 
 The binaries will be installed under $GOPATH/bin/, they may be copied to any SONiC switch and run there.
 
-You can also build a debian package and install it:
-
-    git clone https://github.com/Azure/sonic-telemetry.git
-    pushd sonic-telemetry
-    dpkg-buildpackage -rfakeroot -b -us -uc
-    popd
-
 ### Running
 * See [SONiC gRPC telemetry](./doc/grpc_telemetry.md) for how to run dial-in mode system telemetry server
 * See [SONiC telemetry in dial-out mode](./doc/dialout.md) for how to run dial-out mode system telemetry client

dialout/dialout_client/dialout_client.go

@@ -6,7 +6,7 @@ import (
 	"errors"
 	"fmt"
 	spb "github.com/Azure/sonic-telemetry/proto"
-	sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
+	//sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
 	"github.com/go-redis/redis"
 	log "github.com/golang/glog"
 	gpb "github.com/openconfig/gnmi/proto/gnmi"
@@ -16,6 +16,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"net"
+	sdc "test/sonic-telemetry-new-pfcwd/sonic_data_client"
 	//"reflect"
 	"strconv"
 	"strings"

dialout/dialout_client/dialout_client_test.go

@@ -28,10 +28,12 @@ import (
 	"testing"
 	"time"
 
-	sds "github.com/Azure/sonic-telemetry/dialout/dialout_server"
+	//sds "github.com/Azure/sonic-telemetry/dialout/dialout_server"
 	spb "github.com/Azure/sonic-telemetry/proto"
-	sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
+	sds "test/sonic-telemetry-new-pfcwd/dialout/dialout_server"
+	//sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
 	gclient "github.com/openconfig/gnmi/client/gnmi"
+	sdc "test/sonic-telemetry-new-pfcwd/sonic_data_client"
 )
 
 var clientTypes = []string{gclient.Type}
@@ -166,6 +168,14 @@ func prepareConfigDb(t *testing.T) {
 	}
 	mpi_alias_map := loadConfig(t, "", countersPortAliasMapByte)
 	loadConfigDB(t, rclient, mpi_alias_map)
+
+	fileName = "../../testdata/CONFIG_PFCWD_PORTS.txt"
+	configPfcwdByte, err := ioutil.ReadFile(fileName)
+	if err != nil {
+		t.Fatalf("read file %v err: %v", fileName, err)
+	}
+	mpi_pfcwd_map := loadConfig(t, "", configPfcwdByte)
+	loadConfigDB(t, rclient, mpi_pfcwd_map)
 }
 
 func prepareDb(t *testing.T) {

dialout/dialout_server_cli/dialout_server_cli.go

@@ -10,8 +10,9 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 
-	ds "github.com/Azure/sonic-telemetry/dialout/dialout_server"
+	//ds "github.com/Azure/sonic-telemetry/dialout/dialout_server"
 	testcert "github.com/Azure/sonic-telemetry/testdata/tls"
+	ds "test/sonic-telemetry-new-pfcwd/dialout/dialout_server"
 )
 
 var (

gnmi_server/client_subscribe.go

@@ -12,7 +12,10 @@ import (
 	"google.golang.org/grpc/codes"
 
 	//spb "github.com/Azure/sonic-telemetry/proto"
-	sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
+	//sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
+	sdc "test/sonic-telemetry-new-pfcwd/sonic_data_client"
+	vdc "test/sonic-telemetry-new-pfcwd/virtual_database_client"
+
 	gnmipb "github.com/openconfig/gnmi/proto/gnmi"
 )
 
@@ -101,6 +104,7 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 	}
 
 	var target string
+	fmt.Printf("Subscribe Prefix: %v\n", target)
 	prefix := c.subscribe.GetPrefix()
 	if prefix == nil {
 		return grpc.Errorf(codes.Unimplemented, "No target specified in prefix")
@@ -119,6 +123,8 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 	var dc sdc.Client
 	if target == "OTHERS" {
 		dc, err = sdc.NewNonDbClient(paths, prefix)
+	} else if target == "SONiC_DB" {
+		dc, err = vdc.NewDbClient(paths, prefix)
 	} else {
 		dc, err = sdc.NewDbClient(paths, prefix)
 	}
@@ -226,6 +232,11 @@ func (c *Client) send(stream gnmipb.GNMI_SubscribeServer) error {
 				c.errors++
 				return err
 			}
+		case vdc.Value:
+			if resp, err = vdc.ValToResp(v); err != nil {
+				c.errors++
+				return err
+			}
 		default:
 			log.V(1).Infof("Unknown data type %v for %s in queue", items[0], c)
 			c.errors++

gnmi_server/server.go

@@ -15,7 +15,9 @@ import (
 	"google.golang.org/grpc/reflection"
 	"google.golang.org/grpc/status"
 
-	sdc "github.com/Azure/sonic-telemetry/sonic_data_client"
+	sdc "test/sonic-telemetry-new-pfcwd/sonic_data_client"
+	vdc "test/sonic-telemetry-new-pfcwd/virtual_database_client"
+
 	gnmipb "github.com/openconfig/gnmi/proto/gnmi"
 )
 
@@ -77,6 +79,11 @@ func (srv *Server) Serve() error {
 	return srv.s.Serve(srv.lis)
 }
 
+func (srv *Server) Stop() {
+	s := srv.s
+	s.Stop()
+}
+
 // Address returns the port the Server is listening to.
 func (srv *Server) Address() string {
 	addr := srv.lis.Addr().String()
@@ -174,6 +181,8 @@ func (s *Server) Get(ctx context.Context, req *gnmipb.GetRequest) (*gnmipb.GetRe
 	var dc sdc.Client
 	if target == "OTHERS" {
 		dc, err = sdc.NewNonDbClient(paths, prefix)
+	} else if target == "SONiC_DB" {
+		dc, err = vdc.NewDbClient(paths, prefix)
 	} else {
 		dc, err = sdc.NewDbClient(paths, prefix)
 	}
@@ -186,6 +195,10 @@ func (s *Server) Get(ctx context.Context, req *gnmipb.GetRequest) (*gnmipb.GetRe
 		return nil, status.Error(codes.NotFound, err.Error())
 	}
 
+	if target == "SONiC_DB" {
+		notifications = make([]*gnmipb.Notification, len(spbValues))
+	}
+
 	for index, spbValue := range spbValues {
 		update := &gnmipb.Update{
 			Path: spbValue.GetPath(),