fix: observe grpc client (#339)

* fix: observe grpc client

Signed-off-by: gatici <[email protected]>

* fix: correct the NF name in comment

Signed-off-by: gatici <[email protected]>

* chore: update version

Signed-off-by: gatici <[email protected]>

* refactor: move implementation under service

Signed-off-by: gatici <[email protected]>

* move if clause related to MANAGED_BY_CONFIG_POD env value to service/init.go

Signed-off-by: gatici <[email protected]>

* remove else clause and fix typos

Signed-off-by: gatici <[email protected]>

* update go.mod and go.sum

Signed-off-by: gatici <[email protected]>

---------

Signed-off-by: gatici <[email protected]>

Author: gatici

VERSION

@@ -1 +1 @@
-1.6.2-dev
+1.6.2

factory/config.go

@@ -237,7 +237,7 @@ func (r *RoutingConfig) GetVersion() string {
 	return ""
 }
 
-func (c *Config) updateConfig(commChannel chan *protos.NetworkSliceResponse) bool {
+func (c *Config) UpdateConfig(commChannel chan *protos.NetworkSliceResponse) bool {
 	for {
 		rsp := <-commChannel
 		logger.GrpcLog.Infof("received updateConfig in the smf app: %+v \n", rsp)

factory/factory.go

@@ -14,7 +14,6 @@ import (
 	"os"
 	"sync"
 
-	"github.com/omec-project/config5g/proto/client"
 	"github.com/omec-project/smf/logger"
 	"gopkg.in/yaml.v2"
 )
@@ -26,7 +25,11 @@ var (
 	SmfConfigSyncLock sync.Mutex
 )
 
-// TODO: Support configuration update from REST api
+// InitConfigFactory gets the NrfConfig and subscribes the config pod.
+// This observes the GRPC client availability and connection status in a loop.
+// When the GRPC server pod is restarted, GRPC connection status stuck in idle.
+// If GRPC client does not exist, creates it. If client exists but GRPC connectivity is not ready,
+// then it closes the existing client start a new client.
 func InitConfigFactory(f string) error {
 	if content, err := os.ReadFile(f); err != nil {
 		return err
@@ -45,15 +48,7 @@ func InitConfigFactory(f string) error {
 			enableKafka := true
 			SmfConfig.Configuration.KafkaInfo.EnableKafka = &enableKafka
 		}
-
-		roc := os.Getenv("MANAGED_BY_CONFIG_POD")
-		if roc == "true" {
-			gClient := client.ConnectToConfigServer(SmfConfig.Configuration.WebuiUri)
-			commChannel := gClient.PublishOnConfigChange(false)
-			go SmfConfig.updateConfig(commChannel)
-		}
 	}
-
 	return nil
 }
 

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826
 	github.com/omec-project/aper v1.2.1
-	github.com/omec-project/config5g v1.5.1
+	github.com/omec-project/config5g v1.5.4
 	github.com/omec-project/nas v1.4.2
 	github.com/omec-project/ngap v1.3.0
 	github.com/omec-project/openapi v1.3.1
@@ -79,7 +79,7 @@ require (
 	golang.org/x/sys v0.26.0 // indirect
 	golang.org/x/text v0.19.0 // indirect
 	google.golang.org/grpc v1.67.1 // indirect
-	google.golang.org/protobuf v1.34.2 // indirect
+	google.golang.org/protobuf v1.35.1 // indirect
 	gopkg.in/h2non/gock.v1 v1.1.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	inet.af/netaddr v0.0.0-20220811202034-502d2d690317 // indirect

go.sum

@@ -225,8 +225,8 @@ github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
 github.com/omec-project/aper v1.2.1 h1:A4KhlTMta5mUA1DX2njOjOn0SuFi5WoWnSqdbRnl9Q8=
 github.com/omec-project/aper v1.2.1/go.mod h1:CbCZ0uNmcD3XA4YM6wBwpqP1PYtGAaC3RqLL4B5gG2U=
-github.com/omec-project/config5g v1.5.1 h1:JaVgr76tnjJIb7Uoesv5a9GI72NdOXtCvfukj0/ONio=
-github.com/omec-project/config5g v1.5.1/go.mod h1:o04ZdwGcM7tbGjuT5t/WzYSKLXOSnFl6vH7b6BGAspU=
+github.com/omec-project/config5g v1.5.4 h1:5JMw5Fsr5qyLZpQi3IZQaQPj78QJMFQXDVS3QDMOY9Y=
+github.com/omec-project/config5g v1.5.4/go.mod h1:HOvQtmi79f8cw35AiFHWHDoCTuZbXfMjeFJWgtPbwaI=
 github.com/omec-project/nas v1.4.2 h1:V+DKwxmxRrwMmpNQ3+kx7WQjqY1XqbWDIeQXWIof+Gk=
 github.com/omec-project/nas v1.4.2/go.mod h1:cAXzuNN0dVDXgwg2cbptQjwTKP4llXPV8jrU4LVp7H0=
 github.com/omec-project/ngap v1.3.0 h1:NuAc6GZRup6E3jg+fGTy8aR4VkOlB1mnkaaYr6OGmMc=
@@ -519,8 +519,8 @@ google.golang.org/protobuf v1.23.1-0.20200526195155-81db48ad09cc/go.mod h1:EGpAD
 google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
-google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
+google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
+google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

service/init.go

@@ -18,6 +18,8 @@ import (
 	"time"
 
 	aperLogger "github.com/omec-project/aper/logger"
+	grpcClient "github.com/omec-project/config5g/proto/client"
+	protos "github.com/omec-project/config5g/proto/sdcoreConfig"
 	nasLogger "github.com/omec-project/nas/logger"
 	ngapLogger "github.com/omec-project/ngap/logger"
 	openapiLogger "github.com/omec-project/openapi/logger"
@@ -139,9 +141,65 @@ func (smf *SMF) Initialize(c *cli.Context) error {
 		}()
 	}
 
+	if os.Getenv("MANAGED_BY_CONFIG_POD") == "true" {
+		logger.InitLog.Infoln("MANAGED_BY_CONFIG_POD is true")
+		go manageGrpcClient(factory.SmfConfig.Configuration.WebuiUri)
+	}
 	return nil
 }
 
+// manageGrpcClient connects the config pod GRPC server and subscribes the config changes.
+// Then it updates SMF configuration.
+func manageGrpcClient(webuiUri string) {
+	var configChannel chan *protos.NetworkSliceResponse
+	var client grpcClient.ConfClient
+	var stream protos.ConfigService_NetworkSliceSubscribeClient
+	var err error
+	count := 0
+	for {
+		if client != nil {
+			if client.CheckGrpcConnectivity() != "ready" {
+				time.Sleep(time.Second * 30)
+				count++
+				if count > 5 {
+					err = client.GetConfigClientConn().Close()
+					if err != nil {
+						logger.InitLog.Infof("failing ConfigClient is not closed properly: %+v", err)
+					}
+					client = nil
+					count = 0
+				}
+				logger.InitLog.Infoln("checking the connectivity readiness")
+				continue
+			}
+
+			if stream == nil {
+				stream, err = client.SubscribeToConfigServer()
+				if err != nil {
+					logger.InitLog.Infof("failing SubscribeToConfigServer: %+v", err)
+					continue
+				}
+			}
+
+			if configChannel == nil {
+				configChannel = client.PublishOnConfigChange(true, stream)
+				logger.InitLog.Infoln("PublishOnConfigChange is triggered")
+				go factory.SmfConfig.UpdateConfig(configChannel)
+				logger.InitLog.Infoln("SMF updateConfig is triggered")
+			}
+		} else {
+			client, err = grpcClient.ConnectToConfigServer(webuiUri)
+			stream = nil
+			configChannel = nil
+			logger.InitLog.Infoln("connecting to config server")
+			if err != nil {
+				logger.InitLog.Errorf("%+v", err)
+			}
+			continue
+		}
+	}
+}
+
 func (smf *SMF) setLogLevel() {
 	if factory.SmfConfig.Logger == nil {
 		logger.InitLog.Warnln("SMF config without log level setting")
@@ -263,10 +321,9 @@ func (smf *SMF) Start() {
 	context.InitSMFUERouting(&factory.UERoutingConfig)
 
 	// Wait for additional/updated config from config pod
-	roc := os.Getenv("MANAGED_BY_CONFIG_POD")
-	if roc == "true" {
-		logger.InitLog.Infoln("configuration is managed by Config Pod")
-		logger.InitLog.Infoln("waiting for initial configuration from config pod")
+	if os.Getenv("MANAGED_BY_CONFIG_POD") == "true" {
+		logger.InitLog.Infof("configuration is managed by Config Pod")
+		logger.InitLog.Infof("waiting for initial configuration from config pod")
 
 		// Main thread should be blocked for config update from ROC
 		// Future config update from ROC can be handled via background go-routine.