feat: add router and ensembler webhook (#395)

* feat: add router and ensembler webhook

* validation for invalid event type

* add id and move inside webhook goroutine

* add webhook ondeployed in create & update router endpoint

* change event type format

* fix dependency

* modify event webhook for deploy and undeploy router

* go mod tidy

* fix linter

* Refactor away specific webhook functions

* Fix incorrect expected unit test result

* Fix mocked methods on ensembler webhook client

* Remove redundant struct definitions

* Refactor positioning of webhook calls

* Remove mocked webhook client from deployment controller test

* Test removing unit test case

* Reorder webhook call when updating routers

* Update go mod file

* Refactor webhook request body

* Fix lint comment by renaming struct name

---------

Co-authored-by: ewezy <[email protected]>

.gitignore

@@ -1,3 +1,4 @@
 .DS_Store
 config-dev-w-creds.yaml
 environments-dev-w-creds.yaml
+.idea/

api/go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/GoogleCloudPlatform/spark-on-k8s-operator v0.0.0-20221025152940-c261df66a006
 	github.com/antihax/optional v1.0.0
 	github.com/caraml-dev/merlin v0.0.0
-	github.com/caraml-dev/mlp v1.12.0
+	github.com/caraml-dev/mlp v1.13.2
 	github.com/caraml-dev/turing/engines/experiment v0.0.0
 	github.com/caraml-dev/turing/engines/router v0.0.0
 	github.com/caraml-dev/universal-prediction-interface v0.3.6
@@ -66,6 +66,7 @@ require (
 	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/ajg/form v1.5.1 // indirect
 	github.com/andybalholm/brotli v1.0.5 // indirect
+	github.com/avast/retry-go/v4 v4.6.0 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.30.6-0.20240906182417-827d25db0048 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.4 // indirect
 	github.com/aws/aws-sdk-go-v2/config v1.17.8 // indirect
@@ -267,7 +268,6 @@ replace (
 
 	github.com/caraml-dev/merlin => github.com/caraml-dev/merlin/api v0.0.0-20240313065547-6778bd14c119
 	github.com/caraml-dev/merlin-pyspark-app => github.com/caraml-dev/merlin/python/batch-predictor v0.0.0-20240313065547-6778bd14c119
-	github.com/caraml-dev/mlp => github.com/deadlycoconuts/mlp v0.0.0-20240917090435-d94d92572eac
 
 	github.com/caraml-dev/turing/engines/experiment => ../engines/experiment
 	github.com/caraml-dev/turing/engines/router => ../engines/router

api/go.sum

@@ -72,6 +72,8 @@ github.com/antihax/optional v1.0.0 h1:xK2lYat7ZLaVVcIuj82J8kIro4V6kDe0AUDFboUCwc
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
+github.com/avast/retry-go/v4 v4.6.0 h1:K9xNA+KeB8HHc2aWFuLb25Offp+0iVRXEvFx8IinRJA=
+github.com/avast/retry-go/v4 v4.6.0/go.mod h1:gvWlPhBVsvBbLkVGDg/KwvBv0bEkCOLRRSHKIr2PyOE=
 github.com/aws/aws-sdk-go v1.17.7/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go-v2 v1.16.16/go.mod h1:SwiyXi/1zTUZ6KIAmLK5V5ll8SiURNUYOqTerZPaF9k=
 github.com/aws/aws-sdk-go-v2 v1.30.6-0.20240906182417-827d25db0048 h1:wXvkIvYQ3EPVO5MhCoEv2u5LDwfWp+kLTQMIGyyvi/0=
@@ -125,6 +127,8 @@ github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMU
 github.com/buger/jsonparser v1.1.1/go.mod h1:6RYKKt7H4d4+iWqouImQ9R2FZql3VbhNgx27UK13J/0=
 github.com/caraml-dev/merlin/api v0.0.0-20240313065547-6778bd14c119 h1:8zt/6B7ySOokLa7WDpjaybBOL8x5C35nbMIrKFMsZcg=
 github.com/caraml-dev/merlin/api v0.0.0-20240313065547-6778bd14c119/go.mod h1:b2MbvZaVUSJQaoF0AuEgwTVVYK8lPHCIvnYpfvnjkQY=
+github.com/caraml-dev/mlp v1.13.2 h1:N3lk+ToQ281duZImQLTQ28uJtmoc9Zkxx1CR94rS15U=
+github.com/caraml-dev/mlp v1.13.2/go.mod h1:9kPooDSYsVu5q/z2K4T9uu08RGyiFNbCAFnQVBMJxOk=
 github.com/caraml-dev/universal-prediction-interface v0.3.6 h1:G/D4aukfjLECl8armJqFy/R2+0u/f4AiurSFqAo33uQ=
 github.com/caraml-dev/universal-prediction-interface v0.3.6/go.mod h1:e0qmFOXQxx8HFg5ObYyQO3WVnrqsr5v5JApFmeF7eJo=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -170,8 +174,6 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/deadlycoconuts/mlp v0.0.0-20240917090435-d94d92572eac h1:M6dR1d3O+xY4suQPumkbNO1Ie70M6WoYFaoW+cjKT+8=
-github.com/deadlycoconuts/mlp v0.0.0-20240917090435-d94d92572eac/go.mod h1:9kPooDSYsVu5q/z2K4T9uu08RGyiFNbCAFnQVBMJxOk=
 github.com/denisenkom/go-mssqldb v0.0.0-20190515213511-eb9f6a1743f3/go.mod h1:zAg7JM8CkOJ43xKXIj7eRO9kmWm/TW578qo+oDO6tuM=
 github.com/dgraph-io/ristretto v0.0.1 h1:cJwdnj42uV8Jg4+KLrYovLiCgIfz9wtWm6E6KA+1tLs=
 github.com/dgraph-io/ristretto v0.0.1/go.mod h1:T40EBc7CJke8TkpiYfGGKAeFjSaxuFXhuXRyumBd6RE=

api/turing/api/base_controller.go

@@ -5,6 +5,8 @@ import (
 	val "github.com/go-playground/validator/v10"
 	"github.com/gorilla/schema"
 
+	"github.com/caraml-dev/turing/api/turing/webhook"
+
 	"github.com/caraml-dev/turing/api/turing/models"
 )
 
@@ -16,19 +18,21 @@ type Controller interface {
 // BaseController implements common methods that may be shared by all API controllers
 type BaseController struct {
 	*AppContext
-	decoder   *schema.Decoder
-	validator *val.Validate
+	decoder       *schema.Decoder
+	validator     *val.Validate
+	webhookClient webhook.Client
 }
 
 // NewBaseController returns a new instance of BaseController
-func NewBaseController(ctx *AppContext, validator *val.Validate) BaseController {
+func NewBaseController(ctx *AppContext, validator *val.Validate, webhookClient webhook.Client) BaseController {
 	decoder := schema.NewDecoder()
 	decoder.IgnoreUnknownKeys(true)
 
 	return BaseController{
-		AppContext: ctx,
-		decoder:    decoder,
-		validator:  validator,
+		AppContext:    ctx,
+		decoder:       decoder,
+		validator:     validator,
+		webhookClient: webhookClient,
 	}
 }
 

api/turing/api/deployment_controller.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -25,6 +26,8 @@ func (c RouterDeploymentController) deployOrRollbackRouter(
 	router *models.Router,
 	routerVersion *models.RouterVersion,
 ) error {
+	ctx := context.Background()
+
 	// Get the router environment
 	environment, err := c.MLPService.GetEnvironment(router.EnvironmentName)
 	if err != nil {
@@ -48,7 +51,7 @@ func (c RouterDeploymentController) deployOrRollbackRouter(
 		"starting deployment for router %s version %d", router.Name, routerVersion.Version))
 
 	// Deploy the given router version
-	endpoint, err := c.deployRouterVersion(project, environment, routerVersion, eventsCh)
+	endpoint, err := c.deployRouterVersion(ctx, project, environment, routerVersion, eventsCh)
 
 	// Start accumulating non-critical errors
 	errorStrings := make([]string, 0)
@@ -93,6 +96,7 @@ func (c RouterDeploymentController) deployOrRollbackRouter(
 		}
 
 		err = errors.New(strings.Join(errorStrings, ". "))
+
 		return err
 	}
 
@@ -149,6 +153,7 @@ func (c RouterDeploymentController) writeDeploymentEvents(
 // (current version reference, status, endpoint, etc.) are not in the scope of this method.
 // This method returns the new router endpoint (if successful) and any error.
 func (c RouterDeploymentController) deployRouterVersion(
+	_ context.Context,
 	project *mlp.Project,
 	environment *merlin.Environment,
 	routerVersion *models.RouterVersion,
@@ -273,6 +278,7 @@ func (c RouterDeploymentController) deployRouterVersion(
 	// Deploy succeeded - update version's status to deployed and return endpoint
 	routerVersion.Status = models.RouterVersionStatusDeployed
 	_, err = c.RouterVersionsService.Save(routerVersion)
+
 	return endpoint, err
 }
 

api/turing/api/deployment_controller_test.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"errors"
@@ -21,6 +22,8 @@ import (
 )
 
 func TestDeployVersionSuccess(t *testing.T) {
+	ctx := context.TODO()
+
 	testEnv := "test-env"
 	environment := &merlin.Environment{Name: testEnv}
 	project := &mlp.Project{ID: 1, Name: "test-project"}
@@ -191,7 +194,7 @@ func TestDeployVersionSuccess(t *testing.T) {
 
 			// Run deploy and test that the router version's status is deployed and the endpoint
 			// returned by deploy version is expected.
-			endpoint, err := ctrl.deployRouterVersion(project, environment, data.routerVersion, eventsCh)
+			endpoint, err := ctrl.deployRouterVersion(ctx, project, environment, data.routerVersion, eventsCh)
 			assert.NoError(t, err)
 			assert.Equal(t, models.RouterVersionStatusDeployed, data.routerVersion.Status)
 			assert.Equal(t, "test-url", endpoint)

api/turing/api/ensembler_images_api_test.go

@@ -7,13 +7,15 @@ import (
 	"testing"
 
 	"github.com/caraml-dev/mlp/api/client"
+	"github.com/stretchr/testify/mock"
+
 	"github.com/caraml-dev/turing/api/turing/api/request"
 	"github.com/caraml-dev/turing/api/turing/imagebuilder"
 	"github.com/caraml-dev/turing/api/turing/models"
 	"github.com/caraml-dev/turing/api/turing/service"
 	"github.com/caraml-dev/turing/api/turing/service/mocks"
 	"github.com/caraml-dev/turing/api/turing/validation"
-	"github.com/stretchr/testify/mock"
+	webhookMock "github.com/caraml-dev/turing/api/turing/webhook/mocks"
 )
 
 func TestEnsemblerImagesController_ListImages(t *testing.T) {
@@ -477,6 +479,7 @@ func TestEnsemblerImagesController_ListImages(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			validator, _ := validation.NewValidator(nil)
+			mockWebhookClient := webhookMock.NewClient(t)
 
 			c := EnsemblerImagesController{
 				BaseController: NewBaseController(
@@ -486,6 +489,7 @@ func TestEnsemblerImagesController_ListImages(t *testing.T) {
 						EnsemblerImagesService: tt.ensemblerImageService(),
 					},
 					validator,
+					mockWebhookClient,
 				),
 			}
 			if got := c.ListImages(tt.args.in0, tt.args.vars, tt.args.in2); !reflect.DeepEqual(got, tt.want) {
@@ -818,6 +822,7 @@ func TestEnsemblerImagesController_BuildImage(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			validator, _ := validation.NewValidator(nil)
+			mockWebhookClient := webhookMock.NewClient(t)
 
 			c := EnsemblerImagesController{
 				BaseController: NewBaseController(
@@ -827,6 +832,7 @@ func TestEnsemblerImagesController_BuildImage(t *testing.T) {
 						EnsemblerImagesService: tt.ensemblerImageService(),
 					},
 					validator,
+					mockWebhookClient,
 				),
 			}
 			if got := c.BuildImage(tt.args.in0, tt.args.vars, tt.args.body); !reflect.DeepEqual(got, tt.want) {

api/turing/api/ensemblers_api.go

@@ -9,8 +9,10 @@ import (
 	mlp "github.com/caraml-dev/mlp/api/client"
 
 	"github.com/caraml-dev/turing/api/turing/api/request"
+	"github.com/caraml-dev/turing/api/turing/log"
 	"github.com/caraml-dev/turing/api/turing/models"
 	"github.com/caraml-dev/turing/api/turing/service"
+	"github.com/caraml-dev/turing/api/turing/webhook"
 )
 
 type EnsemblersController struct {
@@ -62,12 +64,16 @@ func (c EnsemblersController) GetEnsembler(
 }
 
 func (c EnsemblersController) CreateEnsembler(
-	_ *http.Request,
+	req *http.Request,
 	vars RequestVars,
 	body interface{},
 ) *Response {
-	var errResp *Response
-	var project *mlp.Project
+	var (
+		ctx     = req.Context()
+		errResp *Response
+		project *mlp.Project
+	)
+
 	if project, errResp = c.getProjectFromRequestVars(vars); errResp != nil {
 		return errResp
 	}
@@ -80,14 +86,23 @@ func (c EnsemblersController) CreateEnsembler(
 		return InternalServerError("unable to save the ensembler", err.Error())
 	}
 
+	// call webhook for ensembler creation event
+	if errWebhook := c.webhookClient.TriggerWebhooks(ctx, webhook.OnEnsemblerCreated, ensembler); errWebhook != nil {
+		log.Warnf(
+			"Error triggering webhook for event %s, ensembler id: %d, %v",
+			webhook.OnEnsemblerCreated, ensembler.GetID(), errWebhook,
+		)
+	}
+
 	return Created(ensembler)
 }
 
 func (c EnsemblersController) UpdateEnsembler(
-	_ *http.Request,
+	req *http.Request,
 	vars RequestVars,
 	body interface{},
 ) *Response {
+	ctx := req.Context()
 	options := EnsemblersPathOptions{}
 
 	if err := c.ParseVars(&options, vars); err != nil {
@@ -149,14 +164,23 @@ func (c EnsemblersController) UpdateEnsembler(
 		}
 	}
 
+	// call webhook for ensembler update event
+	if errWebhook := c.webhookClient.TriggerWebhooks(ctx, webhook.OnEnsemblerUpdated, ensembler); errWebhook != nil {
+		log.Warnf(
+			"Error triggering webhook for event %s, ensembler id: %d, %v",
+			webhook.OnEnsemblerUpdated, ensembler.GetID(), errWebhook,
+		)
+	}
+
 	return Ok(ensembler)
 }
 
 func (c EnsemblersController) DeleteEnsembler(
-	_ *http.Request,
+	req *http.Request,
 	vars RequestVars,
 	_ interface{},
 ) *Response {
+	ctx := req.Context()
 	options := EnsemblersPathOptions{}
 
 	if err := c.ParseVars(&options, vars); err != nil {
@@ -213,6 +237,14 @@ func (c EnsemblersController) DeleteEnsembler(
 		return Error(httpStatus, "failed to delete the ensembler", err.Error())
 	}
 
+	// call webhook for ensembler deletion event
+	if errWebhook := c.webhookClient.TriggerWebhooks(ctx, webhook.OnEnsemblerDeleted, ensembler); errWebhook != nil {
+		log.Warnf(
+			"Error triggering webhook for event %s, ensembler id: %d, %v",
+			webhook.OnEnsemblerDeleted, ensembler.GetID(), errWebhook,
+		)
+	}
+
 	return Ok(map[string]int{"id": int(ensembler.GetID())})
 }