Support gnmi set, add some testcases and fix some bugs

dialout/dialout_client/dialout_client.go

@@ -21,6 +21,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"os"
 )
 
 const (
@@ -94,7 +95,6 @@ func (d Destination) Validate() error {
 
 // Global config for all clients
 type ClientConfig struct {
-	SrcIp          string
 	RetryInterval  time.Duration
 	Encoding       gpb.Encoding
 	Unidirectional bool        // by default, no reponse from remote server
@@ -127,6 +127,7 @@ type clientSubscription struct {
 	sendMsg   uint64
 	recvMsg   uint64
 	errors    uint64
+	restart   bool //whether need restart connect to server flag
 }
 
 // Client handles execution of the telemetry publish service.
@@ -263,9 +264,14 @@ func newClient(ctx context.Context, dest Destination) (*Client, error) {
 	opts := []grpc.DialOption{
 		grpc.WithBlock(),
 	}
+
 	if clientCfg.TLS != nil {
 		opts = append(opts, grpc.WithTransportCredentials(credentials.NewTLS(clientCfg.TLS)))
+	} else {
+		opts = append(opts, grpc.WithInsecure())
+		log.V(2).Infof("gRPC without TLS")
 	}
+
 	conn, err := grpc.DialContext(ctx, dest.Addrs, opts...)
 	if err != nil {
 		return nil, fmt.Errorf("Dial to (%s, timeout %v): %v", dest, timeout, err)
@@ -297,6 +303,7 @@ restart: //Remote server might go down, in that case we restart with next destin
 	cs.q = queue.NewPriorityQueue(1, false)
 	cs.opened = true
 	cs.client = nil
+	cs.restart = false
 	cs.cMu.Unlock()
 
 	cs.conTryCnt++
@@ -337,48 +344,54 @@ restart: //Remote server might go down, in that case we restart with next destin
 
 	switch cs.reportType {
 	case Periodic:
+		ticker := time.NewTicker(cs.interval)
 		for {
 			select {
-			default:
-				spbValues, err := cs.dc.Get(nil)
-				if err != nil {
-					// TODO: need to inform
-					log.V(2).Infof("Data read error %v for %v", err, cs)
-					continue
-					//return nil, status.Error(codes.NotFound, err.Error())
-				}
-				var updates []*gpb.Update
-				var spbValue *spb.Value
-				for _, spbValue = range spbValues {
-					update := &gpb.Update{
-						Path: spbValue.GetPath(),
-						Val:  spbValue.GetVal(),
+			case <-ticker.C:
+				go func() {
+					spbValues, err := cs.dc.Get(nil)
+					if err != nil {
+						// TODO: need to inform
+						log.V(2).Infof("Data read error %v for %v", err, cs)
+						//continue
+						//return nil, status.Error(codes.NotFound, err.Error())
+						return
 					}
-					updates = append(updates, update)
-				}
-				rs := &gpb.SubscribeResponse_Update{
-					Update: &gpb.Notification{
-						Timestamp: spbValue.GetTimestamp(),
-						Prefix:    cs.prefix,
-						Update:    updates,
-					},
-				}
-				response := &gpb.SubscribeResponse{Response: rs}
-
-				log.V(6).Infof("cs %s sending \n\t%v \n To %s", cs.name, response, dest)
-				err = pub.Send(response)
-				if err != nil {
-					log.V(1).Infof("Client %v pub Send error:%v, cs.conTryCnt %v", cs.name, err, cs.conTryCnt)
-					cs.Close()
-					// Retry
-					goto restart
-				}
-				log.V(6).Infof("cs %s to  %s done", cs.name, dest)
-				cs.sendMsg++
-				c.sendMsg++
+					var updates []*gpb.Update
+					var spbValue *spb.Value
+					for _, spbValue = range spbValues {
+						update := &gpb.Update{
+							Path: spbValue.GetPath(),
+							Val:  spbValue.GetVal(),
+						}
+						updates = append(updates, update)
+					}
+					rs := &gpb.SubscribeResponse_Update{
+						Update: &gpb.Notification{
+							Timestamp: spbValue.GetTimestamp(),
+							Prefix:    cs.prefix,
+							Update:    updates,
+						},
+					}
+					response := &gpb.SubscribeResponse{Response: rs}
 
-				time.Sleep(cs.interval)
+					log.V(6).Infof("cs %s sending \n\t%v \n To %s", cs.name, response, dest)
+					err = pub.Send(response)
+					if err != nil {
+						log.V(1).Infof("Client %v pub Send error:%v, cs.conTryCnt %v", cs.name, err, cs.conTryCnt)
+						cs.restart = true
+						cs.Close()
+						// Retry
+						//goto restart
+					}
+					log.V(6).Infof("cs %s to  %s done", cs.name, dest)
+					cs.sendMsg++
+					c.sendMsg++
+				}()
 			case <-cs.stop:
+				if cs.restart == true {
+					goto restart
+				}
 				log.V(1).Infof("%v exiting publishRun routine for destination %s", cs, dest)
 				return
 			}
@@ -470,6 +483,7 @@ func processTelemetryClientConfig(ctx context.Context, redisDb *redis.Client, ke
 	}
 
 	log.V(2).Infof("Processing %v %v", tableKey, fv)
+
 	configMu.Lock()
 	defer configMu.Unlock()
 
@@ -482,8 +496,6 @@ func processTelemetryClientConfig(ctx context.Context, redisDb *redis.Client, ke
 		} else {
 			for field, value := range fv {
 				switch field {
-				case "src_ip":
-					clientCfg.SrcIp = value
 				case "retry_interval":
 					//TODO: check validity of the interval
 					itvl, err := strconv.ParseUint(value, 10, 64)
@@ -595,17 +607,25 @@ func processTelemetryClientConfig(ctx context.Context, redisDb *redis.Client, ke
 					cs.prefix = &gpb.Path{
 						Target: value,
 					}
+					deviceName, err := os.Hostname()
+					if err != nil {
+						log.V(2).Infof("Get hostname error: %v", err)
+					} else {
+						cs.prefix.Origin = deviceName
+					}
 				case "paths":
-					paths := strings.Split(value, ",")
-					for _, path := range paths {
-						pp, err := ygot.StringToPath(path, ygot.StructuredPath)
+					ps := strings.Split(value, ",")
+					newPaths := []*gpb.Path{}
+					for _, p := range ps {
+						pp, err := ygot.StringToPath(p, ygot.StructuredPath)
 						if err != nil {
 							log.V(2).Infof("Invalid paths %v", value)
 							return fmt.Errorf("Invalid paths %v", value)
 						}
 						// append *gpb.Path
-						cs.paths = append(cs.paths, pp)
+						newPaths = append(newPaths, pp)
 					}
+					cs.paths = newPaths
 				default:
 					log.V(2).Infof("Invalid field %v value %v", field, value)
 					return fmt.Errorf("Invalid field %v value %v", field, value)

dialout/dialout_client/dialout_client_test.go

@@ -144,7 +144,7 @@ func prepareDb(t *testing.T) {
 	mpi_name_map := loadConfig(t, "COUNTERS_PORT_NAME_MAP", countersPortNameMapByte)
 	loadDB(t, rclient, mpi_name_map)
 
-	fileName = "../../testdata/COUNTERS:Ethernet68.txt"
+	fileName = "../../testdata/COUNTERS-Ethernet68.txt"
 	countersEthernet68Byte, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		t.Fatalf("read file %v err: %v", fileName, err)
@@ -153,7 +153,7 @@ func prepareDb(t *testing.T) {
 	mpi_counter := loadConfig(t, "COUNTERS:oid:0x1000000000039", countersEthernet68Byte)
 	loadDB(t, rclient, mpi_counter)
 
-	fileName = "../../testdata/COUNTERS:Ethernet1.txt"
+	fileName = "../../testdata/COUNTERS-Ethernet1.txt"
 	countersEthernet1Byte, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		t.Fatalf("read file %v err: %v", fileName, err)
@@ -163,7 +163,7 @@ func prepareDb(t *testing.T) {
 	loadDB(t, rclient, mpi_counter)
 
 	// "Ethernet64:0": "oid:0x1500000000092a"  : queue counter, to work as data noise
-	fileName = "../../testdata/COUNTERS:oid:0x1500000000092a.txt"
+	fileName = "../../testdata/COUNTERS-oid-0x1500000000092a.txt"
 	counters92aByte, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		t.Fatalf("read file %v err: %v", fileName, err)
@@ -265,22 +265,22 @@ func TestGNMIDialOutPublish(t *testing.T) {
 	}
 	_ = countersPortNameMapByte
 
-	fileName = "../../testdata/COUNTERS:Ethernet68.txt"
+	fileName = "../../testdata/COUNTERS-Ethernet68.txt"
 	countersEthernet68Byte, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		t.Fatalf("read file %v err: %v", fileName, err)
 	}
 	_ = countersEthernet68Byte
 
-	fileName = "../../testdata/COUNTERS:Ethernet_wildcard.txt"
+	fileName = "../../testdata/COUNTERS-Ethernet_wildcard.txt"
 	countersEthernetWildcardByte, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		t.Fatalf("read file %v err: %v", fileName, err)
 	}
 
 	_ = countersEthernetWildcardByte
 
-	fileName = "../../testdata/COUNTERS:Ethernet_wildcard_PFC_7_RX.txt"
+	fileName = "../../testdata/COUNTERS-Ethernet_wildcard_PFC_7_RX.txt"
 	countersEthernetWildcardPfcByte, err := ioutil.ReadFile(fileName)
 	if err != nil {
 		t.Fatalf("read file %v err: %v", fileName, err)
@@ -289,7 +289,6 @@ func TestGNMIDialOutPublish(t *testing.T) {
 	_ = countersEthernetWildcardPfcByte
 
 	clientCfg := ClientConfig{
-		SrcIp:          "",
 		RetryInterval:  5 * time.Second,
 		Encoding:       pb.Encoding_JSON_IETF,
 		Unidirectional: true,

dialout/dialout_client_cli/dialout_client_cli.go

@@ -15,17 +15,20 @@ import (
 
 var (
 	clientCfg = dc.ClientConfig{
-		SrcIp:          "",
 		RetryInterval:  30 * time.Second,
 		Encoding:       gpb.Encoding_JSON_IETF,
 		Unidirectional: true,
-		TLS:            &tls.Config{},
 	}
+
+	tlsCfg = tls.Config{}
+
+	tlsDisable bool
 )
 
 func init() {
-	flag.StringVar(&clientCfg.TLS.ServerName, "server_name", "", "When set, use this hostname to verify server certificate during TLS handshake.")
-	flag.BoolVar(&clientCfg.TLS.InsecureSkipVerify, "insecure", false, "When set, client will not verify the server certificate during TLS handshake.")
+	flag.StringVar(&tlsCfg.ServerName, "server_name", "", "When set, use this hostname to verify server certificate during TLS handshake.")
+	flag.BoolVar(&tlsCfg.InsecureSkipVerify, "skip_verify", false, "When set, client will not verify the server certificate during TLS handshake.")
+	flag.BoolVar(&tlsDisable, "insecure", false, "Without TLS, only for testing")
 	flag.DurationVar(&clientCfg.RetryInterval, "retry_interval", 30*time.Second, "Interval at which client tries to reconnect to destination servers")
 	flag.BoolVar(&clientCfg.Unidirectional, "unidirectional", true, "No repesponse from server is expected")
 }
@@ -41,6 +44,10 @@ func main() {
 		cancel()
 	}()
 	log.V(1).Infof("Starting telemetry publish client")
+	if !tlsDisable {
+		clientCfg.TLS = &tlsCfg
+		log.V(1).Infof("TLS enable")
+	}
 	err := dc.DialOutRun(ctx, &clientCfg)
 	log.V(1).Infof("Exiting telemetry publish client: %v", err)
 	log.Flush()

dialout/dialout_server_cli/dialout_server_cli.go

@@ -11,7 +11,6 @@ import (
 	"google.golang.org/grpc/credentials"
 
 	ds "github.com/Azure/sonic-telemetry/dialout/dialout_server"
-	testcert "github.com/Azure/sonic-telemetry/testdata/tls"
 )
 
 var (
@@ -20,7 +19,7 @@ var (
 	caCert            = flag.String("ca_crt", "", "CA certificate for client certificate validation. Optional.")
 	serverCert        = flag.String("server_crt", "", "TLS server certificate")
 	serverKey         = flag.String("server_key", "", "TLS server private key")
-	insecure          = flag.Bool("insecure", false, "Skip providing TLS cert and key, for testing only!")
+	insecure          = flag.Bool("insecure", false, "Without TLS, for testing only!")
 	allowNoClientCert = flag.Bool("allow_no_client_auth", false, "When set, telemetry server will request but not require a client certificate.")
 )
 
@@ -35,12 +34,8 @@ func main() {
 	var certificate tls.Certificate
 	var err error
 
-	if *insecure {
-		certificate, err = testcert.NewCert()
-		if err != nil {
-			log.Exitf("could not load server key pair: %s", err)
-		}
-	} else {
+	var opts []grpc.ServerOption
+	if !*insecure {
 		switch {
 		case *serverCert == "":
 			log.Errorf("serverCert must be set.")
@@ -53,32 +48,32 @@ func main() {
 		if err != nil {
 			log.Exitf("could not load server key pair: %s", err)
 		}
-	}
-
-	tlsCfg := &tls.Config{
-		ClientAuth:   tls.RequireAndVerifyClientCert,
-		Certificates: []tls.Certificate{certificate},
-	}
-	if *allowNoClientCert {
-		// RequestClientCert will ask client for a certificate but won't
-		// require it to proceed. If certificate is provided, it will be
-		// verified.
-		tlsCfg.ClientAuth = tls.RequestClientCert
-	}
 
-	if *caCert != "" {
-		ca, err := ioutil.ReadFile(*caCert)
-		if err != nil {
-			log.Exitf("could not read CA certificate: %s", err)
+		tlsCfg := &tls.Config{
+			ClientAuth:   tls.RequireAndVerifyClientCert,
+			Certificates: []tls.Certificate{certificate},
 		}
-		certPool := x509.NewCertPool()
-		if ok := certPool.AppendCertsFromPEM(ca); !ok {
-			log.Exit("failed to append CA certificate")
+		if *allowNoClientCert {
+			// RequestClientCert will ask client for a certificate but won't
+			// require it to proceed. If certificate is provided, it will be
+			// verified.
+			tlsCfg.ClientAuth = tls.RequestClientCert
+		}
+
+		if *caCert != "" {
+			ca, err := ioutil.ReadFile(*caCert)
+			if err != nil {
+				log.Exitf("could not read CA certificate: %s", err)
+			}
+			certPool := x509.NewCertPool()
+			if ok := certPool.AppendCertsFromPEM(ca); !ok {
+				log.Exit("failed to append CA certificate")
+			}
+			tlsCfg.ClientCAs = certPool
 		}
-		tlsCfg.ClientCAs = certPool
+		opts = []grpc.ServerOption{grpc.Creds(credentials.NewTLS(tlsCfg))}
 	}
 
-	opts := []grpc.ServerOption{grpc.Creds(credentials.NewTLS(tlsCfg))}
 	cfg := &ds.Config{}
 	cfg.Port = int64(*port)
 	s, err := ds.NewServer(cfg, opts)