mirror of
https://github.com/fluxcd/flagger.git
synced 2026-04-15 06:57:34 +00:00
Add load test runner service
- embed rakyll/hey in the runner container image
This commit is contained in:
stefanprodan
@@ -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"]
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user