Repository: google/gnxi (language: python)

Issue:
subscription is not implemented in gnmi_target
Hi, 

I am tried to use gnmi_subscribe with gnmi_target. it seems subscribe option is not implemented. I get below error

gnmi_subscribe.go:127] Error using STREAM mode: rpc error: code = Unimplemented desc = Subscribe is not implemented.

Any help bringing up the subscription implementation in the target?


Discussion / hints:

Fix patch:
diff --git a/gnmi/server.go b/gnmi/server.go
index 042d4d61..9a14e262 100644
--- a/gnmi/server.go
+++ b/gnmi/server.go
@@ -20,7 +20,6 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -54,6 +53,11 @@ var (
 	subscribeSync      = &pb.SubscribeResponse{Response: &pb.SubscribeResponse_SyncResponse{SyncResponse: true}}
 )
 
+const (
+	maxStreamSampleInterval = 60 * time.Second
+	minStreamSampleInterval = 1 * time.Second
+)
+
 // Server struct maintains the data structure for device config and implements the interface of gnmi server. It supports Capabilities, Get, and Set APIs.
 // Typical usage:
 //
@@ -638,7 +642,7 @@ func (s *Server) InternalUpdate(fp func(config ygot.ValidatedGoStruct) error) er
 func (s *Server) Subscribe(stream pb.GNMI_SubscribeServer) error {
 	var err error
 
-	c := streamClient{stream: stream}
+	c := &streamClient{stream: stream}
 	c.sr, err = stream.Recv()
 
 	switch {
@@ -673,17 +677,43 @@ func (s *Server) Subscribe(stream pb.GNMI_SubscribeServer) error {
 
 	switch mode {
 	case pb.SubscriptionList_ONCE:
-		go func() {
-			s.doOnceSubscription(&c)
-		}()
+		go s.doOnceSubscription(c)
 	case pb.SubscriptionList_POLL:
 		return status.Error(codes.Unimplemented, "mode POLL is not implemented.")
 	case pb.SubscriptionList_STREAM:
-		return status.Error(codes.Unimplemented, "mode SAMPLE is not implemented.")
+
+		for _, sub := range c.sr.GetSubscribe().GetSubscription() {
+			// Check for only Sample subscriptions, with valid paths and interval value.
+			if mode := sub.GetMode(); mode != pb.SubscriptionMode_SAMPLE {
+				return status.Errorf(codes.Unimplemented, "subscription mode %v not implemented", mode)
+			}
+			interval := sub.GetSampleInterval()
+			if interval > uint64(maxStreamSampleInterval.Nanoseconds()) {
+				return status.Errorf(codes.InvalidArgument, "maximum supported sampling interval is %d", maxStreamSampleInterval)
+			}
+			if interval < uint64(minStreamSampleInterval.Nanoseconds()) && interval != 0 {
+				return status.Errorf(codes.InvalidArgument, "minumum supported sampling interval is %d", minStreamSampleInterval)
+			}
+			fullPath := sub.GetPath()
+			prefix := c.sr.GetSubscribe().GetPrefix()
+			if prefix != nil {
+				fullPath = gnmiFullPath(prefix, fullPath)
+			}
+			nodes, err := ytypes.GetNode(s.model.schemaTreeRoot, s.config, fullPath, &ytypes.GetHandleWildcards{})
+			if len(nodes) == 0 || err != nil || util.IsValueNil(nodes[0].Data) {
+				return status.Errorf(codes.InvalidArgument, "path %v not found: %v", fullPath, err)
+			}
+		}
+		// Closing the done channel makes the spawed subroutines exit.
+		done := make(chan bool)
+		defer close(done)
+		for _, sub := range c.sr.GetSubscribe().GetSubscription() {
+			go s.doSampleSubscription(c, sub, done)
+		}
 	default:
 		return status.Errorf(codes.InvalidArgument, "subscription mode %v not recognized", mode)
 	}
-	go s.doSendSubscriptionMsgs(&c)
+	go s.doSendSubscriptionMsgs(c)
 	return <-errC
 }
 
@@ -698,6 +728,55 @@ type streamClient struct {
 // subscribeSyncToken signals doSendSubscriptionMsgs to send subscribeSync.
 type subscribeSyncToken struct{}
 
+// doSampleSubscription processes a STREAM Sampling Subscription.
+// It pushes Notification message in the queue.
+// On error or when the channel is closed, this routine exits.
+func (s *Server) doSampleSubscription(c *streamClient, sub *pb.Subscription, done <-chan bool) {
+	prefix := c.sr.GetSubscribe().GetPrefix()
+	fullPath := sub.GetPath()
+	if prefix != nil {
+		fullPath = gnmiFullPath(prefix, fullPath)
+	}
+	if !c.sr.GetSubscribe().GetUpdatesOnly() {
+		n, err := s.subscriptionUpdates(fullPath)
+		if err != nil {
+			return
+		}
+		c.msgQ.Insert(n)
+	}
+	c.msgQ.Insert(subscribeSyncToken{})
+
+	updateInterval := time.Nanosecond * time.Duration(sub.GetSampleInterval())
+	if updateInterval == 0 {
+		updateInterval = minStreamSampleInterval
+	}
+	updateTicker := time.NewTicker(updateInterval)
+	defer updateTicker.Stop()
+	for {
+		select {
+		case <-updateTicker.C:
+			n, err := s.subscriptionUpdates(fullPath)
+			if err != nil {
+				return
+			}
+			c.msgQ.Insert(n)
+		case <-done:
+			return
+		}
+	}
+}
+
+// subscriptionUpdates returns a Notification message for the path.
+func (s *Server) subscriptionUpdates(fullPath *pb.Path) (*pb.Notification, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	updates, err := s.updatesFromNode(fullPath)
+	return &pb.Notification{
+		Timestamp: time.Now().UnixNano(),
+		Update:    updates,
+	}, err
+}
+
 // doOnceSubscription processes a ONCE Subscription. It produces a single
 // Notification message for all subscribed paths.
 func (s *Server) doOnceSubscription(c *streamClient) {
@@ -733,20 +812,7 @@ func (s *Server) doOnceSubscription(c *streamClient) {
 // doSendSubscriptionMsgs monitors the message queue and sends
 // Subscription Responses to the client.
 func (s *Server) doSendSubscriptionMsgs(c *streamClient) {
-	subscribeTimeout := time.NewTimer(time.Second * 5)
-	subscribeTimeout.Stop()
-
-	// If a send doesn't occur within the timeout.
-	done := make(chan struct{})
-	defer close(done)
-	go func() {
-		select {
-		case <-subscribeTimeout.C:
-			c.errC <- errors.New("subscription timed out while sending")
-		case <-done:
-		}
-	}()
-
+	var response *pb.SubscribeResponse
 	for {
 		item, _, err := c.msgQ.Next(c.stream.Context())
 		if err != nil {
@@ -758,24 +824,19 @@ func (s *Server) doSendSubscriptionMsgs(c *streamClient) {
 			return
 		}
 		if _, ok := item.(subscribeSyncToken); ok {
-			if err = c.stream.Send(subscribeSync); err != nil {
-				c.errC <- err
+			response = subscribeSync
+		} else {
+			n, ok := item.(*pb.Notification)
+			if !ok || n == nil {
+				c.errC <- status.Errorf(codes.Internal, "invalid notification message: %v", item)
 				return
 			}
-			continue
-		}
-		n, ok := item.(*pb.Notification)
-		if !ok || n == nil {
-			c.errC <- status.Errorf(codes.Internal, "invalid notification message: %v", item)
-			return
-		}
-		response := &pb.SubscribeResponse{
-			Response: &pb.SubscribeResponse_Update{
-				Update: n,
-			},
+			response = &pb.SubscribeResponse{
+				Response: &pb.SubscribeResponse_Update{
+					Update: n,
+				},
+			}
 		}
-		subscribeTimeout.Reset(time.Second * 5)
-		defer subscribeTimeout.Stop()
 		err = c.stream.Send(response)
 		if err != nil {
 			c.errC <- err


Test patch:
diff --git a/gnmi/server_test.go b/gnmi/server_test.go
index d834b245..41f58786 100644
--- a/gnmi/server_test.go
+++ b/gnmi/server_test.go
@@ -1481,6 +1481,198 @@ func runTestSubscribeOnce(t *testing.T, s *Server, subscribptions []*pb.Subscrip
 	}
 }
 
+func TestSubscribeSample(t *testing.T) {
+	jsonConfigRoot := `{
+		"openconfig-system:system": {
+			"openconfig-openflow:openflow": {
+				"agent": {
+					"state": {
+						"failure-mode": "SECURE",
+						"max-backoff": 10
+					}
+				}
+			}
+		}
+}`
+	pathAgentState := &pb.Path{
+		Elem: []*pb.PathElem{
+			&pb.PathElem{Name: "system"},
+			&pb.PathElem{Name: "openflow"},
+			&pb.PathElem{Name: "agent"},
+			&pb.PathElem{Name: "state"},
+		}}
+	pathAgentFailureMode := proto.Clone(pathAgentState).(*pb.Path)
+	pathAgentFailureMode.Elem = append(pathAgentFailureMode.Elem, &pb.PathElem{Name: "failure-mode"})
+	pathAgentMaxBackoff := proto.Clone(pathAgentState).(*pb.Path)
+	pathAgentMaxBackoff.Elem = append(pathAgentMaxBackoff.Elem, &pb.PathElem{Name: "max-backoff"})
+	pathAgentFoo := proto.Clone(pathAgentState).(*pb.Path)
+	pathAgentFoo.Elem = append(pathAgentMaxBackoff.Elem, &pb.PathElem{Name: "foo"})
+
+	secsToNanoSecs := func(s int) uint64 { return uint64(s * 1000000000) }
+
+	s, err := NewServer(model, []byte(jsonConfigRoot), nil)
+	if err != nil {
+		t.Fatalf("error in creating server: %v", err)
+	}
+
+	tests := []struct {
+		desc              string
+		pathPrefix        *pb.Path
+		subscription      *pb.Subscription
+		timeout           time.Duration
+		updatesOnly       bool
+		wantUpdates       []*pb.Update
+		wantNotifications int
+	}{{
+		desc: "Single Sample Subscription",
+		subscription: &pb.Subscription{
+			Mode:           pb.SubscriptionMode_SAMPLE,
+			SampleInterval: secsToNanoSecs(1),
+			Path:           pathAgentFailureMode},
+		timeout:           time.Millisecond * 3500,
+		wantNotifications: 4,
+		wantUpdates: []*pb.Update{
+			&pb.Update{
+				Path: pathAgentFailureMode,
+				Val: &pb.TypedValue{
+					Value: &pb.TypedValue_StringVal{StringVal: "SECURE"}}},
+		},
+	}, {
+		desc: "Single Subscribe to Update Only with 0 interval",
+		subscription: &pb.Subscription{
+			Mode:           pb.SubscriptionMode_SAMPLE,
+			SampleInterval: 0,
+			Path:           pathAgentState},
+		timeout:           time.Millisecond * 3500,
+		wantNotifications: 3,
+		updatesOnly:       true,
+		wantUpdates: []*pb.Update{
+			&pb.Update{
+				Path: pathAgentFailureMode,
+				Val: &pb.TypedValue{
+					Value: &pb.TypedValue_StringVal{StringVal: "SECURE"}}},
+			&pb.Update{
+				Path: pathAgentMaxBackoff,
+				Val:  &pb.TypedValue{Value: &pb.TypedValue_UintVal{UintVal: uint64(10)}}},
+		},
+	}, {
+		desc:       "Subscribe with prefix",
+		pathPrefix: pathAgentState,
+		subscription: &pb.Subscription{
+			Mode:           pb.SubscriptionMode_SAMPLE,
+			SampleInterval: secsToNanoSecs(2),
+			Path:           &pb.Path{Elem: []*pb.PathElem{&pb.PathElem{Name: "failure-mode"}}}},
+		timeout:           time.Millisecond * 4500,
+		wantNotifications: 3,
+		wantUpdates: []*pb.Update{
+			&pb.Update{
+				Path: pathAgentFailureMode,
+				Val: &pb.TypedValue{
+					Value: &pb.TypedValue_StringVal{StringVal: "SECURE"}}},
+		},
+	}}
+
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			runTestSubscribeSample(t, s, test.subscription, test.pathPrefix, test.updatesOnly, test.timeout, test.wantNotifications, test.wantUpdates)
+		})
+	}
+}
+
+// runTestSubscribeSample requests STREAM sampling subscription, and compares the returned Notifications.
+func runTestSubscribeSample(t *testing.T, s *Server, subscription *pb.Subscription, pathPrefix *pb.Path, updatesOnly bool, timeout time.Duration, wantNotifications int, wantUpdates []*pb.Update,
+) {
+	req := &pb.SubscribeRequest{
+		Request: &pb.SubscribeRequest_Subscribe{
+			Subscribe: &pb.SubscriptionList{
+				Prefix:       pathPrefix,
+				Mode:         pb.SubscriptionList_STREAM,
+				UpdatesOnly:  updatesOnly,
+				Subscription: []*pb.Subscription{subscription},
+			},
+		},
+	}
+
+	sampleInterval := subscription.GetSampleInterval()
+	if sampleInterval == 0 {
+		sampleInterval = uint64(minStreamSampleInterval.Nanoseconds())
+	}
+	interval := time.Nanosecond * time.Duration(sampleInterval)
+
+	errC := make(chan error)
+	doneC := make(chan bool)
+	defer close(errC)
+	msgQ := coalesce.NewQueue()
+	c := &streamClient{sr: req, stream: nil, errC: errC, msgQ: msgQ}
+
+	go s.doSampleSubscription(c, subscription, doneC)
+
+	time.Sleep(timeout)
+	close(doneC)
+	msgQ.Close()
+
+	gotNotifications := 0
+	lastNotificationTimestamp := int64(time.Now().UnixNano())
+
+	if !updatesOnly && wantNotifications > 0 {
+		msg, _, err := c.msgQ.Next(context.Background())
+		if err != nil {
+			t.Fatalf("Error getting initial Notification from the queue: %v", err)
+		}
+		n, ok := msg.(*pb.Notification)
+		if !ok || n == nil {
+			t.Fatalf("wanted Notification message in queue, got: %v", msg)
+		}
+		lastNotificationTimestamp = n.GetTimestamp()
+		gotNotifications = gotNotifications + 1
+		if diff := cmp.Diff(n.GetUpdate(), wantUpdates, protocmp.Transform(), cmpopts.SortSlices(updateLess)); diff != "" {
+			t.Errorf("Initial Notification Updates diff:\n%v", diff)
+		}
+	}
+
+	msg, _, err := c.msgQ.Next(context.Background())
+	if err != nil {
+		t.Fatalf("Error getting sync_response from the queue: %v", err)
+	}
+	if _, ok := msg.(subscribeSyncToken); !ok {
+		t.Fatalf("did not receive sync_response message")
+	}
+
+	for {
+		msg, _, err := c.msgQ.Next(context.Background())
+		if err != nil {
+			if coalesce.IsClosedQueue(err) {
+				break
+			} else {
+				t.Fatalf("Error getting Notifications from the queue: %v", err)
+			}
+		}
+
+		if _, ok := msg.(subscribeSyncToken); ok {
+			t.Fatalf("received unexpected sync_response message")
+		}
+
+		n, ok := msg.(*pb.Notification)
+		if !ok || n == nil {
+			t.Fatalf("wanted Notification message in queue, got: %v", msg)
+		}
+		gotNotifications = gotNotifications + 1
+		timeDiff := time.Nanosecond * time.Duration(n.GetTimestamp()-lastNotificationTimestamp)
+		lastNotificationTimestamp = n.GetTimestamp()
+		if (timeDiff - interval) > interval {
+			t.Errorf("Notification messages not within sampling interval: %v", timeDiff)
+		}
+		if diff := cmp.Diff(n.GetUpdate(), wantUpdates, protocmp.Transform(), cmpopts.SortSlices(updateLess)); diff != "" {
+			t.Errorf("Notification Updates diff:\n%v", diff)
+		}
+	}
+
+	if diff := cmp.Diff(wantNotifications, gotNotifications); diff != "" {
+		t.Errorf("wanted %d Notifications, got %d", wantNotifications, gotNotifications)
+	}
+
+}
+
 // updateLess compares 2 Update messages by the string comparison of their Paths.
 func updateLess(a, b *pb.Update) bool {
 	pathA, err := ygot.PathToString(a.GetPath())