Telemetry support for streaming events

azure-pipelines.yml

@@ -19,6 +19,10 @@ resources:
     type: github
     name: sonic-net/sonic-mgmt-common
     endpoint: sonic-net
+  - repository: sonic-swss-common
+    type: github
+    name: sonic-net/sonic-swss-common
+    endpoint: sonic-net
 
 stages:
 - stage: Build
@@ -47,7 +51,12 @@ stages:
     - checkout: sonic-mgmt-common
       clean: true
       submodules: recursive
-      displayName: 'Checkout code'
+      displayName: 'Checkout sonic-mgmt-common'
+
+    - checkout: sonic-swss-common
+      clean: true
+      submodules: recursive
+      displayName: 'Checkout sonic-swss-common'
 
     - task: DownloadPipelineArtifact@2
       inputs:
@@ -81,6 +90,16 @@ stages:
         sudo dpkg -i ../target/debs/buster/libyang*1.0.73*.deb
       displayName: "Install dependency"
 
+    - script: |
+        # LIBSWSSCOMMON
+        sudo apt-get -y purge libhiredis-dev libnl-3-dev libnl-route-3-dev
+        sudo dpkg -i ../target/debs/buster/libnl-3-200_*.deb
+        sudo dpkg -i ../target/debs/buster/libnl-genl-3-200_*.deb
+        sudo dpkg -i ../target/debs/buster/libnl-route-3-200_*.deb
+        sudo dpkg -i ../target/debs/buster/libnl-nf-3-200_*.deb
+        sudo dpkg -i ../target/debs/buster/libhiredis0.14_*.deb
+      displayName: "Install libswsscommon dependencies"
+
     - script: |
         set -ex
         # Install .NET CORE
@@ -90,6 +109,25 @@ stages:
         sudo apt-get install -y dotnet-sdk-5.0
       displayName: "Install .NET CORE"
 
+    - task: DownloadPipelineArtifact@2
+      inputs:
+        source: specific
+        project: build
+        pipeline: Azure.sonic-swss-common
+        artifact: sonic-swss-common
+        runVersion: 'latestFromBranch'
+        runBranch: 'refs/heads/master'
+      displayName: "Download sonic-swss-common"
+
+    - script: |
+        set -ex
+        # LIBSWSSCOMMON
+        sudo dpkg -i libswsscommon_1.0.0_amd64.deb
+        sudo dpkg -i libswsscommon-dev_1.0.0_amd64.deb
+        sudo dpkg -i python3-swsscommon_1.0.0_amd64.deb
+      workingDirectory: $(Pipeline.Workspace)/
+      displayName: 'Install libswsscommon package'
+
     - script: |
         set -ex
         ls -l

gnmi_server/client_subscribe.go

@@ -30,14 +30,31 @@ type Client struct {
 	// Wait for all sub go routine to finish
 	w     sync.WaitGroup
 	fatal bool
+	logLevel   int
 }
 
+// Syslog level for error
+const logLevelError int = 3
+const logLevelDebug int = 7
+const logLevelMax int = logLevelDebug
+
 // NewClient returns a new initialized client.
 func NewClient(addr net.Addr) *Client {
 	pq := queue.NewPriorityQueue(1, false)
 	return &Client{
 		addr: addr,
 		q:    pq,
+		logLevel: logLevelError,
+	}
+}
+
+func (c *Client) setLogLevel(lvl int) {
+	if (lvl >= 0) {
+		if lvl < logLevelMax {
+			c.logLevel = lvl
+		} else {
+			c.logLevel = logLevelMax
+		}
 	}
 }
 
@@ -121,8 +138,12 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 	}
 	var dc sdc.Client
 
+	mode := c.subscribe.GetMode()
+
 	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)
 	} else if _, ok, _, _ := sdc.IsTargetDb(target); ok {
 		dc, err = sdc.NewDbClient(paths, prefix)
 	} else {
@@ -134,7 +155,7 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 		return grpc.Errorf(codes.NotFound, "%v", err)
 	}
 
-	switch mode := c.subscribe.GetMode(); mode {
+	switch mode {
 	case gnmipb.SubscriptionList_STREAM:
 		c.stop = make(chan struct{}, 1)
 		c.w.Add(1)
@@ -155,7 +176,7 @@ func (c *Client) Run(stream gnmipb.GNMI_SubscribeServer) (err error) {
 
 	log.V(1).Infof("Client %s running", c)
 	go c.recv(stream)
-	err = c.send(stream)
+	err = c.send(stream, dc)
 	c.Close()
 	// Wait until all child go routines exited
 	c.w.Wait()
@@ -226,8 +247,9 @@ func (c *Client) recv(stream gnmipb.GNMI_SubscribeServer) {
 }
 
 // send runs until process Queue returns an error.
-func (c *Client) send(stream gnmipb.GNMI_SubscribeServer) error {
+func (c *Client) send(stream gnmipb.GNMI_SubscribeServer, dc sdc.Client) error {
 	for {
+		var val *sdc.Value
 		items, err := c.q.Get(1)
 
 		if items == nil {
@@ -241,12 +263,14 @@ func (c *Client) send(stream gnmipb.GNMI_SubscribeServer) error {
 		}
 
 		var resp *gnmipb.SubscribeResponse
+
 		switch v := items[0].(type) {
 		case sdc.Value:
 			if resp, err = sdc.ValToResp(v); err != nil {
 				c.errors++
 				return err
 			}
+			val = &v;
 		default:
 			log.V(1).Infof("Unknown data type %v for %s in queue", items[0], c)
 			c.errors++
@@ -257,8 +281,11 @@ func (c *Client) send(stream gnmipb.GNMI_SubscribeServer) error {
 		if err != nil {
 			log.V(1).Infof("Client %s sending error:%v", c, err)
 			c.errors++
+			dc.FailedSend()
 			return err
 		}
+
+		dc.SentOne(val)
 		log.V(5).Infof("Client %s done sending, msg count %d, msg %v", c, c.sendMsg, resp)
 	}
 }

gnmi_server/server.go

@@ -47,6 +47,7 @@ type Config struct {
 	// Port for the Server to listen on. If 0 or unset the Server will pick a port
 	// for this Server.
 	Port     int64
+	LogLevel int
 	UserAuth AuthTypes
 }
 
@@ -234,6 +235,8 @@ func (s *Server) Subscribe(stream gnmipb.GNMI_SubscribeServer) error {
 
 	c := NewClient(pr.Addr)
 
+	c.setLogLevel(s.config.LogLevel)
+
 	s.cMu.Lock()
 	if oc, ok := s.clients[c.String()]; ok {
 		log.V(2).Infof("Delete duplicate client %s", oc)

sonic_data_client/db_client.go

@@ -51,6 +51,12 @@ type Client interface {
 
 	// Close provides implemenation for explicit cleanup of Client
 	Close() error
+
+	// callbacks on send failed
+	FailedSend()
+
+	// callback on sent 
+	SentOne(*Value)
 }
 
 type Stream interface {
@@ -271,7 +277,6 @@ func (c *DbClient) PollRun(q *queue.PriorityQueue, poll chan struct{}, w *sync.W
 				SyncResponse: false,
 				Val:          val,
 			}
-
 			c.q.Put(Value{spbv})
 			log.V(6).Infof("Added spbv #%v", spbv)
 		}
@@ -488,7 +493,7 @@ func populateDbtablePath(prefix, path *gnmipb.Path, pathG2S *map[*gnmipb.Path][]
 	}
 
 	if targetDbName == "COUNTERS_DB" {
-        err := initCountersPortNameMap()
+		err := initCountersPortNameMap()
 		if err != nil {
 			return err
 		}
@@ -1230,6 +1235,12 @@ func (c *DbClient) Capabilities() []gnmipb.ModelData {
 	return nil
 }
 
+func (c *DbClient) SentOne(val *Value) {
+}
+
+func (c *DbClient) FailedSend() {
+}
+
 // validateSampleInterval validates the sampling interval of the given subscription.
 func validateSampleInterval(sub *gnmipb.Subscription) (time.Duration, error) {
 	requestedInterval := time.Duration(sub.GetSampleInterval())