[FSSDK-9632] feat: Implement notification consistency across agent no…

…des (#399)

* start basic implementation

* resolve conflict

* change notification.go logic

* modify notification handler

* Update config

* add redis pubsub struct

* update config

* implement & use redis center as notification.center

* use opticlient from middleware

* rename notification handler name

* update config management

* update notidication sender

* cleaner syncup code

* update test

* add notification filtering

* fix bug

* rename syncer struct

* not using pointers

* fix test

* implement data channel

* update tests

* add unit tests for default receiver

* add unit tests for redis receiver

* add unit test

* add comments

* clean up

* fix license header year

* update config

* use default response writer

* filter on sdk keys

* update unit test

* fix failing acceptance test

* rename

cmd/optimizely/main.go

@@ -267,7 +267,7 @@ func main() {
 
 	ctx, cancel := context.WithCancel(context.Background()) // Create default service context
 	sg := server.NewGroup(ctx, conf.Server)                 // Create a new server group to manage the individual http listeners
-	optlyCache := optimizely.NewCache(ctx, conf.Client, sdkMetricsRegistry)
+	optlyCache := optimizely.NewCache(ctx, *conf, sdkMetricsRegistry)
 	optlyCache.Init(conf.SDKKeys)
 
 	// goroutine to check for signals to gracefully shutdown listeners
@@ -281,7 +281,7 @@ func main() {
 		cancel()
 	}()
 
-	apiRouter := routers.NewDefaultAPIRouter(optlyCache, conf.API, agentMetricsRegistry)
+	apiRouter := routers.NewDefaultAPIRouter(optlyCache, *conf, agentMetricsRegistry)
 	adminRouter := routers.NewAdminRouter(*conf)
 
 	log.Info().Str("version", conf.Version).Msg("Starting services.")

config.yaml

@@ -243,3 +243,17 @@ runtime:
     ## To just read the current rate, pass rate < 0.
     ## (For n>1 the details of sampling may change.)
     mutexProfileFraction: 0
+
+## synchronization should be enabled when multiple replicas of agent is deployed
+## if notification synchronization is enabled, then the active notification event-stream API
+## will get the notifications from multiple replicas
+synchronization:
+    pubsub:
+        redis:
+            host: "redis.demo.svc:6379"
+            password: ""
+            database: 0
+            channel: "optimizely-sync"
+    notification:
+        enable: false
+        default: "redis"

config/config.go

@@ -126,6 +126,20 @@ func NewDefaultConfig() *AgentConfig {
 		Webhook: WebhookConfig{
 			Port: "8085",
 		},
+		Synchronization: SyncConfig{
+			Pubsub: map[string]interface{}{
+				"redis": map[string]interface{}{
+					"host":     "localhost:6379",
+					"password": "",
+					"database": 0,
+					"channel":  "optimizely-notifications",
+				},
+			},
+			Notification: NotificationConfig{
+				Enable:  false,
+				Default: "redis",
+			},
+		},
 	}
 
 	return &config
@@ -139,14 +153,27 @@ type AgentConfig struct {
 
 	SDKKeys []string `yaml:"sdkKeys" json:"sdkKeys"`
 
-	Admin   AdminConfig   `json:"admin"`
-	API     APIConfig     `json:"api"`
-	Log     LogConfig     `json:"log"`
-	Tracing TracingConfig `json:"tracing"`
-	Client  ClientConfig  `json:"client"`
-	Runtime RuntimeConfig `json:"runtime"`
-	Server  ServerConfig  `json:"server"`
-	Webhook WebhookConfig `json:"webhook"`
+	Admin           AdminConfig   `json:"admin"`
+	API             APIConfig     `json:"api"`
+	Log             LogConfig     `json:"log"`
+	Tracing         TracingConfig `json:"tracing"`
+	Client          ClientConfig  `json:"client"`
+	Runtime         RuntimeConfig `json:"runtime"`
+	Server          ServerConfig  `json:"server"`
+	Webhook         WebhookConfig `json:"webhook"`
+	Synchronization SyncConfig    `json:"synchronization"`
+}
+
+// SyncConfig contains Synchronization configuration for the multiple Agent nodes
+type SyncConfig struct {
+	Pubsub       map[string]interface{} `json:"pubsub"`
+	Notification NotificationConfig     `json:"notification"`
+}
+
+// NotificationConfig contains Notification Synchronization configuration for the multiple Agent nodes
+type NotificationConfig struct {
+	Enable  bool   `json:"enable"`
+	Default string `json:"default"`
 }
 
 // HTTPSDisabledWarning is logged when keyfile and certfile are not provided in server configuration

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/golang-jwt/jwt/v4 v4.5.0
 	github.com/google/uuid v1.3.0
 	github.com/lestrrat-go/jwx v0.9.0
-	github.com/optimizely/go-sdk v1.8.4-0.20230515121609-7ffed835c991
+	github.com/optimizely/go-sdk v1.8.4-0.20230911163718-b10e161e39b8
 	github.com/orcaman/concurrent-map v1.0.0
 	github.com/prometheus/client_golang v1.11.0
 	github.com/rakyll/statik v0.1.7

go.sum

@@ -244,8 +244,8 @@ github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
 github.com/onsi/gomega v1.18.1/go.mod h1:0q+aL8jAiMXy9hbwj2mr5GziHiwhAIQpFmmtT5hitRs=
-github.com/optimizely/go-sdk v1.8.4-0.20230515121609-7ffed835c991 h1:bRoRDKRa7EgSTCEb54qaDuU6IDegQKQumun8buDV/cY=
-github.com/optimizely/go-sdk v1.8.4-0.20230515121609-7ffed835c991/go.mod h1:06VK8mwwQTEh7QzP+qivf16tXtXEpoeblqtlhfvWEgk=
+github.com/optimizely/go-sdk v1.8.4-0.20230911163718-b10e161e39b8 h1:1LhZsu7IB7LR3PzwIzfP56cdOkUAKRXxW1wljd352sg=
+github.com/optimizely/go-sdk v1.8.4-0.20230911163718-b10e161e39b8/go.mod h1:zITWqffjOXsae/Z0PlCN5kgJRgJF/0g/k8RBEsxNrxg=
 github.com/orcaman/concurrent-map v1.0.0 h1:I/2A2XPCb4IuQWcQhBhSwGfiuybl/J0ev9HDbW65HOY=
 github.com/orcaman/concurrent-map v1.0.0/go.mod h1:Lu3tH6HLW3feq74c2GC+jIMS/K2CFcDWnWD9XkenwhI=
 github.com/pelletier/go-toml/v2 v2.0.6 h1:nrzqCb7j9cDFj2coyLNLaZuJTLjWjlaz6nvTvIwycIU=