From f204fe53f464b612c1cbe14ce1759265e527c209 Mon Sep 17 00:00:00 2001
From: stefanprodan <stefan.prodan@gmail.com>
Date: Wed, 24 Jul 2019 16:14:22 +0300
Subject: [PATCH] Implement canary gating API with in-memory storage POST
 /gate/[check|open|close]

---
 cmd/loadtester/main.go   |  4 +-
 pkg/loadtester/gate.go   | 31 +++++++++++++++
 pkg/loadtester/server.go | 82 +++++++++++++++++++++++++++++++++++++++-
 3 files changed, 115 insertions(+), 2 deletions(-)
 create mode 100644 pkg/loadtester/gate.go

diff --git a/cmd/loadtester/main.go b/cmd/loadtester/main.go
index 4e29ffa4..74c4d09b 100644
--- a/cmd/loadtester/main.go
+++ b/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)
 }
diff --git a/pkg/loadtester/gate.go b/pkg/loadtester/gate.go
new file mode 100644
index 00000000..fdacdd7e
--- /dev/null
+++ b/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
+}
diff --git a/pkg/loadtester/server.go b/pkg/loadtester/server.go
index ca907756..3ddc7903 100644
--- a/pkg/loadtester/server.go
+++ b/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 {