Implement canary gating API with in-memory storage

POST /gate/[check|open|close]

cmd/loadtester/main.go

@@ -47,5 +47,7 @@ func main() {
 	go taskRunner.Start(100*time.Millisecond, stopCh)
 
 	logger.Infof("Starting load tester v%s API on port %s", VERSION, port)
-	loadtester.ListenAndServe(port, time.Minute, logger, taskRunner, stopCh)
+
+	gateStorage := loadtester.NewGateStorage("in-memory")
+	loadtester.ListenAndServe(port, time.Minute, logger, taskRunner, gateStorage, stopCh)
 }

pkg/loadtester/gate.go

@@ -0,0 +1,31 @@
+package loadtester
+
+import "sync"
+
+type GateStorage struct {
+	backend string
+	data    *sync.Map
+}
+
+func NewGateStorage(backend string) *GateStorage {
+	return &GateStorage{
+		backend: backend,
+		data:    new(sync.Map),
+	}
+}
+
+func (gs *GateStorage) open(key string) {
+	gs.data.Store(key, true)
+}
+
+func (gs *GateStorage) close(key string) {
+	gs.data.Store(key, false)
+}
+
+func (gs *GateStorage) isOpen(key string) (locked bool) {
+	val, ok := gs.data.LoadOrStore(key, false)
+	if ok {
+		return val.(bool)
+	}
+	return
+}

pkg/loadtester/server.go

@@ -14,7 +14,7 @@ import (
 )
 
 // ListenAndServe starts a web server and waits for SIGTERM
-func ListenAndServe(port string, timeout time.Duration, logger *zap.SugaredLogger, taskRunner *TaskRunner, stopCh <-chan struct{}) {
+func ListenAndServe(port string, timeout time.Duration, logger *zap.SugaredLogger, taskRunner *TaskRunner, gate *GateStorage, stopCh <-chan struct{}) {
 	mux := http.DefaultServeMux
 	mux.Handle("/metrics", promhttp.Handler())
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
@@ -29,6 +29,86 @@ func ListenAndServe(port string, timeout time.Duration, logger *zap.SugaredLogge
 		w.WriteHeader(http.StatusForbidden)
 		w.Write([]byte("Forbidden"))
 	})
+	mux.HandleFunc("/gate/check", func(w http.ResponseWriter, r *http.Request) {
+		body, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			logger.Error("reading the request body failed", zap.Error(err))
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		defer r.Body.Close()
+
+		canary := &flaggerv1.CanaryWebhookPayload{}
+		err = json.Unmarshal(body, canary)
+		if err != nil {
+			logger.Error("decoding the request body failed", zap.Error(err))
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		canaryName := fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)
+		approved := gate.isOpen(canaryName)
+		if approved {
+			w.WriteHeader(http.StatusOK)
+			w.Write([]byte("Approved"))
+		} else {
+			w.WriteHeader(http.StatusForbidden)
+			w.Write([]byte("Forbidden"))
+		}
+
+		logger.Infof("%s gate check: approved %v", canaryName, approved)
+	})
+
+	mux.HandleFunc("/gate/open", func(w http.ResponseWriter, r *http.Request) {
+		body, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			logger.Error("reading the request body failed", zap.Error(err))
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		defer r.Body.Close()
+
+		canary := &flaggerv1.CanaryWebhookPayload{}
+		err = json.Unmarshal(body, canary)
+		if err != nil {
+			logger.Error("decoding the request body failed", zap.Error(err))
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		canaryName := fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)
+		gate.open(canaryName)
+
+		w.WriteHeader(http.StatusAccepted)
+
+		logger.Infof("%s gate opened", canaryName)
+	})
+
+	mux.HandleFunc("/gate/close", func(w http.ResponseWriter, r *http.Request) {
+		body, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			logger.Error("reading the request body failed", zap.Error(err))
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+		defer r.Body.Close()
+
+		canary := &flaggerv1.CanaryWebhookPayload{}
+		err = json.Unmarshal(body, canary)
+		if err != nil {
+			logger.Error("decoding the request body failed", zap.Error(err))
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		canaryName := fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)
+		gate.close(canaryName)
+
+		w.WriteHeader(http.StatusAccepted)
+
+		logger.Infof("%s gate closed", canaryName)
+	})
+
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		body, err := ioutil.ReadAll(r.Body)
 		if err != nil {