github/flagger
1
0
Fork 0
mirror of https://github.com/fluxcd/flagger.git synced 2026-04-15 06:57:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Canary-release anything behind K8s service

Browse Source 
Resolves #371

---

This adds the support for `corev1.Service` as the `targetRef.kind`, so that we can use Flagger just for canary analysis and traffic-shifting on existing and pre-created services. Flagger doesn't touch deployments and HPAs in this mode.

This is useful for keeping your full-control on the resources backing the service to be canary-released, including pods(behind a ClusterIP service) and external services(behind an ExternalName service).

Major use-case in my mind are:

- Canary-release a K8s cluster. You create two clusters and a master cluster. In the master cluster, you create two `ExternalName` services pointing to (the hostname of the loadbalancer of the targeted app instance in) each cluster. Flagger runs on the master cluster and helps safely rolling-out a new K8s cluster by doing a canary release on the `ExternalName` service.
- You want annotations and labels added to the service for integrating with things like external lbs(without extending Flagger to support customizing any aspect of the K8s service it manages

**Design**:

A canary release on a K8s service is almost the same as one on a K8s deployment. The only fundamental difference is that it operates only on a set of K8s services.

For example, one may start by creating two Helm releases for `podinfo-blue` and `podinfo-green`, and a K8s service `podinfo`. The `podinfo` service should initially have the same `Spec` as that of  `podinfo-blue`.

On a new release, you update `podinfo-green`, then trigger Flagger by updating the K8s service `podinfo` so that it points to pods or `externalName` as declared in `podinfo-green`. Flagger does the rest. The end result is the traffic to `podinfo` is gradually and safely shifted from `podinfo-blue` to `podinfo-green`.

**How it works**:

Under the hood, Flagger maintains two K8s services, `podinfo-primary` and `podinfo-canary`. Compared to canaries on K8s deployments, it doesn't create the service named `podinfo`, as it is already provided by YOU.

Once Flagger detects the change in the `podinfo` service, it updates the `podinfo-canary` service and the routes, then analyzes the canary. On successful analysis, it promotes the canary service to the `podinfo-primary` service. You expose the `podinfo` service via any L7 ingress solution or a service mesh so that the traffic is managed by Flagger for safe deployments.

**Giving it a try**:

To give it a try, create a `Canary` as usual, but its `targetRef` pointed to a K8s service:

```
apiVersion: flagger.app/v1alpha3
kind: Canary
metadata:
  name: podinfo
spec:
  provider: kubernetes
  targetRef:
    apiVersion: core/v1
    kind: Service
    name: podinfo
  service:
    port: 9898
  canaryAnalysis:
    # schedule interval (default 60s)
    interval: 10s
    # max number of failed checks before rollback
    threshold: 2
    # number of checks to run before rollback
    iterations: 2
    # Prometheus checks based on
    # http_request_duration_seconds histogram
    metrics: []
```

Create a K8s service named `podinfo`, and update it. Now watch for the services `podinfo`, `podinfo-primary`, `podinfo-canary`.

Flagger tracks `podinfo` service for changes. Upon any change, it reconciles `podinfo-primary` and `podinfo-canary` services. `podinfo-canary` always replicate the latest `podinfo`. In contract, `podinfo-primary` replicates the latest successful `podinfo-canary`.

**Notes**:

- For the canary cluster use-case, we would need to write a K8s operator to, e.g. for App Mesh, sync `ExternalName` services to AppMesh `VirtualNode`s. But that's another story!
This commit is contained in:
Yusuke Kuoka
2019-11-27 09:07:29 +09:00
parent 446a2b976c
commit 1ba595bc6f
19 changed files with 1016 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.circleci/config.yml
+11
View File
@@ -92,6 +92,17 @@ jobs:
- run: test/e2e-kubernetes.sh
- run: test/e2e-kubernetes-tests.sh
e2e-kubernetes-svc-testing:
machine: true
steps:
- checkout
- attach_workspace:
at: /tmp/bin
- run: test/container-build.sh
- run: test/e2e-kind.sh
- run: test/e2e-kubernetes.sh
- run: test/e2e-kubernetes-svc-tests.sh
e2e-smi-istio-testing:
machine: true
steps:
pkg/canary/controller.go
+3 -1
View File
@@ -1,6 +1,8 @@
package canary
import "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
import (
"github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
)
type Controller interface {
IsPrimaryReady(canary *v1alpha3.Canary) (bool, error)
pkg/canary/deployment.go → pkg/canary/deployment_controller.go
+6 -25
View File
@@ -6,7 +6,6 @@ import (
"io"
"github.com/google/go-cmp/cmp"
"github.com/mitchellh/hashstructure"
"go.uber.org/zap"
appsv1 "k8s.io/api/apps/v1"
hpav1 "k8s.io/api/autoscaling/v2beta1"
@@ -34,6 +33,7 @@ type DeploymentController struct {
// scales to zero the canary deployment and returns the pod selector label and container ports
func (c *DeploymentController) Initialize(cd *flaggerv1.Canary, skipLivenessChecks bool) (label string, ports map[string]int32, err error) {
primaryName := fmt.Sprintf("%s-primary", cd.Spec.TargetRef.Name)
label, ports, err = c.createPrimaryDeployment(cd)
if err != nil {
return "", ports, fmt.Errorf("creating deployment %s.%s failed: %v", primaryName, cd.Namespace, err)
@@ -143,30 +143,7 @@ func (c *DeploymentController) HasTargetChanged(cd *flaggerv1.Canary) (bool, err
return false, fmt.Errorf("deployment %s.%s query error %v", targetName, cd.Namespace, err)
}
if cd.Status.LastAppliedSpec == "" {
return true, nil
}
newHash, err := hashstructure.Hash(canary.Spec.Template, nil)
if err != nil {
return false, fmt.Errorf("hash error %v", err)
}
// do not trigger a canary deployment on manual rollback
if cd.Status.LastPromotedSpec == fmt.Sprintf("%d", newHash) {
return false, nil
}
if cd.Status.LastAppliedSpec != fmt.Sprintf("%d", newHash) {
return true, nil
}
return false, nil
}
// HaveDependenciesChanged returns true if the canary configmaps or secrets have changed
func (c *DeploymentController) HaveDependenciesChanged(cd *flaggerv1.Canary) (bool, error) {
return c.configTracker.HasConfigChanged(cd)
return hasSpecChanged(cd, canary.Spec.Template)
}
// Scale sets the canary deployment replicas
@@ -425,6 +402,10 @@ var sidecars = map[string]bool{
"envoy": true,
}
func (c *DeploymentController) HaveDependenciesChanged(cd *flaggerv1.Canary) (bool, error) {
return c.configTracker.HasConfigChanged(cd)
}
// getPorts returns a list of all container ports
func (c *DeploymentController) getPorts(cd *flaggerv1.Canary, deployment *appsv1.Deployment) (map[string]int32, error) {
ports := make(map[string]int32)
pkg/canary/deployment_test.go → pkg/canary/deployment_controller_test.go
View File
pkg/canary/factory.go
+7
View File
@@ -41,10 +41,17 @@ func (factory *Factory) Controller(kind string) Controller {
FlaggerClient: factory.flaggerClient,
},
}
serviceCtrl := &ServiceController{
logger: factory.logger,
kubeClient: factory.kubeClient,
flaggerClient: factory.flaggerClient,
}
switch {
case kind == "Deployment":
return deploymentCtrl
case kind == "Service":
return serviceCtrl
default:
return deploymentCtrl
}
pkg/canary/service_controller.go
+135
View File
@@ -0,0 +1,135 @@
package canary
import (
"fmt"
ex "github.com/pkg/errors"
"go.uber.org/zap"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
)
// ServiceController is managing the operations for Kubernetes service kind
type ServiceController struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
}
// SetStatusFailedChecks updates the canary failed checks counter
func (c *ServiceController) SetStatusFailedChecks(cd *flaggerv1.Canary, val int) error {
return setStatusFailedChecks(c.flaggerClient, cd, val)
}
// SetStatusWeight updates the canary status weight value
func (c *ServiceController) SetStatusWeight(cd *flaggerv1.Canary, val int) error {
return setStatusWeight(c.flaggerClient, cd, val)
}
// SetStatusIterations updates the canary status iterations value
func (c *ServiceController) SetStatusIterations(cd *flaggerv1.Canary, val int) error {
return setStatusIterations(c.flaggerClient, cd, val)
}
// SetStatusPhase updates the canary status phase
func (c *ServiceController) SetStatusPhase(cd *flaggerv1.Canary, phase flaggerv1.CanaryPhase) error {
return setStatusPhase(c.flaggerClient, cd, phase)
}
var _ Controller = &ServiceController{}
// Initialize creates the primary deployment, hpa,
// scales to zero the canary deployment and returns the pod selector label and container ports
func (c *ServiceController) Initialize(cd *flaggerv1.Canary, skipLivenessChecks bool) (label string, ports map[string]int32, err error) {
return "", nil, nil
}
// Promote copies target's spec from canary to primary
func (c *ServiceController) Promote(cd *flaggerv1.Canary) error {
targetName := cd.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
canary, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("service %s.%s not found", targetName, cd.Namespace)
}
return fmt.Errorf("service %s.%s query error %v", targetName, cd.Namespace, err)
}
primary, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(primaryName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("service %s.%s not found", primaryName, cd.Namespace)
}
return fmt.Errorf("service %s.%s query error %v", primaryName, cd.Namespace, err)
}
primaryCopy := canary.DeepCopy()
primaryCopy.ObjectMeta.Name = primary.ObjectMeta.Name
if primaryCopy.Spec.Type == "ClusterIP" {
primaryCopy.Spec.ClusterIP = primary.Spec.ClusterIP
}
primaryCopy.ObjectMeta.ResourceVersion = primary.ObjectMeta.ResourceVersion
primaryCopy.ObjectMeta.UID = primary.ObjectMeta.UID
// apply update
_, err = c.kubeClient.CoreV1().Services(cd.Namespace).Update(primaryCopy)
if err != nil {
return fmt.Errorf("updating service %s.%s spec failed: %v",
primaryCopy.GetName(), primaryCopy.Namespace, err)
}
return nil
}
// HasServiceChanged returns true if the canary service spec has changed
func (c *ServiceController) HasTargetChanged(cd *flaggerv1.Canary) (bool, error) {
targetName := cd.Spec.TargetRef.Name
canary, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return false, fmt.Errorf("service %s.%s not found", targetName, cd.Namespace)
}
return false, fmt.Errorf("service %s.%s query error %v", targetName, cd.Namespace, err)
}
return hasSpecChanged(cd, canary.Spec)
}
// Scale sets the canary deployment replicas
func (c *ServiceController) Scale(cd *flaggerv1.Canary, replicas int32) error {
return nil
}
func (c *ServiceController) ScaleFromZero(cd *flaggerv1.Canary) error {
return nil
}
func (c *ServiceController) SyncStatus(cd *flaggerv1.Canary, status flaggerv1.CanaryStatus) error {
dep, err := c.kubeClient.CoreV1().Services(cd.Namespace).Get(cd.Spec.TargetRef.Name, metav1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("service %s.%s not found", cd.Spec.TargetRef.Name, cd.Namespace)
}
return ex.Wrap(err, "SyncStatus service query error")
}
return syncCanaryStatus(c.flaggerClient, cd, status, dep.Spec, func(cdCopy *flaggerv1.Canary) {})
}
func (c *ServiceController) HaveDependenciesChanged(cd *flaggerv1.Canary) (bool, error) {
return false, nil
}
func (c *ServiceController) IsPrimaryReady(cd *flaggerv1.Canary) (bool, error) {
return true, nil
}
func (c *ServiceController) IsCanaryReady(cd *flaggerv1.Canary) (bool, error) {
return true, nil
}
pkg/canary/spec.go
+30
View File
@@ -0,0 +1,30 @@
package canary
import (
"fmt"
"github.com/mitchellh/hashstructure"
"github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
)
func hasSpecChanged(cd *v1alpha3.Canary, spec interface{}) (bool, error) {
if cd.Status.LastAppliedSpec == "" {
return true, nil
}
newHash, err := hashstructure.Hash(spec, nil)
if err != nil {
return false, fmt.Errorf("hash error %v", err)
}
// do not trigger a canary deployment on manual rollback
if cd.Status.LastPromotedSpec == fmt.Sprintf("%d", newHash) {
return false, nil
}
if cd.Status.LastAppliedSpec != fmt.Sprintf("%d", newHash) {
return true, nil
}
return false, nil
}
pkg/canary/status.go
+35 -12
View File
@@ -5,6 +5,7 @@ import (
"github.com/mitchellh/hashstructure"
ex "github.com/pkg/errors"
"github.com/weaveworks/flagger/pkg/client/clientset/versioned"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -28,7 +29,13 @@ func (c *DeploymentController) SyncStatus(cd *flaggerv1.Canary, status flaggerv1
return ex.Wrap(err, "SyncStatus configs query error")
}
hash, err := hashstructure.Hash(dep.Spec.Template, nil)
return syncCanaryStatus(c.flaggerClient, cd, status, dep.Spec.Template, func(cdCopy *flaggerv1.Canary) {
cdCopy.Status.TrackedConfigs = configs
})
}
func syncCanaryStatus(flaggerClient versioned.Interface, cd *flaggerv1.Canary, status flaggerv1.CanaryStatus, canaryResource interface{}, setAll func(cdCopy *flaggerv1.Canary)) error {
hash, err := hashstructure.Hash(canaryResource, nil)
if err != nil {
return ex.Wrap(err, "SyncStatus hash error")
}
@@ -37,7 +44,7 @@ func (c *DeploymentController) SyncStatus(cd *flaggerv1.Canary, status flaggerv1
err = retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
var selErr error
if !firstTry {
cd, selErr = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
cd, selErr = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
if selErr != nil {
return selErr
}
@@ -49,13 +56,13 @@ func (c *DeploymentController) SyncStatus(cd *flaggerv1.Canary, status flaggerv1
cdCopy.Status.Iterations = status.Iterations
cdCopy.Status.LastAppliedSpec = fmt.Sprintf("%d", hash)
cdCopy.Status.LastTransitionTime = metav1.Now()
cdCopy.Status.TrackedConfigs = configs
setAll(cdCopy)
if ok, conditions := MakeStatusConditions(cd.Status, status.Phase); ok {
cdCopy.Status.Conditions = conditions
}
_, err = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
_, err = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
firstTry = false
return
})
@@ -67,11 +74,15 @@ func (c *DeploymentController) SyncStatus(cd *flaggerv1.Canary, status flaggerv1
// SetStatusFailedChecks updates the canary failed checks counter
func (c *DeploymentController) SetStatusFailedChecks(cd *flaggerv1.Canary, val int) error {
return setStatusFailedChecks(c.flaggerClient, cd, val)
}
func setStatusFailedChecks(flaggerClient versioned.Interface, cd *flaggerv1.Canary, val int) error {
firstTry := true
err := retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
var selErr error
if !firstTry {
cd, selErr = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
cd, selErr = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
if selErr != nil {
return selErr
}
@@ -80,7 +91,7 @@ func (c *DeploymentController) SetStatusFailedChecks(cd *flaggerv1.Canary, val i
cdCopy.Status.FailedChecks = val
cdCopy.Status.LastTransitionTime = metav1.Now()
_, err = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
_, err = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
firstTry = false
return
})
@@ -92,11 +103,15 @@ func (c *DeploymentController) SetStatusFailedChecks(cd *flaggerv1.Canary, val i
// SetStatusWeight updates the canary status weight value
func (c *DeploymentController) SetStatusWeight(cd *flaggerv1.Canary, val int) error {
return setStatusWeight(c.flaggerClient, cd, val)
}
func setStatusWeight(flaggerClient versioned.Interface, cd *flaggerv1.Canary, val int) error {
firstTry := true
err := retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
var selErr error
if !firstTry {
cd, selErr = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
cd, selErr = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
if selErr != nil {
return selErr
}
@@ -105,7 +120,7 @@ func (c *DeploymentController) SetStatusWeight(cd *flaggerv1.Canary, val int) er
cdCopy.Status.CanaryWeight = val
cdCopy.Status.LastTransitionTime = metav1.Now()
_, err = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
_, err = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
firstTry = false
return
})
@@ -117,11 +132,15 @@ func (c *DeploymentController) SetStatusWeight(cd *flaggerv1.Canary, val int) er
// SetStatusIterations updates the canary status iterations value
func (c *DeploymentController) SetStatusIterations(cd *flaggerv1.Canary, val int) error {
return setStatusIterations(c.flaggerClient, cd, val)
}
func setStatusIterations(flaggerClient versioned.Interface, cd *flaggerv1.Canary, val int) error {
firstTry := true
err := retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
var selErr error
if !firstTry {
cd, selErr = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
cd, selErr = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
if selErr != nil {
return selErr
}
@@ -131,7 +150,7 @@ func (c *DeploymentController) SetStatusIterations(cd *flaggerv1.Canary, val int
cdCopy.Status.Iterations = val
cdCopy.Status.LastTransitionTime = metav1.Now()
_, err = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
_, err = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
firstTry = false
return
})
@@ -144,11 +163,15 @@ func (c *DeploymentController) SetStatusIterations(cd *flaggerv1.Canary, val int
// SetStatusPhase updates the canary status phase
func (c *DeploymentController) SetStatusPhase(cd *flaggerv1.Canary, phase flaggerv1.CanaryPhase) error {
return setStatusPhase(c.flaggerClient, cd, phase)
}
func setStatusPhase(flaggerClient versioned.Interface, cd *flaggerv1.Canary, phase flaggerv1.CanaryPhase) error {
firstTry := true
err := retry.RetryOnConflict(retry.DefaultBackoff, func() (err error) {
var selErr error
if !firstTry {
cd, selErr = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
cd, selErr = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).Get(cd.GetName(), metav1.GetOptions{})
if selErr != nil {
return selErr
}
@@ -171,7 +194,7 @@ func (c *DeploymentController) SetStatusPhase(cd *flaggerv1.Canary, phase flagge
cdCopy.Status.Conditions = conditions
}
_, err = c.flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
_, err = flaggerClient.FlaggerV1alpha3().Canaries(cd.Namespace).UpdateStatus(cdCopy)
firstTry = false
return
})
pkg/canary/tracker.go
+1 -1
View File
@@ -16,7 +16,7 @@ import (
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
)
// configTracker is managing the operations for Kubernetes ConfigMaps and Secrets
// ConfigTracker is managing the operations for Kubernetes ConfigMaps and Secrets
type ConfigTracker struct {
KubeClient kubernetes.Interface
FlaggerClient clientset.Interface
pkg/controller/controller_test.go
+54
View File
@@ -10,6 +10,7 @@ import (
hpav2 "k8s.io/api/autoscaling/v2beta1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/tools/record"
@@ -52,6 +53,7 @@ func SetupMocks(c *flaggerv1.Canary) Mocks {
// init kube clientset and register mock objects
kubeClient := fake.NewSimpleClientset(
newTestDeployment(),
newTestService(),
newTestHPA(),
NewTestConfigMap(),
NewTestConfigMapEnv(),
@@ -555,6 +557,58 @@ func newTestDeploymentV2() *appsv1.Deployment {
return d
}
func newTestService() *corev1.Service {
d := &corev1.Service{
TypeMeta: metav1.TypeMeta{APIVersion: appsv1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: corev1.ServiceSpec{
Selector: map[string]string{
"app": "podinfo",
},
Type: corev1.ServiceTypeClusterIP,
Ports: []corev1.ServicePort{
{
Name: "http",
Port: 9898,
Protocol: corev1.ProtocolTCP,
TargetPort: intstr.FromString("http"),
},
},
},
}
return d
}
func newTestServiceV2() *corev1.Service {
d := &corev1.Service{
TypeMeta: metav1.TypeMeta{APIVersion: appsv1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: corev1.ServiceSpec{
Selector: map[string]string{
"app": "podinfo-v2",
},
Type: corev1.ServiceTypeClusterIP,
Ports: []corev1.ServicePort{
{
Name: "http",
Port: 9898,
Protocol: corev1.ProtocolTCP,
TargetPort: intstr.FromString("http"),
},
},
},
}
return d
}
func newTestHPA() *hpav2.HorizontalPodAutoscaler {
h := &hpav2.HorizontalPodAutoscaler{
TypeMeta: metav1.TypeMeta{APIVersion: hpav2.SchemeGroupVersion.String()},
pkg/controller/scheduler.go
+4 -4
View File
@@ -117,7 +117,7 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh
meshRouter := c.routerFactory.MeshRouter(provider)
// create or update ClusterIP services
if err := c.routerFactory.KubernetesRouter(labelSelector, map[string]string{}, ports).Reconcile(cd); err != nil {
if err := c.routerFactory.KubernetesRouter(cd.Spec.TargetRef.Kind, labelSelector, map[string]string{}, ports).Reconcile(cd); err != nil {
c.recordEventWarningf(cd, "%v", err)
return
}
@@ -585,12 +585,12 @@ func (c *Controller) shouldAdvance(cd *flaggerv1.Canary, canaryController canary
return true, nil
}
newDep, err := canaryController.HasTargetChanged(cd)
newTarget, err := canaryController.HasTargetChanged(cd)
if err != nil {
return false, err
}
if newDep {
return newDep, nil
if newTarget {
return newTarget, nil
}
newCfg, err := canaryController.HaveDependenciesChanged(cd)
pkg/controller/scheduler_svc_test.go
+166
View File
@@ -0,0 +1,166 @@
package controller
import (
"testing"
hpav1 "k8s.io/api/autoscaling/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
)
func TestScheduler_ServicePromotion(t *testing.T) {
mocks := SetupMocks(newTestServiceCanary())
// init
mocks.ctrl.advanceCanary("podinfo", "default", true)
// check initialized status
c, err := mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})
if err != nil {
t.Fatal(err.Error())
}
if c.Status.Phase != flaggerv1.CanaryPhaseInitialized {
t.Errorf("Got canary state %v wanted %v", c.Status.Phase, flaggerv1.CanaryPhaseInitialized)
}
// update
svc2 := newTestServiceV2()
_, err = mocks.kubeClient.CoreV1().Services("default").Update(svc2)
if err != nil {
t.Fatal(err.Error())
}
// detect service spec changes
mocks.ctrl.advanceCanary("podinfo", "default", true)
primaryWeight, canaryWeight, mirrored, err := mocks.router.GetRoutes(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
primaryWeight = 60
canaryWeight = 40
err = mocks.router.SetRoutes(mocks.canary, primaryWeight, canaryWeight, mirrored)
if err != nil {
t.Fatal(err.Error())
}
// advance
mocks.ctrl.advanceCanary("podinfo", "default", true)
// check progressing status
c, err = mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})
if err != nil {
t.Fatal(err.Error())
}
if c.Status.Phase != flaggerv1.CanaryPhaseProgressing {
t.Errorf("Got canary state %v wanted %v", c.Status.Phase, flaggerv1.CanaryPhaseProgressing)
}
// promote
mocks.ctrl.advanceCanary("podinfo", "default", true)
// check promoting status
c, err = mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})
if err != nil {
t.Fatal(err.Error())
}
if c.Status.Phase != flaggerv1.CanaryPhasePromoting {
t.Errorf("Got canary state %v wanted %v", c.Status.Phase, flaggerv1.CanaryPhasePromoting)
}
// finalise
mocks.ctrl.advanceCanary("podinfo", "default", true)
primaryWeight, canaryWeight, mirrored, err = mocks.router.GetRoutes(mocks.canary)
if err != nil {
t.Fatal(err.Error())
}
if primaryWeight != 100 {
t.Errorf("Got primary route %v wanted %v", primaryWeight, 100)
}
if canaryWeight != 0 {
t.Errorf("Got canary route %v wanted %v", canaryWeight, 0)
}
if mirrored != false {
t.Errorf("Got mirrored %v wanted %v", mirrored, false)
}
primarySvc, err := mocks.kubeClient.CoreV1().Services("default").Get("podinfo-primary", metav1.GetOptions{})
if err != nil {
t.Fatal(err.Error())
}
primaryLabelValue := primarySvc.Spec.Selector["app"]
canaryLabelValue := svc2.Spec.Selector["app"]
if primaryLabelValue != canaryLabelValue {
t.Errorf("Got primary selector label value %v wanted %v", primaryLabelValue, canaryLabelValue)
}
// check finalising status
c, err = mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})
if err != nil {
t.Fatal(err.Error())
}
if c.Status.Phase != flaggerv1.CanaryPhaseFinalising {
t.Errorf("Got canary state %v wanted %v", c.Status.Phase, flaggerv1.CanaryPhaseFinalising)
}
// scale canary to zero
mocks.ctrl.advanceCanary("podinfo", "default", true)
c, err = mocks.flaggerClient.FlaggerV1alpha3().Canaries("default").Get("podinfo", metav1.GetOptions{})
if err != nil {
t.Fatal(err.Error())
}
if c.Status.Phase != flaggerv1.CanaryPhaseSucceeded {
t.Errorf("Got canary state %v wanted %v", c.Status.Phase, flaggerv1.CanaryPhaseSucceeded)
}
}
func newTestServiceCanary() *flaggerv1.Canary {
cd := &flaggerv1.Canary{
TypeMeta: metav1.TypeMeta{APIVersion: flaggerv1.SchemeGroupVersion.String()},
ObjectMeta: metav1.ObjectMeta{
Namespace: "default",
Name: "podinfo",
},
Spec: flaggerv1.CanarySpec{
TargetRef: hpav1.CrossVersionObjectReference{
Name: "podinfo",
APIVersion: "core/v1",
Kind: "Service",
},
Service: flaggerv1.CanaryService{
Port: 9898,
},
CanaryAnalysis: flaggerv1.CanaryAnalysis{
Threshold: 10,
StepWeight: 10,
MaxWeight: 50,
Metrics: []flaggerv1.CanaryMetric{
{
Name: "istio_requests_total",
Threshold: 99,
Interval: "1m",
},
{
Name: "istio_request_duration_seconds_bucket",
Threshold: 500,
Interval: "1m",
},
},
},
},
}
return cd
}
pkg/router/factory.go
+20 -3
View File
@@ -34,9 +34,9 @@ func NewFactory(kubeConfig *restclient.Config, kubeClient kubernetes.Interface,
}
}
// KubernetesRouter returns a ClusterIP service router
func (factory *Factory) KubernetesRouter(labelSelector string, annotations map[string]string, ports map[string]int32) *KubernetesRouter {
return &KubernetesRouter{
// KubernetesDeploymentRouter returns a ClusterIP service router
func (factory *Factory) KubernetesRouter(kind string, labelSelector string, annotations map[string]string, ports map[string]int32) KubernetesRouter {
deploymentRouter := &KubernetesDeploymentRouter{
logger: factory.logger,
flaggerClient: factory.flaggerClient,
kubeClient: factory.kubeClient,
@@ -44,6 +44,23 @@ func (factory *Factory) KubernetesRouter(labelSelector string, annotations map[s
annotations: annotations,
ports: ports,
}
serviceRouter := &KubernetesServiceRouter{
logger: factory.logger,
flaggerClient: factory.flaggerClient,
kubeClient: factory.kubeClient,
labelSelector: labelSelector,
annotations: annotations,
ports: ports,
}
switch {
case kind == "Deployment":
return deploymentRouter
case kind == "Service":
return serviceRouter
default:
return deploymentRouter
}
}
// MeshRouter returns a service mesh router
pkg/router/kubernetes.go
+22 -148
View File
@@ -1,161 +1,35 @@
package router
import (
"fmt"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"go.uber.org/zap"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/kubernetes"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
)
// KubernetesRouter is managing ClusterIP services
type KubernetesRouter struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
labelSelector string
annotations map[string]string
ports map[string]int32
// KubernetesDeploymentRouter is managing ClusterIP services
type KubernetesRouter interface {
// Reconcile creates or updates K8s services to prepare for the canary release
Reconcile(canary *flaggerv1.Canary) error
}
// Reconcile creates or updates the primary and canary services
func (c *KubernetesRouter) Reconcile(canary *flaggerv1.Canary) error {
targetName := canary.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
canaryName := fmt.Sprintf("%s-canary", targetName)
// main svc
err := c.reconcileService(canary, targetName, primaryName)
if err != nil {
return err
func buildService(canary *flaggerv1.Canary, name string, src *corev1.Service) *corev1.Service {
svc := src.DeepCopy()
svc.ObjectMeta.Name = name
svc.ObjectMeta.Namespace = canary.Namespace
svc.ObjectMeta.OwnerReferences = []metav1.OwnerReference{
*metav1.NewControllerRef(canary, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
}
_, exists := svc.ObjectMeta.Annotations["kubectl.kubernetes.io/last-applied-configuration"]
if exists {
// Leaving this results in updates from flagger to this svc never succeed due to resourceVersion mismatch:
// Operation cannot be fulfilled on services "mysvc-canary": the object has been modified; please apply your changes to the latest version and try again
delete(svc.ObjectMeta.Annotations, "kubectl.kubernetes.io/last-applied-configuration")
}
// canary svc
err = c.reconcileService(canary, canaryName, targetName)
if err != nil {
return err
}
// primary svc
err = c.reconcileService(canary, primaryName, primaryName)
if err != nil {
return err
}
return nil
}
func (c *KubernetesRouter) SetRoutes(canary *flaggerv1.Canary, primaryRoute int, canaryRoute int) error {
return nil
}
func (c *KubernetesRouter) GetRoutes(canary *flaggerv1.Canary) (primaryRoute int, canaryRoute int, err error) {
return 0, 0, nil
}
func (c *KubernetesRouter) reconcileService(canary *flaggerv1.Canary, name string, target string) error {
portName := canary.Spec.Service.PortName
if portName == "" {
portName = "http"
}
targetPort := intstr.IntOrString{
Type: intstr.Int,
IntVal: canary.Spec.Service.Port,
}
if canary.Spec.Service.TargetPort.String() != "0" {
targetPort = canary.Spec.Service.TargetPort
}
svcSpec := corev1.ServiceSpec{
Type: corev1.ServiceTypeClusterIP,
Selector: map[string]string{c.labelSelector: target},
Ports: []corev1.ServicePort{
{
Name: portName,
Protocol: corev1.ProtocolTCP,
Port: canary.Spec.Service.Port,
TargetPort: targetPort,
},
},
}
for n, p := range c.ports {
cp := corev1.ServicePort{
Name: n,
Protocol: corev1.ProtocolTCP,
Port: p,
TargetPort: intstr.IntOrString{
Type: intstr.Int,
IntVal: p,
},
}
svcSpec.Ports = append(svcSpec.Ports, cp)
}
svc, err := c.kubeClient.CoreV1().Services(canary.Namespace).Get(name, metav1.GetOptions{})
if errors.IsNotFound(err) {
svc = &corev1.Service{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: canary.Namespace,
Labels: map[string]string{c.labelSelector: name},
Annotations: c.annotations,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(canary, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: svcSpec,
}
_, err = c.kubeClient.CoreV1().Services(canary.Namespace).Create(svc)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s.%s created", svc.GetName(), canary.Namespace)
return nil
}
if err != nil {
return fmt.Errorf("service %s query error %v", name, err)
}
if svc != nil {
sortPorts := func(a, b interface{}) bool {
return a.(corev1.ServicePort).Port < b.(corev1.ServicePort).Port
}
portsDiff := cmp.Diff(svcSpec.Ports, svc.Spec.Ports, cmpopts.SortSlices(sortPorts))
selectorsDiff := cmp.Diff(svcSpec.Selector, svc.Spec.Selector)
if portsDiff != "" || selectorsDiff != "" {
svcClone := svc.DeepCopy()
svcClone.Spec.Ports = svcSpec.Ports
svcClone.Spec.Selector = svcSpec.Selector
_, err = c.kubeClient.CoreV1().Services(canary.Namespace).Update(svcClone)
if err != nil {
return fmt.Errorf("service %s update error %v", name, err)
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s updated", svc.GetName())
}
}
return nil
return svc
}
pkg/router/kubernetes_deployment.go
+182
View File
@@ -0,0 +1,182 @@
package router
import (
"fmt"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"go.uber.org/zap"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/kubernetes"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
)
// KubernetesDeploymentRouter is managing ClusterIP services
type KubernetesDeploymentRouter struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
labelSelector string
annotations map[string]string
ports map[string]int32
}
// Reconcile creates or updates the primary and canary services
func (c *KubernetesDeploymentRouter) Reconcile(canary *flaggerv1.Canary) error {
targetName := canary.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
canaryName := fmt.Sprintf("%s-canary", targetName)
// main svc
err := c.reconcileService(canary, targetName, primaryName)
if err != nil {
return err
}
// canary svc
err = c.reconcileService(canary, canaryName, targetName)
if err != nil {
return err
}
// primary svc
err = c.reconcileService(canary, primaryName, primaryName)
if err != nil {
return err
}
return nil
}
func (c *KubernetesDeploymentRouter) SetRoutes(canary *flaggerv1.Canary, primaryRoute int, canaryRoute int) error {
return nil
}
func (c *KubernetesDeploymentRouter) GetRoutes(canary *flaggerv1.Canary) (primaryRoute int, canaryRoute int, err error) {
return 0, 0, nil
}
func (c *KubernetesDeploymentRouter) reconcileService(canary *flaggerv1.Canary, name string, target string) error {
portName := canary.Spec.Service.PortName
if portName == "" {
portName = "http"
}
targetPort := intstr.IntOrString{
Type: intstr.Int,
IntVal: canary.Spec.Service.Port,
}
if canary.Spec.Service.TargetPort.String() != "0" {
targetPort = canary.Spec.Service.TargetPort
}
svcSpec := corev1.ServiceSpec{
Type: corev1.ServiceTypeClusterIP,
Selector: map[string]string{c.labelSelector: target},
Ports: []corev1.ServicePort{
{
Name: portName,
Protocol: corev1.ProtocolTCP,
Port: canary.Spec.Service.Port,
TargetPort: targetPort,
},
},
}
for n, p := range c.ports {
cp := corev1.ServicePort{
Name: n,
Protocol: corev1.ProtocolTCP,
Port: p,
TargetPort: intstr.IntOrString{
Type: intstr.Int,
IntVal: p,
},
}
svcSpec.Ports = append(svcSpec.Ports, cp)
}
svc, err := c.kubeClient.CoreV1().Services(canary.Namespace).Get(name, metav1.GetOptions{})
if errors.IsNotFound(err) {
svc = &corev1.Service{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: canary.Namespace,
Labels: map[string]string{c.labelSelector: name},
Annotations: c.annotations,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(canary, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: svcSpec,
}
_, err = c.kubeClient.CoreV1().Services(canary.Namespace).Create(svc)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s.%s created", svc.GetName(), canary.Namespace)
return nil
}
if err != nil {
return fmt.Errorf("service %s query error %v", name, err)
}
if svc != nil {
sortPorts := func(a, b interface{}) bool {
return a.(corev1.ServicePort).Port < b.(corev1.ServicePort).Port
}
portsDiff := cmp.Diff(svcSpec.Ports, svc.Spec.Ports, cmpopts.SortSlices(sortPorts))
selectorsDiff := cmp.Diff(svcSpec.Selector, svc.Spec.Selector)
if portsDiff != "" || selectorsDiff != "" {
svcClone := svc.DeepCopy()
svcClone.Spec.Ports = svcSpec.Ports
svcClone.Spec.Selector = svcSpec.Selector
_, err = c.kubeClient.CoreV1().Services(canary.Namespace).Update(svcClone)
if err != nil {
return fmt.Errorf("service %s update error %v", name, err)
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s updated", svc.GetName())
}
}
return nil
}
func (c *KubernetesDeploymentRouter) createService(canary *flaggerv1.Canary, name string, src *corev1.Service) error {
svc := buildService(canary, name, src)
if svc.Spec.Type == "ClusterIP" {
// Reset and let K8s assign the IP. Otherwise we get an error due to the IP is already assigned
svc.Spec.ClusterIP = ""
}
// Let K8s set this. Otherwise K8s API complains with "resourceVersion should not be set on objects to be created"
svc.ObjectMeta.ResourceVersion = ""
_, err := c.kubeClient.CoreV1().Services(canary.Namespace).Create(svc)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s.%s created", svc.GetName(), canary.Namespace)
return nil
}
pkg/router/kubernetes_test.go → pkg/router/kubernetes_deployment_test.go
+3 -3
View File
@@ -8,7 +8,7 @@ import (
func TestServiceRouter_Create(t *testing.T) {
mocks := setupfakeClients()
router := &KubernetesRouter{
router := &KubernetesDeploymentRouter{
kubeClient: mocks.kubeClient,
flaggerClient: mocks.flaggerClient,
logger: mocks.logger,
@@ -48,7 +48,7 @@ func TestServiceRouter_Create(t *testing.T) {
func TestServiceRouter_Update(t *testing.T) {
mocks := setupfakeClients()
router := &KubernetesRouter{
router := &KubernetesDeploymentRouter{
kubeClient: mocks.kubeClient,
flaggerClient: mocks.flaggerClient,
logger: mocks.logger,
@@ -90,7 +90,7 @@ func TestServiceRouter_Update(t *testing.T) {
func TestServiceRouter_Undo(t *testing.T) {
mocks := setupfakeClients()
router := &KubernetesRouter{
router := &KubernetesDeploymentRouter{
kubeClient: mocks.kubeClient,
flaggerClient: mocks.flaggerClient,
logger: mocks.logger,
pkg/router/kubernetes_service.go
+124
View File
@@ -0,0 +1,124 @@
package router
import (
"fmt"
"go.uber.org/zap"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
)
// KubernetesServiceRouter manages ClusterIP services
type KubernetesServiceRouter struct {
kubeClient kubernetes.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
labelSelector string
annotations map[string]string
ports map[string]int32
}
// Reconcile creates or updates the primary and canary services to prepare for the canary release process targeted on the K8s service
func (c *KubernetesServiceRouter) Reconcile(canary *flaggerv1.Canary) error {
targetName := canary.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
canaryName := fmt.Sprintf("%s-canary", targetName)
svc, err := c.kubeClient.CoreV1().Services(canary.Namespace).Get(targetName, metav1.GetOptions{})
if err != nil {
return err
}
// canary svc
err = c.reconcileCanaryService(canary, canaryName, svc)
if err != nil {
return err
}
// primary svc
err = c.reconcilePrimaryService(canary, primaryName, svc)
if err != nil {
return err
}
return nil
}
func (c *KubernetesServiceRouter) SetRoutes(canary *flaggerv1.Canary, primaryRoute int, canaryRoute int) error {
return nil
}
func (c *KubernetesServiceRouter) GetRoutes(canary *flaggerv1.Canary) (primaryRoute int, canaryRoute int, err error) {
return 0, 0, nil
}
func (c *KubernetesServiceRouter) reconcileCanaryService(canary *flaggerv1.Canary, name string, src *corev1.Service) error {
current, err := c.kubeClient.CoreV1().Services(canary.Namespace).Get(name, metav1.GetOptions{})
if errors.IsNotFound(err) {
return c.createService(canary, name, src)
}
if err != nil {
return fmt.Errorf("service %s query error %v", name, err)
}
new := buildService(canary, name, src)
if new.Spec.Type == "ClusterIP" {
// We can't change this immutable field
new.Spec.ClusterIP = current.Spec.ClusterIP
}
// We can't change this immutable field
new.ObjectMeta.UID = current.ObjectMeta.UID
new.ObjectMeta.ResourceVersion = current.ObjectMeta.ResourceVersion
_, err = c.kubeClient.CoreV1().Services(canary.Namespace).Update(new)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s.%s updated", new.GetName(), canary.Namespace)
return nil
}
func (c *KubernetesServiceRouter) reconcilePrimaryService(canary *flaggerv1.Canary, name string, src *corev1.Service) error {
_, err := c.kubeClient.CoreV1().Services(canary.Namespace).Get(name, metav1.GetOptions{})
if errors.IsNotFound(err) {
return c.createService(canary, name, src)
}
if err != nil {
return fmt.Errorf("service %s query error %v", name, err)
}
return nil
}
func (c *KubernetesServiceRouter) createService(canary *flaggerv1.Canary, name string, src *corev1.Service) error {
svc := buildService(canary, name, src)
if svc.Spec.Type == "ClusterIP" {
// Reset and let K8s assign the IP. Otherwise we get an error due to the IP is already assigned
svc.Spec.ClusterIP = ""
}
// Let K8s set this. Otherwise K8s API complains with "resourceVersion should not be set on objects to be created"
svc.ObjectMeta.ResourceVersion = ""
_, err := c.kubeClient.CoreV1().Services(canary.Namespace).Create(svc)
if err != nil {
return err
}
c.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("Service %s.%s created", svc.GetName(), canary.Namespace)
return nil
}
test/e2e-kubernetes-tests-svc.sh
Executable
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env bash
# This script runs e2e tests for Blue/Green initialization, analysis and promotion
# Prerequisites: Kubernetes Kind, Kustomize
set -o errexit
REPO_ROOT=$(git rev-parse --show-toplevel)
export KUBECONFIG="$(kind get kubeconfig-path --name="kind")"
echo '>>> Creating test namespace'
kubectl create namespace test
echo '>>> Installing the load tester'
kubectl apply -k ${REPO_ROOT}/kustomize/tester
kubectl -n test rollout status deployment/flagger-loadtester
echo '>>> Initialising canary'
kubectl apply -f ${REPO_ROOT}/test/e2e-workload.yaml
kubectl apply -n test -f - <<EOS
apiVersion: v1
kind: Service
metadata:
name: podinfo
spec:
ports:
- name: http
port: 9898
protocol: TCP
targetPort: http
selector:
app: podinfo
type: ClusterIP
EOS
cat <<EOF | kubectl apply -f -
apiVersion: flagger.app/v1alpha3
kind: Canary
metadata:
name: podinfo
namespace: test
spec:
provider: kubernetes
targetRef:
apiVersion: core/v1
kind: Service
name: podinfo
progressDeadlineSeconds: 60
service:
port: 9898
canaryAnalysis:
interval: 15s
threshold: 10
iterations: 5
metrics:
- name: request-success-rate
threshold: 99
interval: 1m
- name: request-duration
threshold: 500
interval: 30s
webhooks:
- name: "gate"
type: confirm-rollout
url: http://flagger-loadtester.test/gate/approve
- name: acceptance-test
type: pre-rollout
url: http://flagger-loadtester.test/
timeout: 10s
metadata:
type: bash
cmd: "curl -sd 'test' http://podinfo-canary/token | grep token"
- name: load-test
url: http://flagger-loadtester.test/
timeout: 5s
metadata:
type: cmd
cmd: "hey -z 10m -q 10 -c 2 http://podinfo-canary.test/"
logCmdOutput: "true"
EOF
echo '>>> Waiting for primary to be ready'
kubectl -n test rollout status deploy/podinfo
retries=50
count=0
ok=false
until ${ok}; do
kubectl -n test get canary/podinfo | grep 'Initialized' && ok=true || ok=false
sleep 5
count=$(($count + 1))
if [[ ${count} -eq ${retries} ]]; then
kubectl -n flagger-system logs deployment/flagger
echo "No more retries left"
exit 1
fi
done
echo '✔ Canary initialization test passed'
echo '>>> Initializing secondary'
kubectl apply -f ${REPO_ROOT}/test/e2e-workload-v2.yaml
echo '>>> Waiting for secondary to be ready'
kubectl -n test rollout status deploy/podinfo-v2
echo '>>> Triggering canary deployment'
kubectl apply -n test -f - <<EOS
apiVersion: v1
kind: Service
metadata:
name: podinfo
spec:
ports:
- name: http
port: 9898
protocol: TCP
targetPort: http
selector:
app: podinfo-v2
type: ClusterIP
EOS
echo '>>> Waiting for canary promotion'
retries=50
count=0
ok=false
until ${ok}; do
kubectl -n test describe service/podinfo-primary | grep 'podinfo-v2' && ok=true || ok=false
sleep 10
kubectl -n flagger-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 flagger-system logs deployment/flagger
echo "No more retries left"
exit 1
fi
done
echo '✔ Canary promotion test passed'
kubectl -n flagger-system logs deployment/flagger
test/e2e-workload-v2.yaml
+68
View File
@@ -0,0 +1,68 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: podinfo-v2
namespace: test
labels:
app: podinfo-v2
spec:
minReadySeconds: 5
revisionHistoryLimit: 5
progressDeadlineSeconds: 60
strategy:
rollingUpdate:
maxUnavailable: 0
type: RollingUpdate
selector:
matchLabels:
app: podinfo-v2
template:
metadata:
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9797"
labels:
app: podinfo-v2
spec:
containers:
- name: podinfod
image: stefanprodan/podinfo:3.1.1
imagePullPolicy: IfNotPresent
ports:
- name: http
containerPort: 9898
protocol: TCP
- name: http-metrics
containerPort: 9797
protocol: TCP
- name: grpc
containerPort: 9999
protocol: TCP
command:
- ./podinfo
- --port=9898
- --port-metrics=9797
- --grpc-port=9999
- --grpc-service-name=podinfo
- --level=info
- --random-delay=false
- --random-error=false
livenessProbe:
httpGet:
port: 9898
path: /healthz
initialDelaySeconds: 5
timeoutSeconds: 5
readinessProbe:
httpGet:
port: 9898
path: /readyz
initialDelaySeconds: 5
timeoutSeconds: 5
resources:
limits:
cpu: 1000m
memory: 128Mi
requests:
cpu: 1m
memory: 16Mi