diff --git a/Dockerfile.loadtest b/Dockerfile.loadtest new file mode 100644 index 00000000..ab233d3d --- /dev/null +++ b/Dockerfile.loadtest @@ -0,0 +1,44 @@ +FROM golang:1.11 AS hey-builder + +RUN mkdir -p /go/src/github.com/rakyll/hey/ + +WORKDIR /go/src/github.com/rakyll/hey + +ADD https://github.com/rakyll/hey/archive/v0.1.1.tar.gz . + +RUN tar xzf v0.1.1.tar.gz --strip 1 + +RUN go get ./... + +RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \ + go install -ldflags '-w -extldflags "-static"' \ + /go/src/github.com/rakyll/hey + +FROM golang:1.11 AS builder + +RUN mkdir -p /go/src/github.com/stefanprodan/flagger/ + +WORKDIR /go/src/github.com/stefanprodan/flagger + +COPY . . + +RUN go test -race ./pkg/loadtest/ + +RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o loadtest ./cmd/loadtest/* + +FROM alpine:3.8 + +RUN addgroup -S app \ + && adduser -S -g app app \ + && apk --no-cache add ca-certificates curl + +WORKDIR /home/app + +COPY --from=hey-builder /go/bin/hey /usr/local/bin/hey +COPY --from=builder /go/src/github.com/stefanprodan/flagger/loadtest . + +RUN chown -R app:app ./ + +USER app + +ENTRYPOINT ["./loadtest"] \ No newline at end of file diff --git a/cmd/loadtest/main.go b/cmd/loadtest/main.go new file mode 100644 index 00000000..3a453d4d --- /dev/null +++ b/cmd/loadtest/main.go @@ -0,0 +1,41 @@ +package main + +import ( + "flag" + "github.com/knative/pkg/signals" + "github.com/stefanprodan/flagger/pkg/loadtest" + "github.com/stefanprodan/flagger/pkg/logging" + "log" + "time" +) + +var ( + logLevel string + port string + timeout time.Duration +) + +func init() { + flag.StringVar(&logLevel, "log-level", "debug", "Log level can be: debug, info, warning, error.") + flag.StringVar(&port, "port", "9090", "Port to listen on.") + flag.DurationVar(&timeout, "timeout", time.Hour, "Command exec timeout.") +} + +func main() { + flag.Parse() + + logger, err := logging.NewLogger(logLevel) + if err != nil { + log.Fatalf("Error creating logger: %v", err) + } + defer logger.Sync() + + stopCh := signals.SetupSignalHandler() + + taskRunner := loadtest.NewTaskRunner(logger, timeout) + + go taskRunner.Start(100*time.Millisecond, stopCh) + + logger.Infof("Starting HTTP server on port %s", port) + loadtest.ListenAndServe(port, time.Minute, logger, taskRunner, stopCh) +} diff --git a/pkg/loadtest/runner.go b/pkg/loadtest/runner.go new file mode 100644 index 00000000..5f24b0e7 --- /dev/null +++ b/pkg/loadtest/runner.go @@ -0,0 +1,101 @@ +package loadtest + +import ( + "context" + "encoding/hex" + "go.uber.org/zap" + "hash/fnv" + "os/exec" + "sync" + "sync/atomic" + "time" +) + +type TaskRunner struct { + logger *zap.SugaredLogger + timeout time.Duration + todoTasks *sync.Map + runningTasks *sync.Map + totalExecs uint64 +} + +type Task struct { + Canary string + Command string +} + +func (t Task) Hash() string { + fnvHash := fnv.New32() + fnvBytes := fnvHash.Sum([]byte(t.Canary + t.Command)) + return hex.EncodeToString(fnvBytes[:]) +} + +func NewTaskRunner(logger *zap.SugaredLogger, timeout time.Duration) *TaskRunner { + return &TaskRunner{ + logger: logger, + todoTasks: new(sync.Map), + runningTasks: new(sync.Map), + timeout: timeout, + } +} + +func (tr *TaskRunner) Add(task Task) { + tr.todoTasks.Store(task.Hash(), task) +} + +func (tr *TaskRunner) GetTotalExecs() uint64 { + return atomic.LoadUint64(&tr.totalExecs) +} + +func (tr *TaskRunner) runAll() { + tr.todoTasks.Range(func(key interface{}, value interface{}) bool { + task := value.(Task) + go func(t Task) { + // remove task from the to do list + tr.todoTasks.Delete(t.Hash()) + + // check if task is already running, if not run the task's command + if _, exists := tr.runningTasks.Load(t.Hash()); !exists { + // save the task in the running list + tr.runningTasks.Store(t.Hash(), t) + + // create timeout context + ctx, cancel := context.WithTimeout(context.Background(), tr.timeout) + defer cancel() + + // increment the total exec counter + atomic.AddUint64(&tr.totalExecs, 1) + + tr.logger.With("canary", t.Canary).Infof("command starting %s", t.Command) + cmd := exec.CommandContext(ctx, "sh", "-c", t.Command) + + // execute task + out, err := cmd.CombinedOutput() + if err != nil { + tr.logger.With("canary", t.Canary).Errorf("command failed %s %v %s", t.Command, err, out) + } else { + tr.logger.With("canary", t.Canary).Infof("command finished %s", t.Command) + } + + // remove task from the running list + tr.runningTasks.Delete(t.Hash()) + } else { + tr.logger.With("canary", t.Canary).Infof("command skipped %s is already running", t.Command) + } + }(task) + return true + }) +} + +func (tr *TaskRunner) Start(interval time.Duration, stopCh <-chan struct{}) { + tickChan := time.NewTicker(interval).C + for { + select { + case <-tickChan: + tr.runAll() + case <-stopCh: + tr.logger.Info("shutting down the task runner") + return + } + } +} diff --git a/pkg/loadtest/runner_test.go b/pkg/loadtest/runner_test.go new file mode 100644 index 00000000..8194a9b0 --- /dev/null +++ b/pkg/loadtest/runner_test.go @@ -0,0 +1,52 @@ +package loadtest + +import ( + "github.com/stefanprodan/flagger/pkg/logging" + "testing" + "time" +) + +func TestTaskRunner_Start(t *testing.T) { + stop := make(chan struct{}) + logger, _ := logging.NewLogger("debug") + tr := NewTaskRunner(logger, time.Hour) + + go tr.Start(10*time.Millisecond, stop) + + task1 := Task{ + Canary: "podinfo.default", + Command: "sleep 0.6", + } + task2 := Task{ + Canary: "podinfo.default", + Command: "sleep 0.7", + } + + tr.Add(task1) + tr.Add(task2) + + time.Sleep(100 * time.Millisecond) + + tr.Add(task1) + tr.Add(task2) + + time.Sleep(100 * time.Millisecond) + + tr.Add(task1) + tr.Add(task2) + + if tr.GetTotalExecs() != 2 { + t.Errorf("Got total executed commands %v wanted %v", tr.GetTotalExecs(), 2) + } + + time.Sleep(time.Second) + + tr.Add(task1) + tr.Add(task2) + + time.Sleep(time.Second) + + if tr.GetTotalExecs() != 4 { + t.Errorf("Got total executed commands %v wanted %v", tr.GetTotalExecs(), 4) + } +} diff --git a/pkg/loadtest/server.go b/pkg/loadtest/server.go new file mode 100644 index 00000000..b9801276 --- /dev/null +++ b/pkg/loadtest/server.go @@ -0,0 +1,85 @@ +package loadtest + +import ( + "context" + "encoding/json" + "fmt" + "io/ioutil" + "net/http" + "time" + + "github.com/prometheus/client_golang/prometheus/promhttp" + flaggerv1 "github.com/stefanprodan/flagger/pkg/apis/flagger/v1alpha3" + "go.uber.org/zap" +) + +// ListenAndServe starts a web server and waits for SIGTERM +func ListenAndServe(port string, timeout time.Duration, logger *zap.SugaredLogger, taskRunner *TaskRunner, stopCh <-chan struct{}) { + mux := http.DefaultServeMux + mux.Handle("/metrics", promhttp.Handler()) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte("OK")) + }) + mux.HandleFunc("/", 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() + + payload := &flaggerv1.CanaryWebhookPayload{} + err = json.Unmarshal(body, payload) + if err != nil { + logger.Error("decoding the request body failed", zap.Error(err)) + w.WriteHeader(http.StatusBadRequest) + return + } + + if len(payload.Metadata) > 0 { + if cmd, ok := payload.Metadata["cmd"]; ok { + taskRunner.Add(Task{ + Canary: fmt.Sprintf("%s.%s", payload.Name, payload.Namespace), + Command: cmd, + }) + } else { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte("cmd not found in metadata")) + return + } + } else { + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte("metadata not found in payload")) + return + } + + w.WriteHeader(http.StatusAccepted) + }) + srv := &http.Server{ + Addr: ":" + port, + Handler: mux, + ReadTimeout: 5 * time.Second, + WriteTimeout: 1 * time.Minute, + IdleTimeout: 15 * time.Second, + } + + // run server in background + go func() { + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + logger.Fatalf("HTTP server crashed %v", err) + } + }() + + // wait for SIGTERM or SIGINT + <-stopCh + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + + if err := srv.Shutdown(ctx); err != nil { + logger.Errorf("HTTP server graceful shutdown failed %v", err) + } else { + logger.Info("HTTP server stopped") + } +}