flagger/pkg/router/istio.go

361 lines
11 KiB
Go

package router
import (
"fmt"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3"
istiov1alpha3 "github.com/weaveworks/flagger/pkg/apis/istio/v1alpha3"
clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned"
"go.uber.org/zap"
"k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/apis/meta/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/kubernetes"
)
// IstioRouter is managing Istio virtual services
type IstioRouter struct {
kubeClient kubernetes.Interface
istioClient clientset.Interface
flaggerClient clientset.Interface
logger *zap.SugaredLogger
}
// Sync creates or updates the Istio virtual service
func (ir *IstioRouter) Sync(canary *flaggerv1.Canary) error {
targetName := canary.Spec.TargetRef.Name
primaryName := fmt.Sprintf("%s-primary", targetName)
// set hosts and add the ClusterIP service host if it doesn't exists
hosts := canary.Spec.Service.Hosts
var hasServiceHost bool
for _, h := range hosts {
if h == targetName {
hasServiceHost = true
break
}
}
if !hasServiceHost {
hosts = append(hosts, targetName)
}
// set gateways and add the mesh gateway if it doesn't exists
gateways := canary.Spec.Service.Gateways
var hasMeshGateway bool
for _, g := range gateways {
if g == "mesh" {
hasMeshGateway = true
break
}
}
if !hasMeshGateway {
gateways = append(gateways, "mesh")
}
// create destinations with primary weight 100% and canary weight 0%
canaryRoute := []istiov1alpha3.DestinationWeight{
{
Destination: istiov1alpha3.Destination{
Host: primaryName,
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: 100,
},
{
Destination: istiov1alpha3.Destination{
Host: fmt.Sprintf("%s-canary", targetName),
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: 0,
},
}
newSpec := istiov1alpha3.VirtualServiceSpec{
Hosts: hosts,
Gateways: gateways,
Http: []istiov1alpha3.HTTPRoute{
{
Match: canary.Spec.Service.Match,
Rewrite: canary.Spec.Service.Rewrite,
Timeout: canary.Spec.Service.Timeout,
Retries: canary.Spec.Service.Retries,
CorsPolicy: canary.Spec.Service.CorsPolicy,
AppendHeaders: addHeaders(canary),
Route: canaryRoute,
},
},
}
if len(canary.Spec.CanaryAnalysis.Match) > 0 {
canaryMatch := mergeMatchConditions(canary.Spec.CanaryAnalysis.Match, canary.Spec.Service.Match)
newSpec.Http = []istiov1alpha3.HTTPRoute{
{
Match: canaryMatch,
Rewrite: canary.Spec.Service.Rewrite,
Timeout: canary.Spec.Service.Timeout,
Retries: canary.Spec.Service.Retries,
CorsPolicy: canary.Spec.Service.CorsPolicy,
AppendHeaders: addHeaders(canary),
Route: canaryRoute,
},
{
Match: canary.Spec.Service.Match,
Rewrite: canary.Spec.Service.Rewrite,
Timeout: canary.Spec.Service.Timeout,
Retries: canary.Spec.Service.Retries,
CorsPolicy: canary.Spec.Service.CorsPolicy,
AppendHeaders: addHeaders(canary),
Route: []istiov1alpha3.DestinationWeight{
{
Destination: istiov1alpha3.Destination{
Host: primaryName,
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: 100,
},
},
},
}
}
virtualService, err := ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Get(targetName, metav1.GetOptions{})
// insert
if errors.IsNotFound(err) {
virtualService = &istiov1alpha3.VirtualService{
ObjectMeta: metav1.ObjectMeta{
Name: targetName,
Namespace: canary.Namespace,
OwnerReferences: []metav1.OwnerReference{
*metav1.NewControllerRef(canary, schema.GroupVersionKind{
Group: flaggerv1.SchemeGroupVersion.Group,
Version: flaggerv1.SchemeGroupVersion.Version,
Kind: flaggerv1.CanaryKind,
}),
},
},
Spec: newSpec,
}
_, err = ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Create(virtualService)
if err != nil {
return fmt.Errorf("VirtualService %s.%s create error %v", targetName, canary.Namespace, err)
}
ir.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("VirtualService %s.%s created", virtualService.GetName(), canary.Namespace)
return nil
}
if err != nil {
return fmt.Errorf("VirtualService %s.%s query error %v", targetName, canary.Namespace, err)
}
// update service but keep the original destination weights
if virtualService != nil {
if diff := cmp.Diff(newSpec, virtualService.Spec, cmpopts.IgnoreTypes(istiov1alpha3.DestinationWeight{})); diff != "" {
vtClone := virtualService.DeepCopy()
vtClone.Spec = newSpec
if len(virtualService.Spec.Http) > 0 {
vtClone.Spec.Http[0].Route = virtualService.Spec.Http[0].Route
}
_, err = ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Update(vtClone)
if err != nil {
return fmt.Errorf("VirtualService %s.%s update error %v", targetName, canary.Namespace, err)
}
ir.logger.With("canary", fmt.Sprintf("%s.%s", canary.Name, canary.Namespace)).
Infof("VirtualService %s.%s updated", virtualService.GetName(), canary.Namespace)
}
}
return nil
}
// GetRoutes returns the destinations weight for primary and canary
func (ir *IstioRouter) GetRoutes(canary *flaggerv1.Canary) (
primaryWeight int,
canaryWeight int,
err error,
) {
targetName := canary.Spec.TargetRef.Name
vs := &istiov1alpha3.VirtualService{}
vs, err = ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Get(targetName, v1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
err = fmt.Errorf("VirtualService %s.%s not found", targetName, canary.Namespace)
return
}
err = fmt.Errorf("VirtualService %s.%s query error %v", targetName, canary.Namespace, err)
return
}
var httpRoute istiov1alpha3.HTTPRoute
for _, http := range vs.Spec.Http {
for _, r := range http.Route {
if r.Destination.Host == fmt.Sprintf("%s-canary", targetName) {
httpRoute = http
break
}
}
}
for _, route := range httpRoute.Route {
if route.Destination.Host == fmt.Sprintf("%s-primary", targetName) {
primaryWeight = route.Weight
}
if route.Destination.Host == fmt.Sprintf("%s-canary", targetName) {
canaryWeight = route.Weight
}
}
if primaryWeight == 0 && canaryWeight == 0 {
err = fmt.Errorf("VirtualService %s.%s does not contain routes for %s-primary and %s-canary",
targetName, canary.Namespace, targetName, targetName)
}
return
}
// SetRoutes updates the destinations weight for primary and canary
func (ir *IstioRouter) SetRoutes(
canary *flaggerv1.Canary,
primaryWeight int,
canaryWeight int,
) error {
targetName := canary.Spec.TargetRef.Name
vs, err := ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Get(targetName, v1.GetOptions{})
if err != nil {
if errors.IsNotFound(err) {
return fmt.Errorf("VirtualService %s.%s not found", targetName, canary.Namespace)
}
return fmt.Errorf("VirtualService %s.%s query error %v", targetName, canary.Namespace, err)
}
vsCopy := vs.DeepCopy()
// weighted routing (progressive canary)
vsCopy.Spec.Http = []istiov1alpha3.HTTPRoute{
{
Match: canary.Spec.Service.Match,
Rewrite: canary.Spec.Service.Rewrite,
Timeout: canary.Spec.Service.Timeout,
Retries: canary.Spec.Service.Retries,
CorsPolicy: canary.Spec.Service.CorsPolicy,
AppendHeaders: addHeaders(canary),
Route: []istiov1alpha3.DestinationWeight{
{
Destination: istiov1alpha3.Destination{
Host: fmt.Sprintf("%s-primary", targetName),
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: primaryWeight,
},
{
Destination: istiov1alpha3.Destination{
Host: fmt.Sprintf("%s-canary", targetName),
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: canaryWeight,
},
},
},
}
// fix routing (A/B testing)
if len(canary.Spec.CanaryAnalysis.Match) > 0 {
// merge the common routes with the canary ones
canaryMatch := mergeMatchConditions(canary.Spec.CanaryAnalysis.Match, canary.Spec.Service.Match)
vsCopy.Spec.Http = []istiov1alpha3.HTTPRoute{
{
Match: canaryMatch,
Rewrite: canary.Spec.Service.Rewrite,
Timeout: canary.Spec.Service.Timeout,
Retries: canary.Spec.Service.Retries,
CorsPolicy: canary.Spec.Service.CorsPolicy,
AppendHeaders: addHeaders(canary),
Route: []istiov1alpha3.DestinationWeight{
{
Destination: istiov1alpha3.Destination{
Host: fmt.Sprintf("%s-primary", targetName),
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: primaryWeight,
},
{
Destination: istiov1alpha3.Destination{
Host: fmt.Sprintf("%s-canary", targetName),
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: canaryWeight,
},
},
},
{
Match: canary.Spec.Service.Match,
Rewrite: canary.Spec.Service.Rewrite,
Timeout: canary.Spec.Service.Timeout,
Retries: canary.Spec.Service.Retries,
CorsPolicy: canary.Spec.Service.CorsPolicy,
AppendHeaders: addHeaders(canary),
Route: []istiov1alpha3.DestinationWeight{
{
Destination: istiov1alpha3.Destination{
Host: fmt.Sprintf("%s-primary", targetName),
Port: istiov1alpha3.PortSelector{
Number: uint32(canary.Spec.Service.Port),
},
},
Weight: primaryWeight,
},
},
},
}
}
vs, err = ir.istioClient.NetworkingV1alpha3().VirtualServices(canary.Namespace).Update(vsCopy)
if err != nil {
return fmt.Errorf("VirtualService %s.%s update failed: %v", targetName, canary.Namespace, err)
}
return nil
}
// addHeaders applies headers before forwarding a request to the destination service
// compatible with Istio 1.0.x and 1.1.0
func addHeaders(canary *flaggerv1.Canary) (headers map[string]string) {
if canary.Spec.Service.Headers != nil &&
canary.Spec.Service.Headers.Request != nil &&
len(canary.Spec.Service.Headers.Request.Add) > 0 {
headers = canary.Spec.Service.Headers.Request.Add
}
return
}
// mergeMatchConditions appends the URI match rules to canary conditions
func mergeMatchConditions(canary, defaults []istiov1alpha3.HTTPMatchRequest) []istiov1alpha3.HTTPMatchRequest {
for i := range canary {
for _, d := range defaults {
if d.Uri != nil {
canary[i].Uri = d.Uri
}
}
}
return canary
}