diff --git a/cmd/flagger/main.go b/cmd/flagger/main.go index ed2fa659..fc3cad5b 100644 --- a/cmd/flagger/main.go +++ b/cmd/flagger/main.go @@ -2,6 +2,9 @@ package main import ( "flag" + "log" + "time" + _ "github.com/istio/glog" clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned" informers "github.com/weaveworks/flagger/pkg/client/informers/externalversions" @@ -9,6 +12,7 @@ import ( "github.com/weaveworks/flagger/pkg/logging" "github.com/weaveworks/flagger/pkg/metrics" "github.com/weaveworks/flagger/pkg/notifier" + "github.com/weaveworks/flagger/pkg/router" "github.com/weaveworks/flagger/pkg/server" "github.com/weaveworks/flagger/pkg/signals" "github.com/weaveworks/flagger/pkg/version" @@ -17,8 +21,6 @@ import ( _ "k8s.io/client-go/plugin/pkg/client/auth/gcp" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" - "log" - "time" ) var ( @@ -126,6 +128,8 @@ func main() { // start HTTP server go server.ListenAndServe(port, 3*time.Second, logger, stopCh) + routerFactory := router.NewFactory(cfg, kubeClient, flaggerClient, logger, meshClient) + c := controller.NewController( kubeClient, meshClient, @@ -135,6 +139,7 @@ func main() { metricsServer, logger, slack, + routerFactory, meshProvider, version.VERSION, ) diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index 6fbc2b91..813a2719 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -2,10 +2,12 @@ package controller import ( "fmt" - "github.com/weaveworks/flagger/pkg/metrics" "sync" "time" + "github.com/weaveworks/flagger/pkg/metrics" + "github.com/weaveworks/flagger/pkg/router" + "github.com/google/go-cmp/cmp" flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3" clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned" @@ -45,6 +47,7 @@ type Controller struct { observer metrics.Observer recorder metrics.Recorder notifier *notifier.Slack + routerFactory *router.Factory meshProvider string } @@ -57,6 +60,7 @@ func NewController( metricServer string, logger *zap.SugaredLogger, notifier *notifier.Slack, + routerFactory *router.Factory, meshProvider string, version string, ) *Controller { @@ -100,6 +104,7 @@ func NewController( observer: metrics.NewObserver(metricServer), recorder: recorder, notifier: notifier, + routerFactory: routerFactory, meshProvider: meshProvider, } diff --git a/pkg/controller/scheduler.go b/pkg/controller/scheduler.go index 4c699f73..b5f67c53 100644 --- a/pkg/controller/scheduler.go +++ b/pkg/controller/scheduler.go @@ -2,12 +2,13 @@ package controller import ( "fmt" - "github.com/weaveworks/flagger/pkg/router" "strings" "time" + "github.com/weaveworks/flagger/pkg/router" + flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3" - "k8s.io/apimachinery/pkg/apis/meta/v1" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) // scheduleCanaries synchronises the canary map with the jobs map, @@ -96,11 +97,10 @@ func (c *Controller) advanceCanary(name string, namespace string, skipLivenessCh } // init routers - routerFactory := router.NewFactory(c.kubeClient, c.flaggerClient, c.logger, c.istioClient) - meshRouter := routerFactory.MeshRouter(c.meshProvider) + meshRouter := c.routerFactory.MeshRouter(c.meshProvider) // create or update ClusterIP services - if err := routerFactory.KubernetesRouter().Reconcile(cd); err != nil { + if err := c.routerFactory.KubernetesRouter().Reconcile(cd); err != nil { c.recordEventWarningf(cd, "%v", err) return } diff --git a/pkg/router/factory.go b/pkg/router/factory.go index 4a879392..8a318a9b 100644 --- a/pkg/router/factory.go +++ b/pkg/router/factory.go @@ -1,23 +1,28 @@ package router import ( + "context" + clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned" "go.uber.org/zap" "k8s.io/client-go/kubernetes" + restclient "k8s.io/client-go/rest" ) type Factory struct { + kubeConfig *restclient.Config kubeClient kubernetes.Interface meshClient clientset.Interface flaggerClient clientset.Interface logger *zap.SugaredLogger } -func NewFactory(kubeClient kubernetes.Interface, +func NewFactory(kubeConfig *restclient.Config, kubeClient kubernetes.Interface, flaggerClient clientset.Interface, logger *zap.SugaredLogger, meshClient clientset.Interface) *Factory { return &Factory{ + kubeConfig: kubeConfig, meshClient: meshClient, kubeClient: kubeClient, flaggerClient: flaggerClient, @@ -36,18 +41,26 @@ func (factory *Factory) KubernetesRouter() *KubernetesRouter { // MeshRouter returns a service mesh router (Istio or AppMesh) func (factory *Factory) MeshRouter(provider string) Interface { - if provider == "appmesh" { + switch provider { + case "appmesh": return &AppMeshRouter{ logger: factory.logger, flaggerClient: factory.flaggerClient, kubeClient: factory.kubeClient, appmeshClient: factory.meshClient, } - } - return &IstioRouter{ - logger: factory.logger, - flaggerClient: factory.flaggerClient, - kubeClient: factory.kubeClient, - istioClient: factory.meshClient, + case "supergloo": + supergloo, err := NewSuperglooRouter(context.TODO(), factory.flaggerClient, factory.logger, factory.kubeConfig) + if err != nil { + panic("TODO") + } + return supergloo + default: + return &IstioRouter{ + logger: factory.logger, + flaggerClient: factory.flaggerClient, + kubeClient: factory.kubeClient, + istioClient: factory.meshClient, + } } } diff --git a/pkg/router/supergloo.go b/pkg/router/supergloo.go index 226cd4f0..2b0ecdfd 100644 --- a/pkg/router/supergloo.go +++ b/pkg/router/supergloo.go @@ -8,45 +8,50 @@ import ( "github.com/solo-io/solo-kit/pkg/api/v1/clients/factory" "github.com/solo-io/solo-kit/pkg/api/v1/clients/kube" solokitcore "github.com/solo-io/solo-kit/pkg/api/v1/resources/core" + solokiterror "github.com/solo-io/solo-kit/pkg/errors" gloov1 "github.com/solo-io/gloo/projects/gloo/pkg/api/v1" supergloov1 "github.com/solo-io/supergloo/pkg/api/v1" flaggerv1 "github.com/weaveworks/flagger/pkg/apis/flagger/v1alpha3" clientset "github.com/weaveworks/flagger/pkg/client/clientset/versioned" "go.uber.org/zap" - "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" ) // SuperglooRouter is managing Istio virtual services type SuperglooRouter struct { - kubeClient kubernetes.Interface - rrClient supergloov1.RoutingRuleClient flaggerClient clientset.Interface logger *zap.SugaredLogger + targetMesh solokitcore.ResourceRef } -func (ir *SuperglooRouter) foo() error { +func NewSuperglooRouter(ctx context.Context, flaggerClient clientset.Interface, logger *zap.SugaredLogger, cfg *rest.Config) (*SuperglooRouter, error) { - sharedCache := kube.NewKubeCache(context.TODO()) - RoutingRuleClient, err := supergloov1.NewRoutingRuleClient(&factory.KubeResourceClientFactory{ - Crd: supergloov1.RoutingRuleCrd, - // Cfg: ir.kubeClient, - SharedCache: sharedCache, + sharedCache := kube.NewKubeCache(ctx) + routingRuleClient, err := supergloov1.NewRoutingRuleClient(&factory.KubeResourceClientFactory{ + Crd: supergloov1.RoutingRuleCrd, + Cfg: cfg, + SharedCache: sharedCache, + SkipCrdCreation: true, }) if err != nil { - return fmt.Errorf("creating RoutingRule client %v", err) + return nil, fmt.Errorf("creating RoutingRule client %v", err) } - if err := RoutingRuleClient.Register(); err != nil { - return err + if err := routingRuleClient.Register(); err != nil { + return nil, err } - ir.rrClient = RoutingRuleClient - return nil + // TODO(yuval-k): un hard code this + targetMesh := solokitcore.ResourceRef{ + Namespace: "supergloo-system", + Name: "yuval", + } + return &SuperglooRouter{rrClient: routingRuleClient, logger: logger, flaggerClient: flaggerClient, targetMesh: targetMesh}, nil } // Reconcile creates or updates the Istio virtual service func (ir *SuperglooRouter) Reconcile(canary *flaggerv1.Canary) error { - return nil + return ir.SetRoutes(canary, 100, 0) } // GetRoutes returns the destinations weight for primary and canary @@ -55,6 +60,32 @@ func (ir *SuperglooRouter) GetRoutes(canary *flaggerv1.Canary) ( canaryWeight int, err error, ) { + targetName := canary.Spec.TargetRef.Name + var rr *supergloov1.RoutingRule + rr, err = ir.rrClient.Read(canary.Namespace, targetName, solokitclients.ReadOpts{}) + if err != nil { + return + } + traffic := rr.GetSpec().GetTrafficShifting() + if traffic == nil { + err = fmt.Errorf("target rule is not for traffic shifting") + return + } + dests := traffic.GetDestinations().GetDestinations() + for _, dest := range dests { + if dest.GetDestination().GetUpstream().Name == upstreamName(canary.Namespace, fmt.Sprintf("%s-primary", targetName), canary.Spec.Service.Port) { + primaryWeight = int(dest.Weight) + } + if dest.GetDestination().GetUpstream().Name == upstreamName(canary.Namespace, fmt.Sprintf("%s-canary", targetName), canary.Spec.Service.Port) { + canaryWeight = int(dest.Weight) + } + } + + if primaryWeight == 0 && canaryWeight == 0 { + err = fmt.Errorf("RoutingRule %s.%s does not contain routes for %s-primary and %s-canary", + targetName, canary.Namespace, targetName, targetName) + } + return } @@ -73,7 +104,41 @@ func (ir *SuperglooRouter) SetRoutes( // and is the same as targetName := canary.Spec.TargetRef.Name + destinations := []*gloov1.WeightedDestination{} + if primaryWeight != 0 { + destinations = append(destinations, &gloov1.WeightedDestination{ + Destination: &gloov1.Destination{ + Upstream: solokitcore.ResourceRef{ + Name: upstreamName(canary.Namespace, fmt.Sprintf("%s-primary", targetName), canary.Spec.Service.Port), + Namespace: "supergloo-system", + }, + }, + Weight: uint32(primaryWeight), + }) + } + + if canaryWeight != 0 { + destinations = append(destinations, &gloov1.WeightedDestination{ + Destination: &gloov1.Destination{ + Upstream: solokitcore.ResourceRef{ + Name: upstreamName(canary.Namespace, fmt.Sprintf("%s-canary", targetName), canary.Spec.Service.Port), + Namespace: "supergloo-system", + }, + }, + Weight: uint32(canaryWeight), + }) + } + + if len(destinations) == 0 { + return fmt.Errorf("RoutingRule %s.%s update failed: no valid weights", targetName, canary.Namespace) + } + rule := &supergloov1.RoutingRule{ + Metadata: solokitcore.Metadata{ + Name: targetName, + Namespace: canary.Namespace, + }, + TargetMesh: &ir.targetMesh, DestinationSelector: &supergloov1.PodSelector{ SelectorType: &supergloov1.PodSelector_UpstreamSelector_{ UpstreamSelector: &supergloov1.PodSelector_UpstreamSelector{ @@ -88,36 +153,25 @@ func (ir *SuperglooRouter) SetRoutes( RuleType: &supergloov1.RoutingRuleSpec_TrafficShifting{ TrafficShifting: &supergloov1.TrafficShifting{ Destinations: &gloov1.MultiDestination{ - Destinations: []*gloov1.WeightedDestination{ - { - Destination: &gloov1.Destination{ - Upstream: solokitcore.ResourceRef{ - Name: upstreamName(canary.Namespace, fmt.Sprintf("%s-primary", targetName), canary.Spec.Service.Port), - Namespace: "supergloo-system", - }, - }, - Weight: uint32(primaryWeight), - }, { - Destination: &gloov1.Destination{ - Upstream: solokitcore.ResourceRef{ - Name: upstreamName(canary.Namespace, fmt.Sprintf("%s-canary", targetName), canary.Spec.Service.Port), - Namespace: "supergloo-system", - }, - }, - Weight: uint32(canaryWeight), - }, - }, + Destinations: destinations, }, }, }, }, } - // TODO: read to get the resource version + if oldRr, err := ir.rrClient.Read(rule.Metadata.Namespace, rule.Metadata.Name, solokitclients.ReadOpts{}); err != nil { + // ignore not exist errors.. + if !solokiterror.IsNotExist(err) { + return fmt.Errorf("RoutingRule %s.%s read failed: %v", targetName, canary.Namespace, err) + } + } else { + rule.Metadata.ResourceVersion = oldRr.Metadata.ResourceVersion + } + _, err := ir.rrClient.Write(rule, solokitclients.WriteOpts{OverwriteExisting: true}) if err != nil { return fmt.Errorf("RoutingRule %s.%s update failed: %v", targetName, canary.Namespace, err) - } return nil }