Add endpoint circuit breaker

api/api.go

@@ -333,6 +333,7 @@ func (a *ApplicationHandler) BuildControlPlaneRoutes() *chi.Mux {
 								e.With(handler.RequireEnabledProject()).Delete("/", handler.DeleteEndpoint)
 								e.With(handler.RequireEnabledProject()).Put("/expire_secret", handler.ExpireSecret)
 								e.With(handler.RequireEnabledProject()).Put("/pause", handler.PauseEndpoint)
+								e.With(handler.RequireEnabledProject()).Post("/activate", handler.ActivateEndpoint)
 							})
 						})
 

api/handlers/configuration.go

@@ -17,24 +17,24 @@ import (
 )
 
 func (h *Handler) GetConfiguration(w http.ResponseWriter, r *http.Request) {
-	config, err := postgres.NewConfigRepo(h.A.DB).LoadConfiguration(r.Context())
+	configuration, err := postgres.NewConfigRepo(h.A.DB).LoadConfiguration(r.Context())
 	if err != nil && !errors.Is(err, datastore.ErrConfigNotFound) {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
 	}
 
-	configResponse := []*models.ConfigurationResponse{}
-	if config != nil {
-		if config.StoragePolicy.Type == datastore.S3 {
+	var configResponse []*models.ConfigurationResponse
+	if configuration != nil {
+		if configuration.StoragePolicy.Type == datastore.S3 {
 			policy := &datastore.S3Storage{}
-			policy.Bucket = config.StoragePolicy.S3.Bucket
-			policy.Endpoint = config.StoragePolicy.S3.Endpoint
-			policy.Region = config.StoragePolicy.S3.Region
-			config.StoragePolicy.S3 = policy
+			policy.Bucket = configuration.StoragePolicy.S3.Bucket
+			policy.Endpoint = configuration.StoragePolicy.S3.Endpoint
+			policy.Region = configuration.StoragePolicy.S3.Region
+			configuration.StoragePolicy.S3 = policy
 		}
 
 		c := &models.ConfigurationResponse{
-			Configuration: config,
+			Configuration: configuration,
 			ApiVersion:    convoy.GetVersion(),
 		}
 
@@ -61,14 +61,14 @@ func (h *Handler) CreateConfiguration(w http.ResponseWriter, r *http.Request) {
 		NewConfig:  &newConfig,
 	}
 
-	config, err := cc.Run(r.Context())
+	configuration, err := cc.Run(r.Context())
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
 	}
 
 	c := &models.ConfigurationResponse{
-		Configuration: config,
+		Configuration: configuration,
 		ApiVersion:    convoy.GetVersion(),
 	}
 
@@ -92,14 +92,14 @@ func (h *Handler) UpdateConfiguration(w http.ResponseWriter, r *http.Request) {
 		Config:     &newConfig,
 	}
 
-	config, err := uc.Run(r.Context())
+	configuration, err := uc.Run(r.Context())
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
 	}
 
 	c := &models.ConfigurationResponse{
-		Configuration: config,
+		Configuration: configuration,
 		ApiVersion:    convoy.GetVersion(),
 	}
 

api/handlers/endpoint.go

@@ -3,6 +3,9 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"github.com/frain-dev/convoy/pkg/circuit_breaker"
+	"github.com/frain-dev/convoy/pkg/msgpack"
 	"net/http"
 
 	"github.com/frain-dev/convoy/api/models"
@@ -207,9 +210,37 @@ func (h *Handler) GetEndpoints(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// fetch keys from redis and mutate endpoints slice
+	keys := make([]string, len(endpoints))
+	for i := 0; i < len(endpoints); i++ {
+		keys[i] = fmt.Sprintf("breaker:%s", endpoints[i].UID)
+	}
+
+	cbs, err := h.A.Redis.MGet(r.Context(), keys...).Result()
+	if err != nil {
+		_ = render.Render(w, r, util.NewServiceErrResponse(err))
+		return
+	}
+
+	for i := 0; i < len(cbs); i++ {
+		if cbs[i] != nil {
+			str, ok := cbs[i].(string)
+			if ok {
+				var c circuit_breaker.CircuitBreaker
+				asBytes := []byte(str)
+				innerErr := msgpack.DecodeMsgPack(asBytes, &c)
+				if innerErr != nil {
+					continue
+				}
+				endpoints[i].FailureRate = c.FailureRate
+			}
+		}
+	}
+
 	resp := models.NewListResponse(endpoints, func(endpoint datastore.Endpoint) models.EndpointResponse {
 		return models.EndpointResponse{Endpoint: &endpoint}
 	})
+
 	serverResponse := util.NewServerResponse(
 		"Endpoints fetched successfully",
 		models.PagedResponse{Content: &resp, Pagination: &paginationData}, http.StatusOK)
@@ -410,7 +441,7 @@ func (h *Handler) ExpireSecret(w http.ResponseWriter, r *http.Request) {
 // PauseEndpoint
 //
 //	@Summary		Pause endpoint
-//	@Description	This endpoint toggles an endpoint status between the active and paused states
+//	@Description	Toggles an endpoint's status between active and paused states
 //	@Id				PauseEndpoint
 //	@Tags			Endpoints
 //	@Accept			json
@@ -458,6 +489,57 @@ func (h *Handler) PauseEndpoint(w http.ResponseWriter, r *http.Request) {
 	util.WriteResponse(w, r, resBytes, http.StatusAccepted)
 }
 
+// ActivateEndpoint
+//
+//	@Summary		Activate endpoint
+//	@Description	Activated an inactive endpoint
+//	@Id				PauseEndpoint
+//	@Tags			Endpoints
+//	@Accept			json
+//	@Produce		json
+//	@Param			projectID	path		string	true	"Project ID"
+//	@Param			endpointID	path		string	true	"Endpoint ID"
+//	@Success		202			{object}	util.ServerResponse{data=models.EndpointResponse}
+//	@Failure		400,401,404	{object}	util.ServerResponse{data=Stub}
+//	@Security		ApiKeyAuth
+//	@Router			/v1/projects/{projectID}/endpoints/{endpointID}/activate [post]
+func (h *Handler) ActivateEndpoint(w http.ResponseWriter, r *http.Request) {
+	project, err := h.retrieveProject(r)
+	if err != nil {
+		_ = render.Render(w, r, util.NewErrorResponse(err.Error(), http.StatusBadRequest))
+		return
+	}
+
+	aes := services.ActivateEndpointService{
+		EndpointRepo: postgres.NewEndpointRepo(h.A.DB, h.A.Cache),
+		ProjectID:    project.UID,
+		EndpointId:   chi.URLParam(r, "endpointID"),
+	}
+
+	endpoint, err := aes.Run(r.Context())
+	if err != nil {
+		_ = render.Render(w, r, util.NewServiceErrResponse(err))
+		return
+	}
+
+	resp := &models.EndpointResponse{Endpoint: endpoint}
+	serverResponse := util.NewServerResponse("endpoint status successfully activated", resp, http.StatusAccepted)
+
+	rb, err := json.Marshal(serverResponse)
+	if err != nil {
+		_ = render.Render(w, r, util.NewServiceErrResponse(err))
+		return
+	}
+
+	resBytes, err := h.RM.VersionResponse(r, rb, "UpdateEndpoint")
+	if err != nil {
+		_ = render.Render(w, r, util.NewServiceErrResponse(err))
+		return
+	}
+
+	util.WriteResponse(w, r, resBytes, http.StatusAccepted)
+}
+
 func (h *Handler) retrieveEndpoint(ctx context.Context, endpointID, projectID string) (*datastore.Endpoint, error) {
 	endpointRepo := postgres.NewEndpointRepo(h.A.DB, h.A.Cache)
 	return endpointRepo.FindEndpointByID(ctx, endpointID, projectID)

api/models/models.go

@@ -154,7 +154,7 @@ type PortalLinkResponse struct {
 	DeletedAt         null.Time                  `json:"deleted_at,omitempty"`
 }
 
-// NewListResponse is generic function for looping over
+// NewListResponse is a generic function for looping over
 // a slice of type M and returning a slice of type T
 func NewListResponse[T, M any](items []M, fn func(item M) T) []T {
 	results := make([]T, 0)

api/server_suite_test.go

@@ -128,10 +128,13 @@ func buildServer() *ApplicationHandler {
 	noopCache := ncache.NewNoopCache()
 	r, _ := rlimiter.NewRedisLimiter(cfg.Redis.BuildDsn())
 
+	rd, _ := rdb.NewClient(cfg.Redis.BuildDsn())
+
 	ah, _ := NewApplicationHandler(
 		&types.APIOptions{
 			DB:       db,
 			Queue:    newQueue,
+			Redis:    rd.Client(),
 			Logger:   logger,
 			Cache:    noopCache,
 			Rate:     r,

api/testdb/seed.go

@@ -557,21 +557,23 @@ func SeedUser(db database.Database, email, password string) (*datastore.User, er
 }
 
 func SeedConfiguration(db database.Database) (*datastore.Configuration, error) {
-	config := &datastore.Configuration{
-		UID:                ulid.Make().String(),
-		IsAnalyticsEnabled: true,
-		IsSignupEnabled:    true,
-		StoragePolicy:      &datastore.DefaultStoragePolicy,
+	c := &datastore.Configuration{
+		UID:                  ulid.Make().String(),
+		IsAnalyticsEnabled:   true,
+		IsSignupEnabled:      true,
+		StoragePolicy:        &datastore.DefaultStoragePolicy,
+		RetentionPolicy:      &datastore.DefaultRetentionPolicy,
+		CircuitBreakerConfig: &datastore.DefaultCircuitBreakerConfiguration,
 	}
 
 	// Seed Data
 	configRepo := postgres.NewConfigRepo(db)
-	err := configRepo.CreateConfiguration(context.TODO(), config)
+	err := configRepo.CreateConfiguration(context.TODO(), c)
 	if err != nil {
 		return nil, err
 	}
 
-	return config, nil
+	return c, nil
 }
 
 func SeedDevice(db database.Database, g *datastore.Project, endpointID string) error {

api/types/types.go

@@ -9,13 +9,15 @@ import (
 	"github.com/frain-dev/convoy/internal/pkg/limiter"
 	"github.com/frain-dev/convoy/pkg/log"
 	"github.com/frain-dev/convoy/queue"
+	"github.com/redis/go-redis/v9"
 )
 
 type ContextKey string
 
 type APIOptions struct {
 	FFlag    *fflag.FFlag
 	DB       database.Database
+	Redis    redis.UniversalClient
 	Queue    queue.Queuer
 	Logger   log.StdLogger
 	Cache    cache.Cache

cmd/agent/agent.go

@@ -2,7 +2,6 @@ package agent
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"os/signal"
 	"time"
@@ -114,38 +113,35 @@ func AddAgentCommand(a *cli.App) *cobra.Command {
 	cmd.Flags().Uint32Var(&workerPort, "worker-port", 0, "Worker port")
 	cmd.Flags().Uint32Var(&ingestPort, "ingest-port", 0, "Ingest port")
 
-	cmd.Flags().StringVar(&logLevel, "log-level", "", "scheduler log level")
+	cmd.Flags().StringVar(&logLevel, "log-level", "", "Log level")
 	cmd.Flags().IntVar(&consumerPoolSize, "consumers", -1, "Size of the consumers pool.")
 	cmd.Flags().IntVar(&interval, "interval", 10, "the time interval, measured in seconds to update the in-memory store from the database")
 	cmd.Flags().StringVar(&executionMode, "mode", "", "Execution Mode (one of events, retry and default)")
 
 	return cmd
 }
 
-func startServerComponent(ctx context.Context, a *cli.App) error {
+func startServerComponent(_ context.Context, a *cli.App) error {
+	lo := a.Logger.(*log.Logger)
+	lo.SetPrefix("agent")
+
 	cfg, err := config.Get()
 	if err != nil {
-		a.Logger.WithError(err).Fatal("Failed to load configuration")
+		lo.WithError(err).Fatal("Failed to load configuration")
 	}
 
 	start := time.Now()
-	a.Logger.Info("Starting Convoy data plane ...")
+	lo.Info("Starting Convoy data plane")
 
 	apiKeyRepo := postgres.NewAPIKeyRepo(a.DB, a.Cache)
 	userRepo := postgres.NewUserRepo(a.DB, a.Cache)
 	portalLinkRepo := postgres.NewPortalLinkRepo(a.DB, a.Cache)
 	err = realm_chain.Init(&cfg.Auth, apiKeyRepo, userRepo, portalLinkRepo, a.Cache)
 	if err != nil {
-		a.Logger.WithError(err).Fatal("failed to initialize realm chain")
-	}
-
-	flag, err := fflag.NewFFlag(&cfg)
-	if err != nil {
-		a.Logger.WithError(err).Fatal("failed to create fflag controller")
+		lo.WithError(err).Fatal("failed to initialize realm chain")
 	}
 
-	lo := a.Logger.(*log.Logger)
-	lo.SetPrefix("api server")
+	flag := fflag.NewFFlag(&cfg)
 
 	lvl, err := log.ParseLevel(cfg.Logger.Level)
 	if err != nil {
@@ -164,6 +160,7 @@ func startServerComponent(ctx context.Context, a *cli.App) error {
 			Logger:   lo,
 			Cache:    a.Cache,
 			Rate:     a.Rate,
+			Redis:    a.Redis,
 			Licenser: a.Licenser,
 		})
 	if err != nil {
@@ -172,7 +169,7 @@ func startServerComponent(ctx context.Context, a *cli.App) error {
 
 	srv.SetHandler(evHandler.BuildDataPlaneRoutes())
 
-	fmt.Printf("Started convoy server in %s\n", time.Since(start))
+	lo.Infof("Started convoy server in %s", time.Since(start))
 
 	httpConfig := cfg.Server.HTTP
 	if httpConfig.SSL {
@@ -181,7 +178,7 @@ func startServerComponent(ctx context.Context, a *cli.App) error {
 		return nil
 	}
 
-	fmt.Printf("Starting Convoy Agent on port %v\n", cfg.Server.HTTP.AgentPort)
+	lo.Infof("Starting Convoy Agent on port %v", cfg.Server.HTTP.AgentPort)
 
 	go func() {
 		srv.Listen()

cmd/ff/feature_flags.go

@@ -20,10 +20,8 @@ func AddFeatureFlagsCommand() *cobra.Command {
 			if err != nil {
 				log.WithError(err).Fatal("Error fetching the config.")
 			}
-			f, err := fflag2.NewFFlag(&cfg)
-			if err != nil {
-				return err
-			}
+
+			f := fflag2.NewFFlag(&cfg)
 			return f.ListFeatures()
 		},
 		PersistentPostRun: func(cmd *cobra.Command, args []string) {},