TranslClient: Use new translib subscription APIs (#122)

* Allow data clients to send full gnmipb.Notification

Added gnmipb.Notification as a member of the Value messgae object
defined by sonic_internal.proto. Data clients now can fill a complete
Notification object instead of timestamp, path, TypedValue separately.

* Use origin to identify translib yang paths

Use TranslClient to handle the subscribe requests when origin is
"openconfig". Fallback to the existing target based identification
logic if origin is not given.

* Use new translib subscribe APIs in TranslClient

StreamRun() enhancements:
 - Create a new SubscribeSession in the beginning and pass it to all
   further translib API calls
 - Call translib.IsSubscribeSupported() to reasolve preferences of the
   requested paths. This also splits the target_defined request into
   on_change and sample paths.
 - Use one translib.Subscribe() call passing all on_change enabled
   paths, if there are any. This will handle initial updates and
   subsequent on_change notifications.
 - Use one translib.Stream() call for each sample subscribe path, if
   present. This will be called in a loop at every sample interval.
 - Maintain the ygot objects received for each translib.Stream() call
   in a cache. Diff the current set of objects with the objects from
   the previous iteration to resolve deleted paths and modified values
   (when suppress_redundant is enabled).

PollRun() and OnceRun() enhancements:
 - Call translib.Stream() with each subscribed path to generate
   notification data.
 - In poll mode, this is repeated when a poll message is received

* Increase gnmi_server gotest timeout to 20m

Makefile

@@ -96,7 +96,7 @@ $(ENVFILE):
 
 check_gotest: $(DBCONFG) $(ENVFILE)
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(GO) test -race -coverprofile=coverage-config.txt -covermode=atomic -v github.com/sonic-net/sonic-gnmi/sonic_db_config
-	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(TESTENV) $(GO) test -race -coverprofile=coverage-gnmi.txt -covermode=atomic -mod=vendor $(BLD_FLAGS) -v github.com/sonic-net/sonic-gnmi/gnmi_server -coverpkg ../...
+	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(TESTENV) $(GO) test -race -timeout 20m -coverprofile=coverage-gnmi.txt -covermode=atomic -mod=vendor $(BLD_FLAGS) -v github.com/sonic-net/sonic-gnmi/gnmi_server -coverpkg ../...
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(TESTENV) $(GO) test -coverprofile=coverage-dialcout.txt -covermode=atomic -mod=vendor $(BLD_FLAGS) -v github.com/sonic-net/sonic-gnmi/dialout/dialout_client
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(GO) test -race -coverprofile=coverage-data.txt -covermode=atomic -mod=vendor -v github.com/sonic-net/sonic-gnmi/sonic_data_client
 	sudo CGO_LDFLAGS="$(CGO_LDFLAGS)" CGO_CXXFLAGS="$(CGO_CXXFLAGS)" $(GO) test -race -coverprofile=coverage-dbus.txt -covermode=atomic -mod=vendor -v github.com/sonic-net/sonic-gnmi/sonic_service_client

gnmi_server/client_subscribe.go

@@ -10,6 +10,7 @@ import (
 	log "github.com/golang/glog"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	sdc "github.com/sonic-net/sonic-gnmi/sonic_data_client"
 	gnmipb "github.com/openconfig/gnmi/proto/gnmi"
 )
@@ -128,23 +129,23 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 		return grpc.Errorf(codes.InvalidArgument, "first message must be SubscriptionList: %q", query)
 	}
 
-	var target string
 	prefix := c.subscribe.GetPrefix()
-	if prefix == nil {
-		return grpc.Errorf(codes.Unimplemented, "No target specified in prefix")
-	} else {
-		target = prefix.GetTarget()
-		// TODO: add data client support for fetching non-db data
-		if target == "" {
-			return grpc.Errorf(codes.Unimplemented, "Empty target data not supported yet")
-		}
-	}
+	origin := prefix.GetOrigin()
+	target := prefix.GetTarget()
 
 	paths, err := c.populateDbPathSubscrition(c.subscribe)
 	if err != nil {
 		return grpc.Errorf(codes.NotFound, "Invalid subscription path: %v %q", err, query)
 	}
 
+	if o, err := ParseOrigin(paths); err != nil {
+		return err // origin conflict within paths
+	} else if len(origin) == 0 {
+		origin = o // Use origin from paths if not given in prefix
+	} else if len(o) != 0 && o != origin {
+		return status.Error(codes.InvalidArgument, "Origin conflict between prefix and paths")
+	}
+
 	if connectionKey, valid = connectionManager.Add(c.addr, query.String()); !valid {
 		return grpc.Errorf(codes.Unavailable, "Server connections are at capacity.")
 	}
@@ -155,7 +156,18 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 
 	mode := c.subscribe.GetMode()
 
-	if target == "OTHERS" {
+	log.V(3).Infof("mode=%v, origin=%q, target=%q", mode, origin, target)
+
+	if origin == "openconfig" {
+		dc, err = sdc.NewTranslClient(prefix, paths, ctx, extensions, sdc.TranslWildcardOption{})
+	} else if len(origin) != 0 {
+		return grpc.Errorf(codes.Unimplemented, "Unsupported origin: %s", origin)
+	} else if target == "" {
+		// This and subsequent conditions handle target based path identification
+		// when origin == "". As per the spec it should have been treated as "openconfig".
+		// But we take a deviation and stick to legacy logic for backward compatibility
+		return grpc.Errorf(codes.Unimplemented, "Empty target data not supported")
+	} else if target == "OTHERS" {
 		dc, err = sdc.NewNonDbClient(paths, prefix)
 	} else if ((target == "EVENTS") && (mode == gnmipb.SubscriptionList_STREAM)) {
 		dc, err = sdc.NewEventClient(paths, prefix, c.logLevel)