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

basics seem working!

Browse Source
This commit is contained in:
Yuval Kohavi
2019-04-16 15:10:08 -04:00
parent e5612bca50
commit 868482c240
5 changed files with 129 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/flagger/main.go
+7 -2
View File
@@ -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,
)
pkg/controller/controller.go
+6 -1
View File
@@ -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,
}
pkg/controller/scheduler.go
+5 -5
View File
@@ -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
}
pkg/router/factory.go
+21 -8
View File
@@ -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,
}
}
}
pkg/router/supergloo.go
+90 -36
View File
@@ -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
}