mirror of
https://github.com/fluxcd/flagger.git
synced 2026-04-15 06:57:34 +00:00
Merge pull request #507 from ernst01/loadtester-concord-support
loadtester: add concord test support
This commit is contained in:
Stefan Prodan
@@ -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.
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user