Refactor Message Queue implementation (#64)

* Refactor management service message queue

* Refactor treatment service message queue

* Update plugin

* Bump golangci-lint version

* More fixes

* Refactor implementation to message queue packages

Author: terryyylim

.github/workflows/plugins.yml

@@ -25,7 +25,7 @@ on:
 
 env:
   GO_VERSION: 1.18
-  GO_LINT_VERSION: v1.45.2
+  GO_LINT_VERSION: v1.51.2
 
 jobs:
   lint-go:

.github/workflows/xp.yml

@@ -26,7 +26,7 @@ on:
 env:
   ARTIFACT_RETENTION_DAYS: 7
   GO_VERSION: 1.18
-  GO_LINT_VERSION: v1.45.2
+  GO_LINT_VERSION: v1.51.2
 
 jobs:
   lint-python:

Makefile

@@ -98,7 +98,7 @@ compile-protos: | $(protoc_dir)
 .PHONY: setup
 setup:
 	@echo "> Initializing dependencies ..."
-	@test -x ${GOPATH}/bin/golangci-lint || go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.40.1
+	@test -x ${GOPATH}/bin/golangci-lint || go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.51.2
 
 	@echo "Setting up dev tools..."
 	@test -x "$(which pre-commit)" || pip install pre-commit

api/schema.yaml

@@ -9,13 +9,26 @@ components:
         topic_name:
           type: string
           description: Topic name of the PubSub subscription
+    MessageQueueKind:
+      description: Kind of message queue
+      type: string
+      enum:
+        - noop
+        - pubsub
+    MessageQueueConfig:
+      type: object
+      properties:
+        kind:
+          $ref: '#/components/schemas/MessageQueueKind'
+        pub_sub:
+          $ref: '#/components/schemas/PubSub'
     SegmenterConfig:
       type: object
     TreatmentServiceConfig:
       type: object
       properties:
-        pub_sub:
-          $ref: '#/components/schemas/PubSub'
+        message_queue_config:
+          $ref: '#/components/schemas/MessageQueueConfig'
         segmenter_config:
           $ref: '#/components/schemas/SegmenterConfig'
     SelectedTreatmentData:

clients/testutils/mocks/management/ManagementClientInterface.go

@@ -0,0 +1,26 @@
+package messagequeue
+
+// MessageQueueKind describes the message queue for transmitting event updates to and fro Treatment Service
+type MessageQueueKind = string
+
+const (
+	// NoopMQ is a No-Op Message Queue
+	NoopMQ MessageQueueKind = ""
+	// PubSubMQ is a PubSub Message Queue
+	PubSubMQ MessageQueueKind = "pubsub"
+)
+
+type MessageQueueConfig struct {
+	// The type of Message Queue for event updates
+	Kind MessageQueueKind `default:""`
+
+	// PubSubConfig captures the config related to publishing and subscribing to a PubSub Message Queue
+	PubSubConfig *PubSubConfig
+}
+
+type PubSubConfig struct {
+	Project   string `json:"project" default:"dev" validate:"required"`
+	TopicName string `json:"topic_name" default:"xp-update" validate:"required"`
+	// PubSubTimeoutSeconds is the duration beyond which subscribing to a topic will time out
+	PubSubTimeoutSeconds int `json:"pub_sub_timeout_seconds" default:"30" validate:"required"`
+}

clients/testutils/mocks/treatment/TreatmentClientInterface.go

@@ -7,6 +7,7 @@ import (
 	"github.com/caraml-dev/xp/management-service/config"
 	mw "github.com/caraml-dev/xp/management-service/middleware"
 	"github.com/caraml-dev/xp/management-service/services"
+	"github.com/caraml-dev/xp/management-service/services/messagequeue"
 )
 
 type AppContext struct {
@@ -29,11 +30,7 @@ func NewAppContext(db *gorm.DB, authorizer *mw.Authorizer, cfg *config.Config) (
 	}
 
 	// Init Services
-	pubSubConfig := config.PubSubConfig{
-		Project:   cfg.PubSubConfig.Project,
-		TopicName: cfg.PubSubConfig.TopicName,
-	}
-	pubSubPublisherService, err := services.NewPubSubPublisherService(&pubSubConfig)
+	messageQueueService, err := messagequeue.NewMessageQueueService(*cfg.MessageQueueConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -76,7 +73,7 @@ func NewAppContext(db *gorm.DB, authorizer *mw.Authorizer, cfg *config.Config) (
 		treatmentSvc,
 		treatmentHistorySvc,
 		validationService,
-		pubSubPublisherService,
+		messageQueueService,
 		configurationSvc,
 	)
 

common/api/schema/schema.go

@@ -8,9 +8,11 @@ import (
 	"github.com/stretchr/testify/require"
 	"gorm.io/gorm"
 
+	common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
 	"github.com/caraml-dev/xp/management-service/config"
 	mw "github.com/caraml-dev/xp/management-service/middleware"
 	"github.com/caraml-dev/xp/management-service/services"
+	"github.com/caraml-dev/xp/management-service/services/messagequeue"
 	"github.com/caraml-dev/xp/management-service/services/mocks"
 )
 
@@ -30,9 +32,11 @@ func TestNewAppContext(t *testing.T) {
 		MLPConfig: &config.MLPConfig{
 			URL: "http://mlp.example.com/api/merlin/v1",
 		},
-		PubSubConfig: &config.PubSubConfig{
-			Project:   "test",
-			TopicName: "update",
+		MessageQueueConfig: &common_mq_config.MessageQueueConfig{
+			PubSubConfig: &common_mq_config.PubSubConfig{
+				Project:   "test",
+				TopicName: "update",
+			},
 		},
 		ValidationConfig: config.ValidationConfig{
 			ValidationUrlTimeoutSeconds: 5,
@@ -53,7 +57,7 @@ func TestNewAppContext(t *testing.T) {
 	validationService, err := services.NewValidationService(cfg.ValidationConfig)
 	require.NoError(t, err)
 
-	pubSubPublisherService, _ := services.NewPubSubPublisherService(cfg.PubSubConfig)
+	messageQueueService, _ := messagequeue.NewMessageQueueService(*cfg.MessageQueueConfig)
 
 	expHistSvc := services.NewExperimentHistoryService(db)
 	expSvc := services.NewExperimentService(&allServices, db)
@@ -87,10 +91,10 @@ func TestNewAppContext(t *testing.T) {
 			return segmenterSvc, nil
 		},
 	)
-	// Patch PubSub publisher service
-	monkey.Patch(services.NewPubSubPublisherService,
-		func(pubsubConfig *config.PubSubConfig) (services.PubSubPublisherService, error) {
-			return pubSubPublisherService, nil
+	// Patch MessageQueue service
+	monkey.Patch(messagequeue.NewMessageQueueService,
+		func(messageQueueConfig common_mq_config.MessageQueueConfig) (messagequeue.MessageQueueService, error) {
+			return messageQueueService, nil
 		},
 	)
 	// Patch New MLP Service to validate the input and return the mock service object
@@ -116,7 +120,7 @@ func TestNewAppContext(t *testing.T) {
 		TreatmentService:         treatmentSvc,
 		TreatmentHistoryService:  treatmentHistSvc,
 		ValidationService:        validationService,
-		PubSubPublisherService:   pubSubPublisherService,
+		MessageQueueService:      messageQueueService,
 		ConfigurationService:     configurationSvc,
 	}
 	monkey.Patch(services.NewServices,
@@ -131,7 +135,7 @@ func TestNewAppContext(t *testing.T) {
 			treatmentService services.TreatmentService,
 			treatmentHistoryService services.TreatmentHistoryService,
 			validationService services.ValidationService,
-			publisherService services.PubSubPublisherService,
+			messageQueueService messagequeue.MessageQueueService,
 			configurationService services.ConfigurationService,
 		) services.Services {
 			return allServices

common/messagequeue/config.go

@@ -8,6 +8,7 @@ import (
 	"github.com/gojek/mlp/api/pkg/instrumentation/sentry"
 
 	common_config "github.com/caraml-dev/xp/common/config"
+	common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
 )
 
 type Config struct {
@@ -18,7 +19,7 @@ type Config struct {
 	AuthorizationConfig *AuthorizationConfig
 	DbConfig            *DatabaseConfig
 	MLPConfig           *MLPConfig
-	PubSubConfig        *PubSubConfig
+	MessageQueueConfig  *common_mq_config.MessageQueueConfig
 	SegmenterConfig     map[string]interface{}
 	ValidationConfig    ValidationConfig
 	DeploymentConfig    DeploymentConfig
@@ -53,13 +54,6 @@ type MLPConfig struct {
 	URL string
 }
 
-// PubSubConfig captures the config for the Google PubSub client, to publish messages
-// about changes in the experimentation data
-type PubSubConfig struct {
-	Project   string `default:"dev"`
-	TopicName string `default:"xp-update"`
-}
-
 // ValidationConfig captures the config related to the validation of schemas
 type ValidationConfig struct {
 	ValidationUrlTimeoutSeconds int `default:"5"`

management-service/appcontext/appcontext.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	common_mq_config "github.com/caraml-dev/xp/common/messagequeue"
 	"github.com/gojek/mlp/api/pkg/instrumentation/newrelic"
 	"github.com/gojek/mlp/api/pkg/instrumentation/sentry"
 	"github.com/stretchr/testify/assert"
@@ -38,9 +39,13 @@ func TestDefaultConfigs(t *testing.T) {
 		MLPConfig: &MLPConfig{
 			URL: "",
 		},
-		PubSubConfig: &PubSubConfig{
-			Project:   "dev",
-			TopicName: "xp-update",
+		MessageQueueConfig: &common_mq_config.MessageQueueConfig{
+			Kind: "",
+			PubSubConfig: &common_mq_config.PubSubConfig{
+				Project:              "dev",
+				TopicName:            "xp-update",
+				PubSubTimeoutSeconds: 30,
+			},
 		},
 		ValidationConfig: ValidationConfig{
 			ValidationUrlTimeoutSeconds: 5,
@@ -109,9 +114,13 @@ func TestLoadConfigFiles(t *testing.T) {
 				MLPConfig: &MLPConfig{
 					URL: "test-mlp-url",
 				},
-				PubSubConfig: &PubSubConfig{
-					Project:   "test-pubsub-project",
-					TopicName: "test-pubsub-topic",
+				MessageQueueConfig: &common_mq_config.MessageQueueConfig{
+					Kind: "pubsub",
+					PubSubConfig: &common_mq_config.PubSubConfig{
+						Project:              "test-pubsub-project",
+						TopicName:            "test-pubsub-topic",
+						PubSubTimeoutSeconds: 30,
+					},
 				},
 				ValidationConfig: ValidationConfig{
 					ValidationUrlTimeoutSeconds: 5,

management-service/appcontext/appcontext_test.go

@@ -29,9 +29,11 @@ DbConfig:
 MLPConfig:
   URL: http://localhost:8080/api/v1
 
-PubSubConfig:
-  Project: dev
-  TopicName: xp-update
+MessageQueueConfig:
+  # Kind: pubsub
+  PubSubConfig:
+    Project: dev
+    TopicName: xp-update
 
 NewRelicConfig:
   Enabled: false