Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

PRE-NEG changes #11

Merged
merged 2 commits into from
Oct 11, 2017
Merged

PRE-NEG changes #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
db8666f
move ControllerContext from glbc
freehan Oct 10, 2017
7da4efe
make translator public
freehan Oct 10, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions cmd/glbc/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ import (
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"

"k8s.io/ingress-gce/pkg/backends"
"k8s.io/ingress-gce/pkg/context"
"k8s.io/ingress-gce/pkg/controller"
"k8s.io/ingress-gce/pkg/loadbalancers"
"k8s.io/ingress-gce/pkg/storage"
Expand Down Expand Up @@ -249,10 +250,11 @@ func main() {
SvcPort: intstr.FromInt(int(port)),
}

var namer *utils.Namer
var cloud *gce.GCECloud
if *inCluster || *useRealCloud {
// Create cluster manager
namer, err := newNamer(kubeClient, *clusterName, controller.DefaultFirewallName)
namer, err = newNamer(kubeClient, *clusterName, controller.DefaultFirewallName)
if err != nil {
glog.Fatalf("%v", err)
}
Expand Down Expand Up @@ -287,7 +289,7 @@ func main() {
clusterManager = controller.NewFakeClusterManager(*clusterName, controller.DefaultFirewallName).ClusterManager
}

ctx := controller.NewControllerContext(kubeClient, *watchNamespace, *resyncPeriod)
ctx := context.NewControllerContext(kubeClient, *watchNamespace, *resyncPeriod, false)

// Start loadbalancer controller
lbc, err := controller.NewLoadBalancerController(kubeClient, ctx, clusterManager)
Expand Down
60 changes: 60 additions & 0 deletions pkg/context/context.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
/*
Copyright 2017 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package context

import (
informerv1 "k8s.io/client-go/informers/core/v1"
informerv1beta1 "k8s.io/client-go/informers/extensions/v1beta1"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
"time"
)

// ControllerContext holds
type ControllerContext struct {
IngressInformer cache.SharedIndexInformer
ServiceInformer cache.SharedIndexInformer
PodInformer cache.SharedIndexInformer
NodeInformer cache.SharedIndexInformer
EndpointInformer cache.SharedIndexInformer
// Stop is the stop channel shared among controllers
StopCh chan struct{}
}

func NewControllerContext(kubeClient kubernetes.Interface, namespace string, resyncPeriod time.Duration, enableEndpointsInformer bool) *ControllerContext {
context := &ControllerContext{
IngressInformer: informerv1beta1.NewIngressInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
ServiceInformer: informerv1.NewServiceInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
PodInformer: informerv1.NewPodInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
NodeInformer: informerv1.NewNodeInformer(kubeClient, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
StopCh: make(chan struct{}),
}
if enableEndpointsInformer {
context.EndpointInformer = informerv1.NewEndpointsInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
}
return context
}

func (ctx *ControllerContext) Start() {
go ctx.IngressInformer.Run(ctx.StopCh)
go ctx.ServiceInformer.Run(ctx.StopCh)
go ctx.PodInformer.Run(ctx.StopCh)
go ctx.NodeInformer.Run(ctx.StopCh)
if ctx.EndpointInformer != nil {
go ctx.EndpointInformer.Run(ctx.StopCh)
}
}
43 changes: 7 additions & 36 deletions pkg/controller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,13 @@ import (
apiv1 "k8s.io/api/core/v1"
extensions "k8s.io/api/extensions/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
informerv1 "k8s.io/client-go/informers/core/v1"
informerv1beta1 "k8s.io/client-go/informers/extensions/v1beta1"
"k8s.io/client-go/kubernetes"
scheme "k8s.io/client-go/kubernetes/scheme"
unversionedcore "k8s.io/client-go/kubernetes/typed/core/v1"
listers "k8s.io/client-go/listers/core/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"

"k8s.io/ingress-gce/pkg/context"
"k8s.io/ingress-gce/pkg/firewalls"
"k8s.io/ingress-gce/pkg/loadbalancers"
)
Expand All @@ -55,33 +53,6 @@ var (
storeSyncPollPeriod = 5 * time.Second
)

// ControllerContext holds
type ControllerContext struct {
IngressInformer cache.SharedIndexInformer
ServiceInformer cache.SharedIndexInformer
PodInformer cache.SharedIndexInformer
NodeInformer cache.SharedIndexInformer
// Stop is the stop channel shared among controllers
StopCh chan struct{}
}

func NewControllerContext(kubeClient kubernetes.Interface, namespace string, resyncPeriod time.Duration) *ControllerContext {
return &ControllerContext{
IngressInformer: informerv1beta1.NewIngressInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
ServiceInformer: informerv1.NewServiceInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
PodInformer: informerv1.NewPodInformer(kubeClient, namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
NodeInformer: informerv1.NewNodeInformer(kubeClient, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}),
StopCh: make(chan struct{}),
}
}

func (ctx *ControllerContext) Start() {
go ctx.IngressInformer.Run(ctx.StopCh)
go ctx.ServiceInformer.Run(ctx.StopCh)
go ctx.PodInformer.Run(ctx.StopCh)
go ctx.NodeInformer.Run(ctx.StopCh)
}

// LoadBalancerController watches the kubernetes api and adds/removes services
// from the loadbalancer, via loadBalancerConfig.
type LoadBalancerController struct {
Expand All @@ -101,7 +72,7 @@ type LoadBalancerController struct {
recorder record.EventRecorder
nodeQueue *taskQueue
ingQueue *taskQueue
tr *GCETranslator
Translator *GCETranslator
stopCh chan struct{}
// stopLock is used to enforce only a single call to Stop is active.
// Needed because we allow stopping through an http endpoint and
Expand All @@ -120,7 +91,7 @@ type LoadBalancerController struct {
// - clusterManager: A ClusterManager capable of creating all cloud resources
// required for L7 loadbalancing.
// - resyncPeriod: Watchers relist from the Kubernetes API server this often.
func NewLoadBalancerController(kubeClient kubernetes.Interface, ctx *ControllerContext, clusterManager *ClusterManager) (*LoadBalancerController, error) {
func NewLoadBalancerController(kubeClient kubernetes.Interface, ctx *context.ControllerContext, clusterManager *ClusterManager) (*LoadBalancerController, error) {
eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartLogging(glog.Infof)
eventBroadcaster.StartRecordingToSink(&unversionedcore.EventSinkImpl{
Expand Down Expand Up @@ -196,7 +167,7 @@ func NewLoadBalancerController(kubeClient kubernetes.Interface, ctx *ControllerC
// Nodes are updated every 10s and we don't care, so no update handler.
})

lbc.tr = &GCETranslator{&lbc}
lbc.Translator = &GCETranslator{&lbc}
lbc.tlsLoader = &apiServerTLSLoader{client: lbc.client}
glog.V(3).Infof("Created new loadbalancer controller")

Expand Down Expand Up @@ -285,8 +256,8 @@ func (lbc *LoadBalancerController) sync(key string) (err error) {
return err
}

allNodePorts := lbc.tr.toNodePorts(&allIngresses)
gceNodePorts := lbc.tr.toNodePorts(&gceIngresses)
allNodePorts := lbc.Translator.toNodePorts(&allIngresses)
gceNodePorts := lbc.Translator.toNodePorts(&gceIngresses)
lbNames := lbc.ingLister.Store.ListKeys()
lbs, err := lbc.toRuntimeInfo(gceIngresses)
if err != nil {
Expand Down Expand Up @@ -368,7 +339,7 @@ func (lbc *LoadBalancerController) sync(key string) (err error) {
return syncError
}

if urlMap, err := lbc.tr.toURLMap(&ing); err != nil {
if urlMap, err := lbc.Translator.toURLMap(&ing); err != nil {
syncError = fmt.Errorf("%v, convert to url map error %v", syncError, err)
} else if err := l7.UpdateUrlMap(urlMap); err != nil {
lbc.recorder.Eventf(&ing, apiv1.EventTypeWarning, "UrlMap", err.Error())
Expand Down
3 changes: 2 additions & 1 deletion pkg/controller/controller_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ import (
"k8s.io/client-go/kubernetes/fake"
"k8s.io/kubernetes/pkg/api"

"k8s.io/ingress-gce/pkg/context"
"k8s.io/ingress-gce/pkg/firewalls"
"k8s.io/ingress-gce/pkg/loadbalancers"
"k8s.io/ingress-gce/pkg/utils"
Expand All @@ -53,7 +54,7 @@ func defaultBackendName(clusterName string) string {
// newLoadBalancerController create a loadbalancer controller.
func newLoadBalancerController(t *testing.T, cm *fakeClusterManager) *LoadBalancerController {
kubeClient := fake.NewSimpleClientset()
ctx := NewControllerContext(kubeClient, api_v1.NamespaceAll, 1*time.Second)
ctx := context.NewControllerContext(kubeClient, api_v1.NamespaceAll, 1*time.Second, true)
lb, err := NewLoadBalancerController(kubeClient, ctx, cm.ClusterManager)
if err != nil {
t.Fatalf("%v", err)
Expand Down
10 changes: 5 additions & 5 deletions pkg/controller/utils_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ func TestZoneListing(t *testing.T) {
"zone-2": {"n2"},
}
addNodes(lbc, zoneToNode)
zones, err := lbc.tr.ListZones()
zones, err := lbc.Translator.ListZones()
if err != nil {
t.Errorf("Failed to list zones: %v", err)
}
Expand Down Expand Up @@ -99,7 +99,7 @@ func TestProbeGetter(t *testing.T) {
}
addPods(lbc, nodePortToHealthCheck, api_v1.NamespaceDefault)
for p, exp := range nodePortToHealthCheck {
got, err := lbc.tr.GetProbe(p)
got, err := lbc.Translator.GetProbe(p)
if err != nil || got == nil {
t.Errorf("Failed to get probe for node port %v: %v", p, err)
} else if getProbePath(got) != exp {
Expand All @@ -121,7 +121,7 @@ func TestProbeGetterNamedPort(t *testing.T) {
pod.Spec.Containers[0].ReadinessProbe.Handler.HTTPGet.Port = intstr.IntOrString{Type: intstr.String, StrVal: "test"}
}
for p, exp := range nodePortToHealthCheck {
got, err := lbc.tr.GetProbe(p)
got, err := lbc.Translator.GetProbe(p)
if err != nil || got == nil {
t.Errorf("Failed to get probe for node port %v: %v", p, err)
} else if getProbePath(got) != exp {
Expand Down Expand Up @@ -172,7 +172,7 @@ func TestProbeGetterCrossNamespace(t *testing.T) {
addPods(lbc, nodePortToHealthCheck, api_v1.NamespaceDefault)

for p, exp := range nodePortToHealthCheck {
got, err := lbc.tr.GetProbe(p)
got, err := lbc.Translator.GetProbe(p)
if err != nil || got == nil {
t.Errorf("Failed to get probe for node port %v: %v", p, err)
} else if getProbePath(got) != exp {
Expand Down Expand Up @@ -254,7 +254,7 @@ func addNodes(lbc *LoadBalancerController, zoneToNode map[string][]string) {
lbc.nodeLister.Indexer.Add(n)
}
}
lbc.CloudClusterManager.instancePool.Init(lbc.tr)
lbc.CloudClusterManager.instancePool.Init(lbc.Translator)
}

func getProbePath(p *api_v1.Probe) string {
Expand Down