Skip to content

Commit 649e5f6

Browse files
deadlycoconutsdeadlycoconuts
authored
fix(xp-treatment): Add field tags to poller configs and make XP plugin pass poller configs (#88)
* Add field tags to poller configs * Add poller config to plugin * Update message queue config with json tags * Update xp plugin to use message queue config from manager instead of xp management * Reorder imports * Change poller interval field to int * Reorder imports * Reorder imports * Refactor poller into a service
1 parent 397e59b commit 649e5f6

File tree

12 files changed

+66
-82
lines changed

12 files changed

+66
-82
lines changed

‎common/messagequeue/config.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,10 @@ const (
1212

1313
type MessageQueueConfig struct {
1414
// The type of Message Queue for event updates
15-
Kind MessageQueueKind `default:""`
15+
Kind MessageQueueKind `json:"kind" default:""`
1616

1717
// PubSubConfig captures the config related to publishing and subscribing to a PubSub Message Queue
18-
PubSubConfig *PubSubConfig
18+
PubSubConfig *PubSubConfig `json:"pub_sub_config"`
1919
}
2020

2121
type PubSubConfig struct {

‎plugins/turing/config/config.go

+11-8
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66

77
"github.com/caraml-dev/mlp/api/pkg/instrumentation/newrelic"
88
"github.com/caraml-dev/mlp/api/pkg/instrumentation/sentry"
9+
common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
910
"github.com/caraml-dev/xp/treatment-service/config"
1011
"github.com/go-playground/validator/v10"
1112
)
@@ -46,14 +47,16 @@ type TreatmentServicePluginConfig struct {
4647
Port int `json:"port" default:"8080"`
4748
PubSubTimeoutSeconds int `json:"pub_sub_timeout_seconds" validate:"required"`
4849

49-
AssignedTreatmentLogger config.AssignedTreatmentLoggerConfig `json:"assigned_treatment_logger"`
50-
DebugConfig config.DebugConfig `json:"debug_config"`
51-
DeploymentConfig config.DeploymentConfig `json:"deployment_config"`
52-
ManagementService config.ManagementServiceConfig `json:"management_service"`
53-
MonitoringConfig config.Monitoring `json:"monitoring_config"`
54-
SwaggerConfig config.SwaggerConfig `json:"swagger_config"`
55-
NewRelicConfig newrelic.Config `json:"new_relic_config"`
56-
SentryConfig sentry.Config `json:"sentry_config"`
50+
AssignedTreatmentLogger config.AssignedTreatmentLoggerConfig `json:"assigned_treatment_logger"`
51+
DebugConfig config.DebugConfig `json:"debug_config"`
52+
DeploymentConfig config.DeploymentConfig `json:"deployment_config"`
53+
MessageQueueConfig common_mq_config.MessageQueueConfig `json:"message_queue_config"`
54+
ManagementService config.ManagementServiceConfig `json:"management_service"`
55+
MonitoringConfig config.Monitoring `json:"monitoring_config"`
56+
SwaggerConfig config.SwaggerConfig `json:"swagger_config"`
57+
NewRelicConfig newrelic.Config `json:"new_relic_config"`
58+
SentryConfig sentry.Config `json:"sentry_config"`
59+
ManagementServicePollerConfig config.ManagementServicePollerConfig `json:"management_service_poller_config"`
5760
}
5861

5962
type Variable struct {

‎plugins/turing/manager/experiment_manager.go

+13-31
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@ import (
1414

1515
xpclient "github.com/caraml-dev/xp/clients/management"
1616
"github.com/caraml-dev/xp/common/api/schema"
17-
common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
1817
_config "github.com/caraml-dev/xp/plugins/turing/config"
1918
"github.com/caraml-dev/xp/treatment-service/config"
2019
"github.com/go-playground/validator/v10"
@@ -142,37 +141,20 @@ func (em *experimentManager) MakeTreatmentServicePluginConfig(
142141
projectID int,
143142
) (*config.Config, error) {
144143
pluginConfig := &config.Config{
145-
Port: em.TreatmentServicePluginConfig.Port,
146-
ProjectIds: []string{strconv.Itoa(projectID)},
147-
AssignedTreatmentLogger: em.TreatmentServicePluginConfig.AssignedTreatmentLogger,
148-
DebugConfig: em.TreatmentServicePluginConfig.DebugConfig,
149-
DeploymentConfig: em.TreatmentServicePluginConfig.DeploymentConfig,
150-
ManagementService: em.TreatmentServicePluginConfig.ManagementService,
151-
MonitoringConfig: em.TreatmentServicePluginConfig.MonitoringConfig,
152-
SwaggerConfig: em.TreatmentServicePluginConfig.SwaggerConfig,
153-
NewRelicConfig: em.TreatmentServicePluginConfig.NewRelicConfig,
154-
SentryConfig: em.TreatmentServicePluginConfig.SentryConfig,
155-
SegmenterConfig: *treatmentServiceConfig.SegmenterConfig,
156-
}
157-
messageQueueKind := *treatmentServiceConfig.MessageQueueConfig.Kind
158-
switch messageQueueKind {
159-
case schema.MessageQueueKindPubsub:
160-
pluginConfig.MessageQueueConfig = common_mq_config.MessageQueueConfig{
161-
Kind: "pubsub",
162-
PubSubConfig: &common_mq_config.PubSubConfig{
163-
Project: *treatmentServiceConfig.MessageQueueConfig.PubSub.Project,
164-
TopicName: *treatmentServiceConfig.MessageQueueConfig.PubSub.TopicName,
165-
PubSubTimeoutSeconds: em.TreatmentServicePluginConfig.PubSubTimeoutSeconds,
166-
},
167-
}
168-
case schema.MessageQueueKindNoop:
169-
pluginConfig.MessageQueueConfig = common_mq_config.MessageQueueConfig{
170-
Kind: "",
171-
}
172-
default:
173-
return nil, fmt.Errorf("invalid message queue kind (%s) was provided", messageQueueKind)
144+
Port: em.TreatmentServicePluginConfig.Port,
145+
ProjectIds: []string{strconv.Itoa(projectID)},
146+
AssignedTreatmentLogger: em.TreatmentServicePluginConfig.AssignedTreatmentLogger,
147+
DebugConfig: em.TreatmentServicePluginConfig.DebugConfig,
148+
DeploymentConfig: em.TreatmentServicePluginConfig.DeploymentConfig,
149+
MessageQueueConfig: em.TreatmentServicePluginConfig.MessageQueueConfig,
150+
ManagementService: em.TreatmentServicePluginConfig.ManagementService,
151+
MonitoringConfig: em.TreatmentServicePluginConfig.MonitoringConfig,
152+
SwaggerConfig: em.TreatmentServicePluginConfig.SwaggerConfig,
153+
NewRelicConfig: em.TreatmentServicePluginConfig.NewRelicConfig,
154+
SentryConfig: em.TreatmentServicePluginConfig.SentryConfig,
155+
SegmenterConfig: *treatmentServiceConfig.SegmenterConfig,
156+
ManagementServicePollerConfig: em.TreatmentServicePluginConfig.ManagementServicePollerConfig,
174157
}
175-
176158
return pluginConfig, nil
177159
}
178160

‎plugins/turing/manager/experiment_manager_test.go

+8
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,14 @@ func TestNewExperimentManager(t *testing.T) {
6262
"environment_type": "dev",
6363
"max_go_routines": 200
6464
},
65+
"message_queue_config": {
66+
"kind": "dev",
67+
"pub_sub_config": {
68+
"project":"dev",
69+
"topic_name":"xp-update",
70+
"pub_sub_timeout_seconds": 30
71+
}
72+
},
6573
"management_service": {
6674
"authorization_enabled": true,
6775
"url": "http://xp-management.global.io/api/xp/v1"

‎plugins/turing/runner/experiment_runner.go

+3
Original file line numberDiff line numberDiff line change
@@ -257,6 +257,9 @@ func (er *experimentRunner) startBackgroundServices(
257257
}
258258
}()
259259
}
260+
if er.appContext.PollerService != nil {
261+
er.appContext.PollerService.Start()
262+
}
260263
}
261264

262265
func (er *experimentRunner) getRequestParams(

‎treatment-service/appcontext/appcontext.go

+7
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ type AppContext struct {
2424

2525
AssignedTreatmentLogger *monitoring.AssignedTreatmentLogger
2626
LocalStorage *models.LocalStorage
27+
PollerService *services.PollerService
2728
}
2829

2930
func NewAppContext(cfg *config.Config) (*AppContext, error) {
@@ -122,6 +123,11 @@ func NewAppContext(cfg *config.Config) (*AppContext, error) {
122123
return nil, err
123124
}
124125

126+
var pollerService *services.PollerService
127+
if cfg.ManagementServicePollerConfig.Enabled {
128+
pollerService = services.NewPollerService(cfg.ManagementServicePollerConfig, localStorage)
129+
}
130+
125131
appContext := &AppContext{
126132
ExperimentService: experimentSvc,
127133
MetricService: metricService,
@@ -131,6 +137,7 @@ func NewAppContext(cfg *config.Config) (*AppContext, error) {
131137
AssignedTreatmentLogger: logger,
132138
MessageQueueService: messageQueueService,
133139
LocalStorage: localStorage,
140+
PollerService: pollerService,
134141
}
135142

136143
return appContext, nil

‎treatment-service/config/config.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,6 @@ package config
33
import (
44
"fmt"
55
"strconv"
6-
"time"
76

87
"github.com/caraml-dev/mlp/api/pkg/instrumentation/newrelic"
98
"github.com/caraml-dev/mlp/api/pkg/instrumentation/sentry"
@@ -96,8 +95,8 @@ type ManagementServiceConfig struct {
9695
}
9796

9897
type ManagementServicePollerConfig struct {
99-
Enabled bool `default:"false"`
100-
PollInterval time.Duration `default:"30s"`
98+
Enabled bool `json:"enabled" default:"false"`
99+
PollIntervalSeconds int `json:"poll_interval" default:"30"`
101100
}
102101

103102
func (c *Config) GetProjectIds() []models.ProjectId {

‎treatment-service/config/config_test.go

+4-5
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@ package config
22

33
import (
44
"testing"
5-
"time"
65

76
"github.com/caraml-dev/mlp/api/pkg/instrumentation/newrelic"
87
"github.com/caraml-dev/mlp/api/pkg/instrumentation/sentry"
@@ -66,8 +65,8 @@ func TestDefaultConfigs(t *testing.T) {
6665
SentryConfig: sentry.Config{Enabled: false, Labels: emptyStringMap},
6766
SegmenterConfig: make(map[string]interface{}),
6867
ManagementServicePollerConfig: ManagementServicePollerConfig{
69-
Enabled: false,
70-
PollInterval: 30 * time.Second,
68+
Enabled: false,
69+
PollIntervalSeconds: 30,
7170
},
7271
}
7372
cfg, err := Load()
@@ -133,8 +132,8 @@ func TestLoadMultipleConfigs(t *testing.T) {
133132
SentryConfig: sentry.Config{Enabled: true, DSN: "my.amazing.sentry.dsn", Labels: map[string]string{"app": "xp-treatment-service"}},
134133
SegmenterConfig: map[string]interface{}{"s2_ids": map[string]interface{}{"mins2celllevel": 9, "maxs2celllevel": 15}},
135134
ManagementServicePollerConfig: ManagementServicePollerConfig{
136-
Enabled: false,
137-
PollInterval: 30 * time.Second,
135+
Enabled: false,
136+
PollIntervalSeconds: 30,
138137
},
139138
}
140139

‎treatment-service/go.mod

-4
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@ require (
1414
github.com/deepmap/oapi-codegen v1.11.0
1515
github.com/getkin/kin-openapi v0.94.0
1616
github.com/go-chi/chi/v5 v5.0.7
17-
github.com/go-playground/validator/v10 v10.11.1
1817
github.com/golang-collections/collections v0.0.0-20130729185459-604e922904d3
1918
github.com/golang/geo v0.0.0-20210211234256-740aa86cb551
2019
github.com/google/go-cmp v0.6.0
@@ -104,8 +103,6 @@ require (
104103
github.com/go-openapi/jsonpointer v0.19.6 // indirect
105104
github.com/go-openapi/jsonreference v0.20.2 // indirect
106105
github.com/go-openapi/swag v0.22.3 // indirect
107-
github.com/go-playground/locales v0.14.0 // indirect
108-
github.com/go-playground/universal-translator v0.18.0 // indirect
109106
github.com/go-viper/mapstructure/v2 v2.0.0 // indirect
110107
github.com/goccy/go-json v0.9.11 // indirect
111108
github.com/gofrs/flock v0.8.1 // indirect
@@ -139,7 +136,6 @@ require (
139136
github.com/klauspost/asmfmt v1.3.2 // indirect
140137
github.com/klauspost/compress v1.17.4 // indirect
141138
github.com/klauspost/cpuid/v2 v2.0.9 // indirect
142-
github.com/leodido/go-urn v1.2.1 // indirect
143139
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
144140
github.com/magiconair/properties v1.8.7 // indirect
145141
github.com/mailru/easyjson v0.7.7 // indirect

‎treatment-service/go.sum

-6
Original file line numberDiff line numberDiff line change
@@ -333,18 +333,13 @@ github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh
333333
github.com/go-openapi/swag v0.21.1/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
334334
github.com/go-openapi/swag v0.22.3 h1:yMBqmnQ0gyZvEb/+KzuWZOXgllrXT4SADYbvDaXHv/g=
335335
github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14=
336-
github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A=
337336
github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
338337
github.com/go-playground/locales v0.13.0/go.mod h1:taPMhCMXrRLJO55olJkUXHZBHCxTMfnGwq/HNwmWNS8=
339-
github.com/go-playground/locales v0.14.0 h1:u50s323jtVGugKlcYeyzC0etD1HifMjqmJqb8WugfUU=
340338
github.com/go-playground/locales v0.14.0/go.mod h1:sawfccIbzZTqEDETgFXqTho0QybSa7l++s0DH+LDiLs=
341339
github.com/go-playground/universal-translator v0.17.0/go.mod h1:UkSxE5sNxxRwHyU+Scu5vgOQjsIJAF8j9muTVoKLVtA=
342-
github.com/go-playground/universal-translator v0.18.0 h1:82dyy6p4OuJq4/CByFNOn/jYrnRPArHwAcmLoJZxyho=
343340
github.com/go-playground/universal-translator v0.18.0/go.mod h1:UvRDBj+xPUEGrFYl+lu/H90nyDXpg0fqeB/AQUGNTVA=
344341
github.com/go-playground/validator/v10 v10.4.1/go.mod h1:nlOn6nFhuKACm19sB/8EGNn9GlaMV7XkbRSipzJ0Ii4=
345342
github.com/go-playground/validator/v10 v10.11.0/go.mod h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
346-
github.com/go-playground/validator/v10 v10.11.1 h1:prmOlTVv+YjZjmRmNSF3VmspqJIxJWXmqUsHwfTRRkQ=
347-
github.com/go-playground/validator/v10 v10.11.1/go.mod h1:i+3WkQ1FvaUjjxh1kSvIA4dMGDBiPU55YFDl0WbKdWU=
348343
github.com/go-sql-driver/mysql v1.3.0 h1:pgwjLi/dvffoP9aabwkT3AKpXQM93QARkjFhDDqC1UE=
349344
github.com/go-sql-driver/mysql v1.3.0/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w=
350345
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -604,7 +599,6 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
604599
github.com/labstack/echo/v4 v4.7.2/go.mod h1:xkCDAdFCIf8jsFQ5NnbK7oqaF/yU1A1X20Ltm0OvSks=
605600
github.com/labstack/gommon v0.3.1/go.mod h1:uW6kP17uPlLJsD3ijUYn3/M5bAxtlZhMI6m3MFxTMTM=
606601
github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII=
607-
github.com/leodido/go-urn v1.2.1 h1:BqpAaACuzVSgi/VLzGZIobT2z4v53pjosyNd9Yv6n/w=
608602
github.com/leodido/go-urn v1.2.1/go.mod h1:zt4jvISO2HfUBqxjfIshjdMTYS56ZS/qv49ictyFfxY=
609603
github.com/lestrrat-go/backoff/v2 v2.0.8/go.mod h1:rHP/q/r9aT27n24JQLa7JhSQZCKBBOiM/uP402WwN8Y=
610604
github.com/lestrrat-go/blackmagic v1.0.0/go.mod h1:TNgH//0vYSs8VXDCfkZLgIrVTTXQELZffUV0tz3MtdQ=

‎treatment-service/server/server.go

+4-13
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,6 @@ type Server struct {
3333
subscribe bool
3434
// cleanup captures all the actions to be executed on server shut down
3535
cleanup []func()
36-
// poller captures the poller instance
37-
poller *Poller
3836
}
3937

4038
// NewServer creates and configures an APIServer serving all application routes.
@@ -108,11 +106,6 @@ func NewServer(configFiles []string) (*Server, error) {
108106
subscribe = true
109107
}
110108

111-
var poller *Poller
112-
if cfg.ManagementServicePollerConfig.Enabled {
113-
poller = NewPoller(cfg.ManagementServicePollerConfig, appCtx.LocalStorage)
114-
}
115-
116109
srv := http.Server{
117110
Addr: cfg.ListenAddress(),
118111
Handler: mux,
@@ -123,7 +116,6 @@ func NewServer(configFiles []string) (*Server, error) {
123116
appContext: appCtx,
124117
subscribe: subscribe,
125118
cleanup: cleanup,
126-
poller: poller,
127119
}, nil
128120
}
129121

@@ -141,11 +133,6 @@ func (srv *Server) Start() {
141133
}()
142134
log.Printf("Listening on %s\n", srv.Addr)
143135

144-
if srv.poller != nil {
145-
log.Println("Starting poller...")
146-
srv.poller.Start()
147-
}
148-
149136
stop := make(chan os.Signal, 1)
150137
signal.Notify(stop, os.Interrupt)
151138

@@ -193,5 +180,9 @@ func (srv *Server) startBackgroundService(errChannel chan error) context.CancelF
193180
}
194181
}()
195182

183+
if srv.appContext.PollerService != nil {
184+
srv.appContext.PollerService.Start()
185+
}
186+
196187
return cancel
197188
}

‎treatment-service/server/poller.go ‎treatment-service/services/poller_service.go

+12-10
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
package server
1+
package services
22

33
import (
44
"log"
@@ -8,31 +8,33 @@ import (
88
"github.com/caraml-dev/xp/treatment-service/models"
99
)
1010

11-
type Poller struct {
11+
type PollerService struct {
1212
pollerConfig config.ManagementServicePollerConfig
1313
localStorage *models.LocalStorage
1414
stopChannel chan struct{}
1515
}
1616

17-
// NewPoller creates a new Poller instance with the given configuration and local storage.
17+
// NewPollerService creates a new PollerService instance with the given configuration and local storage.
1818
// pollerConfig: configuration for the poller
1919
// localStorage: local storage to be used by the poller
20-
func NewPoller(pollerConfig config.ManagementServicePollerConfig, localStorage *models.LocalStorage) *Poller {
21-
return &Poller{
20+
func NewPollerService(pollerConfig config.ManagementServicePollerConfig, localStorage *models.LocalStorage) *PollerService {
21+
return &PollerService{
2222
pollerConfig: pollerConfig,
2323
localStorage: localStorage,
2424
stopChannel: make(chan struct{}),
2525
}
2626
}
2727

28-
func (p *Poller) Start() {
29-
ticker := time.NewTicker(p.pollerConfig.PollInterval)
28+
func (p *PollerService) Start() {
29+
log.Println("Starting management service poller service...")
30+
pollInterval := time.Duration(p.pollerConfig.PollIntervalSeconds) * time.Second
31+
ticker := time.NewTicker(pollInterval)
3032
go func() {
3133
for {
3234
select {
3335
case <-ticker.C:
3436
err := p.Refresh()
35-
log.Printf("Polling at %v with interval %v", time.Now(), p.pollerConfig.PollInterval)
37+
log.Printf("Polling at %v with interval %v", time.Now(), pollInterval)
3638
if err != nil {
3739
log.Printf("Error updating local storage: %v", err)
3840
continue
@@ -45,11 +47,11 @@ func (p *Poller) Start() {
4547
}()
4648
}
4749

48-
func (p *Poller) Stop() {
50+
func (p *PollerService) Stop() {
4951
close(p.stopChannel)
5052
}
5153

52-
func (p *Poller) Refresh() error {
54+
func (p *PollerService) Refresh() error {
5355
err := p.localStorage.Init()
5456
return err
5557
}

0 commit comments

Comments
 (0)