diff --git a/.azure/templates/install-go.yml b/.azure/templates/install-go.yml new file mode 100644 index 00000000..3c0db8ca --- /dev/null +++ b/.azure/templates/install-go.yml @@ -0,0 +1,23 @@ +# Azure DevOps YAML Template: Install Go toolchain +# +# Downloads and installs the requested Go release into /usr/local/go and +# exports it on PATH for subsequent steps in the same job. +# +# Usage: +# - template: .azure/templates/install-go.yml +# parameters: +# version: '1.24.4' + +parameters: +- name: version + type: string + default: '1.24.4' + +steps: +- script: | + set -euo pipefail + wget -q https://go.dev/dl/go${{ parameters.version }}.linux-amd64.tar.gz + sudo tar -C /usr/local -xzf go${{ parameters.version }}.linux-amd64.tar.gz + export PATH=$PATH:/usr/local/go/bin + go version + displayName: 'Install Go ${{ parameters.version }}' diff --git a/.azure/templates/setup-test-env.yml b/.azure/templates/setup-test-env.yml new file mode 100644 index 00000000..64be16c6 --- /dev/null +++ b/.azure/templates/setup-test-env.yml @@ -0,0 +1,50 @@ +# Azure DevOps YAML Template: Prepare a SONiC test environment +# +# Consolidates the preamble shared by every job that runs sonic-gnmi Go tests +# inside the sonic-slave container: checkout sonic-gnmi + sonic-mgmt-common + +# sonic-swss-common, install SONiC dependencies, and build sonic-mgmt-common +# (which generates Go code that sonic-gnmi imports via $(MGMT_COMMON_DIR)/build/yang). +# +# Usage: +# - template: .azure/templates/setup-test-env.yml +# parameters: +# buildBranch: $(BUILD_BRANCH) +# fetchDepth: 0 # optional; 0 means full history + +parameters: +- name: buildBranch + type: string + default: $(BUILD_BRANCH) +- name: fetchDepth + type: number + default: 1 + +steps: +- checkout: self + clean: true + submodules: recursive + fetchDepth: ${{ parameters.fetchDepth }} + displayName: 'Checkout code' + +- checkout: sonic-mgmt-common + clean: true + submodules: recursive + displayName: 'Checkout sonic-mgmt-common' + +- checkout: sonic-swss-common + clean: true + submodules: recursive + displayName: 'Checkout sonic-swss-common' + +- template: install-dependencies.yml + parameters: + buildBranch: ${{ parameters.buildBranch }} + arch: amd64 + installTestDeps: true + +- script: | + set -ex + pushd sonic-mgmt-common + NO_TEST_BINS=1 dpkg-buildpackage -rfakeroot -b -us -uc + popd + displayName: 'Build sonic-mgmt-common' diff --git a/common_utils/context.go b/common_utils/context.go index fdb3011d..740f7e65 100644 --- a/common_utils/context.go +++ b/common_utils/context.go @@ -49,6 +49,8 @@ const ( GNOI_HEALTHZ_ACK GNOI_HEALTHZ_CHECK GNOI_HEALTHZ_COLLECT + GNSI_CREDZ_SET + GNSI_CREDZ_CHECKPOINT DBUS DBUS_FAIL DBUS_APPLY_PATCH_DB @@ -95,6 +97,10 @@ func (c CounterType) String() string { return "GNOI Healthz Check" case GNOI_HEALTHZ_COLLECT: return "GNOI Healthz Collect" + case GNSI_CREDZ_SET: + return "GNSI Credz Set" + case GNSI_CREDZ_CHECKPOINT: + return "GNSI Credz Checkpoint" case DBUS: return "DBUS" case DBUS_FAIL: diff --git a/gnmi_server/db_journal.go b/gnmi_server/db_journal.go new file mode 100644 index 00000000..96c3273b --- /dev/null +++ b/gnmi_server/db_journal.go @@ -0,0 +1,293 @@ +package gnmi + +import ( + "compress/gzip" + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "time" + + log "github.com/golang/glog" + "github.com/redis/go-redis/v9" + + "github.com/Azure/sonic-mgmt-common/translib/db" + sdcfg "github.com/sonic-net/sonic-gnmi/sonic_db_config" +) + +const ( + maxFileSize = 2000000 // Bytes + maxBackups = 1 +) + +type DbJournal struct { + database string + rc *redis.Client + ps *redis.PubSub + notifications <-chan *redis.Message + cache map[string]map[string]string + file *os.File + fileName string + done chan bool +} + +var dbNums = map[string]db.DBNum{ + "CONFIG_DB": db.ConfigDB, + "STATE_DB": db.StateDB, +} + +// NewDbJournal returns a new DbJournal for the specified database. +func NewDbJournal(database string) (*DbJournal, error) { + var err error + journal := &DbJournal{} + journal.database = database + dbNum, ok := dbNums[journal.database] + if !ok { + return nil, errors.New("Invalid database passed into NewDbJournal") + } + + ns, _ := sdcfg.GetDbDefaultNamespace() + addr, _ := sdcfg.GetDbTcpAddr(journal.database, ns) + dbId, _ := sdcfg.GetDbId(journal.database, ns) + journal.rc = db.TransactionalRedisClientWithOpts(&redis.Options{ + Network: "tcp", + Addr: addr, + Password: "", + DB: dbId, + DialTimeout: 0, + }) + + if err = journal.init(); err != nil { + return nil, err + } + + keyspace := fmt.Sprintf("__keyspace@%d__:*", dbNum) + keyevent := fmt.Sprintf("__keyevent@%d__:*", dbNum) + journal.ps = journal.rc.PSubscribe(context.Background(), keyspace, keyevent) + if _, err = journal.ps.Receive(context.Background()); err != nil { + return nil, err + } + + journal.notifications = journal.ps.Channel() + + journal.fileName = filepath.Join(HostVarLogPath, strings.ToLower(journal.database)+".txt") + if journal.file, err = os.OpenFile(journal.fileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644); err != nil { + return nil, err + } + + journal.done = make(chan bool, 1) + + go journal.journal() + log.V(2).Infof("Successfully started the DbJournal for %v", journal.database) + return journal, nil +} + +// Close closes the redis objects and the journal file. +func (dbj *DbJournal) Close() { + if dbj == nil { + return + } + dbj.done <- true +} + +func (dbj *DbJournal) cleanup() { + if dbj == nil { + return + } + if dbj.ps != nil { + dbj.ps.Close() + } + if dbj.rc != nil { + db.CloseRedisClient(dbj.rc) + dbj.rc = nil + } + if dbj.file != nil { + dbj.file.Close() + } + if dbj.cache != nil { + dbj.cache = map[string]map[string]string{} + } + log.V(2).Infof("DbJournal closed successfully!") +} + +// init initializes the journal's cache. +func (dbj *DbJournal) init() error { + if dbj == nil || dbj.rc == nil { + return errors.New("DbJournal: redis client is nil") + } + dbj.cache = map[string]map[string]string{} + keys, kErr := dbj.rc.Keys(context.Background(), "*").Result() + if kErr != nil { + return kErr + } + for _, key := range keys { + entry, eErr := dbj.rc.HGetAll(context.Background(), key).Result() + if eErr != nil { + entry = map[string]string{} + } + dbj.cache[key] = entry + } + return nil +} + +// journal monitors the database notifications and logs events to the file. +func (dbj *DbJournal) journal() { + if dbj == nil { + return + } + defer dbj.cleanup() + var event []string + for { + select { + case msg := <-dbj.notifications: + event = append(event, msg.Payload) + if len(event) != 2 { + continue + } + op := event[0] + table := event[1] + entry := fmt.Sprintf("%v: %v %v", time.Now().Format("2006-01-02.15:04:05.000000"), op, table) + diff, dErr := dbj.updateCache(event) + if dErr != nil { + log.V(0).Infof("Shutting down %v Journal: %v", dbj.database, dErr) + return + } + event = []string{} + + if diff != "" { + entry += " " + diff + } + // If no fields were changed or the operation is a set on a table that contains the DB name, don't log the event. + if (diff == "" && (op == "hset" || op == "hdel")) || (op == "set" && strings.Contains(table, dbj.database)) { + continue + } + + if err := dbj.rotateFile(); err != nil { + log.V(0).Infof("Shutting down DbJournal, failed to manage file rotation: %v", err) + return + } + _, writeErr := dbj.file.Write([]byte(entry + "\n")) + if writeErr != nil { + log.V(0).Infof("Failed to write to DbJournal file: %v", writeErr) + } + case <-dbj.done: + return + } + } +} + +// updateCache updates the cache with the latest database entry and returns the diff. +func (dbj *DbJournal) updateCache(event []string) (string, error) { + op := event[0] + table := event[1] + if dbj == nil || dbj.cache == nil || dbj.rc == nil { + return "", errors.New("nil members present in DbJournal") + } + oldEntry, ok := dbj.cache[table] + if !ok { + oldEntry = map[string]string{} + } + newEntry, err := dbj.rc.HGetAll(context.Background(), table).Result() + if err != nil { + newEntry = map[string]string{} + } + // Update the cache + dbj.cache[table] = newEntry + + if op == "del" { + return "", nil + } + + diff := "" + // Find deleted and changed fields + for k, v := range oldEntry { + newVal, ok := newEntry[k] + if !ok { + diff += "-" + k + " " + continue + } + if newVal != v { + diff += k + "=" + newVal + " " + } + } + + // Find added fields + for k, v := range newEntry { + if _, ok := oldEntry[k]; !ok { + diff += "+" + k + ":" + v + " " + } + } + + return diff, nil +} + +// rotateFile makes sure the journal file is opened correctly and rotates it +// if it exceeds the maximum size. +func (dbj *DbJournal) rotateFile() error { + if dbj == nil { + return errors.New("Couldn't rotate file, DbJournal is nil") + } + fileStat, err := os.Stat(dbj.fileName) + if err != nil || dbj.file == nil { + // File does not exist or it is closed, create/open it + if dbj.file, err = os.OpenFile(dbj.fileName, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644); err != nil { + return err + } + return nil + } + + if fileStat.Size() >= maxFileSize { + // Close the journal file and open it as read-only to copy it + dbj.file.Close() + if dbj.file, err = os.OpenFile(dbj.fileName, os.O_RDONLY, 0644); err != nil { + return err + } + + // Remove a rotated, zipped file if the maxBackups limit is reached + files, err := os.ReadDir(HostVarLogPath) + if err != nil { + return err + } + var count uint + var oldest string + for _, file := range files { + if strings.HasPrefix(file.Name(), strings.ToLower(dbj.database)) && strings.HasSuffix(file.Name(), ".gz") { + count++ + if strings.Compare(file.Name(), oldest) == -1 || oldest == "" { + oldest = file.Name() + } + } + } + if count >= maxBackups { + if err := os.Remove(filepath.Join(HostVarLogPath, oldest)); err != nil { + return err + } + } + + // Compress the file + zipName := filepath.Join(HostVarLogPath, strings.ToLower(dbj.database)+"_"+time.Now().Format("20060102150405")+".gz") + zipFile, err := os.Create(zipName) + if err != nil { + return err + } + defer zipFile.Close() + zipWriter := gzip.NewWriter(zipFile) + defer zipWriter.Close() + + if _, err = io.Copy(zipWriter, dbj.file); err != nil { + return err + } + if err = zipWriter.Flush(); err != nil { + return err + } + + // Recreate the journal file + if dbj.file, err = os.Create(dbj.fileName); err != nil { + return err + } + } + return nil +} diff --git a/gnmi_server/db_journal_test.go b/gnmi_server/db_journal_test.go new file mode 100644 index 00000000..5c64b130 --- /dev/null +++ b/gnmi_server/db_journal_test.go @@ -0,0 +1,201 @@ +package gnmi + +import ( + "os" + "strings" + "testing" + "time" + + "github.com/redis/go-redis/v9" +) + +func TestNewDbJournal(t *testing.T) { + tests := []struct { + desc string + db string + wantErr bool + }{ + { + desc: "Success", + db: "CONFIG_DB", + wantErr: false, + }, + { + desc: "InvalidDb", + db: "INVALID_DB", + wantErr: true, + }, + } + + for _, test := range tests { + t.Run(test.desc, func(t *testing.T) { + journal, err := NewDbJournal(test.db) + if err == nil { + journal.Close() + } + + if test.wantErr != (err != nil) { + t.Fatalf("NewDbJournal did not return the expected error - wantErr=%v, err=%v", test.wantErr, err) + } + }) + } +} + +func TestDbJournalInit(t *testing.T) { + tests := []struct { + desc string + dbj *DbJournal + wantErr bool + }{ + { + desc: "Success", + dbj: nil, + wantErr: false, + }, + { + desc: "NilRedisClient", + dbj: &DbJournal{ + database: "CONFIG_DB", + rc: nil, + }, + wantErr: true, + }, + } + + for _, test := range tests { + t.Run(test.desc, func(t *testing.T) { + var err error + if test.dbj == nil { + if test.dbj, err = NewDbJournal("CONFIG_DB"); err != nil { + t.Fatalf("Failed to create new DbJournal: %v", err) + } + defer test.dbj.Close() + } + err = test.dbj.init() + + if test.wantErr != (err != nil) { + t.Fatalf("init did not return the expected error - wantErr=%v, err=%v", test.wantErr, err) + } + }) + } +} + +func TestDbJournalUpdateCache(t *testing.T) { + tests := []struct { + desc string + dbj *DbJournal + event []string + wantErr bool + }{ + { + desc: "SuccessHSet", + dbj: nil, + event: []string{"hset", "PORT|Ethernet1"}, + wantErr: false, + }, + { + desc: "SuccessDel", + dbj: nil, + event: []string{"del", "PORT|Ethernet1"}, + wantErr: false, + }, + { + desc: "NilCache", + dbj: &DbJournal{ + rc: &redis.Client{}, + cache: nil, + }, + event: []string{"hset", "PORT|Ethernet1"}, + wantErr: true, + }, + { + desc: "NilRedisClient", + dbj: &DbJournal{ + rc: nil, + cache: map[string]map[string]string{}, + }, + event: []string{"hset", "PORT|Ethernet1"}, + wantErr: true, + }, + } + + for _, test := range tests { + t.Run(test.desc, func(t *testing.T) { + var err error + if test.dbj == nil { + if test.dbj, err = NewDbJournal("CONFIG_DB"); err != nil { + t.Fatalf("Failed to create new DbJournal: %v", err) + } + defer test.dbj.Close() + } + _, err = test.dbj.updateCache(test.event) + + if test.wantErr != (err != nil) { + t.Fatalf("init did not return the expected error - wantErr=%v, err=%v", test.wantErr, err) + } + }) + } +} + +func TestDbJournalRotateFile(t *testing.T) { + // Set up a DbJournal + dbj, err := NewDbJournal("CONFIG_DB") + if err != nil { + t.Fatalf("Failed to create NewDbJournal: %v", err) + } + defer dbj.Close() + + // If DbJournal has a nil file pointer, it should be handled by rotateFile() + dbj.file = nil + if err := dbj.rotateFile(); err != nil { + t.Fatalf("Rotate failed because of nil file pointer: %v", err) + } + + // Fill the file a few times to make sure rotate is working correctly + for i := 0; i < maxBackups+2; i++ { + // Make sure the file was created and open it + file, err := os.OpenFile(HostVarLogPath+"/config_db.txt", os.O_APPEND|os.O_CREATE|os.O_RDWR, 0644) + if err != nil { + t.Fatalf("Failed to open DbJournal file: %v", err) + } + + // Fill the file to reach 5MB + if err := file.Truncate(maxFileSize); err != nil { + t.Fatalf("Failed to write to DbJournal file: %v", err) + } + + if err := dbj.rotateFile(); err != nil { + t.Fatalf("rotateFile failed: %v", err) + } + + time.Sleep(1 * time.Second) + + // Make sure the file was rotated + fileStat, err := os.Stat(HostVarLogPath + "/config_db.txt") + if err != nil { + t.Fatalf("Couldn't find DbJournal file: %v", err) + } + if fileStat.Size() >= 10000 { + t.Fatalf("DbJournal file was not rotated: size=%v", fileStat.Size()) + } + } + + zippedFiles := 0 + journalFiles := 0 + files, err := os.ReadDir(HostVarLogPath) + if err != nil { + t.Fatalf("Failed to read HostVarLog dir: %v", err) + } + for _, file := range files { + if file.Name() == "config_db.txt" { + journalFiles++ + } + if strings.HasPrefix(file.Name(), "config_db") && strings.HasSuffix(file.Name(), ".gz") { + zippedFiles++ + } + } + if journalFiles != 1 || zippedFiles != maxBackups { + t.Fatalf("Files not rotated correctly: journalFiles=%v, zippedFiles=%v", journalFiles, zippedFiles) + } + +} diff --git a/gnmi_server/gnsi_pathz_test.go b/gnmi_server/gnsi_pathz_test.go index e759f079..7d8cde82 100644 --- a/gnmi_server/gnsi_pathz_test.go +++ b/gnmi_server/gnsi_pathz_test.go @@ -249,6 +249,7 @@ var pathzRotationTestCases = []struct { desc: "RotatePolicyEmptyUploadRequest", f: func(ctx context.Context, t *testing.T, sc pathz.PathzClient, s *Server) { // 0) Open the streaming RPC. + t.Skip("Flaky due to race between server-side revert and test reset after RotateStreamSendError. Tracking: https://github.com/sonic-net/sonic-gnmi/issues/624") stream, err := sc.Rotate(ctx, grpc.EmptyCallOption{}) if err != nil { t.Fatal(err.Error()) @@ -276,6 +277,7 @@ var pathzRotationTestCases = []struct { { desc: "RotatePolicyEmptyRequest", f: func(ctx context.Context, t *testing.T, sc pathz.PathzClient, s *Server) { + t.Skip("Flaky due to race between server-side revert and test reset after RotateStreamSendError. Tracking: https://github.com/sonic-net/sonic-gnmi/issues/624") stream, err := sc.Rotate(ctx, grpc.EmptyCallOption{}) if err != nil { t.Fatal(err.Error()) @@ -295,6 +297,7 @@ var pathzRotationTestCases = []struct { { desc: "RotatePolicyWrongPolicyProto", f: func(ctx context.Context, t *testing.T, sc pathz.PathzClient, s *Server) { + t.Skip("Flaky due to race between server-side revert and test reset after RotateStreamSendError. Tracking: https://github.com/sonic-net/sonic-gnmi/issues/624") stream, err := sc.Rotate(ctx, grpc.EmptyCallOption{}) if err != nil { t.Fatal(err.Error()) @@ -366,6 +369,7 @@ var pathzRotationTestCases = []struct { { desc: "RotatePolicyNoVersion", f: func(ctx context.Context, t *testing.T, sc pathz.PathzClient, s *Server) { + t.Skip("Flaky due to race between server-side revert and test reset after RotateStreamSendError. Tracking: https://github.com/sonic-net/sonic-gnmi/issues/624") stream, err := sc.Rotate(ctx, grpc.EmptyCallOption{}) if err != nil { t.Fatal(err.Error()) diff --git a/gnmi_server/server.go b/gnmi_server/server.go index 5f7712fb..4887133f 100644 --- a/gnmi_server/server.go +++ b/gnmi_server/server.go @@ -8,12 +8,14 @@ import ( "crypto/x509" "encoding/pem" "errors" + "flag" "fmt" "net" "os" "path/filepath" "strings" "sync" + "syscall" "time" "github.com/Azure/sonic-mgmt-common/translib" @@ -55,6 +57,8 @@ import ( "google.golang.org/grpc/status" ) +var enableConfigDbJournal = flag.Bool("enable_config_db_journal", false, "enable config db journal") + var ( muPath = &sync.RWMutex{} supportedEncodings = []gnmipb.Encoding{gnmipb.Encoding_JSON, gnmipb.Encoding_JSON_IETF, gnmipb.Encoding_PROTO} @@ -65,6 +69,9 @@ const ( authLogPath = "/host_var/log/messages" authzRefreshingInterval = 5 * time.Second ) +const ( + HostVarLogPath = "/var/log" +) // Server manages a single gNMI Server implementation. Each client that connects // via Subscribe or Get will receive a stream of updates based on the requested @@ -97,6 +104,8 @@ type Server struct { gnsiAuthz *GNSIAuthzServer gnsiPathz *GNSIPathzServer ConnectionManager *ConnectionManager + // DB Journals + configDbJournal *DbJournal } // handleOperationalGet handles OPERATIONAL target requests directly with standard gNMI types @@ -211,6 +220,7 @@ type Config struct { ZmqPort string IdleConnDuration int ConfigTableName string + GnmiVrf string Vrf string EnableCrl bool // Path to the directory where image is stored. @@ -480,7 +490,31 @@ func SrvAdvConfig(cfg *Config) ([]grpc.ServerOption, []certprovider.Provider, er } // NewServer returns an initialized Server. -// +func createVrfListener(vrf string, port int64) (net.Listener, error) { + lc := net.ListenConfig{ + Control: func(network, address string, c syscall.RawConn) error { + var err error + ctrlErr := c.Control(func(fd uintptr) { + err = syscall.SetsockoptString(int(fd), syscall.SOL_SOCKET, syscall.SO_BINDTODEVICE, vrf) + }) + if ctrlErr != nil { + return ctrlErr + } + if err != nil { + return fmt.Errorf("failed to bind socket to VRF %s: %v", vrf, err) + } + return nil + }, + } + + listener, err := lc.Listen(context.Background(), "tcp", fmt.Sprintf(":%d", port)) + if err != nil { + return nil, err + } + log.V(1).Infof("Created VRF-bound listener on VRF %s, port %d", vrf, port) + return listener, nil +} + // tlsOpts contains TLS credentials and is used only for the TCP listener. // commonOpts contains interceptors, keepalive params, etc. and is used for both listeners. // @@ -553,7 +587,12 @@ func NewServer(config *Config, tlsOpts []grpc.ServerOption, commonOpts []grpc.Se srv.s = grpc.NewServer(tcpOpts...) reflection.Register(srv.s) - srv.lis, err = net.Listen("tcp", fmt.Sprintf(":%d", config.Port)) + // Create VRF-aware listener if GNMI VRF is specified + if config.GnmiVrf != "" && config.GnmiVrf != "default" { + srv.lis, err = createVrfListener(config.GnmiVrf, config.Port) + } else { + srv.lis, err = net.Listen("tcp", fmt.Sprintf(":%d", config.Port)) + } if err != nil { log.Warningf("Failed to open listener port %d: %v; disabling TCP listener", config.Port, err) srv.s.Stop() @@ -604,6 +643,12 @@ func NewServer(config *Config, tlsOpts []grpc.ServerOption, commonOpts []grpc.Se return nil, errors.New("no listener configured: port must be > 0 or unix_socket must be set") } + if *enableConfigDbJournal { + srv.configDbJournal, err = NewDbJournal("CONFIG_DB") + if err != nil { + return nil, fmt.Errorf("failed to create CONFIG_DB Journal: %v", err) + } + } log.V(1).Infof("Created Server on %s, read-only: %t", srv.Address(), !srv.config.EnableTranslibWrite) return srv, nil } diff --git a/gnmi_server/server_test.go b/gnmi_server/server_test.go index b7832953..8deae61a 100644 --- a/gnmi_server/server_test.go +++ b/gnmi_server/server_test.go @@ -12,6 +12,7 @@ import ( "encoding/pem" "flag" "fmt" + "io" "io/ioutil" "net" "os" @@ -27,6 +28,7 @@ import ( "time" "unsafe" + "github.com/Azure/sonic-mgmt-common/translib/db" "github.com/sonic-net/sonic-gnmi/common_utils" spb "github.com/sonic-net/sonic-gnmi/proto" sgpb "github.com/sonic-net/sonic-gnmi/proto/gnoi" @@ -5923,6 +5925,175 @@ func TestInvalidServer(t *testing.T) { } } +func TestServerConfigGnmiVrf(t *testing.T) { + // Test GNMI server creation with VRF configuration + cfg := &Config{ + Port: 8082, + EnableTranslibWrite: true, + EnableNativeWrite: true, + Threshold: 100, + GnmiVrf: "mgmt", + Vrf: "", + } + s, err := NewServer(cfg, []grpc.ServerOption{}, []grpc.ServerOption{}) + if s != nil { + defer func() { + s.ForceStop() + if s.lis != nil { + s.lis.Close() + } + }() + } + if err != nil { + t.Logf("Expected error when VRF doesn't exist: %v", err) + if !contains(err.Error(), "VRF") && !contains(err.Error(), "bind") && !contains(err.Error(), "BINDTODEVICE") { + t.Logf("Error message: %v", err) + } + return + } + if cfg.GnmiVrf != "mgmt" { + t.Errorf("Expected gnmiVrf to be 'mgmt', got '%s'", cfg.GnmiVrf) + } + if s.lis == nil { + t.Errorf("Expected server listener to be created") + } else { + addr := s.lis.Addr() + if addr == nil { + t.Errorf("Expected server listener to have an address") + } else { + t.Logf("Server successfully bound to address: %s with VRF: %s", addr.String(), s.config.GnmiVrf) + } + } +} + +func TestServerConfigZmqVrf(t *testing.T) { + // Test that ZMQ VRF field is properly set in config + cfg := &Config{ + Port: 8083, + EnableTranslibWrite: true, + EnableNativeWrite: true, + Threshold: 100, + GnmiVrf: "", + Vrf: "mgmt", + } + s, err := NewServer(cfg, []grpc.ServerOption{}, []grpc.ServerOption{}) + if s != nil { + defer func() { + s.ForceStop() + if s.lis != nil { + s.lis.Close() + } + }() + } + if err != nil { + t.Errorf("Failed to create server: %v", err) + return + } + if s.config.Vrf != "mgmt" { + t.Errorf("Expected Vrf to be 'mgmt', got '%s'", s.config.Vrf) + } +} + +func TestServerConfigGnmiVrfEmptyAndDefault(t *testing.T) { + // Test that empty GnmiVrf and "default" GnmiVrf both use default listener + tests := []struct { + name string + gnmiVrf string + }{ + {"empty GnmiVrf", ""}, + {"default GnmiVrf", "default"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := &Config{ + Port: 8084, + EnableTranslibWrite: true, + EnableNativeWrite: true, + Threshold: 100, + GnmiVrf: tt.gnmiVrf, + Vrf: "", + } + s, err := NewServer(cfg, []grpc.ServerOption{}, []grpc.ServerOption{}) + if s != nil { + defer func() { + s.ForceStop() + if s.lis != nil { + s.lis.Close() + } + }() + } + if err != nil { + t.Errorf("Failed to create server with GnmiVrf=%q: %v", tt.gnmiVrf, err) + return + } + if s.lis == nil { + t.Errorf("Expected server listener to be created") + } else { + t.Logf("Server with GnmiVrf=%q created on: %s", tt.gnmiVrf, s.lis.Addr().String()) + } + }) + } +} + +func TestCreateVrfListenerBindToDeviceError(t *testing.T) { + _, err := createVrfListener("nonexistent-vrf-12345", 0) + if err == nil { + t.Skip("Expected error when BINDTODEVICE fails, but got success. System may not support VRFs or the VRF might exist.") + } + if !contains(err.Error(), "bind") && !contains(err.Error(), "VRF") { + t.Logf("Got error (which is expected): %v", err) + } +} + +func TestCreateVrfListenerInvalidPort(t *testing.T) { + _, err := createVrfListener("", 99999) + if err == nil { + t.Fatal("Expected error for invalid port") + } +} + +func TestCreateVrfListenerSuccess(t *testing.T) { + listener, err := createVrfListener("", 0) + if err != nil { + t.Skipf("Could not create VRF listener (this is expected if not running as root or on unsupported system): %v", err) + } + if listener == nil { + t.Fatal("Expected non-nil listener") + } + defer listener.Close() + + addr := listener.Addr() + if addr == nil { + t.Error("Expected non-nil address") + } + t.Logf("Successfully created VRF listener on %s", addr.String()) + + tcpAddr, ok := addr.(*net.TCPAddr) + if !ok { + t.Errorf("Expected TCPAddr, got %T", addr) + } else { + if tcpAddr.Port == 0 { + t.Error("Expected non-zero port to be assigned") + } + t.Logf("Listener bound to port %d", tcpAddr.Port) + } +} + +func contains(s, substr string) bool { + return len(s) >= len(substr) && (s == substr || len(substr) == 0 || + (len(s) > 0 && len(substr) > 0 && containsHelper(s, substr))) +} + +func containsHelper(s, substr string) bool { + for i := 0; i <= len(s)-len(substr); i++ { + if s[i:i+len(substr)] == substr { + return true + } + } + return false +} + func TestParseOrigin(t *testing.T) { var test_paths []*gnmipb.Path var err error @@ -6483,6 +6654,94 @@ func TestGnoiAuthorization(t *testing.T) { s.Stop() } +func TestConfigDbJournal(t *testing.T) { + //Since the enableConfigDbJournal is set to false by default, enable it in UT to execute the tests + val := true + enableConfigDbJournal = &val + + ns, _ := sdcfg.GetDbDefaultNamespace() + rclient := getConfigDbClient(t, ns) + defer db.CloseRedisClient(rclient) + tests := []struct { + desc string + cmd func() + expectedEntry string + }{ + { + desc: "HSetNew", + cmd: func() { + rclient.HSet(context.Background(), "DB_JOURNAL|Test", "new", "test") + }, + expectedEntry: "hset DB_JOURNAL|Test +new:test", + }, + { + desc: "HSetExisting", + cmd: func() { + rclient.HSet(context.Background(), "DB_JOURNAL|Test", "new", "already exists") + }, + expectedEntry: "hset DB_JOURNAL|Test new=already exists", + }, + { + desc: "HDel", + cmd: func() { + rclient.HDel(context.Background(), "DB_JOURNAL|Test", "new") + }, + expectedEntry: "hdel DB_JOURNAL|Test -new", + }, + { + desc: "Set", + cmd: func() { + rclient.Set(context.Background(), "NEW_DBJOURNAL_TABLE", "TEST", 0) + }, + expectedEntry: "set NEW_DBJOURNAL_TABLE", + }, + { + desc: "Del", + cmd: func() { + rclient.Del(context.Background(), "NEW_DBJOURNAL_TABLE") + }, + expectedEntry: "del NEW_DBJOURNAL_TABLE", + }, + } + + s := createServer(t, 8081) + go runServer(t, s) + defer s.Stop() + + // Ensure the keys used in this test are not already in the DB. + rclient.Del(context.Background(), "DB_JOURNAL|Test") + rclient.Del(context.Background(), "NEW_DBJOURNAL_TABLE") + time.Sleep(500 * time.Millisecond) + + for _, test := range tests { + t.Run(test.desc, func(t *testing.T) { + // Clear the DbJournal file + err := os.Remove(HostVarLogPath + "/config_db.txt") + if err != nil { + t.Fatalf("Failed to remove journal file: %v", err) + } + + // Trigger a redis event + test.cmd() + + time.Sleep(500 * time.Millisecond) + + // Verify the contents of the file + file, err := os.Open(HostVarLogPath + "/config_db.txt") + if err != nil { + t.Fatalf("Failed to open file: %v", err) + } + defer file.Close() + data, err := io.ReadAll(file) + if err != nil { + t.Fatalf("Failed to read file: %v", err) + } + if !strings.Contains(string(data), test.expectedEntry) { + t.Fatalf("Incorrect file contents: %s", data) + } + }) + } +} func init() { // Enable logs at UT setup diff --git a/pkg/interceptors/dpuproxy/proxy.go b/pkg/interceptors/dpuproxy/proxy.go index 12c34848..298e44cb 100644 --- a/pkg/interceptors/dpuproxy/proxy.go +++ b/pkg/interceptors/dpuproxy/proxy.go @@ -9,6 +9,7 @@ import ( "github.com/golang/glog" gnoi_file_pb "github.com/openconfig/gnoi/file" + gnoi_os_pb "github.com/openconfig/gnoi/os" system "github.com/openconfig/gnoi/system" "google.golang.org/grpc" "google.golang.org/grpc/codes" @@ -62,6 +63,16 @@ var defaultForwardableMethods = []ForwardableMethod{ Description: "Install package on DPU", Mode: ForwardToDPU, }, + { + FullMethod: "/gnoi.os.OS/Verify", + Description: "Verify current OS version on DPU", + Mode: ForwardToDPU, + }, + { + FullMethod: "/gnoi.os.OS/Activate", + Description: "Activate OS version on DPU", + Mode: ForwardToDPU, + }, // gRPC reflection methods needed for grpcurl to work with DPU headers { FullMethod: "/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", @@ -251,6 +262,46 @@ func (p *DPUProxy) forwardTimeRequest(ctx context.Context, conn *grpc.ClientConn return resp, nil } +// forwardOSVerifyRequest forwards a gNOI OS.Verify request to the DPU. +func (p *DPUProxy) forwardOSVerifyRequest(ctx context.Context, conn *grpc.ClientConn, req interface{}) (interface{}, error) { + verifyReq, ok := req.(*gnoi_os_pb.VerifyRequest) + if !ok { + glog.Errorf("[DPUProxy] Invalid request type for OS.Verify method: %T", req) + return nil, status.Errorf(codes.Internal, + "invalid request type for OS.Verify: expected *os.VerifyRequest, got %T", req) + } + + client := gnoi_os_pb.NewOSClient(conn) + resp, err := client.Verify(ctx, verifyReq) + if err != nil { + glog.Errorf("[DPUProxy] Error forwarding OS.Verify request to DPU: %v", err) + return nil, err + } + + glog.Infof("[DPUProxy] Successfully forwarded OS.Verify to DPU, version: %v", resp.GetVersion()) + return resp, nil +} + +// forwardOSActivateRequest forwards a gNOI OS.Activate request to the DPU. +func (p *DPUProxy) forwardOSActivateRequest(ctx context.Context, conn *grpc.ClientConn, req interface{}) (interface{}, error) { + activateReq, ok := req.(*gnoi_os_pb.ActivateRequest) + if !ok { + glog.Errorf("[DPUProxy] Invalid request type for OS.Activate method: %T", req) + return nil, status.Errorf(codes.Internal, + "invalid request type for OS.Activate: expected *os.ActivateRequest, got %T", req) + } + + client := gnoi_os_pb.NewOSClient(conn) + resp, err := client.Activate(ctx, activateReq) + if err != nil { + glog.Errorf("[DPUProxy] Error forwarding OS.Activate request to DPU: %v", err) + return nil, err + } + + glog.Infof("[DPUProxy] Successfully forwarded OS.Activate to DPU") + return resp, nil +} + // forwardStream forwards a streaming RPC to the DPU. // This implements bidirectional streaming proxy between client and DPU. func (p *DPUProxy) forwardStream(ctx context.Context, conn *grpc.ClientConn, ss grpc.ServerStream, info *grpc.StreamServerInfo) error { @@ -477,6 +528,10 @@ func (p *DPUProxy) UnaryInterceptor() grpc.UnaryServerInterceptor { switch info.FullMethod { case "/gnoi.system.System/Time": return p.forwardTimeRequest(ctx, conn, req) + case "/gnoi.os.OS/Verify": + return p.forwardOSVerifyRequest(ctx, conn, req) + case "/gnoi.os.OS/Activate": + return p.forwardOSActivateRequest(ctx, conn, req) default: // This shouldn't happen due to getForwardingMode check, but handle gracefully glog.Errorf("[DPUProxy] Unknown forwardable method: %s", info.FullMethod) diff --git a/pkg/interceptors/dpuproxy/proxy_test.go b/pkg/interceptors/dpuproxy/proxy_test.go index 95b76416..2aa74deb 100644 --- a/pkg/interceptors/dpuproxy/proxy_test.go +++ b/pkg/interceptors/dpuproxy/proxy_test.go @@ -388,3 +388,30 @@ func TestDPUProxy_StreamInterceptor_NonForwardableMethod(t *testing.T) { t.Errorf("Expected Unimplemented code, got: %v", st.Code()) } } + +// TestAllForwardToDPUMethodsHaveHandlers validates that every method registered +// as ForwardToDPU in defaultForwardableMethods has a corresponding forwarding +// handler implemented. +func TestAllForwardToDPUMethodsHaveHandlers(t *testing.T) { + unaryHandled := map[string]bool{ + "/gnoi.system.System/Time": true, + "/gnoi.os.OS/Verify": true, + "/gnoi.os.OS/Activate": true, + } + + streamHandled := map[string]bool{ + "/gnoi.file.File/Put": true, + "/gnoi.system.System/SetPackage": true, + } + + for _, m := range defaultForwardableMethods { + if m.Mode != ForwardToDPU { + continue + } + if !unaryHandled[m.FullMethod] && !streamHandled[m.FullMethod] { + t.Errorf("ForwardToDPU method %s (%s) has no forwarding handler — "+ + "add a case to UnaryInterceptor or forwardStream AND update this test", + m.FullMethod, m.Description) + } + } +} diff --git a/sonic_data_client/client_test.go b/sonic_data_client/client_test.go index 4b6c8d71..a62abf68 100644 --- a/sonic_data_client/client_test.go +++ b/sonic_data_client/client_test.go @@ -1996,13 +1996,13 @@ func TestMixedDbClientGet(t *testing.T) { } }) - t.Run("NoData_SkipsPath", func(t *testing.T) { + t.Run("NoData_ReturnsError", func(t *testing.T) { values, err := c.Get(nil) - if err != nil { - t.Errorf("expected no error, got %v", err) + if err == nil { + t.Errorf("expected NOT_FOUND error for missing specific-key path, got nil") } if len(values) != 0 { - t.Errorf("expected empty values for missing data, got %d", len(values)) + t.Errorf("expected no values for missing data, got %d", len(values)) } }) } diff --git a/sonic_data_client/mixed_db_client.go b/sonic_data_client/mixed_db_client.go index 1569c1e6..5bf9e140 100644 --- a/sonic_data_client/mixed_db_client.go +++ b/sonic_data_client/mixed_db_client.go @@ -1644,12 +1644,20 @@ func (c *MixedDbClient) Get(w *sync.WaitGroup) ([]*spb.Value, error) { return nil, err } val, err, updateReceived := c.tableData2TypedValue(tblPaths, nil) - if !updateReceived { - continue // Skip paths with no data - } if err != nil { return nil, err } + if !updateReceived { + // Specific paths (with a tableKey or field) that produce no data are + // NOT_FOUND per gNMI spec §3.3.4. Table-only queries legitimately + // return empty data, so skip them quietly. + for _, t := range tblPaths { + if t.tableKey != "" || t.field != "" { + return nil, fmt.Errorf("No valid entry found for path %v", gnmiPath) + } + } + continue + } values = append(values, &spb.Value{ Prefix: c.prefix, Path: gnmiPath, diff --git a/sonic_data_client/super_subscription.go b/sonic_data_client/super_subscription.go new file mode 100644 index 00000000..e4e957f8 --- /dev/null +++ b/sonic_data_client/super_subscription.go @@ -0,0 +1,204 @@ +package client + +import ( + "fmt" + "sync" + "sync/atomic" + "time" + + log "github.com/golang/glog" + "github.com/golang/protobuf/proto" + gnmipb "github.com/openconfig/gnmi/proto/gnmi" + spb "github.com/sonic-net/sonic-gnmi/proto" +) + +var superSubs = superSubscriptions{ + mu: &sync.Mutex{}, + subs: map[*superSubscription]bool{}, +} + +type superSubscriptions struct { + mu *sync.Mutex + subs map[*superSubscription]bool +} + +// superSubscription is used to deduplicate subscriptions. Stream Subscriptions +// become part of a superSubscription and whenever a Sample is processed, the +// response is sent to all clients that are part of the superSubscription. +type superSubscription struct { + mu *sync.RWMutex + clients map[*TranslClient]struct{} + request *gnmipb.SubscriptionList + primaryClient *TranslClient + tickers map[int]*time.Ticker // map of interval duration (nanoseconds) to ticker. + sharedUpdates atomic.Uint64 + exclusiveUpdates atomic.Uint64 +} + +// ------------- Super Subscription Functions ------------- +// createSuperSubscription takes a SubscriptionList and returns a new +// superSubscription for that SubscriptionList. This function expects the +// caller to already hold superSubs.mu before calling createSuperSubscription. +func createSuperSubscription(subscription *gnmipb.SubscriptionList) *superSubscription { + if subscription == nil { + return nil + } + newSuperSub := &superSubscription{ + mu: &sync.RWMutex{}, + clients: map[*TranslClient]struct{}{}, + request: subscription, + primaryClient: nil, + tickers: map[int]*time.Ticker{}, + sharedUpdates: atomic.Uint64{}, + exclusiveUpdates: atomic.Uint64{}, + } + if _, ok := superSubs.subs[newSuperSub]; ok { + // This should never happen. + log.V(0).Infof("Super Subscription (%p) for %v already exists but a new has been created!", newSuperSub, subscription) + } + superSubs.subs[newSuperSub] = true + return newSuperSub +} + +// findSuperSubscription takes a SubscriptionList and tries to find an +// existing superSubscription for that SubscriptionList. If one is found, +// the superSubscription is returned. Else, nil is returned. This function +// expects the caller to already hold superSubs.mu before calling findSuperSubscription. +func findSuperSubscription(subscription *gnmipb.SubscriptionList) *superSubscription { + if subscription == nil { + return nil + } + for sub, _ := range superSubs.subs { + if sub.request == nil { + continue + } + if proto.Equal(sub.request, subscription) { + return sub + } + } + return nil +} + +// deleteSuperSub removes superSub from the superSubs map. +// If the superSub is removed from the TranslClient, there +// should be no remaining references to the superSub. This +// function expects the caller to already hold superSubs.mu +// before calling deleteSuperSubscription. +func deleteSuperSubscription(superSub *superSubscription) { + if superSub == nil { + log.V(0).Info("deleteSuperSubscription called on a nil Super Subscription!") + return + } + tickerCleanup(superSub.tickers) + delete(superSubs.subs, superSub) +} + +// ------------- Super Subscription Methods ------------- +// sendNotifications takes a value and adds it to the notification +// queue for each subscription in the superSubscription. +func (ss *superSubscription) sendNotifications(v *spb.Value) { + if v == nil { + return + } + ss.mu.RLock() + defer ss.mu.RUnlock() + for client, _ := range ss.clients { + value := proto.Clone(v).(*spb.Value) + client.q.Put(Value{value}) + } +} + +// populateTickers populates the ticker_info objects in the intervalToTickerInfoMap with the +// shared tickers. If tickers don't exist yet, they are created. +func (ss *superSubscription) populateTickers(intervalToTickerInfoMap map[int][]*ticker_info) error { + if intervalToTickerInfoMap == nil { + return fmt.Errorf("Invalid intervalToTickerInfoMap passed in: %v", intervalToTickerInfoMap) + } + ss.mu.Lock() + defer ss.mu.Unlock() + if len(ss.tickers) == 0 { + // Create the tickers. + for interval, tInfos := range intervalToTickerInfoMap { + ticker := time.NewTicker(time.Duration(interval) * time.Nanosecond) + ss.tickers[interval] = ticker + for _, tInfo := range tInfos { + tInfo.t = ticker + } + } + return nil + } + // Use the existing tickers. + if len(ss.tickers) != len(intervalToTickerInfoMap) { + return fmt.Errorf("Length of intervalToTickerInfoMap does not match length of existing tickers for Super Subscription! existing tickers=%v, intervalToTickerInfoMap=%v", ss.tickers, intervalToTickerInfoMap) + } + for interval, tInfos := range intervalToTickerInfoMap { + ticker, ok := ss.tickers[interval] + if !ok { + return fmt.Errorf("Interval in intervalToTickerInfoMap not found in existing tickers for Super Subscription! interval=%v", interval) + } + for _, tInfo := range tInfos { + tInfo.t = ticker + } + } + return nil +} +func (ss *superSubscription) String() string { + return fmt.Sprintf("[{%p} NumClients=%d, SharedUpdates=%d, ExclusiveUpdates=%d, Request=%v]", ss, len(ss.clients), ss.sharedUpdates.Load(), ss.exclusiveUpdates.Load(), ss.request) +} + +// ------------- TranslClient Methods ------------- +// isPrimary returns true if the client is the primary client of its superSubscription. +func (c *TranslClient) isPrimary() bool { + if c == nil || c.superSub == nil { + return false + } + c.superSub.mu.RLock() + defer c.superSub.mu.RUnlock() + return c.superSub.primaryClient == c +} + +// leaveSuperSubscription removes the client from the superSubscription. +// If there are no remaining clients in the superSubscription, it is deleted. +func (c *TranslClient) leaveSuperSubscription() { + if c == nil || c.superSub == nil { + return + } + superSubs.mu.Lock() + defer superSubs.mu.Unlock() + c.superSub.mu.Lock() + defer c.superSub.mu.Unlock() + delete(c.superSub.clients, c) + if len(c.superSub.clients) == 0 { + deleteSuperSubscription(c.superSub) + log.V(2).Infof("SuperSubscription (%s) closing!", c.superSub) + } else if c.superSub.primaryClient == c { + // Set a new primary client. + for client := range c.superSub.clients { + c.superSub.primaryClient = client + client.wakeChan <- true + break + } + log.V(2).Infof("SuperSubscription (%s): %p is now the primary client", c.superSub, c.superSub.primaryClient) + } +} + +// addClientToSuperSubscription adds a client to a superSubscription. +func (c *TranslClient) addClientToSuperSubscription(subscription *gnmipb.SubscriptionList) { + if c == nil || subscription == nil { + return + } + superSubs.mu.Lock() + defer superSubs.mu.Unlock() + superSub := findSuperSubscription(subscription) + if superSub == nil { + superSub = createSuperSubscription(subscription) + } + superSub.mu.Lock() + defer superSub.mu.Unlock() + c.superSub = superSub + superSub.clients[c] = struct{}{} + if superSub.primaryClient == nil { + superSub.primaryClient = c + } + log.V(2).Infof("SuperSubscription (%s): added new client=%p", superSub, c) +} diff --git a/sonic_data_client/super_subscription_test.go b/sonic_data_client/super_subscription_test.go new file mode 100644 index 00000000..b118acca --- /dev/null +++ b/sonic_data_client/super_subscription_test.go @@ -0,0 +1,438 @@ +package client + +import ( + "github.com/Workiva/go-datastructures/queue" + pb "github.com/openconfig/gnmi/proto/gnmi" + spb "github.com/sonic-net/sonic-gnmi/proto" + "sync" + "testing" + "time" +) + +var ( + SAMPLE_SUB = &pb.SubscriptionList{ + Prefix: &pb.Path{Origin: "openconfig", Target: "OC_YANG"}, + Mode: pb.SubscriptionList_STREAM, + Encoding: pb.Encoding_PROTO, + Subscription: []*pb.Subscription{ + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_SAMPLE, + SampleInterval: 2000000000, + }, + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_SAMPLE, + SampleInterval: 30000000000, + }, + }, + } + ONCHANGE_SUB = &pb.SubscriptionList{ + Prefix: &pb.Path{Origin: "openconfig", Target: "OC_YANG"}, + Mode: pb.SubscriptionList_STREAM, + Encoding: pb.Encoding_PROTO, + Subscription: []*pb.Subscription{ + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_ON_CHANGE, + }, + }, + } + MIXED_SUB = &pb.SubscriptionList{ + Prefix: &pb.Path{Origin: "openconfig", Target: "OC_YANG"}, + Mode: pb.SubscriptionList_STREAM, + Encoding: pb.Encoding_PROTO, + Subscription: []*pb.Subscription{ + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_ON_CHANGE, + }, + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_SAMPLE, + SampleInterval: 30000000000, + }, + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_TARGET_DEFINED, + }, + }, + } + TD_SUB = &pb.SubscriptionList{ + Prefix: &pb.Path{Origin: "openconfig", Target: "OC_YANG"}, + Mode: pb.SubscriptionList_STREAM, + Encoding: pb.Encoding_PROTO, + Subscription: []*pb.Subscription{ + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_TARGET_DEFINED, + }, + { + Path: &pb.Path{Elem: []*pb.PathElem{{Name: "interfaces"}}}, + Mode: pb.SubscriptionMode_TARGET_DEFINED, + SampleInterval: 4000000000, + }, + }, + } +) + +func clearSuperSubscriptions() { + superSubs.mu.Lock() + defer superSubs.mu.Unlock() + //clear(superSubs.subs) + for k := range superSubs.subs { + delete(superSubs.subs, k) + } +} +func TestCreateSuperSubscription(t *testing.T) { + tests := []struct { + name string + subscription *pb.SubscriptionList + expectNil bool + }{ + { + name: "CreateSuperSubscription", + subscription: SAMPLE_SUB, + expectNil: false, + }, + { + name: "CreateNilSuperSubscription", + subscription: nil, + expectNil: true, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + if superSub := createSuperSubscription(test.subscription); (superSub == nil) != test.expectNil { + t.Errorf("createSuperSubscription returned invalid SuperSubscription: expectNil=%v, got=%v", test.expectNil, superSub) + } + numSuperSubs := len(superSubs.subs) + if test.expectNil && numSuperSubs != 0 { + t.Errorf("createSuperSubscription incorrectly added a SuperSubscription to the map: len(superSubs)=%v", numSuperSubs) + } else if !test.expectNil && numSuperSubs != 1 { + t.Errorf("createSuperSubscription did not add a SuperSubscription to the map: len(superSubs)=%v", numSuperSubs) + } + }) + } +} +func TestFindSuperSubscription(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + createSuperSubscription(SAMPLE_SUB) + createSuperSubscription(MIXED_SUB) + createSuperSubscription(TD_SUB) + tests := []struct { + name string + subscription *pb.SubscriptionList + expectedOk bool + }{ + { + name: "FindExistingSuperSubscription", + subscription: SAMPLE_SUB, + expectedOk: true, + }, + { + name: "FindNonExistingSuperSubscription", + subscription: ONCHANGE_SUB, + expectedOk: false, + }, + { + name: "FindNilSuperSubscription", + subscription: nil, + expectedOk: false, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + superSub := findSuperSubscription(test.subscription) + if test.expectedOk && superSub == nil { + t.Errorf("findSuperSubscription incorrectly returned a nil superSubscription: want=%v, got=%v", test.expectedOk, superSub) + } else if !test.expectedOk && superSub != nil { + t.Errorf("findSuperSubscription incorrectly returned a non-nil superSubscription") + } + }) + } +} +func TestDeleteSuperSubscription(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + tests := []struct { + name string + superSub *superSubscription + }{ + { + name: "DeleteExistingSuperSubscription", + superSub: createSuperSubscription(SAMPLE_SUB), + }, + { + name: "DeleteNonExistingSuperSubscription", + superSub: &superSubscription{}, + }, + { + name: "DeleteNilSuperSubscription", + superSub: nil, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + deleteSuperSubscription(test.superSub) + if test.superSub == nil { + return + } + if _, ok := superSubs.subs[test.superSub]; ok { + t.Errorf("deleteSuperSubscription did not remove %v from the superSubs map: %v", test.superSub, superSubs.subs) + } + }) + } +} +func TestSendNotifications(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + clients := []*TranslClient{} + superSub := createSuperSubscription(SAMPLE_SUB) + for i := 0; i < 5; i++ { + newClient := &TranslClient{ + q: queue.NewPriorityQueue(1, false), + } + clients = append(clients, newClient) + newClient.addClientToSuperSubscription(SAMPLE_SUB) + } + superSub.sendNotifications(&spb.Value{}) + for i := 0; i < 5; i++ { + if _, err := clients[i].q.Get(1); err != nil { + t.Errorf("Error receiving notifications: %v", err) + } + } + // Nil value passed into sendNotifications should be a no-op + superSub.sendNotifications(nil) + for i := 0; i < 5; i++ { + if !clients[i].q.Empty() { + t.Errorf("A notification was incorrectly written to the queue: length=%v", clients[i].q.Len()) + } + } +} +func TestPopulateTickers(t *testing.T) { + tests := []struct { + name string + superSub *superSubscription + ticker_map map[int][]*ticker_info + expectErr bool + expectedLen int + }{ + { + name: "CreateNewTicker", + superSub: &superSubscription{ + mu: &sync.RWMutex{}, + tickers: map[int]*time.Ticker{}, + }, + ticker_map: map[int][]*ticker_info{ + 10000: []*ticker_info{&ticker_info{interval: 10000}, &ticker_info{interval: 10000}}, + 30000: []*ticker_info{&ticker_info{interval: 30000}}, + }, + expectErr: false, + expectedLen: 2, + }, + { + name: "UseExistingTickers", + superSub: &superSubscription{ + mu: &sync.RWMutex{}, + tickers: map[int]*time.Ticker{ + 10000: time.NewTicker(10000 * time.Nanosecond), + 30000: time.NewTicker(30000 * time.Nanosecond), + }, + }, + ticker_map: map[int][]*ticker_info{ + 10000: []*ticker_info{&ticker_info{interval: 10000}, &ticker_info{interval: 10000}}, + 30000: []*ticker_info{&ticker_info{interval: 30000}}, + }, + expectErr: false, + expectedLen: 2, + }, + { + name: "LengthOfTickerMapsDontMatch", + superSub: &superSubscription{ + mu: &sync.RWMutex{}, + tickers: map[int]*time.Ticker{ + 10000: time.NewTicker(10000 * time.Nanosecond), + }, + }, + ticker_map: map[int][]*ticker_info{ + 10000: []*ticker_info{&ticker_info{interval: 10000}, &ticker_info{interval: 10000}}, + 30000: []*ticker_info{&ticker_info{interval: 30000}}, + }, + expectErr: true, + }, + { + name: "TickerMapKeysDontMatch", + superSub: &superSubscription{ + mu: &sync.RWMutex{}, + tickers: map[int]*time.Ticker{ + 10000: time.NewTicker(10000 * time.Nanosecond), + 20000: time.NewTicker(20000 * time.Nanosecond), + }, + }, + ticker_map: map[int][]*ticker_info{ + 10000: []*ticker_info{&ticker_info{interval: 10000}, &ticker_info{interval: 10000}}, + 30000: []*ticker_info{&ticker_info{interval: 30000}}, + }, + expectErr: true, + }, + { + name: "NilTickerMap", + superSub: &superSubscription{ + mu: &sync.RWMutex{}, + tickers: map[int]*time.Ticker{}, + }, + ticker_map: nil, + expectErr: true, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + // Save pointers to the original tickers to ensure they were not overwritten. + origTickers := map[int]*time.Ticker{} + for interval, ticker := range test.superSub.tickers { + origTickers[interval] = ticker + } + err := test.superSub.populateTickers(test.ticker_map) + if err != nil { + if !test.expectErr { + t.Errorf("Unexpected error returned from populateTickers: %v", err) + } + return + } else if test.expectErr { + t.Fatalf("Test expected error but got %v", err) + } + for interval, tInfos := range test.ticker_map { + for _, tInfo := range tInfos { + if tInfo.t == nil { + t.Errorf("Nil ticker in ticker_map") + } + origTicker, ok := origTickers[interval] + if ok && origTicker != tInfo.t { + t.Errorf("Original ticker was overwritten: original=%p, new=%p", origTicker, tInfo.t) + } + } + } + if len(test.superSub.tickers) != test.expectedLen { + t.Errorf("Unexpected number of tickers in the Super Subscription: want=%v, got=%v", test.expectedLen, len(test.superSub.tickers)) + } + }) + } +} +func TestIsPrimary(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + client1 := &TranslClient{q: queue.NewPriorityQueue(1, false)} + client2 := &TranslClient{q: queue.NewPriorityQueue(1, false)} + client1.addClientToSuperSubscription(SAMPLE_SUB) + client2.addClientToSuperSubscription(SAMPLE_SUB) + if !client1.isPrimary() { + t.Errorf("Expected client1 to be the primary client but it is not") + } + if client2.isPrimary() { + t.Errorf("Expected client2 to not be the primary client but it is") + } + client1.superSub.primaryClient = nil + if client1.isPrimary() { + t.Errorf("The primaryClient is nil but client1.isPrimary() returned true") + } + client1.superSub = nil + if client1.isPrimary() { + t.Errorf("The Super Subscription is nil but client1.isPrimary() returned true") + } +} +func TestLeaveSuperSubscription(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + client1 := &TranslClient{q: queue.NewPriorityQueue(1, false), wakeChan: make(chan bool, 1)} + client2 := &TranslClient{q: queue.NewPriorityQueue(1, false), wakeChan: make(chan bool, 1)} + client1.addClientToSuperSubscription(SAMPLE_SUB) + client2.addClientToSuperSubscription(SAMPLE_SUB) + superSub := client1.superSub + client1.leaveSuperSubscription() + if _, ok := superSubs.subs[superSub]; !ok { + t.Errorf("Super Subscription deleted from the global map: %v", superSubs.subs) + } + if len(superSub.clients) != 1 { + t.Errorf("Super Subscription clients not updated correctly: want: length=1, got: length=%v", len(superSub.clients)) + } + if superSub.primaryClient != client2 { + t.Errorf("Super Subscription primary client not set correctly: want=%v, got=%v", client2, superSub.primaryClient) + } + client2.leaveSuperSubscription() + if _, ok := superSubs.subs[superSub]; ok { + t.Errorf("Super Subscription not deleted from the global map: %v", superSubs.subs) + } + if len(superSub.clients) != 0 { + t.Errorf("Super Subscription clients not updated correctly: want: length=0, got: length=%v", len(superSub.clients)) + } + client1.superSub = nil + client1.leaveSuperSubscription() +} +func TestAddClientToSuperSubscription(t *testing.T) { + t.Cleanup(clearSuperSubscriptions) + client1 := &TranslClient{q: queue.NewPriorityQueue(1, false)} + client2 := &TranslClient{q: queue.NewPriorityQueue(1, false)} + client3 := &TranslClient{q: queue.NewPriorityQueue(1, false)} + client1.addClientToSuperSubscription(SAMPLE_SUB) + client2.addClientToSuperSubscription(SAMPLE_SUB) + superSub := client1.superSub + if client1.superSub == nil || len(superSub.clients) != 2 { + t.Fatalf("superSub not created correctly: %v", client1.superSub) + } + if superSub.primaryClient != client1 { + t.Errorf("superSub primary client not set correctly: want=%v, got=%v", client1, superSub.primaryClient) + } + if client2.superSub == nil { + t.Errorf("superSub not created correctly: %v", client2.superSub) + } + if client2.superSub.primaryClient != client1 { + t.Errorf("superSub primary client not set correctly: want=%v, got=%v", client1, client2.superSub.primaryClient) + } + client3.addClientToSuperSubscription(nil) + if client3.superSub != nil { + t.Errorf("superSub created incorrectly: %v", client3.superSub) + } +} + +func TestFindSuperSubscription_NilRequestInMap(t *testing.T) { + // 1. Setup the input subscription we are looking for + targetSub := &pb.SubscriptionList{ + Mode: pb.SubscriptionList_STREAM, + } + + // 2. Initialize or mock superSubs to contain a nil request + superSubs.subs = make(map[*superSubscription]bool) + + // Entry A: The nil request case + badEntry := &superSubscription{request: nil} + superSubs.subs[badEntry] = true + + // Entry B: A valid request that doesn't match + wrongEntry := &superSubscription{ + request: &pb.SubscriptionList{Mode: pb.SubscriptionList_ONCE}, + } + superSubs.subs[wrongEntry] = true + + // 3. Execution + result := findSuperSubscription(targetSub) + + // 4. Validation + if result != nil { + t.Errorf("Expected nil return when no matches exist, even with nil requests in map") + } +} +func TestSuperSubscription_String(t *testing.T) { + // 1. Setup mock data + ss := &superSubscription{ + clients: make(map[*TranslClient]struct{}), + request: &pb.SubscriptionList{ + Mode: pb.SubscriptionList_STREAM, + }, + } + + // Set some values for the atomic counters + ss.sharedUpdates.Store(10) + ss.exclusiveUpdates.Store(5) + + // 2. Execution + ss.String() + +} diff --git a/sonic_data_client/transl_data_client.go b/sonic_data_client/transl_data_client.go index 21e63925..95ac4860 100644 --- a/sonic_data_client/transl_data_client.go +++ b/sonic_data_client/transl_data_client.go @@ -3,9 +3,11 @@ package client import ( "context" + "flag" "fmt" "reflect" "runtime" + "strings" "sync" "time" @@ -24,27 +26,66 @@ import ( ) const ( - DELETE int = 0 - REPLACE int = 1 - UPDATE int = 2 + DELETE int = 0 + REPLACE int = 1 + UPDATE int = 2 + maxWorkers = 2 // max workers for parallel processing of Get/Subscribe requests ) +var useParallelProcessing = flag.Bool("use_parallel_processing", false, "use parallel processing for GET/SUBSCRIBE requests") +var translProcessGetFunc = transutil.TranslProcessGet +var isSubscribeSupportedFunc = translib.IsSubscribeSupported + type TranslClient struct { prefix *gnmipb.Path /* GNMI Path to REST URL Mapping */ path2URI map[*gnmipb.Path]string channel chan struct{} q *queue.PriorityQueue + superSub *superSubscription synced sync.WaitGroup // Control when to send gNMI sync_response w *sync.WaitGroup // wait for all sub go routines to finish mu sync.RWMutex // Mutex for data protection among routines for transl_client ctx context.Context //Contains Auth info and request info + wakeChan chan bool // wakeChan is used to wake up the client to notify it is the new primary client. extensions []*gnmi_extpb.Extension version *translib.Version // Client version; populated by parseVersion() encoding gnmipb.Encoding } +type pathFromGetReq struct { + path *gnmipb.Path + uri string + c *TranslClient +} + +type pathFromSubReq struct { + path string + ts *translSubscriber +} + +func findMin(a, b int) int { + if a < b { + return a + } + return b +} + +func joinErrorsWithNewline(errs []error) error { + //Length of 'err' validation is done before func call. + var errStrings []string + for _, err := range errs { + if err != nil { + errStrings = append(errStrings, err.Error()) + } + } + if len(errStrings) == 0 { + return nil // Return actual nil, not an empty error object + } + // Using "\n" matches the visual output of errors.Join + return fmt.Errorf("%s", strings.Join(errStrings, "\n")) +} func NewTranslClient(prefix *gnmipb.Path, getpaths []*gnmipb.Path, ctx context.Context, extensions []*gnmi_extpb.Extension, opts ...TranslClientOption) (Client, error) { var client TranslClient @@ -65,6 +106,7 @@ func NewTranslClient(prefix *gnmipb.Path, getpaths []*gnmipb.Path, ctx context.C /* Populate GNMI path to REST URL map. */ err = transutil.PopulateClientPaths(prefix, getpaths, &client.path2URI, addWildcardKeys) } + client.wakeChan = make(chan bool, 1) if err != nil { return nil, err @@ -76,6 +118,7 @@ func NewTranslClient(prefix *gnmipb.Path, getpaths []*gnmipb.Path, ctx context.C func (c *TranslClient) Get(w *sync.WaitGroup) ([]*spb.Value, error) { rc, ctx := common_utils.GetContext(c.ctx) c.ctx = ctx + var errs []error var values []*spb.Value ts := time.Now() @@ -83,22 +126,55 @@ func (c *TranslClient) Get(w *sync.WaitGroup) ([]*spb.Value, error) { if version != nil { rc.BundleVersion = version } - /* Iterate through all GNMI paths. */ - for gnmiPath, URIPath := range c.path2URI { - /* Fill values for each GNMI path. */ - val, err := transutil.TranslProcessGet(URIPath, nil, c.ctx) + // Iterate through all GNMI paths in parallel. The max number of workers is equal to + // the number of Openconfig modules in a root query. + numPaths := len(c.path2URI) + pathChan := make(chan pathFromGetReq, numPaths) + valueChan := make(chan *spb.Value, numPaths) + errChan := make(chan error, numPaths) + workerWg := &sync.WaitGroup{} + readerWg := &sync.WaitGroup{} + + numWorkers := 1 + if *useParallelProcessing { + numWorkers = findMin(numPaths, maxWorkers) + } - if err != nil { - return nil, err + // Spawn workers to process the paths in the GET request. + for i := 0; i < numWorkers; i++ { + workerWg.Add(1) + go processGetWorker(pathChan, valueChan, errChan, workerWg) + } + + // Send the paths in the GET request to the workers through the channel. + for gnmiPath, URIPath := range c.path2URI { + pathChan <- pathFromGetReq{path: gnmiPath, uri: URIPath, c: c} + } + close(pathChan) + + // Start goroutines to read the response values and errors from the workers. + readerWg.Add(2) + go func() { + defer readerWg.Done() + for value := range valueChan { + values = append(values, value) + } + }() + go func() { + defer readerWg.Done() + for err := range errChan { + errs = append(errs, err) } + }() - /* Value of each path is added to spb value structure. */ - values = append(values, &spb.Value{ - Prefix: c.prefix, - Path: gnmiPath, - Timestamp: ts.UnixNano(), - Val: val, - }) + // Wait for all goroutines to finish. + workerWg.Wait() + close(valueChan) + close(errChan) + readerWg.Wait() + + if len(errs) != 0 { + return nil, joinErrorsWithNewline(errs) } /* The values structure at the end is returned and then updates in notitications as @@ -110,6 +186,31 @@ func (c *TranslClient) Get(w *sync.WaitGroup) ([]*spb.Value, error) { return values, nil } +// processGetWorker is a worker function to remove paths from a GET request off of the channel and process them. +func processGetWorker(pathChan <-chan pathFromGetReq, valueChan chan<- *spb.Value, errChan chan<- error, wg *sync.WaitGroup) { + defer wg.Done() + for path := range pathChan { + log.Infof("getWorker processing path: %v", path) + val, resp, err := translProcessGetFunc(path.uri, nil, path.c.ctx, path.c.encoding) + if err != nil { + errChan <- err + return + } + + var valueTree ygot.ValidatedGoStruct + if resp != nil { + valueTree = resp.ValueTree + } + v, err := buildValueForGet(path.c.prefix, path.path, path.c.encoding, val, valueTree) + if err != nil { + errChan <- err + return + } + + valueChan <- v + } +} + func (c *TranslClient) Set(delete []*gnmipb.Path, replace []*gnmipb.Update, update []*gnmipb.Update) error { rc, ctx := common_utils.GetContext(c.ctx) c.ctx = ctx @@ -168,6 +269,7 @@ func recoverSubscribe(c *TranslClient) { type ticker_info struct { t *time.Ticker sub *gnmipb.Subscription + interval int pathStr string heartbeat bool } @@ -183,11 +285,9 @@ func getTranslNotificationType(mode gnmipb.SubscriptionMode) translib.Notificati } } -func tickerCleanup(ticker_map map[int][]*ticker_info, c *TranslClient) { - for _, v := range ticker_map { - for _, ti := range v { - ti.t.Stop() - } +func tickerCleanup(tickers map[int]*time.Ticker) { + for _, ticker := range tickers { + ticker.Stop() } } @@ -205,12 +305,8 @@ func (c *TranslClient) StreamRun(q *queue.PriorityQueue, stop chan struct{}, w * enqueFatalMsgTranslib(c, err.Error()) return } + intervalToTickerInfoMap := make(map[int][]*ticker_info) - ticker_map := make(map[int][]*ticker_info) - - defer tickerCleanup(ticker_map, c) - var cases []reflect.SelectCase - cases_map := make(map[int]int) var subscribe_mode gnmipb.SubscriptionMode translPaths := make([]translib.IsSubscribePath, len(subscribe.Subscription)) sampleCache := make(map[string]*ygotCache) @@ -234,13 +330,25 @@ func (c *TranslClient) StreamRun(q *queue.PriorityQueue, stop chan struct{}, w * req.ClientVersion = *c.version } - subSupport, err := translib.IsSubscribeSupported(req) + subSupport, err := isSubscribeSupportedFunc(req) if err != nil { enqueFatalMsgTranslib(c, fmt.Sprintf("Subscribe operation failed with error =%v", err.Error())) return } + numSubWorkers := 1 + if *useParallelProcessing { + numSubWorkers = findMin(len(subscribe.Subscription), maxWorkers) + } + var onChangeSubsString []string + // Spawn routines to process the Sample subscriptions + wg := &sync.WaitGroup{} + subChan := make(chan pathFromSubReq, len(subscribe.Subscription)) + for i := 0; i < numSubWorkers; i++ { + wg.Add(1) + go processSubWorker(subChan, wg) + } for i, pInfo := range subSupport { sub := subscribe.Subscription[pInfo.ID] @@ -259,12 +367,14 @@ func (c *TranslClient) StreamRun(q *queue.PriorityQueue, stop chan struct{}, w * subscribe_mode = gnmipb.SubscriptionMode_ON_CHANGE } else { enqueFatalMsgTranslib(c, fmt.Sprintf("ON_CHANGE Streaming mode invalid for %v", pathStr)) + close(subChan) return } case gnmipb.SubscriptionMode_SAMPLE: subscribe_mode = gnmipb.SubscriptionMode_SAMPLE default: enqueFatalMsgTranslib(c, fmt.Sprintf("Invalid Subscription Mode %d", sub.Mode)) + close(subChan) return } @@ -275,6 +385,7 @@ func (c *TranslClient) StreamRun(q *queue.PriorityQueue, stop chan struct{}, w * if hb := sub.HeartbeatInterval; hb > 0 && hb < uint64(pInfo.MinInterval)*uint64(time.Second) { enqueFatalMsgTranslib(c, fmt.Sprintf("Invalid Heartbeat Interval %ds, minimum interval is %ds", sub.HeartbeatInterval/uint64(time.Second), subSupport[i].MinInterval)) + close(subChan) return } @@ -286,6 +397,7 @@ func (c *TranslClient) StreamRun(q *queue.PriorityQueue, stop chan struct{}, w * interval = minInterval } else if interval < minInterval { enqueFatalMsgTranslib(c, fmt.Sprintf("Invalid SampleInterval %ds, minimum interval is %ds", interval/int(time.Second), pInfo.MinInterval)) + close(subChan) return } @@ -306,79 +418,152 @@ func (c *TranslClient) StreamRun(q *queue.PriorityQueue, stop chan struct{}, w * } // do initial sync & build the cache - ts.doSample(pathStr) + // Add the path to the channel to be processed in parallel + subChan <- pathFromSubReq{path: pathStr, ts: &ts} - addTimer(c, ticker_map, &cases, cases_map, interval, sub, pathStr, false) + addTimer(intervalToTickerInfoMap, interval, sub, pathStr, false) //Heartbeat intervals are valid for SAMPLE in the case suppress_redundant is specified if sub.SuppressRedundant && sub.HeartbeatInterval > 0 { - addTimer(c, ticker_map, &cases, cases_map, int(sub.HeartbeatInterval), sub, pathStr, true) + addTimer(intervalToTickerInfoMap, int(sub.HeartbeatInterval), sub, pathStr, true) } } else if subscribe_mode == gnmipb.SubscriptionMode_ON_CHANGE { onChangeSubsString = append(onChangeSubsString, pathStr) if sub.HeartbeatInterval > 0 { - addTimer(c, ticker_map, &cases, cases_map, int(sub.HeartbeatInterval), sub, pathStr, true) + addTimer(intervalToTickerInfoMap, int(sub.HeartbeatInterval), sub, pathStr, true) } } log.V(6).Infof("End Sub: %v", sub) } + close(subChan) if len(onChangeSubsString) > 0 { ts := translSubscriber{ client: c, session: ss, filterMsgs: subscribe.UpdatesOnly, + sampleWg: wg, } ts.doOnChange(onChangeSubsString) } else { // If at least one ON_CHANGE subscription was present, then // ts.doOnChange() would have sent the sync message. // Explicitly send one here if all are SAMPLE subscriptions. + wg.Wait() enqueueSyncMessage(c) } - cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(c.channel)}) + // Add the subscription to a SuperSubscrption. + c.addClientToSuperSubscription(subscribe) + defer c.leaveSuperSubscription() + + // Get tickers and cases to do samples. + err = c.superSub.populateTickers(intervalToTickerInfoMap) + if err != nil { + errMsg := fmt.Sprintf("Failed to fetch shared tickers from Super Subscription: %v", err) + log.V(0).Info(errMsg) + enqueFatalMsgTranslib(c, errMsg) + return + } + cases, caseIndexToIntervalMap := buildSelectCases(intervalToTickerInfoMap, c.channel) + + // Non-primary clients do not need to listen for ticks unless they're woken up by the Super Subscription. + for !c.isPrimary() { + select { + case <-c.wakeChan: + log.V(2).Infof("Secondary client (%p) waking up: isPrimary=%v, tickers=%v", c, c.isPrimary(), intervalToTickerInfoMap) + continue + case <-c.channel: + log.V(2).Infof("Secondary client (%p) received close signal", c) + return + } + } for { chosen, _, ok := reflect.Select(cases) - if !ok { + if !ok || chosen >= len(caseIndexToIntervalMap) || c.q == nil || c.q.Disposed() { + log.V(2).Infof("TranslClient (%p) exiting StreamRun because an exit signal was received!", c) return } - for _, tick := range ticker_map[cases_map[chosen]] { + // Start goroutines to process the Sample. + ticks := intervalToTickerInfoMap[caseIndexToIntervalMap[chosen]] + tickSubChan := make(chan pathFromSubReq, len(ticks)) + numTickWorkers := 1 + if *useParallelProcessing { + numTickWorkers = findMin(len(ticks), maxWorkers) + } + for i := 0; i < numTickWorkers; i++ { + wg.Add(1) + go processSubWorker(tickSubChan, wg) + } + for _, tick := range ticks { log.V(6).Infof("tick, heartbeat: %t, path: %s\n", tick.heartbeat, c.path2URI[tick.sub.Path]) ts := translSubscriber{ client: c, session: ss, sampleCache: sampleCache[tick.pathStr], filterDups: (!tick.heartbeat && tick.sub.SuppressRedundant), + sampleWg: wg, } - ts.doSample(tick.pathStr) + tickSubChan <- pathFromSubReq{path: tick.pathStr, ts: &ts} } + close(tickSubChan) + wg.Wait() + } +} + +func processSubWorker(subChan <-chan pathFromSubReq, wg *sync.WaitGroup) { + defer wg.Done() + for sub := range subChan { + sub.ts.doSample(sub.path) } } -func addTimer(c *TranslClient, ticker_map map[int][]*ticker_info, cases *[]reflect.SelectCase, cases_map map[int]int, interval int, sub *gnmipb.Subscription, pathStr string, heartbeat bool) { - //Reuse ticker for same sample intervals, otherwise create a new one. - if ticker_map[interval] == nil { - ticker_map[interval] = make([]*ticker_info, 1, 1) - ticker_map[interval][0] = &ticker_info{ - t: time.NewTicker(time.Duration(interval) * time.Nanosecond), +// addTimer adds a new ticker_info with the given interval, sub, and heartbeat to the intervalToTickerInfoMap. +func addTimer(intervalToTickerInfoMap map[int][]*ticker_info, interval int, sub *gnmipb.Subscription, pathStr string, heartbeat bool) { + if tickers, ok := intervalToTickerInfoMap[interval]; ok && tickers != nil { + tickers = append(tickers, &ticker_info{ sub: sub, pathStr: pathStr, + interval: interval, heartbeat: heartbeat, - } - cases_map[len(*cases)] = interval - *cases = append(*cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ticker_map[interval][0].t.C)}) + }) + intervalToTickerInfoMap[interval] = tickers } else { - ticker_map[interval] = append(ticker_map[interval], &ticker_info{ - t: ticker_map[interval][0].t, + intervalToTickerInfoMap[interval] = []*ticker_info{{ sub: sub, pathStr: pathStr, + interval: interval, heartbeat: heartbeat, - }) + }} } } +// buildSelectCases takes in the intervalToTickerInfoMap and a close channel and returns the cases and caseIndexToIntervalMap that include +// those ticker channels. +// - cases is a slice of SelectCase used to call select across all the ticker and close channels. +// - caseIndexToIntervalMap is a map from ticker index within the cases slice to the corresponding interval. It can be used to +// associate an interval with the chosen case during the select operation. +func buildSelectCases(intervalToTickerInfoMap map[int][]*ticker_info, closeChan <-chan struct{}) ([]reflect.SelectCase, map[int]int) { + cases := make([]reflect.SelectCase, 0) + caseIndexToIntervalMap := make(map[int]int) + + for interval, tickers := range intervalToTickerInfoMap { + if len(tickers) < 1 { + continue + } + ticker := tickers[0] + if ticker == nil { + log.V(0).Infof("Failed to build select case for interval %v because ticker is nil!", interval) + continue + } + caseIndexToIntervalMap[len(cases)] = interval + cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ticker.t.C)}) + } + cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(closeChan)}) + return cases, caseIndexToIntervalMap +} + func (c *TranslClient) PollRun(q *queue.PriorityQueue, poll chan struct{}, w *sync.WaitGroup, subscribe *gnmipb.SubscriptionList) { c.w = w defer c.w.Done() @@ -402,12 +587,26 @@ func (c *TranslClient) PollRun(q *queue.PriorityQueue, poll chan struct{}, w *sy } t1 := time.Now() + // Spawn worker threads to process the subscription + numPaths := len(c.path2URI) + wg := &sync.WaitGroup{} + subChan := make(chan pathFromSubReq, numPaths) + numWorkers := 1 + if *useParallelProcessing { + numWorkers = findMin(numPaths, maxWorkers) + } + for i := 0; i < numWorkers; i++ { + wg.Add(1) + go processSubWorker(subChan, wg) + } for _, gnmiPath := range c.path2URI { if synced || !subscribe.UpdatesOnly { ts := translSubscriber{client: c} - ts.doSample(gnmiPath) + subChan <- pathFromSubReq{path: gnmiPath, ts: &ts} } } + close(subChan) + wg.Wait() enqueueSyncMessage(c) synced = true @@ -437,11 +636,24 @@ func (c *TranslClient) OnceRun(q *queue.PriorityQueue, once chan struct{}, w *sy } t1 := time.Now() + // Spawn worker threads to process the subscription + numPaths := len(c.path2URI) + wg := &sync.WaitGroup{} + subChan := make(chan pathFromSubReq, numPaths) + numWorkers := 1 + if *useParallelProcessing { + numWorkers = findMin(numPaths, maxWorkers) + } + for i := 0; i < numWorkers; i++ { + wg.Add(1) + go processSubWorker(subChan, wg) + } for _, gnmiPath := range c.path2URI { ts := translSubscriber{client: c} - ts.doSample(gnmiPath) + subChan <- pathFromSubReq{path: gnmiPath, ts: &ts} } - + close(subChan) + wg.Wait() enqueueSyncMessage(c) log.V(4).Infof("Sync done, once time taken: %v ms", int64(time.Since(t1)/time.Millisecond)) @@ -483,6 +695,98 @@ func getBundleVersion(extensions []*gnmi_extpb.Extension) *string { return nil } +// setPrefixTarget fills prefix taregt for given Notification objects. +func setPrefixTarget(notifs []*gnmipb.Notification, target string) { + for _, n := range notifs { + if n.Prefix == nil { + n.Prefix = &gnmipb.Path{Target: target} + } else { + n.Prefix.Target = target + } + } +} + +// Creates a spb.Value out of data from the translib according to the requested encoding. +func buildValue(prefix *gnmipb.Path, path *gnmipb.Path, enc gnmipb.Encoding, + typedVal *gnmipb.TypedValue, valueTree ygot.ValidatedGoStruct) (*spb.Value, error) { + + switch enc { + case gnmipb.Encoding_JSON, gnmipb.Encoding_JSON_IETF: + if typedVal == nil { + return nil, nil + } + return &spb.Value{ + Prefix: prefix, + Path: path, + Timestamp: time.Now().UnixNano(), + SyncResponse: false, + Val: typedVal, + }, nil + + case gnmipb.Encoding_PROTO: + if valueTree == nil { + return nil, nil + } + + fullPath := transutil.GnmiTranslFullPath(prefix, path) + removeLastPathElem(fullPath) + notifications, err := ygot.TogNMINotifications( + valueTree, + time.Now().UnixNano(), + ygot.GNMINotificationsConfig{ + UsePathElem: true, + PathElemPrefix: fullPath.GetElem(), + }) + if err != nil { + return nil, fmt.Errorf("Cannot convert OC Struct to Notifications: %s", err) + } + if len(notifications) != 1 { + return nil, fmt.Errorf("YGOT returned wrong number of notifications") + } + if len(prefix.Target) != 0 { + // Copy target from reqest.. ygot.TogNMINotifications does not fill it. + setPrefixTarget(notifications, prefix.Target) + } + return &spb.Value{ + Notification: notifications[0], + }, nil + default: + return nil, fmt.Errorf("Unsupported Encoding %s", enc) + } +} + +// buildValueForGet generates a spb.Value for GetRequest. +// Besides the same function as buildValue, it generates spb.Value for nil value as well. +// Return: spb.Value is guaranteed not nil when error is nil. +func buildValueForGet(prefix *gnmipb.Path, path *gnmipb.Path, enc gnmipb.Encoding, + typedVal *gnmipb.TypedValue, valueTree ygot.ValidatedGoStruct) (*spb.Value, error) { + + if spv, err := buildValue(prefix, path, enc, typedVal, valueTree); err != nil || spv != nil { + return spv, err + } + + // spv is nil. Server needs to generate Notification for GetRequest. + switch enc { + case gnmipb.Encoding_JSON, gnmipb.Encoding_JSON_IETF: + return &spb.Value{ + Prefix: prefix, + Path: path, + Timestamp: time.Now().UnixNano(), + SyncResponse: false, + Val: &gnmipb.TypedValue{Value: &gnmipb.TypedValue_JsonIetfVal{JsonIetfVal: []byte("{}")}}, + }, nil + + case gnmipb.Encoding_PROTO: + // The Notification has no update and its prefix is the full path. + fullPath := transutil.GnmiTranslFullPath(prefix, path) + return &spb.Value{ + Notification: &gnmipb.Notification{Prefix: fullPath, Timestamp: time.Now().UnixNano()}, + }, nil + + default: + return nil, fmt.Errorf("Unsupported Encoding %s", enc) + } +} func (c *TranslClient) parseVersion() error { bv := getBundleVersion(c.extensions) if bv == nil { @@ -497,6 +801,10 @@ func (c *TranslClient) parseVersion() error { return fmt.Errorf("Invalid bundle version: %v", *bv) } +func SetUseParallelProcessing(val bool) { + *useParallelProcessing = val +} + type TranslClientOption interface { IsTranslClientOption() } diff --git a/sonic_data_client/transl_data_client_test.go b/sonic_data_client/transl_data_client_test.go new file mode 100644 index 00000000..1b403347 --- /dev/null +++ b/sonic_data_client/transl_data_client_test.go @@ -0,0 +1,705 @@ +package client + +import ( + "context" + "errors" + "github.com/Azure/sonic-mgmt-common/translib" + "github.com/Workiva/go-datastructures/queue" + gnmipb "github.com/openconfig/gnmi/proto/gnmi" + gnmi_extpb "github.com/openconfig/gnmi/proto/gnmi_ext" + "github.com/openconfig/ygot/ygot" + "reflect" + "sync" + "testing" + "time" +) + +// MockGoStruct implements ygot.ValidatedGoStruct for testing +type MockGoStruct struct { + ygot.GoStruct +} + +func (m *MockGoStruct) Validate(...ygot.ValidationOption) error { return nil } +func (m *MockGoStruct) ΛEnumTypeMap() map[string][]reflect.Type { return nil } +func (m *MockGoStruct) ΛBelongingModule() string { return "mock" } + +func TestTranslClient_Get_Full(t *testing.T) { + // Setup parallel flags + pTrue := true + useParallelProcessing = &pTrue + + // Mocking transutil.TranslProcessGet + originalFunc := translProcessGetFunc + defer func() { translProcessGetFunc = originalFunc }() + path1 := &gnmipb.Path{ + Elem: []*gnmipb.PathElem{ + {Name: "openconfig-interfaces"}, + {Name: "interfaces"}, + }, + } + path2 := &gnmipb.Path{ + Elem: []*gnmipb.PathElem{{Name: "openconfig-system"}, {Name: "system"}}, + } + + tests := []struct { + name string + path2URI map[*gnmipb.Path]string + mockErr error + encoding gnmipb.Encoding + expectedCount int + expectError bool + }{ + { + name: "Success_MultiplePaths_Parallel", + path2URI: map[*gnmipb.Path]string{ + path1: "/restconf/data/openconfig-system:system/config/", + path2: "/restconf/data/openconfig-system:system/state/hostname", + }, + encoding: gnmipb.Encoding_JSON_IETF, + mockErr: nil, + expectedCount: 2, + expectError: false, + }, + { + name: "Failure_WorkerError", + path2URI: map[*gnmipb.Path]string{ + path1: "/restconf/data/openconfig-system:system/config/", + }, + mockErr: errors.New("translation failed"), + expectedCount: 0, + expectError: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Mock the translation logic + translProcessGetFunc = func(uriPath string, op *string, ctx context.Context, encoding gnmipb.Encoding) (*gnmipb.TypedValue, *translib.GetResponse, error) { + if tt.mockErr != nil { + return nil, nil, tt.mockErr + } + // Return a dummy value + return &gnmipb.TypedValue{Value: &gnmipb.TypedValue_StringVal{StringVal: "test"}}, nil, nil + } + + c := &TranslClient{ + path2URI: tt.path2URI, + ctx: context.Background(), + encoding: tt.encoding, + } + + // External WG as per function signature + externalWg := &sync.WaitGroup{} + + // Execute + values, err := c.Get(externalWg) + + // Assertions + if tt.expectError { + if err == nil { + t.Errorf("Expected error but got nil") + } + } else { + if err != nil { + t.Errorf("Expected no error but got: %v", err) + } + if len(values) != tt.expectedCount { + t.Errorf("Expected %d values, got %d", tt.expectedCount, len(values)) + } + } + }) + } +} + +func TestTranslClient_StreamRun_Parallel(t *testing.T) { + + // 1. Setup Flags + pTrue := true + useParallelProcessing = &pTrue + + // 2. Mock translib.IsSubscribeSupported + oldSubSupport := isSubscribeSupportedFunc + defer func() { isSubscribeSupportedFunc = oldSubSupport }() + + // 3. Define Paths + path1 := &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "interfaces"}}} + uri1 := "/restconf/data/openconfig-interfaces:interfaces" + + tests := []struct { + name string + subscriptions []*gnmipb.Subscription + isOnChangeSupport bool + preferredType translib.NotificationType + }{ + { + name: "TargetDefined_Resolves_To_OnChange", + subscriptions: []*gnmipb.Subscription{ + { + Path: path1, + Mode: gnmipb.SubscriptionMode_TARGET_DEFINED, + }, + }, + isOnChangeSupport: true, + preferredType: translib.OnChange, + }, + { + name: "TargetDefined_Resolves_To_Sample", + subscriptions: []*gnmipb.Subscription{ + { + Path: path1, + Mode: gnmipb.SubscriptionMode_TARGET_DEFINED, + }, + }, + + isOnChangeSupport: false, + preferredType: translib.Sample, + }, + { + name: "Explicit_OnChange_Supported", + subscriptions: []*gnmipb.Subscription{ + { + Path: path1, + Mode: gnmipb.SubscriptionMode_ON_CHANGE, + }, + }, + + isOnChangeSupport: true, + preferredType: translib.OnChange, + }, + { + name: "Explicit_OnChange_NotSupported_Error", + subscriptions: []*gnmipb.Subscription{ + { + Path: path1, + Mode: gnmipb.SubscriptionMode_ON_CHANGE, + }, + }, + + isOnChangeSupport: false, + preferredType: translib.OnChange, + }, + { + name: "SampleSupport", + subscriptions: []*gnmipb.Subscription{ + { + Path: path1, + Mode: gnmipb.SubscriptionMode_SAMPLE, + }, + }, + isOnChangeSupport: false, + }, + { + name: "Ticker_Trigger_Coverage", + subscriptions: []*gnmipb.Subscription{ + { + Path: path1, + SampleInterval: uint64(1 * time.Millisecond), // Very fast tick + Mode: gnmipb.SubscriptionMode_SAMPLE, + }, + }, + isOnChangeSupport: false, + preferredType: translib.Sample, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Mock returned support info + isSubscribeSupportedFunc = func(req translib.IsSubscribeRequest) ([]*translib.IsSubscribeResponse, error) { + return []*translib.IsSubscribeResponse{ + { + ID: 0, + Path: uri1, + IsOnChangeSupported: tt.isOnChangeSupport, + PreferredType: tt.preferredType, + MinInterval: 1, // uses default + }, + }, nil + } + + // Initialize Client + stopChan := make(chan struct{}) + + q := queue.NewPriorityQueue(10, false) + + c := &TranslClient{ + ctx: context.Background(), + path2URI: map[*gnmipb.Path]string{path1: uri1}, + extensions: []*gnmi_extpb.Extension{}, + } + + subList := &gnmipb.SubscriptionList{ + Subscription: tt.subscriptions, + Encoding: gnmipb.Encoding_JSON_IETF, + } + + externalWg := &sync.WaitGroup{} + externalWg.Add(1) + go c.StreamRun(q, stopChan, externalWg, subList) + time.Sleep(1500 * time.Millisecond) // Your successful 1.5s sleep + close(stopChan) + + // Wait for StreamRun to finish + externalWg.Wait() + + t.Log("Function exited cleanly after processing parallel workers") + }) + } +} + +func TestTranslClient_OnceRun_Coverage(t *testing.T) { + // 1. Setup Flags + pTrue := true + useParallelProcessing = &pTrue + + // 2. Setup Dependencies + q := queue.NewPriorityQueue(10, false) + stopChan := make(chan struct{}) // Required by compiler + externalWg := &sync.WaitGroup{} // Required by compiler + + path1 := &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "interfaces"}}} + uri1 := "/restconf/data/openconfig-interfaces:interfaces" + + // 3. Mock Support + isSubscribeSupportedFunc = func(req translib.IsSubscribeRequest) ([]*translib.IsSubscribeResponse, error) { + return []*translib.IsSubscribeResponse{ + {ID: 0, Path: uri1}, + }, nil + } + + c := &TranslClient{ + ctx: context.Background(), + path2URI: map[*gnmipb.Path]string{path1: uri1}, + extensions: []*gnmi_extpb.Extension{}, + } + + subList := &gnmipb.SubscriptionList{ + Subscription: []*gnmipb.Subscription{ + {Path: path1, Mode: gnmipb.SubscriptionMode_SAMPLE}, + }, + Encoding: gnmipb.Encoding_JSON_IETF, + } + + // 4. Execution + externalWg.Add(1) + go c.OnceRun(q, stopChan, externalWg, subList) + + stopChan <- struct{}{} + + // 5. Cleanup + // OnceRun should finish quickly, but we close stopChan to be safe + time.Sleep(200 * time.Millisecond) + close(stopChan) + externalWg.Wait() + t.Log("Function exited cleanly after processing parallel workers-OnceRun") +} + +func TestPollRun_SuccessFlow(t *testing.T) { + // 1. Setup Dependencies + pollChan := make(chan struct{}) + wg := &sync.WaitGroup{} + pq := queue.NewPriorityQueue(10, false) + + // Mock SubscriptionList + subList := &gnmipb.SubscriptionList{ + Encoding: gnmipb.Encoding_JSON_IETF, + UpdatesOnly: false, + } + path1 := &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "interfaces"}}} + uri1 := "/restconf/data/openconfig-interfaces:interfaces" + + client := &TranslClient{ + path2URI: map[*gnmipb.Path]string{path1: uri1}, + } + + // 2. Run PollRun in a separate goroutine + wg.Add(1) + go client.PollRun(pq, pollChan, wg, subList) + + // 3. Trigger a poll iteration + pollChan <- struct{}{} + + // Give the workers a moment to process + time.Sleep(100 * time.Millisecond) + + // 4. Close the channel to exit the loop and the goroutine + close(pollChan) + + // 5. Wait for PollRun to finish (defer c.w.Done()) + wg.Wait() + t.Log("Function exited cleanly after processing parallel workers-PollRun") + +} + +func TestBuildValue(t *testing.T) { + tests := []struct { + name string + prefix *gnmipb.Path + path *gnmipb.Path + enc gnmipb.Encoding + valueTree ygot.ValidatedGoStruct + wantErr bool + errMsg string + }{ + { + name: "Unsupported Encoding (JSON)", + prefix: &gnmipb.Path{}, + path: &gnmipb.Path{}, + enc: gnmipb.Encoding_JSON, + valueTree: &MockGoStruct{}, + wantErr: true, + errMsg: "Unsupported Encoding JSON", + }, + { + name: "Proto Encoding with Nil ValueTree", + prefix: &gnmipb.Path{}, + path: &gnmipb.Path{}, + enc: gnmipb.Encoding_PROTO, + valueTree: nil, + wantErr: false, + }, + { + name: "Proto Encoding Success with Target", + prefix: &gnmipb.Path{ + Target: "DEVICE_A", + Elem: []*gnmipb.PathElem{{Name: "base"}}, + }, + path: &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "interface"}}}, + enc: gnmipb.Encoding_PROTO, + valueTree: &MockGoStruct{}, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + buildValue(tt.prefix, tt.path, tt.enc, nil, tt.valueTree) + + if tt.wantErr { + t.Log("Function exited cleanly after processing buildVal - Expected err received") + } else { + if tt.valueTree == nil { + t.Log("Function exited cleanly after processing buildVal - Expected valueTree nil received") + } else { + if tt.prefix.Target != "" { + t.Log("Function exited cleanly after processing buildVal - Expected prefix Target non-nil received") + } + } + } + }) + } +} +func TestBuildValueForGet(t *testing.T) { + tests := []struct { + name string + prefix *gnmipb.Path + path *gnmipb.Path + enc gnmipb.Encoding + valueTree ygot.ValidatedGoStruct + wantErr bool + }{ + { + name: "JSON Fallback (Empty Result)", + prefix: &gnmipb.Path{Target: "DUT"}, + path: &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "system"}}}, + enc: gnmipb.Encoding_JSON_IETF, + valueTree: nil, // This will make buildValue return (nil, nil) + wantErr: false, + }, + { + name: "PROTO Fallback (Empty Result)", + prefix: &gnmipb.Path{Target: "DUT"}, + path: &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "interfaces"}}}, + enc: gnmipb.Encoding_PROTO, + valueTree: nil, // buildValue returns nil for Encoding_PROTO if valueTree is nil + wantErr: false, + }, + { + name: "Unsupported Encoding", + enc: gnmipb.Encoding_ASCII, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + buildValueForGet(tt.prefix, tt.path, tt.enc, nil, tt.valueTree) + + if tt.wantErr { + t.Log("Function exited cleanly after processing buildValGet - Expected err received") + return + } + + switch tt.enc { + case gnmipb.Encoding_JSON_IETF: + // Verify fallback JSON structure + t.Log("Function exited cleanly after processing buildValGet with Encoding_JSON_IETF") + break + + case gnmipb.Encoding_PROTO: + // Verify fallback Proto Notification structure + t.Log("Function exited cleanly after processing buildValGet with Encoding_PROTO") + break + + } + }) + } +} + +// TestBuildSelectCases validates the construction of reflect.SelectCase slices +func TestBuildSelectCases(t *testing.T) { + // 1. Setup mock channels and tickers + tick100 := time.NewTicker(100 * time.Millisecond) + tick200 := time.NewTicker(200 * time.Millisecond) + defer tick100.Stop() + defer tick200.Stop() + + closeChan := make(chan struct{}) + + // 2. Define Input Map with edge cases (empty slice and nil ticker) + intervalToTickerInfoMap := map[int][]*ticker_info{ + 100: {&ticker_info{t: tick100}}, + 200: {&ticker_info{t: tick200}}, + 300: {}, // Edge case: Empty slice (should be skipped) + 400: {nil}, // Edge case: Nil ticker (should be skipped) + } + + // 3. Execute the function under test + cases, indexMap := buildSelectCases(intervalToTickerInfoMap, closeChan) + + // 4. Assertions + + // We expect 3 cases: interval 100, interval 200, and the closeChan. + expectedTotalCases := 3 + if len(cases) != expectedTotalCases { + t.Fatalf("Expected %d total cases, but got %d", expectedTotalCases, len(cases)) + } + + // Verify that the very last case is ALWAYS the closeChan + lastCase := cases[len(cases)-1] + if lastCase.Chan.Interface() != closeChan { + t.Logf("The last SelectCase was not the closeChan") + } + if lastCase.Dir != reflect.SelectRecv { + t.Logf("Expected SelectRecv for closeChan, got %v", lastCase.Dir) + } + + // Verify the intervals and their corresponding channels + // Since maps iterate randomly, we loop through the indexMap to verify correctness + foundIntervals := make(map[int]bool) + + for caseIdx, interval := range indexMap { + foundIntervals[interval] = true + + // Check if the SelectCase at this index matches the ticker channel + actualChan := cases[caseIdx].Chan.Interface() + expectedChan := intervalToTickerInfoMap[interval][0].t.C + + if actualChan != expectedChan { + t.Logf("Mismatch at index %d: Case channel does not match interval %d ticker", caseIdx, interval) + } + + if cases[caseIdx].Dir != reflect.SelectRecv { + t.Logf("Case index %d: expected Dir SelectRecv", caseIdx) + } + } + + // Ensure our skipped cases (300, 400) are truly not in the results + if len(foundIntervals) != 2 { + t.Logf("Expected 2 intervals in indexMap (100, 200), but found %d", len(foundIntervals)) + } + if foundIntervals[300] || foundIntervals[400] { + t.Logf("Logic error: indexMap contains intervals that should have been skipped") + } +} +func TestAddTimer(t *testing.T) { + // Setup + pathA := "/system/interfaces/interface/state/counters" + pathB := "/system/processes/process/state/cpu-usage" + + subA := &gnmipb.Subscription{} + subB := &gnmipb.Subscription{} + + tests := []struct { + name string + initialMap map[int][]*ticker_info + interval int + path string + sub *gnmipb.Subscription + heartbeat bool + expectedCount int + checkIndex int + }{ + { + name: "Add to new interval", + initialMap: make(map[int][]*ticker_info), + interval: 10, + path: pathA, + sub: subA, + heartbeat: false, + expectedCount: 1, + checkIndex: 0, + }, + { + name: "Append to existing interval", + initialMap: map[int][]*ticker_info{ + 10: { + {pathStr: pathA, interval: 10}, + }, + }, + interval: 10, + path: pathB, + sub: subB, + heartbeat: true, + expectedCount: 2, + checkIndex: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Act + addTimer(tt.initialMap, tt.interval, tt.sub, tt.path, tt.heartbeat) + + // Assert length + tickers := tt.initialMap[tt.interval] + if len(tickers) != tt.expectedCount { + t.Fatalf("Expected %d tickers for interval %d, got %d", + tt.expectedCount, tt.interval, len(tickers)) + } + + // Assert content of the added/appended item + added := tickers[tt.checkIndex] + if added.pathStr != tt.path { + t.Logf("Expected path %s, got %s", tt.path, added.pathStr) + } + if added.interval != tt.interval { + t.Logf("Expected interval %d, got %d", tt.interval, added.interval) + } + if added.sub != tt.sub { + t.Logf("Subscription pointer mismatch") + } + if added.heartbeat != tt.heartbeat { + t.Logf("Expected heartbeat %v, got %v", tt.heartbeat, added.heartbeat) + } + }) + } +} + +func TestStreamRun_Detailed(t *testing.T) { + // 1. Setup mocks and channels + + stopChan := make(chan struct{}) + wakeChan := make(chan bool, 1) + var wg sync.WaitGroup + wg.Add(1) + + // Mock Priority Queue + mockQ := queue.NewPriorityQueue(10, false) + primaryClient := &TranslClient{} + + // Mock Ticker Channels + //tick10s := make(chan time.Time) + + mockSS := &superSubscription{ + primaryClient: primaryClient} + + ctx, cancel := context.WithTimeout(context.Background(), 5500*time.Millisecond) + defer cancel() + // 2. Initialize the Client + c := &TranslClient{ + ctx: ctx, + channel: stopChan, + wakeChan: wakeChan, + q: mockQ, + superSub: mockSS, + path2URI: make(map[*gnmipb.Path]string), + w: &wg, + } + + // Mock a path and its URI + path := &gnmipb.Path{Elem: []*gnmipb.PathElem{{Name: "system"}}} + c.path2URI[path] = "/restconf/data/system" + + // Mock Subscription List + subList := &gnmipb.SubscriptionList{ + Prefix: &gnmipb.Path{Origin: "openconfig", Target: "OC_YANG"}, + Mode: gnmipb.SubscriptionList_STREAM, + Encoding: gnmipb.Encoding_PROTO, + Subscription: []*gnmipb.Subscription{ + { + Path: path, + Mode: gnmipb.SubscriptionMode_SAMPLE, + SampleInterval: 10 * 1e9, // 10 seconds + }, + }, + } + + go func() { + c.StreamRun(c.q, stopChan, &wg, subList) + }() + c.wakeChan <- false + + time.Sleep(50 * time.Millisecond) + //mockSS.primaryClient = primaryClient + + close(stopChan) + wg.Wait() +} +func TestNewTranslClient(t *testing.T) { + // 1. Setup Mock Data + ctx := context.Background() + prefix := &gnmipb.Path{Target: "device1"} + getPaths := []*gnmipb.Path{ + {Elem: []*gnmipb.PathElem{{Name: "system"}}}, + } + + // Define a custom option if needed, or use a dummy for the type check + type TranslWildcardOption struct{} + + t.Run("Success with valid paths", func(t *testing.T) { + client, err := NewTranslClient(prefix, getPaths, ctx, nil) + + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } + + if client == nil { + t.Fatal("Expected client to be non-nil") + } + + // Verify internal state (casting to access private fields if in same package) + tc := client.(*TranslClient) + if tc.prefix != prefix { + t.Errorf("Prefix mismatch: expected %v, got %v", prefix, tc.prefix) + } + + if tc.path2URI == nil { + t.Error("Expected path2URI map to be initialized") + } + }) + + t.Run("Nil getpaths", func(t *testing.T) { + client, err := NewTranslClient(prefix, nil, ctx, nil) + + if err != nil { + t.Fatalf("Expected no error when getpaths is nil, got %v", err) + } + + tc := client.(*TranslClient) + if tc.path2URI != nil { + t.Error("Expected path2URI to be nil when getpaths is nil") + } + }) +} +func TestTickerCleanup_Coverage(t *testing.T) { + // Test with active tickers + tickers := map[int]*time.Ticker{ + 1: time.NewTicker(time.Millisecond), + 2: time.NewTicker(time.Millisecond), + } + tickerCleanup(tickers) + + // Test with nil map + tickerCleanup(nil) +} diff --git a/sonic_data_client/transl_subscriber.go b/sonic_data_client/transl_subscriber.go index 2b11658b..33a270d9 100644 --- a/sonic_data_client/transl_subscriber.go +++ b/sonic_data_client/transl_subscriber.go @@ -44,6 +44,7 @@ type translSubscriber struct { synced sync.WaitGroup // To signal receipt of sync message from translib rcvdPaths map[string]bool // Paths from received SubscribeResponse msgBuilder notificationBuilder + sampleWg *sync.WaitGroup // WaitGroup to indicate when Sample subscriptions have finished processing } // notificationBuilder creates gnmipb.Notification from a translib.SubscribeResponse @@ -155,6 +156,10 @@ func (ts *translSubscriber) processResponses(q *queue.PriorityQueue) { return } + if ts.sampleWg != nil { + // Wait for any sample subscriptions to finish before sending sync response. + ts.sampleWg.Wait() + } log.V(6).Infof("SENDING SYNC") enqueueSyncMessage(c) syncDone = true @@ -186,7 +191,15 @@ func (ts *translSubscriber) notify(v *translib.SubscribeResponse) error { } spbv := &spb.Value{Notification: msg} - ts.client.q.Put(Value{spbv}) + if ts.client.superSub != nil && v == nil { + ts.client.superSub.sendNotifications(spbv) + ts.client.superSub.sharedUpdates.Add(1) + } else { + ts.client.q.Put(Value{spbv}) + if ts.client.superSub != nil { + ts.client.superSub.exclusiveUpdates.Add(1) + } + } log.V(6).Infof("Added spbv %#v", spbv) return nil } diff --git a/sonic_data_client/transl_subscriber_test.go b/sonic_data_client/transl_subscriber_test.go new file mode 100644 index 00000000..f965cd6a --- /dev/null +++ b/sonic_data_client/transl_subscriber_test.go @@ -0,0 +1,106 @@ +package client + +import ( + "testing" + + "github.com/Azure/sonic-mgmt-common/translib" + "github.com/Workiva/go-datastructures/queue" + spb "github.com/openconfig/gnmi/proto/gnmi" +) + +func TestNotify(t *testing.T) { + tests := []struct { + name string + v *translib.SubscribeResponse + builderMsg *spb.Notification + builderErr error + hasSuperSub bool + expectedErr bool + expectedLen int + }{ + { + name: "Standard notification path", + v: &translib.SubscribeResponse{}, + builderMsg: &spb.Notification{ + Update: []*spb.Update{{Path: &spb.Path{Target: "test"}}}, + }, + expectedErr: false, + expectedLen: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // 1. Initialize the Subscriber and Client pointers + ts := &translSubscriber{} + ts.client = &TranslClient{} + + // 2. Fix the Queue Assignment + ts.client.q = queue.NewPriorityQueue(10, false) + + // 3. Mock the msgBuilder function + ts.msgBuilder = func(resp *translib.SubscribeResponse, s *translSubscriber) (*spb.Notification, error) { + return tt.builderMsg, tt.builderErr + } + + // 4. Handle SuperSub if needed + if tt.hasSuperSub { + ts.client.superSub = &superSubscription{} + } + + // Execute + ts.notify(tt.v) + }) + } +} +func TestNotifyNil(t *testing.T) { + tests := []struct { + name string + v *translib.SubscribeResponse + builderMsg *spb.Notification + builderErr error + hasSuperSub bool + expectedErr bool + expectedLen int + }{ + { + name: "Standard notification path", + v: &translib.SubscribeResponse{}, + builderMsg: &spb.Notification{ + Update: []*spb.Update{{Path: &spb.Path{Target: "test"}}}, + }, + expectedErr: false, + expectedLen: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // 1. Initialize the Subscriber and Client pointers + ts := &translSubscriber{} + ts.client = &TranslClient{} + + // 2. Fix the Queue Assignment + ts.client.q = queue.NewPriorityQueue(10, false) + + // 3. Mock the msgBuilder function + ts.msgBuilder = func(resp *translib.SubscribeResponse, s *translSubscriber) (*spb.Notification, error) { + if resp == nil { + // Return a valid notification to pass the (len == 0) check + return &spb.Notification{ + Update: []*spb.Update{{}}, + }, nil + } + return nil, nil + } + + // 4. Handle SuperSub if needed + if tt.hasSuperSub { + ts.client.superSub = &superSubscription{} + } + + // Execute + ts.notify(tt.v) + }) + } +} diff --git a/sonic_service_client/dbus_client.go b/sonic_service_client/dbus_client.go index 9032481f..8d380ee4 100644 --- a/sonic_service_client/dbus_client.go +++ b/sonic_service_client/dbus_client.go @@ -1,6 +1,7 @@ package host_service import ( + "context" "fmt" "reflect" "strings" @@ -49,8 +50,37 @@ type Service interface { // Docker services APIs LoadDockerImage(image string) error InstallOS(req string) (string, error) + //Credentialz service APIs + SSHMgmtSet(cmd string) error + GLOMEConfigSet(ctx context.Context, cmd string) error + SSHCheckpoint(action CredzCheckpointAction) error + GLOMERestoreCheckpoint(ctx context.Context) error + ConsoleSet(cmd string) error + ConsoleCheckpoint(action CredzCheckpointAction) error } +// Define a function type that matches the DbusApi signature +type DbusApiFunc func(busName, busPath, intName string, timeout int, args ...interface{}) (interface{}, error) + +// Use a variable to hold the implementation. It defaults to the real DbusApi function. +var dbusApiCaller DbusApiFunc = DbusApi + +// NewDbusClientFunc defines the signature for creating a D-Bus client. +type NewDbusClientFunc func() (Service, error) + +// NewDbusClientProvider is a variable that defaults to the real constructor. +// Tests can overwrite this to return a FakeClient. +var NewDbusClientProvider NewDbusClientFunc = NewDbusClient + +type CredzCheckpointAction string + +const ( + CredzCPCreate CredzCheckpointAction = ".create_checkpoint" + CredzCPDelete CredzCheckpointAction = ".delete_checkpoint" + CredzCPRestore CredzCheckpointAction = ".restore_checkpoint" + CredzGlomePushConfig CredzCheckpointAction = ".push_config" +) + type DbusClient struct { busNamePrefix string busPathPrefix string @@ -424,3 +454,96 @@ func (c *DbusClient) HealthzAck(req string) (string, error) { } return strResult, nil } + +func (c *DbusClient) ConsoleSet(cmd string) error { + modName := "gnsi_console" + busName := c.busNamePrefix + modName + busPath := c.busPathPrefix + modName + intName := c.intNamePrefix + modName + ".set" + + common_utils.IncCounter(common_utils.GNSI_CREDZ_SET) + _, err := dbusApiCaller(busName, busPath, intName, 10, cmd) + return err +} + +func (c *DbusClient) SSHMgmtSet(cmd string) error { + modName := "ssh_mgmt" + busName := c.busNamePrefix + modName + busPath := c.busPathPrefix + modName + intName := c.intNamePrefix + modName + ".set" + + common_utils.IncCounter(common_utils.GNSI_CREDZ_SET) + _, err := dbusApiCaller(busName, busPath, intName, 10, cmd) + return err +} + +// GLOMEConfigSet is used to write the GLOME config in the host service file system. +func (c *DbusClient) GLOMEConfigSet(ctx context.Context, cmd string) error { + modName := "glome" + busName := c.busNamePrefix + modName + busPath := c.busPathPrefix + modName + intName := c.intNamePrefix + modName + string(CredzGlomePushConfig) + + common_utils.IncCounter(common_utils.GNSI_CREDZ_SET) + timeout := 10 // Default timeout in seconds. + if deadline, ok := ctx.Deadline(); ok { + remaining := time.Until(deadline) + if remaining <= 0 { + return context.DeadlineExceeded + } + timeout = int(remaining.Seconds()) + if timeout > 10 { + timeout = 10 + } + } + _, err := dbusApiCaller(busName, busPath, intName, timeout, cmd) + return err +} + +func (c *DbusClient) ConsoleCheckpoint(action CredzCheckpointAction) error { + modName := "gnsi_console" + busName := c.busNamePrefix + modName + busPath := c.busPathPrefix + modName + intName := c.intNamePrefix + modName + string(action) + + common_utils.IncCounter(common_utils.GNSI_CREDZ_CHECKPOINT) + _, err := dbusApiCaller(busName, busPath, intName, 10, "") + return err +} + +func (c *DbusClient) SSHCheckpoint(action CredzCheckpointAction) error { + modName := "ssh_mgmt" + busName := c.busNamePrefix + modName + busPath := c.busPathPrefix + modName + intName := c.intNamePrefix + modName + string(action) + + common_utils.IncCounter(common_utils.GNSI_CREDZ_CHECKPOINT) + _, err := dbusApiCaller(busName, busPath, intName, 10, "") + return err +} + +// GLOMERestoreCheckpoint is used to restore the GLOME config metadata to the +// checkpoint state. This is used to rollback the GLOME config in the host +// service file system. +func (c *DbusClient) GLOMERestoreCheckpoint(ctx context.Context) error { + modName := "glome" + busName := c.busNamePrefix + modName + busPath := c.busPathPrefix + modName + intName := c.intNamePrefix + modName + string(CredzCPRestore) + + common_utils.IncCounter(common_utils.GNSI_CREDZ_CHECKPOINT) + // Default timeout in seconds. Set to 5 minutes to give enough time for rollback. + timeout := 300 + if deadline, ok := ctx.Deadline(); ok { + remaining := time.Until(deadline) + if remaining <= 0 { + return context.DeadlineExceeded + } + timeout = int(remaining.Seconds()) + if timeout > 10 { + timeout = 10 + } + } + _, err := dbusApiCaller(busName, busPath, intName, timeout) + return err +} diff --git a/sonic_service_client/dbus_client_test.go b/sonic_service_client/dbus_client_test.go index b8b28c4a..b27b002e 100644 --- a/sonic_service_client/dbus_client_test.go +++ b/sonic_service_client/dbus_client_test.go @@ -1,6 +1,7 @@ package host_service import ( + "context" "errors" "fmt" "github.com/agiledragon/gomonkey/v2" @@ -11,6 +12,7 @@ import ( "reflect" "strings" "testing" + "time" ) func TestNewDbusClient(t *testing.T) { @@ -1656,3 +1658,133 @@ func TestHealthzAck_InvalidReturnType(t *testing.T) { assert.Contains(t, err.Error(), "Invalid result type") assert.Equal(t, "", result) } + +func TestCredentialzDbusMethods(t *testing.T) { + client := &DbusClient{ + busNamePrefix: "org.SONiC.HostService.", + busPathPrefix: "/org/SONiC/HostService/", + intNamePrefix: "org.SONiC.HostService.", + } + // Save the original implementation to restore it later + originalDbusApi := dbusApiCaller + defer func() { dbusApiCaller = originalDbusApi }() + + t.Run("ConsoleSet", func(t *testing.T) { + expectedCmd := "test-password-json" + // Overwrite the caller variable with a mock + dbusApiCaller = func(bus, path, intf string, timeout int, args ...interface{}) (interface{}, error) { + assert.Equal(t, "org.SONiC.HostService.gnsi_console", bus) + assert.Equal(t, "org.SONiC.HostService.gnsi_console.set", intf) + assert.Equal(t, expectedCmd, args[0]) + return nil, nil + } + + err := client.ConsoleSet(expectedCmd) + assert.NoError(t, err) + }) + + t.Run("SSHMgmtSet", func(t *testing.T) { + expectedCmd := "test-ssh-key-json" + dbusApiCaller = func(bus, path, intf string, timeout int, args ...interface{}) (interface{}, error) { + assert.Equal(t, "org.SONiC.HostService.ssh_mgmt", bus) + assert.Equal(t, "org.SONiC.HostService.ssh_mgmt.set", intf) + assert.Equal(t, expectedCmd, args[0]) + return nil, nil + } + + err := client.SSHMgmtSet(expectedCmd) + assert.NoError(t, err) + }) + + t.Run("SSHCheckpoint", func(t *testing.T) { + dbusApiCaller = func(bus, path, intf string, timeout int, args ...interface{}) (interface{}, error) { + assert.Equal(t, "org.SONiC.HostService.ssh_mgmt.create_checkpoint", intf) + assert.Equal(t, "", args[0]) + return nil, nil + } + + err := client.SSHCheckpoint(CredzCPCreate) + assert.NoError(t, err) + }) + + t.Run("GLOMEConfigSetWithContext", func(t *testing.T) { + expectedCmd := "glome-json" + // Create a context with a 5-second deadline to test timeout calculation + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + dbusApiCaller = func(bus, path, intf string, timeout int, args ...interface{}) (interface{}, error) { + assert.Equal(t, "org.SONiC.HostService.glome.push_config", intf) + assert.Equal(t, expectedCmd, args[0]) + // Timeout should be roughly 5 (based on ctx) + assert.True(t, timeout <= 5) + return nil, nil + } + + err := client.GLOMEConfigSet(ctx, expectedCmd) + assert.NoError(t, err) + }) + + t.Run("GLOMERestoreCheckpoint", func(t *testing.T) { + dbusApiCaller = func(bus, path, intf string, timeout int, args ...interface{}) (interface{}, error) { + + assert.Equal(t, "org.SONiC.HostService.glome.restore_checkpoint", intf) + // Restore checkpoint for GLOME often has a high default timeout (300s) + assert.Equal(t, 300, timeout) + return nil, nil + } + err := client.GLOMERestoreCheckpoint(context.Background()) + assert.NoError(t, err) + }) +} + +func TestConsoleCheckpoint(t *testing.T) { + client := &DbusClient{ + busNamePrefix: "org.SONiC.HostService.", + busPathPrefix: "/org/SONiC/HostService/", + intNamePrefix: "org.SONiC.HostService.", + } + originalDbusApi := dbusApiCaller + defer func() { dbusApiCaller = originalDbusApi }() + + tests := []struct { + name string + action CredzCheckpointAction + wantIntf string + }{ + { + name: "Console Create Checkpoint", + action: CredzCPCreate, + wantIntf: "org.SONiC.HostService.gnsi_console.create_checkpoint", + }, + { + name: "Console Delete Checkpoint", + action: CredzCPDelete, + wantIntf: "org.SONiC.HostService.gnsi_console.delete_checkpoint", + }, + { + name: "Console Restore Checkpoint", + action: CredzCPRestore, + wantIntf: "org.SONiC.HostService.gnsi_console.restore_checkpoint", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dbusApiCaller = func(bus, path, intf string, timeout int, args ...interface{}) (interface{}, error) { + assert.Equal(t, "org.SONiC.HostService.gnsi_console", bus) + assert.Equal(t, "/org/SONiC/HostService/gnsi_console", path) + assert.Equal(t, tt.wantIntf, intf) + assert.Equal(t, 10, timeout) + + // Verify ConsoleCheckpoint passes an empty string as the payload + assert.Len(t, args, 1) + assert.Equal(t, "", args[0]) + + return nil, nil + } + + err := client.ConsoleCheckpoint(tt.action) + assert.NoError(t, err) + }) + } +} diff --git a/sonic_service_client/dbus_fake_client.go b/sonic_service_client/dbus_fake_client.go index c8ccc394..66f93f7c 100644 --- a/sonic_service_client/dbus_fake_client.go +++ b/sonic_service_client/dbus_fake_client.go @@ -1,6 +1,7 @@ package host_service import ( + "context" "errors" "fmt" ) @@ -8,6 +9,7 @@ import ( // FakeClient is a mock implementation of the Service interface. type FakeClient struct { CollectResponse string + Command chan []string } func (f *FakeClient) Close() error { return nil } @@ -90,3 +92,33 @@ func (f *FakeClient) HealthzAck(req string) (string, error) { func (f *FakeClientWithError) HealthzAck(req string) (string, error) { return "", fmt.Errorf("simulated dbus error") } + +func (f *FakeClient) ConsoleCheckpoint(action CredzCheckpointAction) error { + f.Command <- []string{"gnsi_console" + string(action), ""} + return nil +} + +func (f *FakeClient) ConsoleSet(cmd string) error { + f.Command <- []string{"gnsi_console.set", cmd} + return nil +} + +func (f *FakeClient) SSHCheckpoint(action CredzCheckpointAction) error { + f.Command <- []string{"ssh_mgmt" + string(action), ""} + return nil +} + +func (f *FakeClient) SSHMgmtSet(cmd string) error { + f.Command <- []string{"ssh_mgmt.set", cmd} + return nil +} + +func (f *FakeClient) GLOMEConfigSet(ctx context.Context, cmd string) error { + f.Command <- []string{"glome" + string(CredzGlomePushConfig), cmd} + return nil +} + +func (f *FakeClient) GLOMERestoreCheckpoint(ctx context.Context) error { + f.Command <- []string{"glome" + string(CredzCPRestore), ""} + return nil +} diff --git a/sonic_service_client/dbus_fake_client_test.go b/sonic_service_client/dbus_fake_client_test.go index fa379abd..4bc21687 100644 --- a/sonic_service_client/dbus_fake_client_test.go +++ b/sonic_service_client/dbus_fake_client_test.go @@ -1,13 +1,16 @@ package host_service import ( + "context" "testing" "github.com/stretchr/testify/assert" ) func TestFakeClientMethods(t *testing.T) { - client := &FakeClient{} + client := &FakeClient{ + Command: make(chan []string, 10), + } assert.NoError(t, client.Close()) assert.NoError(t, client.ConfigReload("test.conf")) @@ -75,4 +78,51 @@ func TestFakeClientMethods(t *testing.T) { assert.Error(t, err) assert.Equal(t, "", output) assert.Equal(t, "request cannot be empty", err.Error()) + + // --- Credentialz Fake Tests --- + + t.Run("SSHCheckpoint", func(t *testing.T) { + err := client.SSHCheckpoint(CredzCPCreate) + assert.NoError(t, err) + msg := <-client.Command + assert.Equal(t, []string{"ssh_mgmt.create_checkpoint", ""}, msg) + }) + + t.Run("SSHMgmtSet", func(t *testing.T) { + testCmd := `{"SshAccountKeys": []}` + err := client.SSHMgmtSet(testCmd) + assert.NoError(t, err) + msg := <-client.Command + assert.Equal(t, []string{"ssh_mgmt.set", testCmd}, msg) + }) + + t.Run("ConsoleCheckpoint", func(t *testing.T) { + err := client.ConsoleCheckpoint(CredzCPRestore) + assert.NoError(t, err) + msg := <-client.Command + assert.Equal(t, []string{"gnsi_console.restore_checkpoint", ""}, msg) + }) + + t.Run("ConsoleSet", func(t *testing.T) { + testCmd := `{"ConsolePasswords": []}` + err := client.ConsoleSet(testCmd) + assert.NoError(t, err) + msg := <-client.Command + assert.Equal(t, []string{"gnsi_console.set", testCmd}, msg) + }) + + t.Run("GLOMEConfigSet", func(t *testing.T) { + testCmd := `{"enabled": true}` + err := client.GLOMEConfigSet(context.Background(), testCmd) + assert.NoError(t, err) + msg := <-client.Command + assert.Equal(t, []string{"glome.push_config", testCmd}, msg) + }) + + t.Run("GLOMERestoreCheckpoint", func(t *testing.T) { + err := client.GLOMERestoreCheckpoint(context.Background()) + assert.NoError(t, err) + msg := <-client.Command + assert.Equal(t, []string{"glome.restore_checkpoint", ""}, msg) + }) } diff --git a/telemetry/telemetry.go b/telemetry/telemetry.go index d6e8aef2..3b3c85da 100644 --- a/telemetry/telemetry.go +++ b/telemetry/telemetry.go @@ -62,6 +62,7 @@ type TelemetryConfig struct { WithMasterArbitration *bool WithSaveOnSet *bool IdleConnDuration *int + GnmiVrf *string Vrf *string EnableCrl *bool CrlExpireDuration *int @@ -76,6 +77,8 @@ type TelemetryConfig struct { AuthPolicyEnabled *bool AuthzPolicyFile *string EnableStreamMultiplexing *bool + MaxRecvMsgSize *int + MaxSendMsgSize *int } func main() { @@ -185,7 +188,8 @@ func setupFlags(fs *flag.FlagSet) (*TelemetryConfig, *gnmi.Config, error) { WithMasterArbitration: fs.Bool("with-master-arbitration", false, "Enables master arbitration policy."), WithSaveOnSet: fs.Bool("with-save-on-set", false, "Enables save-on-set."), IdleConnDuration: fs.Int("idle_conn_duration", 5, "Seconds before server closes idle connections"), - Vrf: fs.String("vrf", "", "VRF name, when zmq_address belong on a VRF, need VRF name to bind ZMQ."), + GnmiVrf: fs.String("gnmi_vrf", "", "VRF name for gNMI server binding."), + Vrf: fs.String("vrf", "", "VRF name for ZMQ client binding."), EnableCrl: fs.Bool("enable_crl", false, "Enable certificate revocation list"), CrlExpireDuration: fs.Int("crl_expire_duration", 86400, "Certificate revocation list cache expire duration"), ImgDirPath: fs.String("img_dir", "/tmp/host_tmp", "Directory path where image will be transferred."), @@ -202,6 +206,8 @@ func setupFlags(fs *flag.FlagSet) (*TelemetryConfig, *gnmi.Config, error) { AuthPolicyEnabled: fs.Bool("authz_policy_enabled", false, "Enable authz policy. Require insecure flag to be false."), AuthzPolicyFile: fs.String("authorization_policy_file", "/keys/authorization_policy.json", "Full path name of the JSON authorization policy file."), EnableStreamMultiplexing: fs.Bool("enable_stream_multiplexing", false, "Allow multiple Subscribe RPCs on a single TCP connection via HTTP/2 stream multiplexing"), + MaxRecvMsgSize: fs.Int("max_recv_msg_size", 4*1024*1024, "Maximum message size in bytes that the server can receive"), + MaxSendMsgSize: fs.Int("max_send_msg_size", 4*1024*1024, "Maximum message size in bytes that the server can send"), } fs.Var(&telemetryCfg.UserAuth, "client_auth", "Client auth mode(s) - none,cert,password") @@ -265,6 +271,7 @@ func setupFlags(fs *flag.FlagSet) (*TelemetryConfig, *gnmi.Config, error) { cfg.Threshold = int(*telemetryCfg.Threshold) cfg.IdleConnDuration = int(*telemetryCfg.IdleConnDuration) cfg.ConfigTableName = *telemetryCfg.ConfigTableName + cfg.GnmiVrf = *telemetryCfg.GnmiVrf cfg.Vrf = *telemetryCfg.Vrf cfg.EnableCrl = *telemetryCfg.EnableCrl cfg.CaCertLnk = *telemetryCfg.CaCertLnk @@ -553,6 +560,11 @@ func startGNMIServer(telemetryCfg *TelemetryConfig, cfg *gnmi.Config, serverCont gnmi.GenerateJwtSecretKey() } + commonOpts = append(commonOpts, + grpc.MaxRecvMsgSize(*telemetryCfg.MaxRecvMsgSize), + grpc.MaxSendMsgSize(*telemetryCfg.MaxSendMsgSize), + ) + // Setup interceptor chain (includes DPU proxy with Redis-based routing) var err error currentServerChain, err = interceptors.NewServerChain() diff --git a/telemetry/telemetry_test.go b/telemetry/telemetry_test.go index 116887cd..657ef606 100644 --- a/telemetry/telemetry_test.go +++ b/telemetry/telemetry_test.go @@ -59,8 +59,14 @@ func TestRunTelemetry(t *testing.T) { func TestFlags(t *testing.T) { originalArgs := os.Args + originalJwtRefreshInt := gnmi.JwtRefreshInt + originalJwtValidInt := gnmi.JwtValidInt + originalCrlExpireDuration := gnmi.GetCrlExpireDuration() defer func() { os.Args = originalArgs + gnmi.JwtRefreshInt = originalJwtRefreshInt + gnmi.JwtValidInt = originalJwtValidInt + gnmi.SetCrlExpireDuration(originalCrlExpireDuration) }() tests := []struct { @@ -69,6 +75,8 @@ func TestFlags(t *testing.T) { expectedThreshold int expectedIdleDur int expectedLogLevel int + expectedGnmiVrf string + expectedVrf string }{ { []string{"cmd", "-port", "9090", "-threshold", "200", "-idle_conn_duration", "10", "-v", "6", "-noTLS"}, @@ -76,6 +84,8 @@ func TestFlags(t *testing.T) { 200, 10, 6, + "", + "", }, { []string{"cmd", "-port", "2020", "-threshold", "500", "-idle_conn_duration", "4", "-v", "0", "-insecure"}, @@ -83,6 +93,8 @@ func TestFlags(t *testing.T) { 500, 4, 0, + "", + "", }, { []string{"cmd", "-port", "5050", "-threshold", "10", "-idle_conn_duration", "3", "-v", "-3", "-noTLS"}, @@ -90,6 +102,17 @@ func TestFlags(t *testing.T) { 10, 3, 2, + "", + "", + }, + { + []string{"cmd", "-port", "8082", "-threshold", "1", "-idle_conn_duration", "1", "-gnmi_vrf", "mgmt", "-vrf", "mgmt", "-noTLS"}, + 8082, + 1, + 1, + 2, + "mgmt", + "mgmt", }, { []string{"cmd", "-port", "8081", "-threshold", "1", "-idle_conn_duration", "1"}, @@ -97,6 +120,8 @@ func TestFlags(t *testing.T) { 1, 1, 2, + "", + "", }, { []string{"cmd", "-port", "8081", "-threshold", "1", "-idle_conn_duration", "1", "-server_crt", "../testdata/certs/testserver.cert"}, @@ -104,6 +129,8 @@ func TestFlags(t *testing.T) { 1, 1, 2, + "", + "", }, } @@ -140,6 +167,14 @@ func TestFlags(t *testing.T) { if *config.LogLevel != test.expectedLogLevel { t.Errorf("Expected log_level to be %d, got %d", test.expectedLogLevel, *config.LogLevel) } + + if *config.GnmiVrf != test.expectedGnmiVrf { + t.Errorf("Expected gnmi_vrf to be %s, got %s", test.expectedGnmiVrf, *config.GnmiVrf) + } + + if *config.Vrf != test.expectedVrf { + t.Errorf("Expected vrf to be %s, got %s", test.expectedVrf, *config.Vrf) + } } } @@ -1416,6 +1451,27 @@ func TestFlagsNoPortNoUnixSocket(t *testing.T) { } } +func TestCertAuthDisabledWhenNoCaCert(t *testing.T) { + // When --client_auth includes "cert" but --ca_crt is empty, cert auth must be + // disabled (not fatal) so the server can still start with the remaining auth modes. + originalArgs := os.Args + defer func() { os.Args = originalArgs }() + + fs := flag.NewFlagSet("testCertAuthDisabledWhenNoCaCert", flag.ContinueOnError) + os.Args = []string{"cmd", "-port", "8080", "-noTLS", "-client_auth", "cert,password"} + + telemetryCfg, _, err := setupFlags(fs) + if err != nil { + t.Fatalf("Unexpected error: %v", err) + } + if telemetryCfg.UserAuth.Enabled("cert") { + t.Error("Expected cert auth to be disabled when ca_crt is empty, but it was still enabled") + } + if !telemetryCfg.UserAuth.Enabled("password") { + t.Error("Expected password auth to remain enabled after cert was disabled") + } +} + func TestMain(m *testing.M) { defer test_utils.MemLeakCheck() m.Run() diff --git a/transl_utils/transl_utils.go b/transl_utils/transl_utils.go index 19adfbad..5aeec747 100644 --- a/transl_utils/transl_utils.go +++ b/transl_utils/transl_utils.go @@ -27,6 +27,10 @@ var ( transLibOpMap map[int]string ) +type TranslibGetFunc func(translib.GetRequest) (translib.GetResponse, error) + +var translibGet TranslibGetFunc = translib.Get + func init() { transLibOpMap = map[int]string{ translib.REPLACE: "REPLACE", @@ -167,42 +171,59 @@ func ConvertToURI(prefix, path *gnmipb.Path, opts ...pathutil.PathValidatorOpt) return ygot.PathToString(fullPath) } +/* GetTranslibFmtType is a helper that converts gnmi Encoding to supported format types in translib */ +func getTranslFmtType(encoding gnmipb.Encoding) translib.TranslibFmtType { + + if encoding == gnmipb.Encoding_PROTO { + return translib.TRANSLIB_FMT_YGOT + } + // default to ietf_json as translib supports either Ygot or ietf_json + return translib.TRANSLIB_FMT_IETF_JSON + +} + /* Fill the values from TransLib. */ -func TranslProcessGet(uriPath string, op *string, ctx context.Context) (*gnmipb.TypedValue, error) { +func TranslProcessGet(uriPath string, op *string, ctx context.Context, encoding gnmipb.Encoding) (*gnmipb.TypedValue, *translib.GetResponse, error) { var jv []byte var data []byte rc, _ := common_utils.GetContext(ctx) - - req := translib.GetRequest{Path: uriPath, User: translib.UserRoles{Name: rc.Auth.User, Roles: rc.Auth.Roles}} + qp := translib.QueryParameters{Content: "all"} + fmtType := getTranslFmtType(encoding) + req := translib.GetRequest{Path: uriPath, FmtType: fmtType, User: translib.UserRoles{Name: rc.Auth.User, Roles: rc.Auth.Roles}, QueryParams: qp} if rc.BundleVersion != nil { nver, err := translib.NewVersion(*rc.BundleVersion) if err != nil { log.V(2).Infof("GET operation failed with error =%v", err.Error()) - return nil, err + return nil, nil, err } req.ClientVersion = nver } if rc.Auth.AuthEnabled { req.AuthEnabled = true } - resp, err1 := translib.Get(req) + resp, err1 := translibGet(req) if isTranslibSuccess(err1) { data = resp.Payload } else { - log.V(2).Infof("GET operation failed with error %v", err1.Error()) - return nil, err1 + log.V(2).Infof("GET operation failed with error =%v, %v", resp.ErrSrc, err1.Error()) + return nil, nil, err1 } - dst := new(bytes.Buffer) - json.Compact(dst, data) - jv = dst.Bytes() + /* When Proto is requested we use ValueTree to generate scalar values in the data_client.*/ + if encoding == gnmipb.Encoding_PROTO { + return nil, &resp, nil + } else { + dst := new(bytes.Buffer) + json.Compact(dst, data) + jv = dst.Bytes() - /* Fill the values into GNMI data structures . */ - return &gnmipb.TypedValue{ - Value: &gnmipb.TypedValue_JsonIetfVal{ - JsonIetfVal: jv, - }}, nil + /* Fill the values into GNMI data structures . */ + return &gnmipb.TypedValue{ + Value: &gnmipb.TypedValue_JsonIetfVal{ + JsonIetfVal: jv, + }}, nil, nil + } } diff --git a/transl_utils/transl_utils_test.go b/transl_utils/transl_utils_test.go index 7b234212..c3093be8 100644 --- a/transl_utils/transl_utils_test.go +++ b/transl_utils/transl_utils_test.go @@ -1,10 +1,14 @@ package transl_utils import ( + "context" "errors" "testing" + "github.com/Azure/sonic-mgmt-common/translib" "github.com/Azure/sonic-mgmt-common/translib/tlerr" + gnmipb "github.com/openconfig/gnmi/proto/gnmi" + "github.com/stretchr/testify/assert" ) func TestToStatus(t *testing.T) { @@ -39,3 +43,48 @@ func TestToStatus(t *testing.T) { Description: "script fail", }) } + +func TestTranslProcessGet_Success_Proto(t *testing.T) { + // 1. Save original function and restore after test + origGet := translibGet + defer func() { translibGet = origGet }() + + // 2. Define the mock behavior + expectedPayload := []byte("mock data") + translibGet = func(req translib.GetRequest) (translib.GetResponse, error) { + return translib.GetResponse{ + Payload: expectedPayload, + }, nil + } + + // 3. Setup context (ensure common_utils.GetContext won't crash) + // You might need to populate the context with Auth/Bundle version if your code reads it + ctx := context.Background() + + // 4. Execute + typedVal, resp, err := TranslProcessGet("/access-list", nil, ctx, gnmipb.Encoding_PROTO) + + // 5. Assertions + assert.NoError(t, err) + assert.Nil(t, typedVal, "PROTO encoding should return nil for TypedValue") + assert.NotNil(t, resp, "PROTO encoding should return the translib response") + assert.Equal(t, expectedPayload, resp.Payload) +} + +func TestTranslProcessGet_Success_JSON(t *testing.T) { + origGet := translibGet + defer func() { translibGet = origGet }() + + translibGet = func(req translib.GetRequest) (translib.GetResponse, error) { + return translib.GetResponse{ + Payload: []byte(`{ "foo": "bar" }`), + }, nil + } + + typedVal, _, err := TranslProcessGet("/any-path", nil, context.Background(), gnmipb.Encoding_JSON_IETF) + + assert.NoError(t, err) + assert.NotNil(t, typedVal) + // Check for compacted JSON (no spaces) + assert.Equal(t, []byte(`{"foo":"bar"}`), typedVal.GetJsonIetfVal()) +}