github/flagger
1
0
Fork 0
mirror of https://github.com/fluxcd/flagger.git synced 2026-09-05 02:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into supergloo-updated

Browse Source
This commit is contained in:
Yuval Kohavi
2019-04-17 18:24:41 -04:00
parent 68d1f583cc 3492b07d9a
commit 156488c8d5
43 changed files with 1683 additions and 784 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.circleci/config.yml
+1 -1
View File
@@ -17,6 +17,6 @@ workflows:
filters:
branches:
ignore:
- gh-pages
- /gh-pages.*/
- /docs-.*/
- /release-.*/
.travis.yml
+11 -12
View File
@@ -1,6 +1,11 @@
sudo: required
language: go
branches:
except:
- /gh-pages.*/
- /docs-.*/
go:
- 1.12.x
@@ -12,13 +17,7 @@ addons:
packages:
- docker-ce
#before_script:
# - go get -u sigs.k8s.io/kind
# - curl https://raw.githubusercontent.com/kubernetes/helm/master/scripts/get | bash
# - curl -LO https://storage.googleapis.com/kubernetes-release/release/$(curl -s https://storage.googleapis.com/kubernetes-release/release/stable.txt)/bin/linux/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin/
script:
- set -e
- make test-fmt
- make test-codegen
- go test -race -coverprofile=coverage.txt -covermode=atomic $(go list ./pkg/...)
@@ -29,16 +28,16 @@ after_success:
echo "PR build, skipping image push";
else
BRANCH_COMMIT=${TRAVIS_BRANCH}-$(echo ${TRAVIS_COMMIT} | head -c7);
docker tag weaveworks/flagger:latest quay.io/weaveworks/flagger:${BRANCH_COMMIT};
echo $DOCKER_PASS | docker login -u=$DOCKER_USER --password-stdin quay.io;
docker push quay.io/weaveworks/flagger:${BRANCH_COMMIT};
docker tag weaveworks/flagger:latest weaveworks/flagger:${BRANCH_COMMIT};
echo $DOCKER_PASS | docker login -u=$DOCKER_USER --password-stdin;
docker push weaveworks/flagger:${BRANCH_COMMIT};
fi
- if [ -z "$TRAVIS_TAG" ]; then
echo "Not a release, skipping image push";
else
docker tag weaveworks/flagger:latest quay.io/weaveworks/flagger:${TRAVIS_TAG};
echo $DOCKER_PASS | docker login -u=$DOCKER_USER --password-stdin quay.io;
docker push quay.io/weaveworks/flagger:$TRAVIS_TAG;
docker tag weaveworks/flagger:latest weaveworks/flagger:${TRAVIS_TAG};
echo $DOCKER_PASS | docker login -u=$DOCKER_USER --password-stdin;
docker push weaveworks/flagger:$TRAVIS_TAG;
fi
- bash <(curl -s https://codecov.io/bash)
- rm coverage.txt
CHANGELOG.md
+17
View File
@@ -2,6 +2,23 @@
All notable changes to this project are documented in this file.
## 0.11.0 (2019-04-17)
Adds pre/post rollout [webhooks](https://docs.flagger.app/how-it-works#webhooks)
#### Features
- Add `pre-rollout` and `post-rollout` webhook types [#147](https://github.com/weaveworks/flagger/pull/147)
#### Improvements
- Unify App Mesh and Istio builtin metric checks [#146](https://github.com/weaveworks/flagger/pull/146)
- Make the pod selector label configurable [#148](https://github.com/weaveworks/flagger/pull/148)
#### Breaking changes
- Set default `mesh` Istio gateway only if no gateway is specified [#141](https://github.com/weaveworks/flagger/pull/141)
## 0.10.0 (2019-03-27)
Adds support for App Mesh
Dockerfile.loadtester
+6 -20
View File
@@ -1,20 +1,4 @@
FROM golang:1.12 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
FROM golang:1.12 AS builder
RUN mkdir -p /go/src/github.com/weaveworks/flagger/
@@ -26,15 +10,17 @@ RUN go test -race ./pkg/loadtester/
RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o loadtester ./cmd/loadtester/*
FROM alpine:3.9
FROM bats/bats:v1.1.0
RUN addgroup -S app \
&& adduser -S -g app app \
&& apk --no-cache add ca-certificates curl
&& apk --no-cache add ca-certificates curl jq
WORKDIR /home/app
COPY --from=hey-builder /go/bin/hey /usr/local/bin/hey
RUN curl -sSLo hey "https://storage.googleapis.com/jblabs/dist/hey_linux_v0.1.2" && \
chmod +x hey && mv hey /usr/local/bin/hey
COPY --from=builder /go/src/github.com/weaveworks/flagger/loadtester .
RUN chown -R app:app ./
Makefile
+2 -2
View File
@@ -88,5 +88,5 @@ reset-test:
kubectl apply -f ./artifacts/canaries
loadtester-push:
docker build -t quay.io/weaveworks/flagger-loadtester:$(LT_VERSION) . -f Dockerfile.loadtester
docker push quay.io/weaveworks/flagger-loadtester:$(LT_VERSION)
docker build -t weaveworks/flagger-loadtester:$(LT_VERSION) . -f Dockerfile.loadtester
docker push weaveworks/flagger-loadtester:$(LT_VERSION)
artifacts/flagger/crd.yaml
+32 -3
View File
@@ -2,6 +2,8 @@ apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: canaries.flagger.app
annotations:
helm.sh/resource-policy: keep
spec:
group: flagger.app
version: v1alpha3
@@ -39,9 +41,9 @@ spec:
properties:
spec:
required:
- targetRef
- service
- canaryAnalysis
- targetRef
- service
- canaryAnalysis
properties:
progressDeadlineSeconds:
type: number
@@ -119,9 +121,36 @@ spec:
properties:
name:
type: string
type:
type: string
enum:
- ""
- pre-rollout
- rollout
- post-rollout
url:
type: string
format: url
timeout:
type: string
pattern: "^[0-9]+(m|s)"
status:
properties:
phase:
type: string
enum:
- ""
- Initialized
- Progressing
- Succeeded
- Failed
canaryWeight:
type: number
failedChecks:
type: number
iterations:
type: number
lastAppliedSpec:
type: string
lastTransitionTime:
type: string
artifacts/flagger/deployment.yaml
+1 -1
View File
@@ -22,7 +22,7 @@ spec:
serviceAccountName: flagger
containers:
- name: flagger
image: quay.io/weaveworks/flagger:0.10.0
image: weaveworks/flagger:0.11.0
imagePullPolicy: IfNotPresent
ports:
- name: http
artifacts/loadtester/deployment.yaml
+1 -1
View File
@@ -17,7 +17,7 @@ spec:
spec:
containers:
- name: loadtester
image: quay.io/stefanprodan/flagger-loadtester:0.2.0
image: weaveworks/flagger-loadtester:0.2.0
imagePullPolicy: IfNotPresent
ports:
- name: http
charts/flagger/Chart.yaml
+2 -2
View File
@@ -1,7 +1,7 @@
apiVersion: v1
name: flagger
version: 0.10.0
appVersion: 0.10.0
version: 0.11.0
appVersion: 0.11.0
kubeVersion: ">=1.11.0-0"
engine: gotpl
description: Flagger is a Kubernetes operator that automates the promotion of canary deployments using Istio routing for traffic shifting and Prometheus metrics for canary analysis.
charts/flagger/templates/crd.yaml
+27
View File
@@ -122,10 +122,37 @@ spec:
properties:
name:
type: string
type:
type: string
enum:
- ""
- pre-rollout
- rollout
- post-rollout
url:
type: string
format: url
timeout:
type: string
pattern: "^[0-9]+(m|s)"
status:
properties:
phase:
type: string
enum:
- ""
- Initialized
- Progressing
- Succeeded
- Failed
canaryWeight:
type: number
failedChecks:
type: number
iterations:
type: number
lastAppliedSpec:
type: string
lastTransitionTime:
type: string
{{- end }}
charts/flagger/values.yaml
+2 -2
View File
@@ -1,8 +1,8 @@
# Default values for flagger.
image:
repository: quay.io/weaveworks/flagger
tag: 0.10.0
repository: weaveworks/flagger
tag: 0.11.0
pullPolicy: IfNotPresent
metricsServer: "http://prometheus:9090"
cmd/flagger/main.go
+11 -2
View File
@@ -3,13 +3,14 @@ package main
import (
"flag"
"log"
"strings"
"time"
_ "github.com/istio/glog"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
informers "github.com/weaveworks/flagger/pkg/client/informers/externalversions"
"github.com/weaveworks/flagger/pkg/controller"
"github.com/weaveworks/flagger/pkg/logging"
"github.com/weaveworks/flagger/pkg/logger"
"github.com/weaveworks/flagger/pkg/metrics"
"github.com/weaveworks/flagger/pkg/notifier"
"github.com/weaveworks/flagger/pkg/router"
@@ -38,6 +39,7 @@ var (
zapEncoding string
namespace string
meshProvider string
selectorLabels string
)
func init() {
@@ -55,12 +57,13 @@ func init() {
flag.StringVar(&zapEncoding, "zap-encoding", "json", "Zap logger encoding.")
flag.StringVar(&namespace, "namespace", "", "Namespace that flagger would watch canary object")
flag.StringVar(&meshProvider, "mesh-provider", "istio", "Service mesh provider, can be istio or appmesh")
flag.StringVar(&selectorLabels, "selector-labels", "app,name,app.kubernetes.io/name", "List of pod labels that Flagger uses to create pod selectors")
}
func main() {
flag.Parse()
logger, err := logging.NewLoggerWithEncoding(logLevel, zapEncoding)
logger, err := logger.NewLoggerWithEncoding(logLevel, zapEncoding)
if err != nil {
log.Fatalf("Error creating logger: %v", err)
}
@@ -103,6 +106,11 @@ func main() {
logger.Fatalf("Error calling Kubernetes API: %v", err)
}
labels := strings.Split(selectorLabels, ",")
if len(labels) < 1 {
logger.Fatalf("At least one selector label is required")
}
logger.Infof("Connected to Kubernetes API %s", ver)
if namespace != "" {
logger.Infof("Watching namespace %s", namespace)
@@ -142,6 +150,7 @@ func main() {
routerFactory,
meshProvider,
version.VERSION,
labels,
)
flaggerInformerFactory.Start(stopCh)
cmd/loadtester/main.go
+2 -2
View File
@@ -3,7 +3,7 @@ package main
import (
"flag"
"github.com/weaveworks/flagger/pkg/loadtester"
"github.com/weaveworks/flagger/pkg/logging"
"github.com/weaveworks/flagger/pkg/logger"
"github.com/weaveworks/flagger/pkg/signals"
"go.uber.org/zap"
"log"
@@ -30,7 +30,7 @@ func init() {
func main() {
flag.Parse()
logger, err := logging.NewLoggerWithEncoding(logLevel, zapEncoding)
logger, err := logger.NewLoggerWithEncoding(logLevel, zapEncoding)
if err != nil {
log.Fatalf("Error creating logger: %v", err)
}
docs/gitbook/how-it-works.md
+42 -15
View File
@@ -2,7 +2,7 @@
[Flagger](https://github.com/weaveworks/flagger) takes a Kubernetes deployment and optionally
a horizontal pod autoscaler \(HPA\) and creates a series of objects
\(Kubernetes deployments, ClusterIP services and Istio virtual services\) to drive the canary analysis and promotion.
\(Kubernetes deployments, ClusterIP services and Istio or App Mesh virtual services\) to drive the canary analysis and promotion.
![Flagger Canary Process](https://raw.githubusercontent.com/weaveworks/flagger/master/docs/diagrams/flagger-canary-hpa.png)
@@ -96,6 +96,9 @@ spec:
app: podinfo
```
Besides `app` Flagger supports `name` and `app.kubernetes.io/name` selectors. If you use a different
convention you can specify your label with the `-selector-labels` flag.
The target deployment should expose a TCP port that will be used by Flagger to create the ClusterIP Service and
the Istio Virtual Service. The container port from the target deployment should match the `service.port` value.
@@ -268,16 +271,22 @@ Gated canary promotion stages:
* check primary and canary deployments status
* halt advancement if a rolling update is underway
* halt advancement if pods are unhealthy
* call pre-rollout webhooks are check results
* halt advancement if any hook returned a non HTTP 2xx result
* increment the failed checks counter
* increase canary traffic weight percentage from 0% to 5% (step weight)
* call webhooks and check results
* call rollout webhooks and check results
* check canary HTTP request success rate and latency
* halt advancement if any metric is under the specified threshold
* increment the failed checks counter
* check if the number of failed checks reached the threshold
* route all traffic to primary
* scale to zero the canary deployment and mark it as failed
* call post-rollout webhooks
* post the analysis result to Slack
* wait for the canary deployment to be updated and start over
* increase canary traffic weight by 5% (step weight) till it reaches 50% (max weight)
* halt advancement if any webhook call fails
* halt advancement while canary request success rate is under the threshold
* halt advancement while canary request duration P99 is over the threshold
* halt advancement if the primary or canary deployment becomes unhealthy
@@ -290,6 +299,8 @@ Gated canary promotion stages:
* route all traffic to primary
* scale to zero the canary deployment
* mark rollout as finished
* call post-rollout webhooks
* post the analysis result to Slack
* wait for the canary deployment to be updated and start over
### Canary Analysis
@@ -524,39 +535,55 @@ rate reaches the 5% threshold, then the canary fails.
When specifying a query, Flagger will run the promql query and convert the result to float64.
Then it compares the query result value with the metric threshold value.
### Webhooks
The canary analysis can be extended with webhooks.
Flagger will call each webhook URL and determine from the response status code (HTTP 2xx) if the canary is failing or not.
The canary analysis can be extended with webhooks. Flagger will call each webhook URL and
determine from the response status code (HTTP 2xx) if the canary is failing or not.
There are three types of hooks:
* Pre-rollout hooks are executed before routing traffic to canary.
The canary advancement is paused if a pre-rollout hook fails and if the number of failures reach the
threshold the canary will be rollback.
* Rollout hooks are executed during the analysis on each iteration before the metric checks.
If a rollout hook call fails the canary advancement is paused and eventfully rolled back.
* Post-rollout hooks are executed after the canary has been promoted or rolled back.
If a post rollout hook fails the error is logged.
Spec:
```yaml
canaryAnalysis:
webhooks:
- name: integration-test
url: http://int-runner.test:8080/
timeout: 30s
metadata:
test: "all"
token: "16688eb5e9f289f1991c"
- name: db-test
- name: "smoke test"
type: pre-rollout
url: http://migration-check.db/query
timeout: 30s
metadata:
key1: "val1"
key2: "val2"
- name: "load test"
type: rollout
url: http://flagger-loadtester.test/
timeout: 15s
metadata:
cmd: "hey -z 1m -q 5 -c 2 http://podinfo-canary.test:9898/"
- name: "notify"
type: post-rollout
url: http://telegram.bot:8080/
timeout: 5s
metadata:
some: "message"
```
> **Note** that the sum of all webhooks timeouts should be lower than the control loop interval.
> **Note** that the sum of all rollout webhooks timeouts should be lower than the analysis interval.
Webhook payload (HTTP POST):
```json
{
"name": "podinfo",
"namespace": "test",
"namespace": "test",
"phase": "Progressing",
"metadata": {
"test": "all",
"token": "16688eb5e9f289f1991c"
@@ -676,4 +703,4 @@ webhooks:
```
When the canary analysis starts, the load tester will initiate a [clone_and_start request](https://github.com/naver/ngrinder/wiki/REST-API-PerfTest)
to the nGrinder server and start a new performance test. the load tester will periodically poll the nGrinder server
for the status of the test, and prevent duplicate requests from being sent in subsequent analysis loops.
for the status of the test, and prevent duplicate requests from being sent in subsequent analysis loops.
docs/gitbook/install/flagger-install-on-google-cloud.md
+6 -4
View File
@@ -186,7 +186,7 @@ Install cert-manager's CRDs:
```bash
CERT_REPO=https://raw.githubusercontent.com/jetstack/cert-manager
kubectl apply -f ${CERT_REPO}/release-0.6/deploy/manifests/00-crds.yaml
kubectl apply -f ${CERT_REPO}/release-0.7/deploy/manifests/00-crds.yaml
```
Create the cert-manager namespace and disable resource validation:
@@ -200,10 +200,12 @@ kubectl label namespace cert-manager certmanager.k8s.io/disable-validation=true
Install cert-manager with Helm:
```bash
helm repo update && helm upgrade -i cert-manager \
helm repo add jetstack https://charts.jetstack.io && \
helm repo update && \
helm upgrade -i cert-manager \
--namespace cert-manager \
--version v0.6.0 \
stable/cert-manager
--version v0.7.0 \
jetstack/cert-manager
```
### Istio Gateway TLS setup
pkg/apis/flagger/v1alpha3/types.go
+17 -3
View File
@@ -148,11 +148,24 @@ type CanaryMetric struct {
Query string `json:"query,omitempty"`
}
// HookType can be pre, post or during rollout
type HookType string
const (
// RolloutHook execute webhook during the canary analysis
RolloutHook HookType = "rollout"
// PreRolloutHook execute webhook before routing traffic to canary
PreRolloutHook HookType = "pre-rollout"
// PreRolloutHook execute webhook after the canary analysis
PostRolloutHook HookType = "post-rollout"
)
// CanaryWebhook holds the reference to external checks used for canary analysis
type CanaryWebhook struct {
Name string `json:"name"`
URL string `json:"url"`
Timeout string `json:"timeout"`
Type HookType `json:"type"`
Name string `json:"name"`
URL string `json:"url"`
Timeout string `json:"timeout"`
// +optional
Metadata *map[string]string `json:"metadata,omitempty"`
}
@@ -161,6 +174,7 @@ type CanaryWebhook struct {
type CanaryWebhookPayload struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
Phase CanaryPhase `json:"phase"`
Metadata map[string]string `json:"metadata,omitempty"`
}
pkg/canary/deployer.go
+357
View File
@@ -0,0 +1,357 @@
package canary
import (
"crypto/rand"
"encoding/base64"
"encoding/json"
"fmt"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
"go.uber.org/zap"
"io"
appsv1 "k8s.io/api/apps/v1"
hpav1 "k8s.io/api/autoscaling/v2beta1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/kubernetes"
)
// Deployer is managing the operations for Kubernetes deployment kind
type Deployer struct {
KubeClient kubernetes.Interface
FlaggerClient clientset.Interface
Logger *zap.SugaredLogger
ConfigTracker ConfigTracker
Labels []string
}
// Initialize creates the primary deployment, hpa,
// scales to zero the canary deployment and returns the pod selector label
func (c *Deployer) Initialize(cd *flaggerv1.Canary) (string, error) {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
label, err := c.createPrimaryDeployment(cd)
if err != nil {
return "", fmt.Errorf("creating deployment %s.%s failed: %v", primaryName, cd.Namespace, err)
}
if cd.Status.Phase == "" {
c.Logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).Infof("Scaling down %s.%s", cd.Spec.TargetRef.Name, cd.Namespace)
if err := c.Scale(cd, 0); err != nil {
return "", err
}
}
if cd.Spec.AutoscalerRef != nil && cd.Spec.AutoscalerRef.Kind == "HorizontalPodAutoscaler" {
if err := c.createPrimaryHpa(cd); err != nil {
return "", fmt.Errorf("creating hpa %s.%s failed: %v", primaryName, cd.Namespace, err)
}
}
return label, nil
}
// Promote copies the pod spec, secrets and config maps from canary to primary
func (c *Deployer) Promote(cd *flaggerv1.Canary) error {
targetName := cd.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
canary, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
label, err := c.getSelectorLabel(canary)
if err != nil {
return fmt.Errorf("invalid label selector! Deployment %s.%s spec.selector.matchLabels must contain selector 'app: %s'",
targetName, cd.Namespace, targetName)
}
primary, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", primaryName, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", primaryName, cd.Namespace, err)
}
// promote secrets and config maps
configRefs, err := c.ConfigTracker.GetTargetConfigs(cd)
if err != nil {
return err
}
if err := c.ConfigTracker.CreatePrimaryConfigs(cd, configRefs); err != nil {
return err
}
primaryCopy := primary.DeepCopy()
primaryCopy.Spec.ProgressDeadlineSeconds = canary.Spec.ProgressDeadlineSeconds
primaryCopy.Spec.MinReadySeconds = canary.Spec.MinReadySeconds
primaryCopy.Spec.RevisionHistoryLimit = canary.Spec.RevisionHistoryLimit
primaryCopy.Spec.Strategy = canary.Spec.Strategy
// update spec with primary secrets and config maps
primaryCopy.Spec.Template.Spec = c.ConfigTracker.ApplyPrimaryConfigs(canary.Spec.Template.Spec, configRefs)
// update pod annotations to ensure a rolling update
annotations, err := c.makeAnnotations(canary.Spec.Template.Annotations)
if err != nil {
return err
}
primaryCopy.Spec.Template.Annotations = annotations
primaryCopy.Spec.Template.Labels = makePrimaryLabels(canary.Spec.Template.Labels, primaryName, label)
_, err = c.KubeClient.AppsV1().Deployments(cd.Namespace).Update(primaryCopy)
if err != nil {
return fmt.Errorf("updating deployment %s.%s template spec failed: %v",
primaryCopy.GetName(), primaryCopy.Namespace, err)
}
return nil
}
// HasDeploymentChanged returns true if the canary deployment pod spec has changed
func (c *Deployer) HasDeploymentChanged(cd *flaggerv1.Canary) (bool, error) {
targetName := cd.Spec.TargetRef.Name
canary, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return false, fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return false, fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
if cd.Status.LastAppliedSpec == "" {
return true, nil
}
newSpec := &canary.Spec.Template.Spec
oldSpecJson, err := base64.StdEncoding.DecodeString(cd.Status.LastAppliedSpec)
if err != nil {
return false, fmt.Errorf("%s.%s decode error %v", cd.Name, cd.Namespace, err)
}
oldSpec := &corev1.PodSpec{}
err = json.Unmarshal(oldSpecJson, oldSpec)
if err != nil {
return false, fmt.Errorf("%s.%s unmarshal error %v", cd.Name, cd.Namespace, err)
}
if diff := cmp.Diff(*newSpec, *oldSpec, cmpopts.IgnoreUnexported(resource.Quantity{})); diff != "" {
//fmt.Println(diff)
return true, nil
}
return false, nil
}
// Scale sets the canary deployment replicas
func (c *Deployer) Scale(cd *flaggerv1.Canary, replicas int32) error {
targetName := cd.Spec.TargetRef.Name
dep, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
depCopy := dep.DeepCopy()
depCopy.Spec.Replicas = int32p(replicas)
_, err = c.KubeClient.AppsV1().Deployments(dep.Namespace).Update(depCopy)
if err != nil {
return fmt.Errorf("scaling %s.%s to %v failed: %v", depCopy.GetName(), depCopy.Namespace, replicas, err)
}
return nil
}
func (c *Deployer) createPrimaryDeployment(cd *flaggerv1.Canary) (string, error) {
targetName := cd.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
canaryDep, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return "", fmt.Errorf("deployment %s.%s not found, retrying", targetName, cd.Namespace)
}
return "", err
}
label, err := c.getSelectorLabel(canaryDep)
if err != nil {
return "", fmt.Errorf("invalid label selector! Deployment %s.%s spec.selector.matchLabels must contain selector 'app: %s'",
targetName, cd.Namespace, targetName)
}
primaryDep, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if errors.IsNotFound(err) {
// create primary secrets and config maps
configRefs, err := c.ConfigTracker.GetTargetConfigs(cd)
if err != nil {
return "", err
}
if err := c.ConfigTracker.CreatePrimaryConfigs(cd, configRefs); err != nil {
return "", err
}
annotations, err := c.makeAnnotations(canaryDep.Spec.Template.Annotations)
if err != nil {
return "", err
}
replicas := int32(1)
if canaryDep.Spec.Replicas != nil && *canaryDep.Spec.Replicas > 0 {
replicas = *canaryDep.Spec.Replicas
}
// create primary deployment
primaryDep = &appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: primaryName,
Labels: canaryDep.Labels,
Namespace: cd.Namespace,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(cd, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: appsv1.DeploymentSpec{
ProgressDeadlineSeconds: canaryDep.Spec.ProgressDeadlineSeconds,
MinReadySeconds: canaryDep.Spec.MinReadySeconds,
RevisionHistoryLimit: canaryDep.Spec.RevisionHistoryLimit,
Replicas: int32p(replicas),
Strategy: canaryDep.Spec.Strategy,
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
label: primaryName,
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: makePrimaryLabels(canaryDep.Spec.Template.Labels, primaryName, label),
Annotations: annotations,
},
// update spec with the primary secrets and config maps
Spec: c.ConfigTracker.ApplyPrimaryConfigs(canaryDep.Spec.Template.Spec, configRefs),
},
},
}
_, err = c.KubeClient.AppsV1().Deployments(cd.Namespace).Create(primaryDep)
if err != nil {
return "", err
}
c.Logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).Infof("Deployment %s.%s created", primaryDep.GetName(), cd.Namespace)
}
return label, nil
}
func (c *Deployer) createPrimaryHpa(cd *flaggerv1.Canary) error {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
hpa, err := c.KubeClient.AutoscalingV2beta1().HorizontalPodAutoscalers(cd.Namespace).Get(cd.Spec.AutoscalerRef.Name, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("HorizontalPodAutoscaler %s.%s not found, retrying",
cd.Spec.AutoscalerRef.Name, cd.Namespace)
}
return err
}
primaryHpaName := fmt.Sprintf("%s-primary", cd.Spec.AutoscalerRef.Name)
primaryHpa, err := c.KubeClient.AutoscalingV2beta1().HorizontalPodAutoscalers(cd.Namespace).Get(primaryHpaName, metav1.GetOptions{})
if errors.IsNotFound(err) {
primaryHpa = &hpav1.HorizontalPodAutoscaler{
ObjectMeta: metav1.ObjectMeta{
Name: primaryHpaName,
Namespace: cd.Namespace,
Labels: hpa.Labels,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(cd, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: hpav1.HorizontalPodAutoscalerSpec{
ScaleTargetRef: hpav1.CrossVersionObjectReference{
Name: primaryName,
Kind: hpa.Spec.ScaleTargetRef.Kind,
APIVersion: hpa.Spec.ScaleTargetRef.APIVersion,
},
MinReplicas: hpa.Spec.MinReplicas,
MaxReplicas: hpa.Spec.MaxReplicas,
Metrics: hpa.Spec.Metrics,
},
}
_, err = c.KubeClient.AutoscalingV2beta1().HorizontalPodAutoscalers(cd.Namespace).Create(primaryHpa)
if err != nil {
return err
}
c.Logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).Infof("HorizontalPodAutoscaler %s.%s created", primaryHpa.GetName(), cd.Namespace)
}
return nil
}
// makeAnnotations appends an unique ID to annotations map
func (c *Deployer) makeAnnotations(annotations map[string]string) (map[string]string, error) {
idKey := "flagger-id"
res := make(map[string]string)
uuid := make([]byte, 16)
n, err := io.ReadFull(rand.Reader, uuid)
if n != len(uuid) || err != nil {
return res, err
}
uuid[8] = uuid[8]&^0xc0 | 0x80
uuid[6] = uuid[6]&^0xf0 | 0x40
id := fmt.Sprintf("%x-%x-%x-%x-%x", uuid[0:4], uuid[4:6], uuid[6:8], uuid[8:10], uuid[10:])
for k, v := range annotations {
if k != idKey {
res[k] = v
}
}
res[idKey] = id
return res, nil
}
// getSelectorLabel returns the selector match label
func (c *Deployer) getSelectorLabel(deployment *appsv1.Deployment) (string, error) {
for _, l := range c.Labels {
if _, ok := deployment.Spec.Selector.MatchLabels[l]; ok {
return l, nil
}
}
return "", fmt.Errorf("selector not found")
}
func makePrimaryLabels(labels map[string]string, primaryName string, label string) map[string]string {
res := make(map[string]string)
for k, v := range labels {
if k != label {
res[k] = v
}
}
res[label] = primaryName
return res
}
func int32p(i int32) *int32 {
return &i
}
pkg/controller/deployer_test.go → pkg/canary/deployer_test.go
+18 -19
View File
@@ -1,4 +1,4 @@
package controller
package canary
import (
"testing"
@@ -8,8 +8,8 @@ import (
)
func TestCanaryDeployer_Sync(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -94,8 +94,8 @@ func TestCanaryDeployer_Sync(t *testing.T) {
}
func TestCanaryDeployer_IsNewSpec(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -106,7 +106,7 @@ func TestCanaryDeployer_IsNewSpec(t *testing.T) {
t.Fatal(err.Error())
}
isNew, err := mocks.deployer.IsNewSpec(mocks.canary)
isNew, err := mocks.deployer.HasDeploymentChanged(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -117,8 +117,8 @@ func TestCanaryDeployer_IsNewSpec(t *testing.T) {
}
func TestCanaryDeployer_Promote(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -162,8 +162,8 @@ func TestCanaryDeployer_Promote(t *testing.T) {
}
func TestCanaryDeployer_IsReady(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Error("Expected primary readiness check to fail")
}
@@ -180,8 +180,8 @@ func TestCanaryDeployer_IsReady(t *testing.T) {
}
func TestCanaryDeployer_SetFailedChecks(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -202,8 +202,8 @@ func TestCanaryDeployer_SetFailedChecks(t *testing.T) {
}
func TestCanaryDeployer_SetState(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -224,8 +224,8 @@ func TestCanaryDeployer_SetState(t *testing.T) {
}
func TestCanaryDeployer_SyncStatus(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -263,8 +263,8 @@ func TestCanaryDeployer_SyncStatus(t *testing.T) {
}
func TestCanaryDeployer_Scale(t *testing.T) {
mocks := SetupMocks(false)
err := mocks.deployer.Sync(mocks.canary)
mocks := SetupMocks()
_, err := mocks.deployer.Initialize(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
@@ -279,5 +279,4 @@ func TestCanaryDeployer_Scale(t *testing.T) {
if *c.Spec.Replicas != 2 {
t.Errorf("Got replicas %v wanted %v", *c.Spec.Replicas, 2)
}
}
pkg/canary/mock.go
+471
View File
@@ -0,0 +1,471 @@
package canary
import (
"github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
fakeFlagger "github.com/weaveworks/flagger/pkg/client/clientset/versioned/fake"
"github.com/weaveworks/flagger/pkg/logger"
"go.uber.org/zap"
appsv1 "k8s.io/api/apps/v1"
hpav1 "k8s.io/api/autoscaling/v1"
hpav2 "k8s.io/api/autoscaling/v2beta1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/fake"
)
type Mocks struct {
canary *v1alpha3.Canary
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
deployer Deployer
logger *zap.SugaredLogger
}
func SetupMocks() Mocks {
// init canary
canary := newTestCanary()
flaggerClient := fakeFlagger.NewSimpleClientset(canary)
// init kube clientset and register mock objects
kubeClient := fake.NewSimpleClientset(
newTestDeployment(),
newTestHPA(),
NewTestConfigMap(),
NewTestConfigMapEnv(),
NewTestConfigMapVol(),
NewTestSecret(),
NewTestSecretEnv(),
NewTestSecretVol(),
)
logger, _ := logger.NewLogger("debug")
deployer := Deployer{
FlaggerClient: flaggerClient,
KubeClient: kubeClient,
Logger: logger,
Labels: []string{"app", "name"},
ConfigTracker: ConfigTracker{
Logger: logger,
KubeClient: kubeClient,
FlaggerClient: flaggerClient,
},
}
return Mocks{
canary: canary,
deployer: deployer,
logger: logger,
flaggerClient: flaggerClient,
kubeClient: kubeClient,
}
}
func NewTestConfigMap() *corev1.ConfigMap {
return &corev1.ConfigMap{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-config-env",
},
Data: map[string]string{
"color": "red",
},
}
}
func NewTestConfigMapV2() *corev1.ConfigMap {
return &corev1.ConfigMap{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-config-env",
},
Data: map[string]string{
"color": "blue",
"output": "console",
},
}
}
func NewTestConfigMapEnv() *corev1.ConfigMap {
return &corev1.ConfigMap{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-config-all-env",
},
Data: map[string]string{
"color": "red",
},
}
}
func NewTestConfigMapVol() *corev1.ConfigMap {
return &corev1.ConfigMap{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-config-vol",
},
Data: map[string]string{
"color": "red",
},
}
}
func NewTestSecret() *corev1.Secret {
return &corev1.Secret{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-secret-env",
},
Type: corev1.SecretTypeOpaque,
Data: map[string][]byte{
"apiKey": []byte("test"),
},
}
}
func NewTestSecretV2() *corev1.Secret {
return &corev1.Secret{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-secret-env",
},
Type: corev1.SecretTypeOpaque,
Data: map[string][]byte{
"apiKey": []byte("test2"),
},
}
}
func NewTestSecretEnv() *corev1.Secret {
return &corev1.Secret{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-secret-all-env",
},
Type: corev1.SecretTypeOpaque,
Data: map[string][]byte{
"apiKey": []byte("test"),
},
}
}
func NewTestSecretVol() *corev1.Secret {
return &corev1.Secret{
TypeMeta: metav1.TypeMeta{APIVersion: corev1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo-secret-vol",
},
Type: corev1.SecretTypeOpaque,
Data: map[string][]byte{
"apiKey": []byte("test"),
},
}
}
func newTestCanary() *v1alpha3.Canary {
cd := &v1alpha3.Canary{
TypeMeta: metav1.TypeMeta{APIVersion: v1alpha3.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: v1alpha3.CanarySpec{
TargetRef: hpav1.CrossVersionObjectReference{
Name: "podinfo",
APIVersion: "apps/v1",
Kind: "Deployment",
},
AutoscalerRef: &hpav1.CrossVersionObjectReference{
Name: "podinfo",
APIVersion: "autoscaling/v2beta1",
Kind: "HorizontalPodAutoscaler",
}, Service: v1alpha3.CanaryService{
Port: 9898,
}, CanaryAnalysis: v1alpha3.CanaryAnalysis{
Threshold: 10,
StepWeight: 10,
MaxWeight: 50,
Metrics: []v1alpha3.CanaryMetric{
{
Name: "istio_requests_total",
Threshold: 99,
Interval: "1m",
},
{
Name: "istio_request_duration_seconds_bucket",
Threshold: 500,
Interval: "1m",
},
},
},
},
}
return cd
}
func newTestDeployment() *appsv1.Deployment {
d := &appsv1.Deployment{
TypeMeta: metav1.TypeMeta{APIVersion: appsv1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: appsv1.DeploymentSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"name": "podinfo",
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"name": "podinfo",
},
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Name: "podinfo",
Image: "quay.io/stefanprodan/podinfo:1.2.0",
Command: []string{
"./podinfo",
"--port=9898",
},
Args: nil,
WorkingDir: "",
Ports: []corev1.ContainerPort{
{
Name: "http",
ContainerPort: 9898,
Protocol: corev1.ProtocolTCP,
},
},
Env: []corev1.EnvVar{
{
Name: "PODINFO_UI_COLOR",
ValueFrom: &corev1.EnvVarSource{
ConfigMapKeyRef: &corev1.ConfigMapKeySelector{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-config-env",
},
Key: "color",
},
},
},
{
Name: "API_KEY",
ValueFrom: &corev1.EnvVarSource{
SecretKeyRef: &corev1.SecretKeySelector{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-secret-env",
},
Key: "apiKey",
},
},
},
},
EnvFrom: []corev1.EnvFromSource{
{
ConfigMapRef: &corev1.ConfigMapEnvSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-config-all-env",
},
},
},
{
SecretRef: &corev1.SecretEnvSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-secret-all-env",
},
},
},
},
VolumeMounts: []corev1.VolumeMount{
{
Name: "config",
MountPath: "/etc/podinfo/config",
ReadOnly: true,
},
{
Name: "secret",
MountPath: "/etc/podinfo/secret",
ReadOnly: true,
},
},
},
},
Volumes: []corev1.Volume{
{
Name: "config",
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-config-vol",
},
},
},
},
{
Name: "secret",
VolumeSource: corev1.VolumeSource{
Secret: &corev1.SecretVolumeSource{
SecretName: "podinfo-secret-vol",
},
},
},
},
},
},
},
}
return d
}
func newTestDeploymentV2() *appsv1.Deployment {
d := &appsv1.Deployment{
TypeMeta: metav1.TypeMeta{APIVersion: appsv1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: appsv1.DeploymentSpec{
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"name": "podinfo",
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: map[string]string{
"name": "podinfo",
},
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{
{
Name: "podinfo",
Image: "quay.io/stefanprodan/podinfo:1.2.1",
Ports: []corev1.ContainerPort{
{
Name: "http",
ContainerPort: 9898,
Protocol: corev1.ProtocolTCP,
},
},
Command: []string{
"./podinfo",
"--port=9898",
},
Env: []corev1.EnvVar{
{
Name: "PODINFO_UI_COLOR",
ValueFrom: &corev1.EnvVarSource{
ConfigMapKeyRef: &corev1.ConfigMapKeySelector{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-config-env",
},
Key: "color",
},
},
},
{
Name: "API_KEY",
ValueFrom: &corev1.EnvVarSource{
SecretKeyRef: &corev1.SecretKeySelector{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-secret-env",
},
Key: "apiKey",
},
},
},
},
EnvFrom: []corev1.EnvFromSource{
{
ConfigMapRef: &corev1.ConfigMapEnvSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-config-all-env",
},
},
},
},
VolumeMounts: []corev1.VolumeMount{
{
Name: "config",
MountPath: "/etc/podinfo/config",
ReadOnly: true,
},
{
Name: "secret",
MountPath: "/etc/podinfo/secret",
ReadOnly: true,
},
},
},
},
Volumes: []corev1.Volume{
{
Name: "config",
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: "podinfo-config-vol",
},
},
},
},
{
Name: "secret",
VolumeSource: corev1.VolumeSource{
Secret: &corev1.SecretVolumeSource{
SecretName: "podinfo-secret-vol",
},
},
},
},
},
},
},
}
return d
}
func newTestHPA() *hpav2.HorizontalPodAutoscaler {
h := &hpav2.HorizontalPodAutoscaler{
TypeMeta: metav1.TypeMeta{APIVersion: hpav2.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: hpav2.HorizontalPodAutoscalerSpec{
ScaleTargetRef: hpav2.CrossVersionObjectReference{
Name: "podinfo",
APIVersion: "apps/v1",
Kind: "Deployment",
},
Metrics: []hpav2.MetricSpec{
{
Type: "Resource",
Resource: &hpav2.ResourceMetricSource{
Name: "cpu",
TargetAverageUtilization: int32p(99),
},
},
},
},
}
return h
}
pkg/canary/ready.go
+112
View File
@@ -0,0 +1,112 @@
package canary
import (
"fmt"
"time"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
appsv1 "k8s.io/api/apps/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// IsPrimaryReady checks the primary deployment status and returns an error if
// the deployment is in the middle of a rolling update or if the pods are unhealthy
// it will return a non retriable error if the rolling update is stuck
func (c *Deployer) IsPrimaryReady(cd *flaggerv1.Canary) (bool, error) {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
primary, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return true, fmt.Errorf("deployment %s.%s not found", primaryName, cd.Namespace)
}
return true, fmt.Errorf("deployment %s.%s query error %v", primaryName, cd.Namespace, err)
}
retriable, err := c.isDeploymentReady(primary, cd.GetProgressDeadlineSeconds())
if err != nil {
return retriable, fmt.Errorf("Halt advancement %s.%s %s", primaryName, cd.Namespace, err.Error())
}
if primary.Spec.Replicas == int32p(0) {
return true, fmt.Errorf("Halt %s.%s advancement primary deployment is scaled to zero",
cd.Name, cd.Namespace)
}
return true, nil
}
// IsCanaryReady checks the primary deployment status and returns an error if
// the deployment is in the middle of a rolling update or if the pods are unhealthy
// it will return a non retriable error if the rolling update is stuck
func (c *Deployer) IsCanaryReady(cd *flaggerv1.Canary) (bool, error) {
targetName := cd.Spec.TargetRef.Name
canary, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return true, fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return true, fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
retriable, err := c.isDeploymentReady(canary, cd.GetProgressDeadlineSeconds())
if err != nil {
if retriable {
return retriable, fmt.Errorf("Halt advancement %s.%s %s", targetName, cd.Namespace, err.Error())
} else {
return retriable, fmt.Errorf("deployment does not have minimum availability for more than %vs",
cd.GetProgressDeadlineSeconds())
}
}
return true, nil
}
// isDeploymentReady determines if a deployment is ready by checking the status conditions
// if a deployment has exceeded the progress deadline it returns a non retriable error
func (c *Deployer) isDeploymentReady(deployment *appsv1.Deployment, deadline int) (bool, error) {
retriable := true
if deployment.Generation <= deployment.Status.ObservedGeneration {
progress := c.getDeploymentCondition(deployment.Status, appsv1.DeploymentProgressing)
if progress != nil {
// Determine if the deployment is stuck by checking if there is a minimum replicas unavailable condition
// and if the last update time exceeds the deadline
available := c.getDeploymentCondition(deployment.Status, appsv1.DeploymentAvailable)
if available != nil && available.Status == "False" && available.Reason == "MinimumReplicasUnavailable" {
from := available.LastUpdateTime
delta := time.Duration(deadline) * time.Second
retriable = !from.Add(delta).Before(time.Now())
}
}
if progress != nil && progress.Reason == "ProgressDeadlineExceeded" {
return false, fmt.Errorf("deployment %q exceeded its progress deadline", deployment.GetName())
} else if deployment.Spec.Replicas != nil && deployment.Status.UpdatedReplicas < *deployment.Spec.Replicas {
return retriable, fmt.Errorf("waiting for rollout to finish: %d out of %d new replicas have been updated",
deployment.Status.UpdatedReplicas, *deployment.Spec.Replicas)
} else if deployment.Status.Replicas > deployment.Status.UpdatedReplicas {
return retriable, fmt.Errorf("waiting for rollout to finish: %d old replicas are pending termination",
deployment.Status.Replicas-deployment.Status.UpdatedReplicas)
} else if deployment.Status.AvailableReplicas < deployment.Status.UpdatedReplicas {
return retriable, fmt.Errorf("waiting for rollout to finish: %d of %d updated replicas are available",
deployment.Status.AvailableReplicas, deployment.Status.UpdatedReplicas)
}
} else {
return true, fmt.Errorf("waiting for rollout to finish: observed deployment generation less then desired generation")
}
return true, nil
}
func (c *Deployer) getDeploymentCondition(
status appsv1.DeploymentStatus,
conditionType appsv1.DeploymentConditionType,
) *appsv1.DeploymentCondition {
for i := range status.Conditions {
c := status.Conditions[i]
if c.Type == conditionType {
return &c
}
}
return nil
}
pkg/canary/status.go
+104
View File
@@ -0,0 +1,104 @@
package canary
import (
"encoding/base64"
"encoding/json"
"fmt"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// SyncStatus encodes the canary pod spec and updates the canary status
func (c *Deployer) SyncStatus(cd *flaggerv1.Canary, status flaggerv1.CanaryStatus) error {
dep, err := c.KubeClient.AppsV1().Deployments(cd.Namespace).Get(cd.Spec.TargetRef.Name, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", cd.Spec.TargetRef.Name, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", cd.Spec.TargetRef.Name, cd.Namespace, err)
}
specJson, err := json.Marshal(dep.Spec.Template.Spec)
if err != nil {
return fmt.Errorf("deployment %s.%s marshal error %v", cd.Spec.TargetRef.Name, cd.Namespace, err)
}
configs, err := c.ConfigTracker.GetConfigRefs(cd)
if err != nil {
return fmt.Errorf("configs query error %v", err)
}
cdCopy := cd.DeepCopy()
cdCopy.Status.Phase = status.Phase
cdCopy.Status.CanaryWeight = status.CanaryWeight
cdCopy.Status.FailedChecks = status.FailedChecks
cdCopy.Status.Iterations = status.Iterations
cdCopy.Status.LastAppliedSpec = base64.StdEncoding.EncodeToString(specJson)
cdCopy.Status.LastTransitionTime = metav1.Now()
cdCopy.Status.TrackedConfigs = configs
cd, err = c.FlaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusFailedChecks updates the canary failed checks counter
func (c *Deployer) SetStatusFailedChecks(cd *flaggerv1.Canary, val int) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.FailedChecks = val
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.FlaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusWeight updates the canary status weight value
func (c *Deployer) SetStatusWeight(cd *flaggerv1.Canary, val int) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.CanaryWeight = val
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.FlaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusIterations updates the canary status iterations value
func (c *Deployer) SetStatusIterations(cd *flaggerv1.Canary, val int) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.Iterations = val
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.FlaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusPhase updates the canary status phase
func (c *Deployer) SetStatusPhase(cd *flaggerv1.Canary, phase flaggerv1.CanaryPhase) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.Phase = phase
cdCopy.Status.LastTransitionTime = metav1.Now()
if phase != flaggerv1.CanaryProgressing {
cdCopy.Status.CanaryWeight = 0
cdCopy.Status.Iterations = 0
}
cd, err := c.FlaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
pkg/controller/tracker.go → pkg/canary/tracker.go
+23 -23
View File
@@ -1,4 +1,4 @@
package controller
package canary
import (
"crypto/sha256"
@@ -16,9 +16,9 @@ import (
// ConfigTracker is managing the operations for Kubernetes ConfigMaps and Secrets
type ConfigTracker struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
KubeClient kubernetes.Interface
FlaggerClient clientset.Interface
Logger *zap.SugaredLogger
}
type ConfigRefType string
@@ -50,7 +50,7 @@ func checksum(data interface{}) string {
// getRefFromConfigMap transforms a Kubernetes ConfigMap into a ConfigRef
// and computes the checksum of the ConfigMap data
func (ct *ConfigTracker) getRefFromConfigMap(name string, namespace string) (*ConfigRef, error) {
config, err := ct.kubeClient.CoreV1().ConfigMaps(namespace).Get(name, metav1.GetOptions{})
config, err := ct.KubeClient.CoreV1().ConfigMaps(namespace).Get(name, metav1.GetOptions{})
if err != nil {
return nil, err
}
@@ -65,7 +65,7 @@ func (ct *ConfigTracker) getRefFromConfigMap(name string, namespace string) (*Co
// getRefFromConfigMap transforms a Kubernetes Secret into a ConfigRef
// and computes the checksum of the Secret data
func (ct *ConfigTracker) getRefFromSecret(name string, namespace string) (*ConfigRef, error) {
secret, err := ct.kubeClient.CoreV1().Secrets(namespace).Get(name, metav1.GetOptions{})
secret, err := ct.KubeClient.CoreV1().Secrets(namespace).Get(name, metav1.GetOptions{})
if err != nil {
return nil, err
}
@@ -75,7 +75,7 @@ func (ct *ConfigTracker) getRefFromSecret(name string, namespace string) (*Confi
secret.Type != corev1.SecretTypeBasicAuth &&
secret.Type != corev1.SecretTypeSSHAuth &&
secret.Type != corev1.SecretTypeTLS {
ct.logger.Debugf("ignoring secret %s.%s type not supported %v", name, namespace, secret.Type)
ct.Logger.Debugf("ignoring secret %s.%s type not supported %v", name, namespace, secret.Type)
return nil, nil
}
@@ -91,7 +91,7 @@ func (ct *ConfigTracker) getRefFromSecret(name string, namespace string) (*Confi
func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]ConfigRef, error) {
res := make(map[string]ConfigRef)
targetName := cd.Spec.TargetRef.Name
targetDep, err := ct.kubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
targetDep, err := ct.KubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return res, fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
@@ -104,7 +104,7 @@ func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]Conf
if cmv := volume.ConfigMap; cmv != nil {
config, err := ct.getRefFromConfigMap(cmv.Name, cd.Namespace)
if err != nil {
ct.logger.Errorf("configMap %s.%s query error %v", cmv.Name, cd.Namespace, err)
ct.Logger.Errorf("configMap %s.%s query error %v", cmv.Name, cd.Namespace, err)
continue
}
if config != nil {
@@ -115,7 +115,7 @@ func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]Conf
if sv := volume.Secret; sv != nil {
secret, err := ct.getRefFromSecret(sv.SecretName, cd.Namespace)
if err != nil {
ct.logger.Errorf("secret %s.%s query error %v", sv.SecretName, cd.Namespace, err)
ct.Logger.Errorf("secret %s.%s query error %v", sv.SecretName, cd.Namespace, err)
continue
}
if secret != nil {
@@ -133,7 +133,7 @@ func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]Conf
name := env.ValueFrom.ConfigMapKeyRef.LocalObjectReference.Name
config, err := ct.getRefFromConfigMap(name, cd.Namespace)
if err != nil {
ct.logger.Errorf("configMap %s.%s query error %v", name, cd.Namespace, err)
ct.Logger.Errorf("configMap %s.%s query error %v", name, cd.Namespace, err)
continue
}
if config != nil {
@@ -143,7 +143,7 @@ func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]Conf
name := env.ValueFrom.SecretKeyRef.LocalObjectReference.Name
secret, err := ct.getRefFromSecret(name, cd.Namespace)
if err != nil {
ct.logger.Errorf("secret %s.%s query error %v", name, cd.Namespace, err)
ct.Logger.Errorf("secret %s.%s query error %v", name, cd.Namespace, err)
continue
}
if secret != nil {
@@ -159,7 +159,7 @@ func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]Conf
name := envFrom.ConfigMapRef.LocalObjectReference.Name
config, err := ct.getRefFromConfigMap(name, cd.Namespace)
if err != nil {
ct.logger.Errorf("configMap %s.%s query error %v", name, cd.Namespace, err)
ct.Logger.Errorf("configMap %s.%s query error %v", name, cd.Namespace, err)
continue
}
if config != nil {
@@ -169,7 +169,7 @@ func (ct *ConfigTracker) GetTargetConfigs(cd *flaggerv1.Canary) (map[string]Conf
name := envFrom.SecretRef.LocalObjectReference.Name
secret, err := ct.getRefFromSecret(name, cd.Namespace)
if err != nil {
ct.logger.Errorf("secret %s.%s query error %v", name, cd.Namespace, err)
ct.Logger.Errorf("secret %s.%s query error %v", name, cd.Namespace, err)
continue
}
if secret != nil {
@@ -221,7 +221,7 @@ func (ct *ConfigTracker) HasConfigChanged(cd *flaggerv1.Canary) (bool, error) {
for _, cfg := range configs {
if trackedConfigs[cfg.GetName()] != cfg.Checksum {
ct.logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).
ct.Logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).
Infof("%s %s has changed", cfg.Type, cfg.Name)
return true, nil
}
@@ -236,7 +236,7 @@ func (ct *ConfigTracker) CreatePrimaryConfigs(cd *flaggerv1.Canary, refs map[str
for _, ref := range refs {
switch ref.Type {
case ConfigRefMap:
config, err := ct.kubeClient.CoreV1().ConfigMaps(cd.Namespace).Get(ref.Name, metav1.GetOptions{})
config, err := ct.KubeClient.CoreV1().ConfigMaps(cd.Namespace).Get(ref.Name, metav1.GetOptions{})
if err != nil {
return err
}
@@ -258,10 +258,10 @@ func (ct *ConfigTracker) CreatePrimaryConfigs(cd *flaggerv1.Canary, refs map[str
}
// update or insert primary ConfigMap
_, err = ct.kubeClient.CoreV1().ConfigMaps(cd.Namespace).Update(primaryConfigMap)
_, err = ct.KubeClient.CoreV1().ConfigMaps(cd.Namespace).Update(primaryConfigMap)
if err != nil {
if errors.IsNotFound(err) {
_, err = ct.kubeClient.CoreV1().ConfigMaps(cd.Namespace).Create(primaryConfigMap)
_, err = ct.KubeClient.CoreV1().ConfigMaps(cd.Namespace).Create(primaryConfigMap)
if err != nil {
return err
}
@@ -270,10 +270,10 @@ func (ct *ConfigTracker) CreatePrimaryConfigs(cd *flaggerv1.Canary, refs map[str
}
}
ct.logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).
ct.Logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).
Infof("ConfigMap %s synced", primaryConfigMap.GetName())
case ConfigRefSecret:
secret, err := ct.kubeClient.CoreV1().Secrets(cd.Namespace).Get(ref.Name, metav1.GetOptions{})
secret, err := ct.KubeClient.CoreV1().Secrets(cd.Namespace).Get(ref.Name, metav1.GetOptions{})
if err != nil {
return err
}
@@ -296,10 +296,10 @@ func (ct *ConfigTracker) CreatePrimaryConfigs(cd *flaggerv1.Canary, refs map[str
}
// update or insert primary Secret
_, err = ct.kubeClient.CoreV1().Secrets(cd.Namespace).Update(primarySecret)
_, err = ct.KubeClient.CoreV1().Secrets(cd.Namespace).Update(primarySecret)
if err != nil {
if errors.IsNotFound(err) {
_, err = ct.kubeClient.CoreV1().Secrets(cd.Namespace).Create(primarySecret)
_, err = ct.KubeClient.CoreV1().Secrets(cd.Namespace).Create(primarySecret)
if err != nil {
return err
}
@@ -308,7 +308,7 @@ func (ct *ConfigTracker) CreatePrimaryConfigs(cd *flaggerv1.Canary, refs map[str
}
}
ct.logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).
ct.Logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).
Infof("Secret %s synced", primarySecret.GetName())
}
}
pkg/controller/controller.go
+12 -9
View File
@@ -5,6 +5,7 @@ import (
"sync"
"time"
"github.com/weaveworks/flagger/pkg/canary"
"github.com/weaveworks/flagger/pkg/metrics"
"github.com/weaveworks/flagger/pkg/router"
@@ -43,7 +44,7 @@ type Controller struct {
logger *zap.SugaredLogger
canaries *sync.Map
jobs map[string]CanaryJob
deployer CanaryDeployer
deployer canary.Deployer
observer metrics.Observer
recorder metrics.Recorder
notifier *notifier.Slack
@@ -63,6 +64,7 @@ func NewController(
routerFactory *router.Factory,
meshProvider string,
version string,
labels []string,
) *Controller {
logger.Debug("Creating event broadcaster")
flaggerscheme.AddToScheme(scheme.Scheme)
@@ -74,14 +76,15 @@ func NewController(
eventRecorder := eventBroadcaster.NewRecorder(
scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
deployer := CanaryDeployer{
logger: logger,
kubeClient: kubeClient,
flaggerClient: flaggerClient,
configTracker: ConfigTracker{
logger: logger,
kubeClient: kubeClient,
flaggerClient: flaggerClient,
deployer := canary.Deployer{
Logger: logger,
KubeClient: kubeClient,
FlaggerClient: flaggerClient,
Labels: labels,
ConfigTracker: canary.ConfigTracker{
Logger: logger,
KubeClient: kubeClient,
FlaggerClient: flaggerClient,
},
}
pkg/controller/controller_test.go
+17 -15
View File
@@ -4,10 +4,11 @@ import (
"github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
istiov1alpha1 "github.com/weaveworks/flagger/pkg/apis/istio/common/v1alpha1"
istiov1alpha3 "github.com/weaveworks/flagger/pkg/apis/istio/v1alpha3"
"github.com/weaveworks/flagger/pkg/canary"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
fakeFlagger "github.com/weaveworks/flagger/pkg/client/clientset/versioned/fake"
informers "github.com/weaveworks/flagger/pkg/client/informers/externalversions"
"github.com/weaveworks/flagger/pkg/logging"
"github.com/weaveworks/flagger/pkg/logger"
"github.com/weaveworks/flagger/pkg/metrics"
"github.com/weaveworks/flagger/pkg/router"
"go.uber.org/zap"
@@ -34,7 +35,7 @@ type Mocks struct {
kubeClient kubernetes.Interface
meshClient clientset.Interface
flaggerClient clientset.Interface
deployer CanaryDeployer
deployer canary.Deployer
observer metrics.Observer
ctrl *Controller
logger *zap.SugaredLogger
@@ -43,11 +44,11 @@ type Mocks struct {
func SetupMocks(abtest bool) Mocks {
// init canary
canary := newTestCanary()
c := newTestCanary()
if abtest {
canary = newTestCanaryAB()
c = newTestCanaryAB()
}
flaggerClient := fakeFlagger.NewSimpleClientset(canary)
flaggerClient := fakeFlagger.NewSimpleClientset(c)
// init kube clientset and register mock objects
kubeClient := fake.NewSimpleClientset(
@@ -61,17 +62,18 @@ func SetupMocks(abtest bool) Mocks {
NewTestSecretVol(),
)
logger, _ := logging.NewLogger("debug")
logger, _ := logger.NewLogger("debug")
// init controller helpers
deployer := CanaryDeployer{
flaggerClient: flaggerClient,
kubeClient: kubeClient,
logger: logger,
configTracker: ConfigTracker{
logger: logger,
kubeClient: kubeClient,
flaggerClient: flaggerClient,
deployer := canary.Deployer{
Logger: logger,
KubeClient: kubeClient,
FlaggerClient: flaggerClient,
Labels: []string{"app", "name"},
ConfigTracker: canary.ConfigTracker{
Logger: logger,
KubeClient: kubeClient,
FlaggerClient: flaggerClient,
},
}
observer := metrics.NewObserver("fake")
@@ -102,7 +104,7 @@ func SetupMocks(abtest bool) Mocks {
meshRouter := rf.MeshRouter("istio")
return Mocks{
canary: canary,
canary: c,
observer: observer,
deployer: deployer,
logger: logger,
pkg/controller/deployer.go
-566
View File
@@ -1,566 +0,0 @@
package controller
import (
"crypto/rand"
"encoding/base64"
"encoding/json"
"fmt"
"io"
"time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
"go.uber.org/zap"
appsv1 "k8s.io/api/apps/v1"
hpav1 "k8s.io/api/autoscaling/v2beta1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/kubernetes"
)
// CanaryDeployer is managing the operations for Kubernetes deployment kind
type CanaryDeployer struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
configTracker ConfigTracker
}
// Promote copies the pod spec, secrets and config maps from canary to primary
func (c *CanaryDeployer) Promote(cd *flaggerv1.Canary) error {
targetName := cd.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
canary, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
primary, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", primaryName, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", primaryName, cd.Namespace, err)
}
// promote secrets and config maps
configRefs, err := c.configTracker.GetTargetConfigs(cd)
if err != nil {
return err
}
if err := c.configTracker.CreatePrimaryConfigs(cd, configRefs); err != nil {
return err
}
primaryCopy := primary.DeepCopy()
primaryCopy.Spec.ProgressDeadlineSeconds = canary.Spec.ProgressDeadlineSeconds
primaryCopy.Spec.MinReadySeconds = canary.Spec.MinReadySeconds
primaryCopy.Spec.RevisionHistoryLimit = canary.Spec.RevisionHistoryLimit
primaryCopy.Spec.Strategy = canary.Spec.Strategy
// update spec with primary secrets and config maps
primaryCopy.Spec.Template.Spec = c.configTracker.ApplyPrimaryConfigs(canary.Spec.Template.Spec, configRefs)
// update pod annotations to ensure a rolling update
annotations, err := c.makeAnnotations(canary.Spec.Template.Annotations)
if err != nil {
return err
}
primaryCopy.Spec.Template.Annotations = annotations
primaryCopy.Spec.Template.Labels = makePrimaryLabels(canary.Spec.Template.Labels, primaryName)
_, err = c.kubeClient.AppsV1().Deployments(cd.Namespace).Update(primaryCopy)
if err != nil {
return fmt.Errorf("updating deployment %s.%s template spec failed: %v",
primaryCopy.GetName(), primaryCopy.Namespace, err)
}
return nil
}
// IsPrimaryReady checks the primary deployment status and returns an error if
// the deployment is in the middle of a rolling update or if the pods are unhealthy
// it will return a non retriable error if the rolling update is stuck
func (c *CanaryDeployer) IsPrimaryReady(cd *flaggerv1.Canary) (bool, error) {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
primary, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return true, fmt.Errorf("deployment %s.%s not found", primaryName, cd.Namespace)
}
return true, fmt.Errorf("deployment %s.%s query error %v", primaryName, cd.Namespace, err)
}
retriable, err := c.isDeploymentReady(primary, cd.GetProgressDeadlineSeconds())
if err != nil {
return retriable, fmt.Errorf("Halt advancement %s.%s %s", primaryName, cd.Namespace, err.Error())
}
if primary.Spec.Replicas == int32p(0) {
return true, fmt.Errorf("Halt %s.%s advancement primary deployment is scaled to zero",
cd.Name, cd.Namespace)
}
return true, nil
}
// IsCanaryReady checks the primary deployment status and returns an error if
// the deployment is in the middle of a rolling update or if the pods are unhealthy
// it will return a non retriable error if the rolling update is stuck
func (c *CanaryDeployer) IsCanaryReady(cd *flaggerv1.Canary) (bool, error) {
targetName := cd.Spec.TargetRef.Name
canary, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return true, fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return true, fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
retriable, err := c.isDeploymentReady(canary, cd.GetProgressDeadlineSeconds())
if err != nil {
if retriable {
return retriable, fmt.Errorf("Halt advancement %s.%s %s", targetName, cd.Namespace, err.Error())
} else {
return retriable, fmt.Errorf("deployment does not have minimum availability for more than %vs",
cd.GetProgressDeadlineSeconds())
}
}
return true, nil
}
// IsNewSpec returns true if the canary deployment pod spec has changed
func (c *CanaryDeployer) IsNewSpec(cd *flaggerv1.Canary) (bool, error) {
targetName := cd.Spec.TargetRef.Name
canary, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return false, fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return false, fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
if cd.Status.LastAppliedSpec == "" {
return true, nil
}
newSpec := &canary.Spec.Template.Spec
oldSpecJson, err := base64.StdEncoding.DecodeString(cd.Status.LastAppliedSpec)
if err != nil {
return false, fmt.Errorf("%s.%s decode error %v", cd.Name, cd.Namespace, err)
}
oldSpec := &corev1.PodSpec{}
err = json.Unmarshal(oldSpecJson, oldSpec)
if err != nil {
return false, fmt.Errorf("%s.%s unmarshal error %v", cd.Name, cd.Namespace, err)
}
if diff := cmp.Diff(*newSpec, *oldSpec, cmpopts.IgnoreUnexported(resource.Quantity{})); diff != "" {
//fmt.Println(diff)
return true, nil
}
return false, nil
}
// ShouldAdvance determines if the canary analysis can proceed
func (c *CanaryDeployer) ShouldAdvance(cd *flaggerv1.Canary) (bool, error) {
if cd.Status.LastAppliedSpec == "" || cd.Status.Phase == flaggerv1.CanaryProgressing {
return true, nil
}
newDep, err := c.IsNewSpec(cd)
if err != nil {
return false, err
}
if newDep {
return newDep, nil
}
newCfg, err := c.configTracker.HasConfigChanged(cd)
if err != nil {
return false, err
}
return newCfg, nil
}
// SetStatusFailedChecks updates the canary failed checks counter
func (c *CanaryDeployer) SetStatusFailedChecks(cd *flaggerv1.Canary, val int) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.FailedChecks = val
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusWeight updates the canary status weight value
func (c *CanaryDeployer) SetStatusWeight(cd *flaggerv1.Canary, val int) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.CanaryWeight = val
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusIterations updates the canary status iterations value
func (c *CanaryDeployer) SetStatusIterations(cd *flaggerv1.Canary, val int) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.Iterations = val
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusWeight updates the canary status weight value
func (c *CanaryDeployer) IncrementStatusIterations(cd *flaggerv1.Canary) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.Iterations = cdCopy.Status.Iterations + 1
cdCopy.Status.LastTransitionTime = metav1.Now()
cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SetStatusPhase updates the canary status phase
func (c *CanaryDeployer) SetStatusPhase(cd *flaggerv1.Canary, phase flaggerv1.CanaryPhase) error {
cdCopy := cd.DeepCopy()
cdCopy.Status.Phase = phase
cdCopy.Status.LastTransitionTime = metav1.Now()
if phase != flaggerv1.CanaryProgressing {
cdCopy.Status.CanaryWeight = 0
cdCopy.Status.Iterations = 0
}
cd, err := c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// SyncStatus encodes the canary pod spec and updates the canary status
func (c *CanaryDeployer) SyncStatus(cd *flaggerv1.Canary, status flaggerv1.CanaryStatus) error {
dep, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(cd.Spec.TargetRef.Name, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", cd.Spec.TargetRef.Name, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", cd.Spec.TargetRef.Name, cd.Namespace, err)
}
specJson, err := json.Marshal(dep.Spec.Template.Spec)
if err != nil {
return fmt.Errorf("deployment %s.%s marshal error %v", cd.Spec.TargetRef.Name, cd.Namespace, err)
}
configs, err := c.configTracker.GetConfigRefs(cd)
if err != nil {
return fmt.Errorf("configs query error %v", err)
}
cdCopy := cd.DeepCopy()
cdCopy.Status.Phase = status.Phase
cdCopy.Status.CanaryWeight = status.CanaryWeight
cdCopy.Status.FailedChecks = status.FailedChecks
cdCopy.Status.Iterations = status.Iterations
cdCopy.Status.LastAppliedSpec = base64.StdEncoding.EncodeToString(specJson)
cdCopy.Status.LastTransitionTime = metav1.Now()
cdCopy.Status.TrackedConfigs = configs
cd, err = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
if err != nil {
return fmt.Errorf("canary %s.%s status update error %v", cdCopy.Name, cdCopy.Namespace, err)
}
return nil
}
// Scale sets the canary deployment replicas
func (c *CanaryDeployer) Scale(cd *flaggerv1.Canary, replicas int32) error {
targetName := cd.Spec.TargetRef.Name
dep, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found", targetName, cd.Namespace)
}
return fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
depCopy := dep.DeepCopy()
depCopy.Spec.Replicas = int32p(replicas)
_, err = c.kubeClient.AppsV1().Deployments(dep.Namespace).Update(depCopy)
if err != nil {
return fmt.Errorf("scaling %s.%s to %v failed: %v", depCopy.GetName(), depCopy.Namespace, replicas, err)
}
return nil
}
// Reconcile creates the primary deployment and hpa
// and scales to zero the canary deployment
func (c *CanaryDeployer) Sync(cd *flaggerv1.Canary) error {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
if err := c.createPrimaryDeployment(cd); err != nil {
return fmt.Errorf("creating deployment %s.%s failed: %v", primaryName, cd.Namespace, err)
}
if cd.Status.Phase == "" {
c.logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).Infof("Scaling down %s.%s", cd.Spec.TargetRef.Name, cd.Namespace)
if err := c.Scale(cd, 0); err != nil {
return err
}
}
if cd.Spec.AutoscalerRef != nil && cd.Spec.AutoscalerRef.Kind == "HorizontalPodAutoscaler" {
if err := c.createPrimaryHpa(cd); err != nil {
return fmt.Errorf("creating hpa %s.%s failed: %v", primaryName, cd.Namespace, err)
}
}
return nil
}
func (c *CanaryDeployer) createPrimaryDeployment(cd *flaggerv1.Canary) error {
targetName := cd.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
canaryDep, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("deployment %s.%s not found, retrying", targetName, cd.Namespace)
}
return err
}
if appSel, ok := canaryDep.Spec.Selector.MatchLabels["app"]; !ok || appSel != canaryDep.Name {
return fmt.Errorf("invalid label selector! Deployment %s.%s spec.selector.matchLabels must contain selector 'app: %s'",
targetName, cd.Namespace, targetName)
}
primaryDep, err := c.kubeClient.AppsV1().Deployments(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if errors.IsNotFound(err) {
// create primary secrets and config maps
configRefs, err := c.configTracker.GetTargetConfigs(cd)
if err != nil {
return err
}
if err := c.configTracker.CreatePrimaryConfigs(cd, configRefs); err != nil {
return err
}
annotations, err := c.makeAnnotations(canaryDep.Spec.Template.Annotations)
if err != nil {
return err
}
replicas := int32(1)
if canaryDep.Spec.Replicas != nil && *canaryDep.Spec.Replicas > 0 {
replicas = *canaryDep.Spec.Replicas
}
// create primary deployment
primaryDep = &appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: primaryName,
Labels: canaryDep.Labels,
Namespace: cd.Namespace,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(cd, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: appsv1.DeploymentSpec{
ProgressDeadlineSeconds: canaryDep.Spec.ProgressDeadlineSeconds,
MinReadySeconds: canaryDep.Spec.MinReadySeconds,
RevisionHistoryLimit: canaryDep.Spec.RevisionHistoryLimit,
Replicas: int32p(replicas),
Strategy: canaryDep.Spec.Strategy,
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{
"app": primaryName,
},
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: makePrimaryLabels(canaryDep.Spec.Template.Labels, primaryName),
Annotations: annotations,
},
// update spec with the primary secrets and config maps
Spec: c.configTracker.ApplyPrimaryConfigs(canaryDep.Spec.Template.Spec, configRefs),
},
},
}
_, err = c.kubeClient.AppsV1().Deployments(cd.Namespace).Create(primaryDep)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).Infof("Deployment %s.%s created", primaryDep.GetName(), cd.Namespace)
}
return nil
}
func (c *CanaryDeployer) createPrimaryHpa(cd *flaggerv1.Canary) error {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
hpa, err := c.kubeClient.AutoscalingV2beta1().HorizontalPodAutoscalers(cd.Namespace).Get(cd.Spec.AutoscalerRef.Name, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("HorizontalPodAutoscaler %s.%s not found, retrying",
cd.Spec.AutoscalerRef.Name, cd.Namespace)
}
return err
}
primaryHpaName := fmt.Sprintf("%s-primary", cd.Spec.AutoscalerRef.Name)
primaryHpa, err := c.kubeClient.AutoscalingV2beta1().HorizontalPodAutoscalers(cd.Namespace).Get(primaryHpaName, metav1.GetOptions{})
if errors.IsNotFound(err) {
primaryHpa = &hpav1.HorizontalPodAutoscaler{
ObjectMeta: metav1.ObjectMeta{
Name: primaryHpaName,
Namespace: cd.Namespace,
Labels: hpa.Labels,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(cd, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: hpav1.HorizontalPodAutoscalerSpec{
ScaleTargetRef: hpav1.CrossVersionObjectReference{
Name: primaryName,
Kind: hpa.Spec.ScaleTargetRef.Kind,
APIVersion: hpa.Spec.ScaleTargetRef.APIVersion,
},
MinReplicas: hpa.Spec.MinReplicas,
MaxReplicas: hpa.Spec.MaxReplicas,
Metrics: hpa.Spec.Metrics,
},
}
_, err = c.kubeClient.AutoscalingV2beta1().HorizontalPodAutoscalers(cd.Namespace).Create(primaryHpa)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", cd.Name, cd.Namespace)).Infof("HorizontalPodAutoscaler %s.%s created", primaryHpa.GetName(), cd.Namespace)
}
return nil
}
// isDeploymentReady determines if a deployment is ready by checking the status conditions
// if a deployment has exceeded the progress deadline it returns a non retriable error
func (c *CanaryDeployer) isDeploymentReady(deployment *appsv1.Deployment, deadline int) (bool, error) {
retriable := true
if deployment.Generation <= deployment.Status.ObservedGeneration {
progress := c.getDeploymentCondition(deployment.Status, appsv1.DeploymentProgressing)
if progress != nil {
// Determine if the deployment is stuck by checking if there is a minimum replicas unavailable condition
// and if the last update time exceeds the deadline
available := c.getDeploymentCondition(deployment.Status, appsv1.DeploymentAvailable)
if available != nil && available.Status == "False" && available.Reason == "MinimumReplicasUnavailable" {
from := available.LastUpdateTime
delta := time.Duration(deadline) * time.Second
retriable = !from.Add(delta).Before(time.Now())
}
}
if progress != nil && progress.Reason == "ProgressDeadlineExceeded" {
return false, fmt.Errorf("deployment %q exceeded its progress deadline", deployment.GetName())
} else if deployment.Spec.Replicas != nil && deployment.Status.UpdatedReplicas < *deployment.Spec.Replicas {
return retriable, fmt.Errorf("waiting for rollout to finish: %d out of %d new replicas have been updated",
deployment.Status.UpdatedReplicas, *deployment.Spec.Replicas)
} else if deployment.Status.Replicas > deployment.Status.UpdatedReplicas {
return retriable, fmt.Errorf("waiting for rollout to finish: %d old replicas are pending termination",
deployment.Status.Replicas-deployment.Status.UpdatedReplicas)
} else if deployment.Status.AvailableReplicas < deployment.Status.UpdatedReplicas {
return retriable, fmt.Errorf("waiting for rollout to finish: %d of %d updated replicas are available",
deployment.Status.AvailableReplicas, deployment.Status.UpdatedReplicas)
}
} else {
return true, fmt.Errorf("waiting for rollout to finish: observed deployment generation less then desired generation")
}
return true, nil
}
func (c *CanaryDeployer) getDeploymentCondition(
status appsv1.DeploymentStatus,
conditionType appsv1.DeploymentConditionType,
) *appsv1.DeploymentCondition {
for i := range status.Conditions {
c := status.Conditions[i]
if c.Type == conditionType {
return &c
}
}
return nil
}
// makeAnnotations appends an unique ID to annotations map
func (c *CanaryDeployer) makeAnnotations(annotations map[string]string) (map[string]string, error) {
idKey := "flagger-id"
res := make(map[string]string)
uuid := make([]byte, 16)
n, err := io.ReadFull(rand.Reader, uuid)
if n != len(uuid) || err != nil {
return res, err
}
uuid[8] = uuid[8]&^0xc0 | 0x80
uuid[6] = uuid[6]&^0xf0 | 0x40
id := fmt.Sprintf("%x-%x-%x-%x-%x", uuid[0:4], uuid[4:6], uuid[6:8], uuid[8:10], uuid[10:])
for k, v := range annotations {
if k != idKey {
res[k] = v
}
}
res[idKey] = id
return res, nil
}
func makePrimaryLabels(labels map[string]string, primaryName string) map[string]string {
idKey := "app"
res := make(map[string]string)
for k, v := range labels {
if k != idKey {
res[k] = v
}
}
res[idKey] = primaryName
return res
}
pkg/controller/scheduler.go
+139 -50
View File
@@ -91,7 +91,8 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
// create primary deployment and hpa if needed
if err := c.deployer.Sync(cd); err != nil {
label, err := c.deployer.Initialize(cd)
if err != nil {
c.recordEventWarningf(cd, "%v", err)
return
}
@@ -100,7 +101,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
meshRouter := c.routerFactory.MeshRouter(c.meshProvider)
// create or update ClusterIP services
if err := c.routerFactory.KubernetesRouter().Reconcile(cd); err != nil {
if err := c.routerFactory.KubernetesRouter(label).Reconcile(cd); err != nil {
c.recordEventWarningf(cd, "%v", err)
return
}
@@ -111,7 +112,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
return
}
shouldAdvance, err := c.deployer.ShouldAdvance(cd)
shouldAdvance, err := c.shouldAdvance(cd)
if err != nil {
c.recordEventWarningf(cd, "%v", err)
return
@@ -240,6 +241,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
}
c.recorder.SetStatus(cd, flaggerv1.CanaryFailed)
c.runPostRolloutHooks(cd, flaggerv1.CanaryFailed)
return
}
@@ -247,6 +249,15 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
// skip check if no traffic is routed to canary
if canaryWeight == 0 {
c.recordEventInfof(cd, "Starting canary analysis for %s.%s", cd.Spec.TargetRef.Name, cd.Namespace)
// run pre-rollout web hooks
if ok := c.runPreRolloutHooks(cd); !ok {
if err := c.deployer.SetStatusFailedChecks(cd, cd.Status.FailedChecks+1); err != nil {
c.recordEventWarningf(cd, "%v", err)
return
}
return
}
} else {
if ok := c.analyseCanary(cd); !ok {
if err := c.deployer.SetStatusFailedChecks(cd, cd.Status.FailedChecks+1); err != nil {
@@ -314,6 +325,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
return
}
c.recorder.SetStatus(cd, flaggerv1.CanarySucceeded)
c.runPostRolloutHooks(cd, flaggerv1.CanarySucceeded)
c.sendNotification(cd, "Canary analysis completed successfully, promotion finished.",
false, false)
return
@@ -380,6 +392,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
return
}
c.recorder.SetStatus(cd, flaggerv1.CanarySucceeded)
c.runPostRolloutHooks(cd, flaggerv1.CanarySucceeded)
c.sendNotification(cd, "Canary analysis completed successfully, promotion finished.",
false, false)
}
@@ -429,6 +442,28 @@ func (c *Controller) shouldSkipAnalysis(cd *flaggerv1.Canary, meshRouter router.
return true
}
func (c *Controller) shouldAdvance(cd *flaggerv1.Canary) (bool, error) {
if cd.Status.LastAppliedSpec == "" || cd.Status.Phase == flaggerv1.CanaryProgressing {
return true, nil
}
newDep, err := c.deployer.HasDeploymentChanged(cd)
if err != nil {
return false, err
}
if newDep {
return newDep, nil
}
newCfg, err := c.deployer.ConfigTracker.HasConfigChanged(cd)
if err != nil {
return false, err
}
return newCfg, nil
}
func (c *Controller) checkCanaryStatus(cd *flaggerv1.Canary, shouldAdvance bool) bool {
c.recorder.SetStatus(cd, cd.Status.Phase)
if cd.Status.Phase == flaggerv1.CanaryProgressing {
@@ -467,24 +502,57 @@ func (c *Controller) checkCanaryStatus(cd *flaggerv1.Canary, shouldAdvance bool)
func (c *Controller) hasCanaryRevisionChanged(cd *flaggerv1.Canary) bool {
if cd.Status.Phase == flaggerv1.CanaryProgressing {
if diff, _ := c.deployer.IsNewSpec(cd); diff {
if diff, _ := c.deployer.HasDeploymentChanged(cd); diff {
return true
}
if diff, _ := c.deployer.configTracker.HasConfigChanged(cd); diff {
if diff, _ := c.deployer.ConfigTracker.HasConfigChanged(cd); diff {
return true
}
}
return false
}
func (c *Controller) runPreRolloutHooks(canary *flaggerv1.Canary) bool {
for _, webhook := range canary.Spec.CanaryAnalysis.Webhooks {
if webhook.Type == flaggerv1.PreRolloutHook {
err := CallWebhook(canary.Name, canary.Namespace, flaggerv1.CanaryProgressing, webhook)
if err != nil {
c.recordEventWarningf(canary, "Halt %s.%s advancement pre-rollout check %s failed %v",
canary.Name, canary.Namespace, webhook.Name, err)
return false
} else {
c.recordEventInfof(canary, "Pre-rollout check %s passed", webhook.Name)
}
}
}
return true
}
func (c *Controller) runPostRolloutHooks(canary *flaggerv1.Canary, phase flaggerv1.CanaryPhase) bool {
for _, webhook := range canary.Spec.CanaryAnalysis.Webhooks {
if webhook.Type == flaggerv1.PostRolloutHook {
err := CallWebhook(canary.Name, canary.Namespace, phase, webhook)
if err != nil {
c.recordEventWarningf(canary, "Post-rollout hook %s failed %v", webhook.Name, err)
return false
} else {
c.recordEventInfof(canary, "Post-rollout check %s passed", webhook.Name)
}
}
}
return true
}
func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool {
// run external checks
for _, webhook := range r.Spec.CanaryAnalysis.Webhooks {
err := CallWebhook(r.Name, r.Namespace, webhook)
if err != nil {
c.recordEventWarningf(r, "Halt %s.%s advancement external check %s failed %v",
r.Name, r.Namespace, webhook.Name, err)
return false
if webhook.Type == "" || webhook.Type == flaggerv1.RolloutHook {
err := CallWebhook(r.Name, r.Namespace, flaggerv1.CanaryProgressing, webhook)
if err != nil {
c.recordEventWarningf(r, "Halt %s.%s advancement external check %s failed %v",
r.Name, r.Namespace, webhook.Name, err)
return false
}
}
}
@@ -494,56 +562,77 @@ func (c *Controller) analyseCanary(r *flaggerv1.Canary) bool {
metric.Interval = r.GetMetricInterval()
}
if metric.Name == "envoy_cluster_upstream_rq" {
val, err := c.observer.GetEnvoySuccessRate(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
if strings.Contains(err.Error(), "no values found") {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
// App Mesh checks
if c.meshProvider == "appmesh" {
if metric.Name == "request-success-rate" || metric.Name == "envoy_cluster_upstream_rq" {
val, err := c.observer.GetEnvoySuccessRate(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
if strings.Contains(err.Error(), "no values found") {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
}
return false
}
return false
}
if float64(metric.Threshold) > val {
c.recordEventWarningf(r, "Halt %s.%s advancement success rate %.2f%% < %v%%",
r.Name, r.Namespace, val, metric.Threshold)
return false
}
}
if metric.Name == "istio_requests_total" {
val, err := c.observer.GetIstioSuccessRate(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
if strings.Contains(err.Error(), "no values found") {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
if float64(metric.Threshold) > val {
c.recordEventWarningf(r, "Halt %s.%s advancement success rate %.2f%% < %v%%",
r.Name, r.Namespace, val, metric.Threshold)
return false
}
return false
}
if float64(metric.Threshold) > val {
c.recordEventWarningf(r, "Halt %s.%s advancement success rate %.2f%% < %v%%",
r.Name, r.Namespace, val, metric.Threshold)
return false
if metric.Name == "request-duration" || metric.Name == "envoy_cluster_upstream_rq_time_bucket" {
val, err := c.observer.GetEnvoyRequestDuration(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
return false
}
t := time.Duration(metric.Threshold) * time.Millisecond
if val > t {
c.recordEventWarningf(r, "Halt %s.%s advancement request duration %v > %v",
r.Name, r.Namespace, val, t)
return false
}
}
}
if metric.Name == "istio_request_duration_seconds_bucket" {
val, err := c.observer.GetIstioRequestDuration(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
return false
// Istio checks
if c.meshProvider == "istio" {
if metric.Name == "request-success-rate" || metric.Name == "istio_requests_total" {
val, err := c.observer.GetIstioSuccessRate(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
if strings.Contains(err.Error(), "no values found") {
c.recordEventWarningf(r, "Halt advancement no values found for metric %s probably %s.%s is not receiving traffic",
metric.Name, r.Spec.TargetRef.Name, r.Namespace)
} else {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
}
return false
}
if float64(metric.Threshold) > val {
c.recordEventWarningf(r, "Halt %s.%s advancement success rate %.2f%% < %v%%",
r.Name, r.Namespace, val, metric.Threshold)
return false
}
}
t := time.Duration(metric.Threshold) * time.Millisecond
if val > t {
c.recordEventWarningf(r, "Halt %s.%s advancement request duration %v > %v",
r.Name, r.Namespace, val, t)
return false
if metric.Name == "request-duration" || metric.Name == "istio_request_duration_seconds_bucket" {
val, err := c.observer.GetIstioRequestDuration(r.Spec.TargetRef.Name, r.Namespace, metric.Name, metric.Interval)
if err != nil {
c.recordEventErrorf(r, "Metrics server %s query failed: %v", c.observer.GetMetricsServer(), err)
return false
}
t := time.Duration(metric.Threshold) * time.Millisecond
if val > t {
c.recordEventWarningf(r, "Halt %s.%s advancement request duration %v > %v",
r.Name, r.Namespace, val, t)
return false
}
}
}
// custom checks
if metric.Query != "" {
val, err := c.observer.GetScalar(metric.Query)
if err != nil {
pkg/controller/webhook.go
+2 -1
View File
@@ -15,10 +15,11 @@ import (
// CallWebhook does a HTTP POST to an external service and
// returns an error if the response status code is non-2xx
func CallWebhook(name string, namespace string, w flaggerv1.CanaryWebhook) error {
func CallWebhook(name string, namespace string, phase flaggerv1.CanaryPhase, w flaggerv1.CanaryWebhook) error {
payload := flaggerv1.CanaryWebhookPayload{
Name: name,
Namespace: namespace,
Phase: phase,
}
if w.Metadata != nil {
pkg/controller/webhook_test.go
+2 -2
View File
@@ -19,7 +19,7 @@ func TestCallWebhook(t *testing.T) {
Metadata: &map[string]string{"key1": "val1"},
}
err := CallWebhook("podinfo", "default", hook)
err := CallWebhook("podinfo", "default", flaggerv1.CanaryProgressing, hook)
if err != nil {
t.Fatal(err.Error())
}
@@ -35,7 +35,7 @@ func TestCallWebhook_StatusCode(t *testing.T) {
URL: ts.URL,
}
err := CallWebhook("podinfo", "default", hook)
err := CallWebhook("podinfo", "default", flaggerv1.CanaryProgressing, hook)
if err == nil {
t.Errorf("Got no error wanted %v", http.StatusInternalServerError)
}
pkg/loadtester/bats.go
+39
View File
@@ -0,0 +1,39 @@
package loadtester
import (
"context"
"fmt"
"os/exec"
)
const TaskTypeBats = "bats"
type BatsTask struct {
TaskBase
command string
logCmdOutput bool
}
func (task *BatsTask) Hash() string {
return hash(task.canary + task.command)
}
func (task *BatsTask) Run(ctx context.Context) (bool, error) {
cmd := exec.CommandContext(ctx, "bash", "-c", task.command)
out, err := cmd.CombinedOutput()
if err != nil {
task.logger.With("canary", task.canary).Errorf("command failed %s %v %s", task.command, err, out)
return false, fmt.Errorf(" %v %v", err, out)
} else {
if task.logCmdOutput {
fmt.Printf("%s\n", out)
}
task.logger.With("canary", task.canary).Infof("command finished %s", task.command)
}
return true, nil
}
func (task *BatsTask) String() string {
return task.command
}
pkg/loadtester/runner_test.go
+2 -2
View File
@@ -1,14 +1,14 @@
package loadtester
import (
"github.com/weaveworks/flagger/pkg/logging"
"github.com/weaveworks/flagger/pkg/logger"
"testing"
"time"
)
func TestTaskRunner_Start(t *testing.T) {
stop := make(chan struct{})
logger, _ := logging.NewLogger("debug")
logger, _ := logger.NewLogger("debug")
tr := NewTaskRunner(logger, time.Hour)
go tr.Start(10*time.Millisecond, stop)
pkg/loadtester/server.go
+21
View File
@@ -44,6 +44,27 @@ func ListenAndServe(port string, timeout time.Duration, logger *zap.SugaredLogge
if !ok {
typ = TaskTypeShell
}
// run bats command (blocking task)
if typ == TaskTypeBats {
bats := BatsTask{
command: payload.Metadata["cmd"],
logCmdOutput: taskRunner.logCmdOutput,
}
ctx, cancel := context.WithTimeout(context.Background(), taskRunner.timeout)
defer cancel()
ok, err := bats.Run(ctx)
if !ok {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
}
w.WriteHeader(http.StatusOK)
return
}
taskFactory, ok := GetTaskFactory(typ)
if !ok {
w.WriteHeader(http.StatusBadRequest)
pkg/loadtester/task.go
+1
View File
@@ -24,6 +24,7 @@ type TaskBase struct {
func (task *TaskBase) Canary() string {
return task.canary
}
func hash(str string) string {
fnvHash := fnv.New32()
fnvBytes := fnvHash.Sum([]byte(str))
pkg/loadtester/task_ngrinder_test.go
+2 -2
View File
@@ -3,7 +3,7 @@ package loadtester
import (
"context"
"fmt"
"github.com/weaveworks/flagger/pkg/logging"
"github.com/weaveworks/flagger/pkg/logger"
"gopkg.in/h2non/gock.v1"
"testing"
"time"
@@ -12,7 +12,7 @@ import (
func TestTaskNGrinder(t *testing.T) {
server := "http://ngrinder:8080"
cloneId := "960"
logger, _ := logging.NewLoggerWithEncoding("debug", "console")
logger, _ := logger.NewLoggerWithEncoding("debug", "console")
canary := "podinfo.default"
taskFactory, ok := GetTaskFactory(TaskTypeNGrinder)
if !ok {
pkg/logging/logger.go → pkg/logger/logger.go
+1 -13
View File
@@ -1,9 +1,6 @@
package logging
package logger
import (
"fmt"
"os"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
)
@@ -64,12 +61,3 @@ func NewLoggerWithEncoding(logLevel, zapEncoding string) (*zap.SugaredLogger, er
}
return logger.Sugar(), nil
}
// Console writes to stdout if the console env var exists
func Console(a ...interface{}) (n int, err error) {
if os.Getenv("console") != "" {
return fmt.Fprintln(os.Stdout, a...)
}
return 0, nil
}
pkg/metrics/envoy.go
+54
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"net/url"
"strconv"
"time"
)
const envoySuccessRateQuery = `
@@ -63,3 +64,56 @@ func (c *Observer) GetEnvoySuccessRate(name string, namespace string, metric str
}
return *rate, nil
}
const envoyRequestDurationQuery = `
histogram_quantile(0.99, sum(rate(
envoy_cluster_upstream_rq_time_bucket{kubernetes_namespace="{{ .Namespace }}",
kubernetes_pod_name=~"{{ .Name }}-[0-9a-zA-Z]+(-[0-9a-zA-Z]+)"}
[{{ .Interval }}])) by (le))
`
// GetEnvoyRequestDuration returns the 99P requests delay using envoy_cluster_upstream_rq_time_bucket metrics
func (c *Observer) GetEnvoyRequestDuration(name string, namespace string, metric string, interval string) (time.Duration, error) {
if c.metricsServer == "fake" {
return 1, nil
}
meta := struct {
Name string
Namespace string
Interval string
}{
name,
namespace,
interval,
}
query, err := render(meta, envoyRequestDurationQuery)
if err != nil {
return 0, err
}
var rate *float64
querySt := url.QueryEscape(query)
result, err := c.queryMetric(querySt)
if err != nil {
return 0, err
}
for _, v := range result.Data.Result {
metricValue := v.Value[1]
switch metricValue.(type) {
case string:
f, err := strconv.ParseFloat(metricValue.(string), 64)
if err != nil {
return 0, err
}
rate = &f
}
}
if rate == nil {
return 0, fmt.Errorf("no values found for metric %s", metric)
}
ms := time.Duration(int64(*rate)) * time.Millisecond
return ms, nil
}
pkg/metrics/envoy_test.go
+23
View File
@@ -26,3 +26,26 @@ func Test_EnvoySuccessRateQueryRender(t *testing.T) {
t.Errorf("\nGot %s \nWanted %s", query, expected)
}
}
func Test_EnvoyRequestDurationQueryRender(t *testing.T) {
meta := struct {
Name string
Namespace string
Interval string
}{
"podinfo",
"default",
"1m",
}
query, err := render(meta, envoyRequestDurationQuery)
if err != nil {
t.Fatal(err)
}
expected := `histogram_quantile(0.99, sum(rate(envoy_cluster_upstream_rq_time_bucket{kubernetes_namespace="default",kubernetes_pod_name=~"podinfo-[0-9a-zA-Z]+(-[0-9a-zA-Z]+)"}[1m])) by (le))`
if query != expected {
t.Errorf("\nGot %s \nWanted %s", query, expected)
}
}
pkg/metrics/observer_test.go
+19
View File
@@ -27,6 +27,25 @@ func TestCanaryObserver_GetEnvoySuccessRate(t *testing.T) {
}
func TestCanaryObserver_GetEnvoyRequestDuration(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.596,"200"]}]}}`
w.Write([]byte(json))
}))
defer ts.Close()
observer := NewObserver(ts.URL)
val, err := observer.GetEnvoyRequestDuration("podinfo", "default", "envoy_cluster_upstream_rq_time_bucket", "1m")
if err != nil {
t.Fatal(err.Error())
}
if val != 200*time.Millisecond {
t.Errorf("Got %v wanted %v", val, 200*time.Millisecond)
}
}
func TestCanaryObserver_GetIstioSuccessRate(t *testing.T) {
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
json := `{"status":"success","data":{"resultType":"vector","result":[{"metric":{},"value":[1545905245.458,"100"]}]}}`
pkg/router/factory.go
+2 -1
View File
@@ -31,11 +31,12 @@ func NewFactory(kubeConfig *restclient.Config, kubeClient kubernetes.Interface,
}
// KubernetesRouter returns a ClusterIP service router
func (factory *Factory) KubernetesRouter() *KubernetesRouter {
func (factory *Factory) KubernetesRouter(label string) *KubernetesRouter {
return &KubernetesRouter{
logger: factory.logger,
flaggerClient: factory.flaggerClient,
kubeClient: factory.kubeClient,
label: label,
}
}
pkg/router/kubernetes.go
+2 -1
View File
@@ -19,6 +19,7 @@ type KubernetesRouter struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
label string
}
// Reconcile creates or updates the primary and canary services
@@ -64,7 +65,7 @@ func (c *KubernetesRouter) reconcileService(canary *flaggerv1.Canary, name strin
svcSpec := corev1.ServiceSpec{
Type: corev1.ServiceTypeClusterIP,
Selector: map[string]string{"app": target},
Selector: map[string]string{c.label: target},
Ports: []corev1.ServicePort{
{
Name: portName,
pkg/router/router_test.go
+2 -2
View File
@@ -6,7 +6,7 @@ import (
istiov1alpha3 "github.com/weaveworks/flagger/pkg/apis/istio/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
fakeFlagger "github.com/weaveworks/flagger/pkg/client/clientset/versioned/fake"
"github.com/weaveworks/flagger/pkg/logging"
"github.com/weaveworks/flagger/pkg/logger"
"go.uber.org/zap"
appsv1 "k8s.io/api/apps/v1"
hpav1 "k8s.io/api/autoscaling/v1"
@@ -35,7 +35,7 @@ func setupfakeClients() fakeClients {
kubeClient := fake.NewSimpleClientset(newMockDeployment(), newMockABTestDeployment())
meshClient := fakeFlagger.NewSimpleClientset()
logger, _ := logging.NewLogger("debug")
logger, _ := logger.NewLogger("debug")
return fakeClients{
canary: canary,
pkg/version/version.go
+1 -1
View File
@@ -1,4 +1,4 @@
package version
var VERSION = "0.10.0"
var VERSION = "0.11.0"
var REVISION = "unknown"
test/README.md
+3 -3
View File
@@ -5,7 +5,7 @@ The e2e testing infrastructure is powered by CircleCI and [Kubernetes Kind](http
CircleCI e2e workflow:
* install latest stable kubectl [e2e-kind.sh](e2e-kind.sh)
* build Kubernetes Kind from master [e2e-kind.sh](e2e-kind.sh)
* install Kubernetes Kind [e2e-kind.sh](e2e-kind.sh)
* create local Kubernetes cluster with kind [e2e-kind.sh](e2e-kind.sh)
* install latest stable Helm CLI [e2e-istio.sh](e2e-istio.sh)
* deploy Tiller on the local cluster [e2e-istio.sh](e2e-istio.sh)
@@ -18,7 +18,7 @@ CircleCI e2e workflow:
* deploy the load tester in the test namespace [e2e-tests.sh](e2e-tests.sh)
* deploy a demo workload (podinfo) in the test namespace [e2e-tests.sh](e2e-tests.sh)
* test the canary initialization [e2e-tests.sh](e2e-tests.sh)
* test the canary analysis and promotion [e2e-tests.sh](e2e-tests.sh)
* test the canary analysis and promotion using weighted traffic and the load testing webhook [e2e-tests.sh](e2e-tests.sh)
* test the A/B testing analysis and promotion using cookies filters and pre/post rollout webhooks [e2e-tests.sh](e2e-tests.sh)
test/e2e-tests.sh
+74 -4
View File
@@ -42,13 +42,13 @@ spec:
canaryAnalysis:
interval: 15s
threshold: 15
maxWeight: 50
maxWeight: 30
stepWeight: 10
metrics:
- name: istio_requests_total
- name: request-success-rate
threshold: 99
interval: 1m
- name: istio_request_duration_seconds_bucket
- name: request-duration
threshold: 500
interval: 30s
- name: "404s percentage"
@@ -83,7 +83,6 @@ spec:
type: cmd
cmd: "hey -z 10m -q 10 -c 2 http://podinfo.test:9898/"
logCmdOutput: "true"
EOF
echo '>>> Waiting for primary to be ready'
@@ -126,6 +125,77 @@ done
echo '✔ Canary promotion test passed'
cat <<EOF | kubectl apply -f -
apiVersion: flagger.app/v1alpha3
kind: Canary
metadata:
name: podinfo
namespace: test
spec:
targetRef:
apiVersion: apps/v1
kind: Deployment
name: podinfo
progressDeadlineSeconds: 60
service:
port: 9898
canaryAnalysis:
interval: 10s
threshold: 5
iterations: 5
match:
- headers:
cookie:
regex: "^(.*?;)?(type=insider)(;.*)?$"
metrics:
- name: request-success-rate
threshold: 99
interval: 1m
- name: request-duration
threshold: 500
interval: 30s
webhooks:
- name: pre
type: pre-rollout
url: http://flagger-loadtester.test/
timeout: 5s
metadata:
type: cmd
cmd: "hey -z 10m -q 10 -c 2 -H 'Cookie: type=insider' http://podinfo-canary.test:9898/"
logCmdOutput: "true"
- name: post
type: post-rollout
url: http://flagger-loadtester.test/
timeout: 15s
metadata:
type: cmd
cmd: "curl -s http://podinfo.test:9898/"
logCmdOutput: "true"
EOF
echo '>>> Triggering A/B testing'
kubectl -n test set image deployment/podinfo podinfod=quay.io/stefanprodan/podinfo:1.4.2
echo '>>> Waiting for A/B testing promotion'
retries=50
count=0
ok=false
until ${ok}; do
kubectl -n test describe deployment/podinfo-primary | grep '1.4.2' && ok=true || ok=false
sleep 10
kubectl -n istio-system logs deployment/flagger --tail 1
count=$(($count + 1))
if [[ ${count} -eq ${retries} ]]; then
kubectl -n test describe deployment/podinfo
kubectl -n test describe deployment/podinfo-primary
kubectl -n istio-system logs deployment/flagger
echo "No more retries left"
exit 1
fi
done
echo '✔ A/B testing promotion test passed'
kubectl -n istio-system logs deployment/flagger
echo '✔ All tests passed'