From 7f590b070162b58147ce27cdb556af313192c246 Mon Sep 17 00:00:00 2001 From: Thiebaud Ernstberger Date: Sat, 14 Mar 2020 13:51:25 -0700 Subject: [PATCH] Issue #477: added concord test support to loadtester + documentation --- docs/gitbook/usage/webhooks.md | 29 +++- pkg/loadtester/concord.go | 242 +++++++++++++++++++++++++++++++++ pkg/loadtester/server.go | 28 ++++ 3 files changed, 298 insertions(+), 1 deletion(-) create mode 100644 pkg/loadtester/concord.go diff --git a/docs/gitbook/usage/webhooks.md b/docs/gitbook/usage/webhooks.md index 288ed19b..0864d95a 100644 --- a/docs/gitbook/usage/webhooks.md +++ b/docs/gitbook/usage/webhooks.md @@ -231,7 +231,7 @@ for the status of the test, and prevent duplicate requests from being sent in su ### Integration Testing -Flagger comes with a testing service that can run Helm tests or Bats tests when configured as a webhook. +Flagger comes with a testing service that can run Helm tests, Bats tests or Concord tests when configured as a webhook. Deploy the Helm test runner in the `kube-system` namespace using the `tiller` service account: @@ -292,6 +292,33 @@ As an alternative to Helm you can use the [Bash Automated Testing System](https: Note that you should create a ConfigMap with your Bats tests and mount it inside the tester container. +You can also configure the test runner to start a [Concord](https://concord.walmartlabs.com/) process. + +```yaml + analysis: + webhooks: + - name: "concord integration test" + type: pre-rollout + url: http://flagger-concordtester.default/ + timeout: 60s + metadata: + type: "concord" + org: "your-concord-org" + project: "your-concord-project" + repo: "your-concord-repo" + entrypoint: "your-concord-entrypoint" + apiKeyPath: "/tmp/concord-api-key" + endpoint: "https://canary-endpoint/" + pollInterval: "5" + pollTimeout: "60" +``` + +`org`, `project`, `repo` and `entrypoint` represents where your test process runs in Concord. +In order to authenticate to Concord, you need to set `apiKeyPath` to a path of a file containing a valid Concord API key + on the `flagger-helmtester` container. This can be done via mounting a Kubernetes secret in the tester's Deployment. +`pollInterval` represents the interval in seconds the web-hook will call Concord to see if the process has finished (Default is 5s). +`pollTimeout` represents the time in seconds the web-hook will try to call Concord before timing out (Default is 30s). + ### Manual Gating For manual approval of a canary deployment you can use the `confirm-rollout` and `confirm-promotion` webhooks. diff --git a/pkg/loadtester/concord.go b/pkg/loadtester/concord.go new file mode 100644 index 00000000..bef452a0 --- /dev/null +++ b/pkg/loadtester/concord.go @@ -0,0 +1,242 @@ +package loadtester + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/ioutil" + "mime/multipart" + "net/http" + "net/url" + "os" + "strconv" + "time" + + "go.uber.org/zap" +) + +// TaskTypeConcord represents the concord type as string +const TaskTypeConcord = "concord" + +// default process values +const concordBasePath = "/api/v1/process" +const defaultPollInterval = 5 +const defaultPollTimeout = 30 + +// concord statuses +const concordStatusSuccess = "FINISHED" +const concordStatusFailed = "FAILED" + +// ConcordTask represents a concord task +type ConcordTask struct { + TaskBase + Command string + Org string + Project string + Repo string + Entrypoint string + APIKeyPath string + Endpoint string + PollInterval time.Duration + PollTimeout time.Duration + BaseURL *url.URL + httpClient *http.Client +} + +// NewConcordTask instantiates a new Concord Task +func NewConcordTask(metadata map[string]string, canary string, logger *zap.SugaredLogger) (*ConcordTask, error) { + var pollIntervalInt, pollTimeoutInt int + + if _, found := metadata["server"]; found == false { + return nil, errors.New("`server` is required with type concord") + } + pURL, err := url.Parse(metadata["server"]) + if err != nil { + return nil, errors.New("failed to create base URL from metadata `concord_url`") + } + if _, found := metadata["org"]; found == false { + return nil, errors.New("`org` is required with type concord") + } + if _, found := metadata["project"]; found == false { + return nil, errors.New("`project` is required with type concord") + } + if _, found := metadata["repo"]; found == false { + return nil, errors.New("`repo` is required with type concord") + } + if _, found := metadata["entrypoint"]; found == false { + return nil, errors.New("`entrypoint` is required with type concord") + } + if _, found := metadata["apiKeyPath"]; found == false { + return nil, errors.New("`apiKeyPath` is required with type concord") + } + _, err = os.Stat(metadata["apiKeyPath"]) + if os.IsNotExist(err) { + return nil, fmt.Errorf("`apiKeyPath` file doesn't exist %s", metadata["apiKeyPath"]) + } + if _, found := metadata["endpoint"]; found == false { + return nil, errors.New("`endpoint` is required with type concord") + } + if _, found := metadata["pollInterval"]; found == false { + pollIntervalInt = defaultPollInterval + } else { + pollIntervalInt, err = strconv.Atoi(metadata["pollInterval"]) + if err != nil { + return nil, errors.New("unable to convert `pollInterval` to int") + } + } + if _, found := metadata["pollTimeout"]; found == false { + pollTimeoutInt = defaultPollTimeout + } else { + pollTimeoutInt, err = strconv.Atoi(metadata["pollTimeout"]) + if err != nil { + return nil, errors.New("unable to convert `pollTimeout` to int") + } + } + + return &ConcordTask{ + TaskBase: TaskBase{ + logger: logger, + }, + BaseURL: pURL, + Org: metadata["org"], + Project: metadata["project"], + Repo: metadata["repo"], + Entrypoint: metadata["entrypoint"], + APIKeyPath: metadata["apiKeyPath"], + Endpoint: metadata["endpoint"], + PollInterval: time.Duration(pollIntervalInt) * time.Second, + PollTimeout: time.Duration(pollTimeoutInt) * time.Second, + httpClient: &http.Client{Timeout: 60 * time.Second}, + }, nil +} + +func (task *ConcordTask) Hash() string { + return hash(task.canary + task.Org + task.Project + task.Repo + task.Entrypoint) +} + +func (task *ConcordTask) String() string { + return fmt.Sprintf("%s %s %s %s", task.Org, task.Project, task.Repo, task.Entrypoint) +} + +func (task *ConcordTask) Run(ctx context.Context) (bool, error) { + instance, err := task.startProcess() + if err != nil { + task.logger.Errorf("failed to start process: %s", err.Error()) + return false, err + } + + return task.checkStatus(ctx, instance, task.PollInterval) +} + +type concordProcess struct { + InstanceID string `json:"instanceId, omitempty"` + ParentInstanceID string `json:"parentInstanceID,omitempty"` + ProjectName string `json:"projectName,omitempty"` + CreatedAt string `json:"createdAt,omitempty"` + Initiator string `json:"initiator,omitempty"` + LastUpdatedAt string `json:"lastUpdatedAt,omitempty"` + Status string `json:"status,omitempty"` + ChildrenIds []string `json:"childrenIds,omitempty"` + OK bool `json:"ok,omitempty"` +} + +func (task *ConcordTask) newRequest(method, path string, contentType string, body io.Reader) (*http.Request, error) { + rel := &url.URL{Path: path} + u := task.BaseURL.ResolveReference(rel) + req, err := http.NewRequest(method, u.String(), body) + if err != nil { + return nil, err + } + + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + + apiKey := "" + dat, err := ioutil.ReadFile(task.APIKeyPath) + if err != nil { + return req, err + } + apiKey = string(dat) + + if apiKey != "" { + req.Header.Set("Authorization", apiKey) + } + + return req, nil +} + +func (task *ConcordTask) do(req *http.Request, v interface{}) (*http.Response, error) { + task.logger.With("canary", task.canary).Infof("calling endpoint %s", req.URL.String()) + resp, err := task.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + err = json.NewDecoder(resp.Body).Decode(v) + return resp, err +} + +func (task *ConcordTask) startProcess() (string, error) { + requestBody := new(bytes.Buffer) + writer := multipart.NewWriter(requestBody) + _ = writer.WriteField("org", task.Org) + _ = writer.WriteField("project", task.Project) + _ = writer.WriteField("repo", task.Repo) + _ = writer.WriteField("entryPoint", task.Entrypoint) + _ = writer.WriteField("arguments.endpoint", task.Endpoint) + + err := writer.Close() + if err != nil { + return "", nil + } + + req, err := task.newRequest(http.MethodPost, concordBasePath, writer.FormDataContentType(), requestBody) + if err != nil { + return "", err + } + + process := concordProcess{} + _, err = task.do(req, &process) + if err != nil { + return "", err + } + + task.logger.With("canary", task.canary).Infof("created process id [%s] OK status is %t", process.InstanceID, process.OK) + return process.InstanceID, nil +} + +func (task *ConcordTask) checkStatus(ctx context.Context, instanceID string, interval time.Duration) (bool, error) { + tickChan := time.NewTicker(interval).C + for { + select { + case <-tickChan: + req, err := task.newRequest(http.MethodGet, fmt.Sprintf("%s/%s", concordBasePath, instanceID), "", nil) + if err != nil { + return false, fmt.Errorf("failed to generate request: %s", err) + } + + process := concordProcess{} + _, err = task.do(req, &process) + if err != nil { + return false, fmt.Errorf("failed checking status: %s", err) + } + task.logger.With("canary", task.canary).Infof("process id [%s] current status is %s", process.InstanceID, process.Status) + + if process.Status == concordStatusSuccess { + return true, nil + } + if process.Status == concordStatusFailed { + return false, fmt.Errorf("concord instanceID: %s failed", instanceID) + } + case <-time.After(task.PollTimeout): + return false, fmt.Errorf("concord process timed out, after %d seconds", int64(task.PollTimeout/time.Second)) + case <-ctx.Done(): + return false, errors.New("context timedout") + } + + } +} diff --git a/pkg/loadtester/server.go b/pkg/loadtester/server.go index f6dc79da..01c73e5d 100644 --- a/pkg/loadtester/server.go +++ b/pkg/loadtester/server.go @@ -288,6 +288,34 @@ func ListenAndServe(port string, timeout time.Duration, logger *zap.SugaredLogge return } + // run concord job (blocking task) + if typ == TaskTypeConcord { + concord, err := NewConcordTask(payload.Metadata, fmt.Sprintf("%s.%s", payload.Name, payload.Namespace), logger) + + if err != nil { + logger.With("canary", payload.Name).Errorf("concord task init error: %s", err) + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), taskRunner.timeout) + defer cancel() + + ok, err := concord.Run(ctx) + if !ok { + if err != nil { + logger.With("canary", payload.Name).Errorf("concord task error: %s", err) + } + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(err.Error())) + return + } + + w.WriteHeader(http.StatusOK) + return + } + taskFactory, ok := GetTaskFactory(typ) if !ok { w.WriteHeader(http.StatusBadRequest)