From 66773fea4b788f5732d0490be191ec151dc97132 Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Tue, 30 May 2017 16:37:49 -0700 Subject: [PATCH 1/8] Rename gce_loadbalancer.go to gce_loadbalancer_external.go --- .../gce/{gce_loadbalancer.go => gce_loadbalancer_external.go} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename pkg/cloudprovider/providers/gce/{gce_loadbalancer.go => gce_loadbalancer_external.go} (100%) diff --git a/pkg/cloudprovider/providers/gce/gce_loadbalancer.go b/pkg/cloudprovider/providers/gce/gce_loadbalancer_external.go similarity index 100% rename from pkg/cloudprovider/providers/gce/gce_loadbalancer.go rename to pkg/cloudprovider/providers/gce/gce_loadbalancer_external.go From b631061f0580c788e6e88f3b5477e807b436f577 Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Tue, 30 May 2017 16:44:04 -0700 Subject: [PATCH 2/8] Rename gce_staticip.go to gce_addresses.go --- .../providers/gce/{gce_staticip.go => gce_addresses.go} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename pkg/cloudprovider/providers/gce/{gce_staticip.go => gce_addresses.go} (100%) diff --git a/pkg/cloudprovider/providers/gce/gce_staticip.go b/pkg/cloudprovider/providers/gce/gce_addresses.go similarity index 100% rename from pkg/cloudprovider/providers/gce/gce_staticip.go rename to pkg/cloudprovider/providers/gce/gce_addresses.go From 2cdaf1f32b1238517ae3526c1e8cde996df0c8e7 Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Thu, 1 Jun 2017 15:22:34 -0700 Subject: [PATCH 3/8] Refactor compute API calls --- pkg/cloudprovider/providers/gce/gce.go | 2 + .../providers/gce/gce_addresses.go | 58 +++++++++---- .../providers/gce/gce_backendservice.go | 40 ++++----- .../providers/gce/gce_forwardingrule.go | 57 +++++++++---- .../providers/gce/gce_healthchecks.go | 21 +---- .../providers/gce/gce_instancegroup.go | 23 ++--- .../providers/gce/gce_instances.go | 33 ++++++++ .../providers/gce/gce_targetpool.go | 84 +++++++++++++++++++ pkg/cloudprovider/providers/gce/gce_util.go | 15 ++++ pkg/cloudprovider/providers/gce/gce_zones.go | 32 ++++++- 10 files changed, 281 insertions(+), 84 deletions(-) create mode 100644 pkg/cloudprovider/providers/gce/gce_targetpool.go diff --git a/pkg/cloudprovider/providers/gce/gce.go b/pkg/cloudprovider/providers/gce/gce.go index 5cb2d7bf447..90e2570a429 100644 --- a/pkg/cloudprovider/providers/gce/gce.go +++ b/pkg/cloudprovider/providers/gce/gce.go @@ -78,6 +78,8 @@ const ( // GCECloud is an implementation of Interface, LoadBalancer and Instances for Google Compute Engine. type GCECloud struct { + // ClusterID contains functionality for getting (and initializing) the ingress-uid. Call GCECloud.Initialize() + // for the cloudprovider to start watching the configmap. ClusterID ClusterID service *compute.Service diff --git a/pkg/cloudprovider/providers/gce/gce_addresses.go b/pkg/cloudprovider/providers/gce/gce_addresses.go index 36aa119e898..877d744ddfa 100644 --- a/pkg/cloudprovider/providers/gce/gce_addresses.go +++ b/pkg/cloudprovider/providers/gce/gce_addresses.go @@ -22,21 +22,20 @@ import ( compute "google.golang.org/api/compute/v1" ) -func newStaticIPMetricContext(request string) *metricContext { +func newAddressMetricContext(request, region string) *metricContext { return &metricContext{ start: time.Now(), - attributes: []string{"staticip_" + request, unusedMetricLabel, unusedMetricLabel}, + attributes: []string{"address_" + request, region, unusedMetricLabel}, } } -// ReserveGlobalStaticIP creates a global static IP. +// ReserveGlobalAddress creates a global address. // Caller is allocated a random IP if they do not specify an ipAddress. If an // ipAddress is specified, it must belong to the current project, eg: an // ephemeral IP associated with a global forwarding rule. -func (gce *GCECloud) ReserveGlobalStaticIP(name, ipAddress string) (address *compute.Address, err error) { - mc := newStaticIPMetricContext("reserve") - op, err := gce.service.GlobalAddresses.Insert(gce.projectID, &compute.Address{Name: name, Address: ipAddress}).Do() - +func (gce *GCECloud) ReserveGlobalAddress(addr *compute.Address) (*compute.Address, error) { + mc := newAddressMetricContext("reserve", "") + op, err := gce.service.GlobalAddresses.Insert(gce.projectID, addr).Do() if err != nil { return nil, mc.Observe(err) } @@ -45,12 +44,12 @@ func (gce *GCECloud) ReserveGlobalStaticIP(name, ipAddress string) (address *com return nil, err } - return gce.service.GlobalAddresses.Get(gce.projectID, name).Do() + return gce.GetGlobalAddress(addr.Name) } -// DeleteGlobalStaticIP deletes a global static IP by name. -func (gce *GCECloud) DeleteGlobalStaticIP(name string) error { - mc := newStaticIPMetricContext("delete") +// DeleteGlobalAddress deletes a global address by name. +func (gce *GCECloud) DeleteGlobalAddress(name string) error { + mc := newAddressMetricContext("delete", "") op, err := gce.service.GlobalAddresses.Delete(gce.projectID, name).Do() if err != nil { return mc.Observe(err) @@ -58,9 +57,40 @@ func (gce *GCECloud) DeleteGlobalStaticIP(name string) error { return gce.waitForGlobalOp(op, mc) } -// GetGlobalStaticIP returns the global static IP by name. -func (gce *GCECloud) GetGlobalStaticIP(name string) (*compute.Address, error) { - mc := newStaticIPMetricContext("get") +// GetGlobalAddress returns the global address by name. +func (gce *GCECloud) GetGlobalAddress(name string) (*compute.Address, error) { + mc := newAddressMetricContext("get", "") v, err := gce.service.GlobalAddresses.Get(gce.projectID, name).Do() return v, mc.Observe(err) } + +// ReserveRegionAddress creates a region address +func (gce *GCECloud) ReserveRegionAddress(addr *compute.Address, region string) (*compute.Address, error) { + mc := newAddressMetricContext("reserve", region) + op, err := gce.service.Addresses.Insert(gce.projectID, region, addr).Do() + if err != nil { + return nil, mc.Observe(err) + } + if err := gce.waitForRegionOp(op, region, mc); err != nil { + return nil, err + } + + return gce.GetRegionAddress(addr.Name, region) +} + +// DeleteRegionAddress deletes a region address by name. +func (gce *GCECloud) DeleteRegionAddress(name, region string) error { + mc := newAddressMetricContext("delete", region) + op, err := gce.service.Addresses.Delete(gce.projectID, region, name).Do() + if err != nil { + return mc.Observe(err) + } + return gce.waitForRegionOp(op, region, mc) +} + +// GetRegionAddress returns the region address by name +func (gce *GCECloud) GetRegionAddress(name, region string) (*compute.Address, error) { + mc := newAddressMetricContext("get", region) + v, err := gce.service.Addresses.Get(gce.projectID, region, name).Do() + return v, mc.Observe(err) +} diff --git a/pkg/cloudprovider/providers/gce/gce_backendservice.go b/pkg/cloudprovider/providers/gce/gce_backendservice.go index 1d79dfde0d3..d460c6157d8 100644 --- a/pkg/cloudprovider/providers/gce/gce_backendservice.go +++ b/pkg/cloudprovider/providers/gce/gce_backendservice.go @@ -23,23 +23,23 @@ import ( compute "google.golang.org/api/compute/v1" ) -func newBackendServiceMetricContext(request string) *metricContext { +func newBackendServiceMetricContext(request, region string) *metricContext { return &metricContext{ start: time.Now(), - attributes: []string{"backendservice_" + request, unusedMetricLabel, unusedMetricLabel}, + attributes: []string{"backendservice_" + request, region, unusedMetricLabel}, } } // GetGlobalBackendService retrieves a backend by name. func (gce *GCECloud) GetGlobalBackendService(name string) (*compute.BackendService, error) { - mc := newBackendServiceMetricContext("get") + mc := newBackendServiceMetricContext("get", "") v, err := gce.service.BackendServices.Get(gce.projectID, name).Do() return v, mc.Observe(err) } // UpdateGlobalBackendService applies the given BackendService as an update to an existing service. func (gce *GCECloud) UpdateGlobalBackendService(bg *compute.BackendService) error { - mc := newBackendServiceMetricContext("update") + mc := newBackendServiceMetricContext("update", "") op, err := gce.service.BackendServices.Update(gce.projectID, bg.Name, bg).Do() if err != nil { return mc.Observe(err) @@ -50,7 +50,7 @@ func (gce *GCECloud) UpdateGlobalBackendService(bg *compute.BackendService) erro // DeleteGlobalBackendService deletes the given BackendService by name. func (gce *GCECloud) DeleteGlobalBackendService(name string) error { - mc := newBackendServiceMetricContext("delete") + mc := newBackendServiceMetricContext("delete", "") op, err := gce.service.BackendServices.Delete(gce.projectID, name).Do() if err != nil { if isHTTPErrorCode(err, http.StatusNotFound) { @@ -64,7 +64,7 @@ func (gce *GCECloud) DeleteGlobalBackendService(name string) error { // CreateGlobalBackendService creates the given BackendService. func (gce *GCECloud) CreateGlobalBackendService(bg *compute.BackendService) error { - mc := newBackendServiceMetricContext("create") + mc := newBackendServiceMetricContext("create", "") op, err := gce.service.BackendServices.Insert(gce.projectID, bg).Do() if err != nil { return mc.Observe(err) @@ -75,7 +75,7 @@ func (gce *GCECloud) CreateGlobalBackendService(bg *compute.BackendService) erro // ListGlobalBackendServices lists all backend services in the project. func (gce *GCECloud) ListGlobalBackendServices() (*compute.BackendServiceList, error) { - mc := newBackendServiceMetricContext("list") + mc := newBackendServiceMetricContext("list", "") // TODO: use PageToken to list all not just the first 500 v, err := gce.service.BackendServices.List(gce.projectID).Do() return v, mc.Observe(err) @@ -85,7 +85,7 @@ func (gce *GCECloud) ListGlobalBackendServices() (*compute.BackendServiceList, e // name, in the given instanceGroup. The instanceGroupLink is the fully // qualified self link of an instance group. func (gce *GCECloud) GetGlobalBackendServiceHealth(name string, instanceGroupLink string) (*compute.BackendServiceGroupHealth, error) { - mc := newBackendServiceMetricContext("get_health") + mc := newBackendServiceMetricContext("get_health", "") groupRef := &compute.ResourceGroupReference{Group: instanceGroupLink} v, err := gce.service.BackendServices.GetHealth(gce.projectID, name, groupRef).Do() return v, mc.Observe(err) @@ -93,25 +93,25 @@ func (gce *GCECloud) GetGlobalBackendServiceHealth(name string, instanceGroupLin // GetRegionBackendService retrieves a backend by name. func (gce *GCECloud) GetRegionBackendService(name, region string) (*compute.BackendService, error) { - mc := newBackendServiceMetricContext("get") + mc := newBackendServiceMetricContext("get", region) v, err := gce.service.RegionBackendServices.Get(gce.projectID, region, name).Do() return v, mc.Observe(err) } // UpdateRegionBackendService applies the given BackendService as an update to an existing service. -func (gce *GCECloud) UpdateRegionBackendService(bg *compute.BackendService) error { - mc := newBackendServiceMetricContext("update") - op, err := gce.service.RegionBackendServices.Update(gce.projectID, bg.Region, bg.Name, bg).Do() +func (gce *GCECloud) UpdateRegionBackendService(bg *compute.BackendService, region string) error { + mc := newBackendServiceMetricContext("update", region) + op, err := gce.service.RegionBackendServices.Update(gce.projectID, region, bg.Name, bg).Do() if err != nil { return mc.Observe(err) } - return gce.waitForRegionOp(op, bg.Region, mc) + return gce.waitForRegionOp(op, region, mc) } // DeleteRegionBackendService deletes the given BackendService by name. func (gce *GCECloud) DeleteRegionBackendService(name, region string) error { - mc := newBackendServiceMetricContext("delete") + mc := newBackendServiceMetricContext("delete", region) op, err := gce.service.RegionBackendServices.Delete(gce.projectID, region, name).Do() if err != nil { if isHTTPErrorCode(err, http.StatusNotFound) { @@ -124,19 +124,19 @@ func (gce *GCECloud) DeleteRegionBackendService(name, region string) error { } // CreateRegionBackendService creates the given BackendService. -func (gce *GCECloud) CreateRegionBackendService(bg *compute.BackendService) error { - mc := newBackendServiceMetricContext("create") - op, err := gce.service.RegionBackendServices.Insert(gce.projectID, bg.Region, bg).Do() +func (gce *GCECloud) CreateRegionBackendService(bg *compute.BackendService, region string) error { + mc := newBackendServiceMetricContext("create", region) + op, err := gce.service.RegionBackendServices.Insert(gce.projectID, region, bg).Do() if err != nil { return mc.Observe(err) } - return gce.waitForRegionOp(op, bg.Region, mc) + return gce.waitForRegionOp(op, region, mc) } // ListRegionBackendServices lists all backend services in the project. func (gce *GCECloud) ListRegionBackendServices(region string) (*compute.BackendServiceList, error) { - mc := newBackendServiceMetricContext("list") + mc := newBackendServiceMetricContext("list", region) // TODO: use PageToken to list all not just the first 500 v, err := gce.service.RegionBackendServices.List(gce.projectID, region).Do() return v, mc.Observe(err) @@ -146,7 +146,7 @@ func (gce *GCECloud) ListRegionBackendServices(region string) (*compute.BackendS // name, in the given instanceGroup. The instanceGroupLink is the fully // qualified self link of an instance group. func (gce *GCECloud) GetRegionalBackendServiceHealth(name, region string, instanceGroupLink string) (*compute.BackendServiceGroupHealth, error) { - mc := newBackendServiceMetricContext("get_health") + mc := newBackendServiceMetricContext("get_health", region) groupRef := &compute.ResourceGroupReference{Group: instanceGroupLink} v, err := gce.service.RegionBackendServices.GetHealth(gce.projectID, region, name, groupRef).Do() return v, mc.Observe(err) diff --git a/pkg/cloudprovider/providers/gce/gce_forwardingrule.go b/pkg/cloudprovider/providers/gce/gce_forwardingrule.go index 0eedf0559cb..de800547ceb 100644 --- a/pkg/cloudprovider/providers/gce/gce_forwardingrule.go +++ b/pkg/cloudprovider/providers/gce/gce_forwardingrule.go @@ -17,7 +17,6 @@ limitations under the License. package gce import ( - "net/http" "time" compute "google.golang.org/api/compute/v1" @@ -44,8 +43,7 @@ func (gce *GCECloud) CreateGlobalForwardingRule(targetProxyLink, ip, name, portR } op, err := gce.service.GlobalForwardingRules.Insert(gce.projectID, rule).Do() if err != nil { - mc.Observe(err) - return nil, err + return nil, mc.Observe(err) } if err = gce.waitForGlobalOp(op, mc); err != nil { return nil, err @@ -56,13 +54,12 @@ func (gce *GCECloud) CreateGlobalForwardingRule(targetProxyLink, ip, name, portR // SetProxyForGlobalForwardingRule links the given TargetHttp(s)Proxy with the given GlobalForwardingRule. // targetProxyLink is the SelfLink of a TargetHttp(s)Proxy. -func (gce *GCECloud) SetProxyForGlobalForwardingRule(fw *compute.ForwardingRule, targetProxyLink string) error { +func (gce *GCECloud) SetProxyForGlobalForwardingRule(forwardingRuleName, targetProxyLink string) error { mc := newForwardingRuleMetricContext("set_proxy", "") op, err := gce.service.GlobalForwardingRules.SetTarget( - gce.projectID, fw.Name, &compute.TargetReference{Target: targetProxyLink}).Do() + gce.projectID, forwardingRuleName, &compute.TargetReference{Target: targetProxyLink}).Do() if err != nil { - mc.Observe(err) - return err + return mc.Observe(err) } return gce.waitForGlobalOp(op, mc) @@ -73,13 +70,7 @@ func (gce *GCECloud) DeleteGlobalForwardingRule(name string) error { mc := newForwardingRuleMetricContext("delete", "") op, err := gce.service.GlobalForwardingRules.Delete(gce.projectID, name).Do() if err != nil { - if isHTTPErrorCode(err, http.StatusNotFound) { - mc.Observe(nil) - return nil - } - - mc.Observe(err) - return err + return mc.Observe(err) } return gce.waitForGlobalOp(op, mc) @@ -99,3 +90,41 @@ func (gce *GCECloud) ListGlobalForwardingRules() (*compute.ForwardingRuleList, e v, err := gce.service.GlobalForwardingRules.List(gce.projectID).Do() return v, mc.Observe(err) } + +// GetRegionForwardingRule returns the RegionalForwardingRule by name & region. +func (gce *GCECloud) GetRegionForwardingRule(name, region string) (*compute.ForwardingRule, error) { + mc := newForwardingRuleMetricContext("get", region) + v, err := gce.service.ForwardingRules.Get(gce.projectID, region, name).Do() + return v, mc.Observe(err) +} + +// ListRegionForwardingRules lists all RegionalForwardingRules in the project & region. +func (gce *GCECloud) ListRegionForwardingRules(region string) (*compute.ForwardingRuleList, error) { + mc := newForwardingRuleMetricContext("list", region) + // TODO: use PageToken to list all not just the first 500 + v, err := gce.service.ForwardingRules.List(gce.projectID, region).Do() + return v, mc.Observe(err) +} + +// CreateRegionForwardingRule creates and returns a +// RegionalForwardingRule that points to the given BackendService +func (gce *GCECloud) CreateRegionForwardingRule(rule *compute.ForwardingRule, region string) error { + mc := newForwardingRuleMetricContext("create", region) + op, err := gce.service.ForwardingRules.Insert(gce.projectID, region, rule).Do() + if err != nil { + return mc.Observe(err) + } + + return gce.waitForRegionOp(op, region, mc) +} + +// DeleteRegionForwardingRule deletes the RegionalForwardingRule by name & region. +func (gce *GCECloud) DeleteRegionForwardingRule(name, region string) error { + mc := newForwardingRuleMetricContext("delete", region) + op, err := gce.service.ForwardingRules.Delete(gce.projectID, region, name).Do() + if err != nil { + return mc.Observe(err) + } + + return gce.waitForRegionOp(op, region, mc) +} diff --git a/pkg/cloudprovider/providers/gce/gce_healthchecks.go b/pkg/cloudprovider/providers/gce/gce_healthchecks.go index 7bd060b54a7..4f5f90c6a1d 100644 --- a/pkg/cloudprovider/providers/gce/gce_healthchecks.go +++ b/pkg/cloudprovider/providers/gce/gce_healthchecks.go @@ -17,7 +17,6 @@ limitations under the License. package gce import ( - "fmt" "time" "k8s.io/kubernetes/pkg/api/v1" @@ -208,28 +207,12 @@ func GetNodesHealthCheckPort() int32 { return lbNodesHealthCheckPort } -// getNodesHealthCheckPath returns the health check path used by the GCE load +// GetNodesHealthCheckPath returns the health check path used by the GCE load // balancers (l4) for performing health checks on nodes. -func getNodesHealthCheckPath() string { +func GetNodesHealthCheckPath() string { return nodesHealthCheckPath } -// makeNodesHealthCheckName returns name of the health check resource used by -// the GCE load balancers (l4) for performing health checks on nodes. -func makeNodesHealthCheckName(clusterID string) string { - return fmt.Sprintf("k8s-%v-node", clusterID) -} - -// MakeHealthCheckFirewallName returns the firewall name used by the GCE load -// balancers (l4) for performing health checks. -func MakeHealthCheckFirewallName(clusterID, hcName string, isNodesHealthCheck bool) string { - if isNodesHealthCheck { - // TODO: Change below fwName to match the proposed schema: k8s-{clusteriD}-{namespace}-{name}-{shortid}-hc. - return makeNodesHealthCheckName(clusterID) + "-http-hc" - } - return "k8s-" + hcName + "-http-hc" -} - // isAtLeastMinNodesHealthCheckVersion checks if a version is higher than // `minNodesHealthCheckVersion`. func isAtLeastMinNodesHealthCheckVersion(vstring string) bool { diff --git a/pkg/cloudprovider/providers/gce/gce_instancegroup.go b/pkg/cloudprovider/providers/gce/gce_instancegroup.go index c25aeb1af08..f04d5676cca 100644 --- a/pkg/cloudprovider/providers/gce/gce_instancegroup.go +++ b/pkg/cloudprovider/providers/gce/gce_instancegroup.go @@ -81,20 +81,16 @@ func (gce *GCECloud) ListInstancesInInstanceGroup(name string, zone string, stat // AddInstancesToInstanceGroup adds the given instances to the given // instance group. -func (gce *GCECloud) AddInstancesToInstanceGroup(name string, zone string, instanceNames []string) error { +func (gce *GCECloud) AddInstancesToInstanceGroup(name string, zone string, instanceRefs []*compute.InstanceReference) error { mc := newInstanceGroupMetricContext("add_instances", zone) - if len(instanceNames) == 0 { + if len(instanceRefs) == 0 { return nil } - // Adding the same instance twice will result in a 4xx error - instances := []*compute.InstanceReference{} - for _, ins := range instanceNames { - instances = append(instances, &compute.InstanceReference{Instance: makeHostURL(gce.projectID, zone, ins)}) - } + op, err := gce.service.InstanceGroups.AddInstances( gce.projectID, zone, name, &compute.InstanceGroupsAddInstancesRequest{ - Instances: instances, + Instances: instanceRefs, }).Do() if err != nil { return mc.Observe(err) @@ -105,21 +101,16 @@ func (gce *GCECloud) AddInstancesToInstanceGroup(name string, zone string, insta // RemoveInstancesFromInstanceGroup removes the given instances from // the instance group. -func (gce *GCECloud) RemoveInstancesFromInstanceGroup(name string, zone string, instanceNames []string) error { +func (gce *GCECloud) RemoveInstancesFromInstanceGroup(name string, zone string, instanceRefs []*compute.InstanceReference) error { mc := newInstanceGroupMetricContext("remove_instances", zone) - if len(instanceNames) == 0 { + if len(instanceRefs) == 0 { return nil } - instances := []*compute.InstanceReference{} - for _, ins := range instanceNames { - instanceLink := makeHostURL(gce.projectID, zone, ins) - instances = append(instances, &compute.InstanceReference{Instance: instanceLink}) - } op, err := gce.service.InstanceGroups.RemoveInstances( gce.projectID, zone, name, &compute.InstanceGroupsRemoveInstancesRequest{ - Instances: instances, + Instances: instanceRefs, }).Do() if err != nil { return mc.Observe(err) diff --git a/pkg/cloudprovider/providers/gce/gce_instances.go b/pkg/cloudprovider/providers/gce/gce_instances.go index 33437a34628..a2f6125be13 100644 --- a/pkg/cloudprovider/providers/gce/gce_instances.go +++ b/pkg/cloudprovider/providers/gce/gce_instances.go @@ -28,6 +28,7 @@ import ( computealpha "google.golang.org/api/compute/v0.beta" compute "google.golang.org/api/compute/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" @@ -35,6 +36,10 @@ import ( "k8s.io/kubernetes/pkg/cloudprovider" ) +const ( + defaultZone = "" +) + func newInstancesMetricContext(request, zone string) *metricContext { return &metricContext{ start: time.Now(), @@ -42,6 +47,34 @@ func newInstancesMetricContext(request, zone string) *metricContext { } } +func splitNodesByZone(nodes []*v1.Node) map[string][]*v1.Node { + zones := make(map[string][]*v1.Node) + for _, n := range nodes { + z := getZone(n) + if z != defaultZone { + zones[z] = append(zones[z], n) + } + } + return zones +} + +func getZone(n *v1.Node) string { + zone, ok := n.Labels[metav1.LabelZoneFailureDomain] + if !ok { + return defaultZone + } + return zone +} + +// ToInstanceReferences returns instance references by links +func (gce *GCECloud) ToInstanceReferences(zone string, instanceNames []string) (refs []*compute.InstanceReference) { + for _, ins := range instanceNames { + instanceLink := makeHostURL(gce.projectID, zone, ins) + refs = append(refs, &compute.InstanceReference{Instance: instanceLink}) + } + return refs +} + // NodeAddresses is an implementation of Instances.NodeAddresses. func (gce *GCECloud) NodeAddresses(_ types.NodeName) ([]v1.NodeAddress, error) { internalIP, err := metadata.Get("instance/network-interfaces/0/ip") diff --git a/pkg/cloudprovider/providers/gce/gce_targetpool.go b/pkg/cloudprovider/providers/gce/gce_targetpool.go new file mode 100644 index 00000000000..10304bfe93b --- /dev/null +++ b/pkg/cloudprovider/providers/gce/gce_targetpool.go @@ -0,0 +1,84 @@ +/* +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 gce + +import ( + "time" + + compute "google.golang.org/api/compute/v1" +) + +func newTargetPoolMetricContext(request, region string) *metricContext { + return &metricContext{ + start: time.Now(), + attributes: []string{"targetpool_" + request, region, unusedMetricLabel}, + } +} + +// GetTargetPool returns the TargetPool by name. +func (gce *GCECloud) GetTargetPool(name, region string) (*compute.TargetPool, error) { + mc := newTargetPoolMetricContext("get", region) + v, err := gce.service.TargetPools.Get(gce.projectID, region, name).Do() + return v, mc.Observe(err) +} + +// CreateTargetPool creates the passed TargetPool +func (gce *GCECloud) CreateTargetPool(tp *compute.TargetPool, region string) (*compute.TargetPool, error) { + mc := newTargetPoolMetricContext("create", region) + op, err := gce.service.TargetPools.Insert(gce.projectID, region, tp).Do() + if err != nil { + return nil, mc.Observe(err) + } + + if err := gce.waitForRegionOp(op, region, mc); err != nil { + return nil, err + } + + return gce.GetTargetPool(tp.Name, region) +} + +// DeleteTargetPool deletes TargetPool by name. +func (gce *GCECloud) DeleteTargetPool(name, region string) error { + mc := newTargetPoolMetricContext("delete", region) + op, err := gce.service.TargetPools.Delete(gce.projectID, region, name).Do() + if err != nil { + return mc.Observe(err) + } + return gce.waitForRegionOp(op, region, mc) +} + +// AddInstancesToTargetPool adds instances by link to the TargetPool +func (gce *GCECloud) AddInstancesToTargetPool(name, region string, instanceRefs []*compute.InstanceReference) error { + add := &compute.TargetPoolsAddInstanceRequest{Instances: instanceRefs} + mc := newTargetPoolMetricContext("add_instances", region) + op, err := gce.service.TargetPools.AddInstance(gce.projectID, region, name, add).Do() + if err != nil { + return mc.Observe(err) + } + return gce.waitForRegionOp(op, region, mc) +} + +// RemoveInstancesToTargetPool removes instances by link to the TargetPool +func (gce *GCECloud) RemoveInstancesFromTargetPool(name, region string, instanceRefs []*compute.InstanceReference) error { + remove := &compute.TargetPoolsRemoveInstanceRequest{Instances: instanceRefs} + mc := newTargetPoolMetricContext("remove_instances", region) + op, err := gce.service.TargetPools.RemoveInstance(gce.projectID, region, name, remove).Do() + if err != nil { + return mc.Observe(err) + } + return gce.waitForRegionOp(op, region, mc) +} diff --git a/pkg/cloudprovider/providers/gce/gce_util.go b/pkg/cloudprovider/providers/gce/gce_util.go index 90b51759f22..bcbd523c889 100644 --- a/pkg/cloudprovider/providers/gce/gce_util.go +++ b/pkg/cloudprovider/providers/gce/gce_util.go @@ -134,3 +134,18 @@ func equalStringSets(x, y []string) bool { yString := sets.NewString(y...) return xString.Equal(yString) } + +func isNotFound(err error) bool { + return isHTTPErrorCode(err, http.StatusNotFound) +} + +func ignoreNotFound(err error) error { + if err == nil || isNotFound(err) { + return nil + } + return err +} + +func isNotFoundOrInUse(err error) bool { + return isNotFound(err) || isInUsedByError(err) +} diff --git a/pkg/cloudprovider/providers/gce/gce_zones.go b/pkg/cloudprovider/providers/gce/gce_zones.go index c8cec5694fb..881fe652fc1 100644 --- a/pkg/cloudprovider/providers/gce/gce_zones.go +++ b/pkg/cloudprovider/providers/gce/gce_zones.go @@ -16,11 +16,41 @@ limitations under the License. package gce -import "k8s.io/kubernetes/pkg/cloudprovider" +import ( + "fmt" + "time" + compute "google.golang.org/api/compute/v1" + + "k8s.io/kubernetes/pkg/cloudprovider" +) + +func newZonesMetricContext(request, region string) *metricContext { + return &metricContext{ + start: time.Now(), + attributes: []string{"zones_" + request, region, unusedMetricLabel}, + } +} + +// GetZone creates a cloudprovider.Zone of the current zone and region func (gce *GCECloud) GetZone() (cloudprovider.Zone, error) { return cloudprovider.Zone{ FailureDomain: gce.localZone, Region: gce.region, }, nil } + +// ListZonesInRegion returns all zones in a GCP region +func (gce *GCECloud) ListZonesInRegion(region string) ([]*compute.Zone, error) { + mc := newZonesMetricContext("list", region) + filter := fmt.Sprintf("region eq %v", gce.getRegionLink(region)) + list, err := gce.service.Zones.List(gce.projectID).Filter(filter).Do() + if err != nil { + return nil, mc.Observe(err) + } + return list.Items, mc.Observe(err) +} + +func (gce *GCECloud) getRegionLink(region string) string { + return fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%v/regions/%v", gce.projectID, region) +} From 1283d655381967531b5aacf788eb06b9b9800356 Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Thu, 1 Jun 2017 15:23:11 -0700 Subject: [PATCH 4/8] Modify external LB logic --- .../gce/gce_loadbalancer_external.go | 335 +++--------------- 1 file changed, 49 insertions(+), 286 deletions(-) diff --git a/pkg/cloudprovider/providers/gce/gce_loadbalancer_external.go b/pkg/cloudprovider/providers/gce/gce_loadbalancer_external.go index b3cf4e7792c..74aefee65ae 100644 --- a/pkg/cloudprovider/providers/gce/gce_loadbalancer_external.go +++ b/pkg/cloudprovider/providers/gce/gce_loadbalancer_external.go @@ -17,13 +17,10 @@ limitations under the License. package gce import ( - "flag" "fmt" - "net" "net/http" "strconv" "strings" - "time" "k8s.io/apimachinery/pkg/types" utilerrors "k8s.io/apimachinery/pkg/util/errors" @@ -37,95 +34,7 @@ import ( compute "google.golang.org/api/compute/v1" ) -type cidrs struct { - ipn netsets.IPNet - isSet bool -} - -var ( - lbSrcRngsFlag cidrs -) - -func newLoadBalancerMetricContext(request, region string) *metricContext { - return &metricContext{ - start: time.Now(), - attributes: []string{"loadbalancer_" + request, region, unusedMetricLabel}, - } -} - -func newTargetPoolMetricContext(request, region string) *metricContext { - return &metricContext{ - start: time.Now(), - attributes: []string{"targetpool_" + request, region, unusedMetricLabel}, - } -} - -func newAddressMetricContext(request, region string) *metricContext { - return &metricContext{ - start: time.Now(), - attributes: []string{"address_" + request, region, unusedMetricLabel}, - } -} - -func init() { - var err error - lbSrcRngsFlag.ipn, err = netsets.ParseIPNets([]string{"130.211.0.0/22", "35.191.0.0/16", "209.85.152.0/22", "209.85.204.0/22"}...) - if err != nil { - panic("Incorrect default GCE L7 source ranges") - } - - flag.Var(&lbSrcRngsFlag, "cloud-provider-gce-lb-src-cidrs", "CIDRS opened in GCE firewall for LB traffic proxy & health checks") -} - -// String is the method to format the flag's value, part of the flag.Value interface. -func (c *cidrs) String() string { - return strings.Join(c.ipn.StringSlice(), ",") -} - -// Set supports a value of CSV or the flag repeated multiple times -func (c *cidrs) Set(value string) error { - // On first Set(), clear the original defaults - if !c.isSet { - c.isSet = true - c.ipn = make(netsets.IPNet) - } else { - return fmt.Errorf("GCE LB CIDRS have already been set") - } - - for _, cidr := range strings.Split(value, ",") { - _, ipnet, err := net.ParseCIDR(cidr) - if err != nil { - return err - } - - c.ipn.Insert(ipnet) - } - return nil -} - -// LoadBalancerSrcRanges contains the ranges of ips used by the GCE load balancers (l4 & L7) -// for proxying client requests and performing health checks. -func LoadBalancerSrcRanges() []string { - return lbSrcRngsFlag.ipn.StringSlice() -} - -// GetLoadBalancer is an implementation of LoadBalancer.GetLoadBalancer -func (gce *GCECloud) GetLoadBalancer(clusterName string, service *v1.Service) (*v1.LoadBalancerStatus, bool, error) { - loadBalancerName := cloudprovider.GetLoadBalancerName(service) - fwd, err := gce.service.ForwardingRules.Get(gce.projectID, gce.region, loadBalancerName).Do() - if err == nil { - status := &v1.LoadBalancerStatus{} - status.Ingress = []v1.LoadBalancerIngress{{IP: fwd.IPAddress}} - - return status, true, nil - } - if isHTTPErrorCode(err, http.StatusNotFound) { - return nil, false, nil - } - return nil, false, err -} - -// EnsureLoadBalancer is an implementation of LoadBalancer.EnsureLoadBalancer. +// ensureExternalLoadBalancer is the external implementation of LoadBalancer.EnsureLoadBalancer. // Our load balancers in GCE consist of four separate GCE resources - a static // IP address, a firewall rule, a target pool, and a forwarding rule. This // function has to manage all of them. @@ -133,7 +42,7 @@ func (gce *GCECloud) GetLoadBalancer(clusterName string, service *v1.Service) (* // Due to an interesting series of design decisions, this handles both creating // new load balancers and updating existing load balancers, recognizing when // each is needed. -func (gce *GCECloud) EnsureLoadBalancer(clusterName string, apiService *v1.Service, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) { +func (gce *GCECloud) ensureExternalLoadBalancer(clusterName string, apiService *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) { if len(nodes) == 0 { return nil, fmt.Errorf("Cannot EnsureLoadBalancer() with no hosts") } @@ -200,10 +109,13 @@ func (gce *GCECloud) EnsureLoadBalancer(clusterName string, apiService *v1.Servi return } if isSafeToReleaseIP { - if err := gce.deleteStaticIP(loadBalancerName, gce.region); err != nil { + if err := gce.DeleteRegionAddress(loadBalancerName, gce.region); err != nil && !isNotFound(err) { glog.Errorf("failed to release static IP %s for load balancer (%v(%v), %v): %v", ipAddress, loadBalancerName, serviceName, gce.region, err) + } else if isNotFound(err) { + glog.V(2).Infof("EnsureLoadBalancer(%v(%v)): address %s is not reserved.", loadBalancerName, serviceName, ipAddress) + } else { + glog.V(2).Infof("EnsureLoadBalancer(%v(%v)): released static IP %s", loadBalancerName, serviceName, ipAddress) } - glog.V(2).Infof("EnsureLoadBalancer(%v(%v)): released static IP %s", loadBalancerName, serviceName, ipAddress) } else { glog.Warningf("orphaning static IP %s during update of load balancer (%v(%v), %v): %v", ipAddress, loadBalancerName, serviceName, gce.region, err) } @@ -329,7 +241,7 @@ func (gce *GCECloud) EnsureLoadBalancer(clusterName string, apiService *v1.Servi // target pool to use local traffic health check. glog.V(2).Infof("Updating from nodes health checks to local traffic health checks for service %v LB %v", apiService.Name, loadBalancerName) if supportsNodesHealthCheck { - hcToDelete = makeHttpHealthCheck(makeNodesHealthCheckName(clusterID), getNodesHealthCheckPath(), GetNodesHealthCheckPort()) + hcToDelete = makeHttpHealthCheck(makeNodesHealthCheckName(clusterID), GetNodesHealthCheckPath(), GetNodesHealthCheckPort()) } tpNeedsUpdate = true } @@ -345,7 +257,7 @@ func (gce *GCECloud) EnsureLoadBalancer(clusterName string, apiService *v1.Servi tpNeedsUpdate = true } if supportsNodesHealthCheck { - hcToCreate = makeHttpHealthCheck(makeNodesHealthCheckName(clusterID), getNodesHealthCheckPath(), GetNodesHealthCheckPort()) + hcToCreate = makeHttpHealthCheck(makeNodesHealthCheckName(clusterID), GetNodesHealthCheckPath(), GetNodesHealthCheckPort()) } } // Now we get to some slightly more interesting logic. @@ -360,18 +272,18 @@ func (gce *GCECloud) EnsureLoadBalancer(clusterName string, apiService *v1.Servi // and something should fail before we recreate it, don't release the // IP. That way we can come back to it later. isSafeToReleaseIP = false - if err := gce.deleteForwardingRule(loadBalancerName, gce.region); err != nil { + if err := gce.DeleteRegionForwardingRule(loadBalancerName, gce.region); err != nil && !isNotFound(err) { return nil, fmt.Errorf("failed to delete existing forwarding rule %s for load balancer update: %v", loadBalancerName, err) } glog.Infof("EnsureLoadBalancer(%v(%v)): deleted forwarding rule", loadBalancerName, serviceName) } if tpExists && tpNeedsUpdate { - // Pass healthchecks to deleteTargetPool to cleanup health checks after cleaning up the target pool itself. + // Pass healthchecks to DeleteExternalTargetPoolAndChecks to cleanup health checks after cleaning up the target pool itself. var hcNames []string if hcToDelete != nil { hcNames = append(hcNames, hcToDelete.Name) } - if err := gce.deleteTargetPool(loadBalancerName, gce.region, hcNames...); err != nil { + if err := gce.DeleteExternalTargetPoolAndChecks(loadBalancerName, gce.region, hcNames...); err != nil { return nil, fmt.Errorf("failed to delete existing target pool %s for load balancer update: %v", loadBalancerName, err) } glog.Infof("EnsureLoadBalancer(%v(%v)): deleted target pool", loadBalancerName, serviceName) @@ -425,8 +337,8 @@ func (gce *GCECloud) EnsureLoadBalancer(clusterName string, apiService *v1.Servi return status, nil } -// UpdateLoadBalancer is an implementation of LoadBalancer.UpdateLoadBalancer. -func (gce *GCECloud) UpdateLoadBalancer(clusterName string, service *v1.Service, nodes []*v1.Node) error { +// updateExternalLoadBalancer is the external implementation of LoadBalancer.UpdateLoadBalancer. +func (gce *GCECloud) updateExternalLoadBalancer(clusterName string, service *v1.Service, nodes []*v1.Node) error { hosts, err := gce.getInstancesByNames(nodeNames(nodes)) if err != nil { return err @@ -445,11 +357,9 @@ func (gce *GCECloud) UpdateLoadBalancer(clusterName string, service *v1.Service, return gce.updateTargetPool(loadBalancerName, existing, hosts) } -// EnsureLoadBalancerDeleted is an implementation of LoadBalancer.EnsureLoadBalancerDeleted. -func (gce *GCECloud) EnsureLoadBalancerDeleted(clusterName string, service *v1.Service) error { +// ensureExternalLoadBalancerDeleted is the external implementation of LoadBalancer.EnsureLoadBalancerDeleted +func (gce *GCECloud) ensureExternalLoadBalancerDeleted(clusterName string, service *v1.Service) error { loadBalancerName := cloudprovider.GetLoadBalancerName(service) - glog.V(2).Infof("EnsureLoadBalancerDeleted(%v, %v, %v, %v, %v)", clusterName, service.Namespace, service.Name, loadBalancerName, - gce.region) var hcNames []string if path, _ := apiservice.GetServiceHealthCheckPathPort(service); path != "" { @@ -473,18 +383,18 @@ func (gce *GCECloud) EnsureLoadBalancerDeleted(clusterName string, service *v1.S } errs := utilerrors.AggregateGoroutines( - func() error { return gce.deleteFirewall(makeFirewallName(loadBalancerName), gce.region) }, + func() error { return ignoreNotFound(gce.DeleteFirewall(makeFirewallName(loadBalancerName))) }, // Even though we don't hold on to static IPs for load balancers, it's // possible that EnsureLoadBalancer left one around in a failed // creation/update attempt, so make sure we clean it up here just in case. - func() error { return gce.deleteStaticIP(loadBalancerName, gce.region) }, + func() error { return ignoreNotFound(gce.DeleteRegionAddress(loadBalancerName, gce.region)) }, func() error { // The forwarding rule must be deleted before either the target pool can, // unfortunately, so we have to do these two serially. - if err := gce.deleteForwardingRule(loadBalancerName, gce.region); err != nil { + if err := ignoreNotFound(gce.DeleteRegionForwardingRule(loadBalancerName, gce.region)); err != nil { return err } - if err := gce.deleteTargetPool(loadBalancerName, gce.region, hcNames...); err != nil { + if err := gce.DeleteExternalTargetPoolAndChecks(loadBalancerName, gce.region, hcNames...); err != nil { return err } return nil @@ -496,62 +406,17 @@ func (gce *GCECloud) EnsureLoadBalancerDeleted(clusterName string, service *v1.S return nil } -func (gce *GCECloud) DeleteForwardingRule(name string) error { - region, err := GetGCERegion(gce.localZone) - if err != nil { - return err - } - return gce.deleteForwardingRule(name, region) -} - -func (gce *GCECloud) deleteForwardingRule(name, region string) error { - mc := newForwardingRuleMetricContext("delete", region) - op, err := gce.service.ForwardingRules.Delete(gce.projectID, region, name).Do() - - if err != nil && isHTTPErrorCode(err, http.StatusNotFound) { - glog.Infof("Forwarding rule %s already deleted. Continuing to delete other resources.", name) - } else if err != nil { - glog.Warningf("Failed to delete forwarding rule %s: got error %s.", name, err.Error()) - return mc.Observe(err) - } else { - if err := gce.waitForRegionOp(op, region, mc); err != nil { - glog.Warningf("Failed waiting for forwarding rule %s to be deleted: got error %s.", - name, err.Error()) - return err - } - } - return nil -} - -// DeleteTargetPool deletes the given target pool. -func (gce *GCECloud) DeleteTargetPool(name string, hcNames ...string) error { - region, err := GetGCERegion(gce.localZone) - if err != nil { - return err - } - return gce.deleteTargetPool(name, region, hcNames...) -} - -func (gce *GCECloud) deleteTargetPool(name, region string, hcNames ...string) error { - mc := newTargetPoolMetricContext("delete", region) - op, err := gce.service.TargetPools.Delete(gce.projectID, region, name).Do() - - if err != nil && isHTTPErrorCode(err, http.StatusNotFound) { +func (gce *GCECloud) DeleteExternalTargetPoolAndChecks(name, region string, hcNames ...string) error { + if err := gce.DeleteTargetPool(name, region); err != nil && isHTTPErrorCode(err, http.StatusNotFound) { glog.Infof("Target pool %s already deleted. Continuing to delete other resources.", name) } else if err != nil { glog.Warningf("Failed to delete target pool %s, got error %s.", name, err.Error()) - return mc.Observe(err) - } else { - if err := gce.waitForRegionOp(op, region, mc); err != nil { - glog.Warningf("Failed waiting for target pool %s to be deleted: got error %s.", - name, err.Error()) - return err - } + return err } // Deletion of health checks is allowed only after the TargetPool reference is deleted for _, hcName := range hcNames { - if err = func() error { + if err := func() error { // Check whether it is nodes health check, which has different name from the load-balancer. isNodesHealthCheck := hcName != name if isNodesHealthCheck { @@ -616,8 +481,10 @@ func (gce *GCECloud) createTargetPool(name, serviceName, ipAddress, region strin gce.sharedResourceLock.Lock() defer gce.sharedResourceLock.Unlock() } - if err := gce.ensureHttpHealthCheckFirewall(serviceName, ipAddress, gce.region, hosts, hc.Name, int32(hc.Port), isNodesHealthCheck); err != nil { - return err + if !gce.OnXPN() { + if err := gce.ensureHttpHealthCheckFirewall(serviceName, ipAddress, region, hosts, hc.Name, int32(hc.Port), isNodesHealthCheck); err != nil { + return err + } } var err error if hc, err = gce.ensureHttpHealthCheck(hc.Name, hc.RequestPath, int32(hc.Port)); err != nil || hc == nil { @@ -639,16 +506,8 @@ func (gce *GCECloud) createTargetPool(name, serviceName, ipAddress, region strin HealthChecks: hcLinks, } - mc := newTargetPoolMetricContext("insert", region) - op, err := gce.service.TargetPools.Insert(gce.projectID, region, pool).Do() - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return mc.Observe(err) - } - if op != nil { - err = gce.waitForRegionOp(op, region, mc) - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return err - } + if _, err := gce.CreateTargetPool(pool, region); err != nil && !isHTTPErrorCode(err, http.StatusConflict) { + return err } return nil } @@ -668,26 +527,13 @@ func (gce *GCECloud) updateTargetPool(loadBalancerName string, existing sets.Str } if len(toAdd) > 0 { - add := &compute.TargetPoolsAddInstanceRequest{Instances: toAdd} - - mc := newTargetPoolMetricContext("update", gce.region) - op, err := gce.service.TargetPools.AddInstance(gce.projectID, gce.region, loadBalancerName, add).Do() - if err != nil { - return mc.Observe(err) - } - if err := gce.waitForRegionOp(op, gce.region, mc); err != nil { + if err := gce.AddInstancesToTargetPool(loadBalancerName, gce.region, toAdd); err != nil { return err } } if len(toRemove) > 0 { - mc := newTargetPoolMetricContext("delete", gce.region) - rm := &compute.TargetPoolsRemoveInstanceRequest{Instances: toRemove} - op, err := gce.service.TargetPools.RemoveInstance(gce.projectID, gce.region, loadBalancerName, rm).Do() - if err != nil { - return mc.Observe(err) - } - if err := gce.waitForRegionOp(op, gce.region, mc); err != nil { + if err := gce.RemoveInstancesFromTargetPool(loadBalancerName, gce.region, toRemove); err != nil { return err } } @@ -695,7 +541,7 @@ func (gce *GCECloud) updateTargetPool(loadBalancerName string, existing sets.Str // Try to verify that the correct number of nodes are now in the target pool. // We've been bitten by a bug here before (#11327) where all nodes were // accidentally removed and want to make similar problems easier to notice. - updatedPool, err := gce.service.TargetPools.Get(gce.projectID, gce.region, loadBalancerName).Do() + updatedPool, err := gce.GetTargetPool(loadBalancerName, gce.region) if err != nil { return err } @@ -851,10 +697,6 @@ func hostURLToComparablePath(hostURL string) string { return hostURL[idx:] } -func makeHealthCheckDescription(serviceName string) string { - return fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, serviceName) -} - func loadBalancerPortRange(ports []v1.ServicePort) (string, error) { if len(ports) == 0 { return "", fmt.Errorf("no ports specified for GCE load balancer") @@ -891,16 +733,12 @@ func translateAffinityType(affinityType v1.ServiceAffinity) string { } } -func makeFirewallName(name string) string { - return fmt.Sprintf("k8s-fw-%s", name) -} - -func makeFirewallDescription(serviceName, ipAddress string) string { - return fmt.Sprintf(`{"kubernetes.io/service-name":"%s", "kubernetes.io/service-ip":"%s"}`, - serviceName, ipAddress) -} - func (gce *GCECloud) firewallNeedsUpdate(name, serviceName, region, ipAddress string, ports []v1.ServicePort, sourceRanges netsets.IPNet) (exists bool, needsUpdate bool, err error) { + if gce.OnXPN() { + glog.V(2).Infoln("firewallNeedsUpdate: Cluster is on XPN network - skipping firewall creation") + return false, false, nil + } + fw, err := gce.service.Firewalls.Get(gce.projectID, makeFirewallName(name)).Do() if err != nil { if isHTTPErrorCode(err, http.StatusNotFound) { @@ -995,54 +833,31 @@ func (gce *GCECloud) createForwardingRule(name, serviceName, region, ipAddress s Target: gce.targetPoolURL(name, region), } - mc := newForwardingRuleMetricContext("create", region) - op, err := gce.service.ForwardingRules.Insert(gce.projectID, region, req).Do() - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return mc.Observe(err) - } - if op != nil { - err = gce.waitForRegionOp(op, region, mc) - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return err - } + if err = gce.CreateRegionForwardingRule(req, region); err != nil && !isHTTPErrorCode(err, http.StatusConflict) { + return err } return nil } func (gce *GCECloud) createFirewall(name, region, desc string, sourceRanges netsets.IPNet, ports []v1.ServicePort, hosts []*gceInstance) error { - mc := newFirewallMetricContext("create") firewall, err := gce.firewallObject(name, region, desc, sourceRanges, ports, hosts) if err != nil { - return mc.Observe(err) + return err } - op, err := gce.service.Firewalls.Insert(gce.projectID, firewall).Do() - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return mc.Observe(err) - } - if op != nil { - err = gce.waitForGlobalOp(op, mc) - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return err - } + if err = gce.CreateFirewall(firewall); err != nil && !isHTTPErrorCode(err, http.StatusConflict) { + return err } return nil } func (gce *GCECloud) updateFirewall(name, region, desc string, sourceRanges netsets.IPNet, ports []v1.ServicePort, hosts []*gceInstance) error { - mc := newFirewallMetricContext("update") firewall, err := gce.firewallObject(name, region, desc, sourceRanges, ports, hosts) if err != nil { - return mc.Observe(err) + return err } - op, err := gce.service.Firewalls.Update(gce.projectID, name, firewall).Do() - if err != nil && !isHTTPErrorCode(err, http.StatusConflict) { - return mc.Observe(err) - } - if op != nil { - err = gce.waitForGlobalOp(op, mc) - if err != nil { - return err - } + + if err = gce.UpdateFirewall(firewall); err != nil && !isHTTPErrorCode(err, http.StatusConflict) { + return err } return nil } @@ -1124,66 +939,14 @@ func (gce *GCECloud) ensureStaticIP(name, serviceName, region, existingIP string addressObj.Address = existingIP } - mc := newAddressMetricContext("create", region) - op, err := gce.service.Addresses.Insert(gce.projectID, region, addressObj).Do() + address, err := gce.ReserveRegionAddress(addressObj, region) if err != nil { if !isHTTPErrorCode(err, http.StatusConflict) { - return "", false, fmt.Errorf("error creating gce static IP address: %v", - mc.Observe(err)) + return "", false, fmt.Errorf("error creating gce static IP address: %v", err) } // StatusConflict == the IP exists already. existed = true } - if op != nil { - err := gce.waitForRegionOp(op, region, mc) - if err != nil { - if !isHTTPErrorCode(err, http.StatusConflict) { - return "", false, fmt.Errorf("error waiting for gce static IP address to be created: %v", err) - } - // StatusConflict == the IP exists already. - existed = true - } - } - // We have to get the address to know which IP was allocated for us. - address, err := gce.service.Addresses.Get(gce.projectID, region, name).Do() - if err != nil { - return "", false, fmt.Errorf("error re-getting gce static IP address: %v", err) - } return address.Address, existed, nil } - -func (gce *GCECloud) deleteFirewall(name, region string) error { - mc := newFirewallMetricContext("delete") - op, err := gce.service.Firewalls.Delete(gce.projectID, name).Do() - - if err != nil && isHTTPErrorCode(err, http.StatusNotFound) { - glog.V(2).Infof("Firewall %s already deleted. Continuing to delete other resources.", name) - } else if err != nil { - glog.Warningf("Failed to delete firewall %s, got error %v", name, err) - return mc.Observe(err) - } else { - if err := gce.waitForGlobalOp(op, mc); err != nil { - glog.Warningf("Failed waiting for Firewall %s to be deleted. Got error: %v", name, err) - return err - } - } - return nil -} - -func (gce *GCECloud) deleteStaticIP(name, region string) error { - mc := newAddressMetricContext("delete", region) - op, err := gce.service.Addresses.Delete(gce.projectID, region, name).Do() - if err != nil && isHTTPErrorCode(err, http.StatusNotFound) { - glog.Infof("Static IP address %s is not reserved", name) - } else if err != nil { - glog.Warningf("Failed to delete static IP address %s, got error %v", name, err) - return mc.Observe(err) - } else { - if err := gce.waitForRegionOp(op, region, mc); err != nil { - glog.Warningf("Failed waiting for address %s to be deleted, got error: %v", name, err) - return err - } - } - return nil -} From 660452dee1981c1397056b75b45eda218f7eb2df Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Thu, 1 Jun 2017 15:24:07 -0700 Subject: [PATCH 5/8] Add internal LB logic --- .../gce/gce_loadbalancer_internal.go | 636 ++++++++++++++++++ .../providers/gce/gce_loadbalancer_naming.go | 103 +++ 2 files changed, 739 insertions(+) create mode 100644 pkg/cloudprovider/providers/gce/gce_loadbalancer_internal.go create mode 100644 pkg/cloudprovider/providers/gce/gce_loadbalancer_naming.go diff --git a/pkg/cloudprovider/providers/gce/gce_loadbalancer_internal.go b/pkg/cloudprovider/providers/gce/gce_loadbalancer_internal.go new file mode 100644 index 00000000000..47b7ed90ee8 --- /dev/null +++ b/pkg/cloudprovider/providers/gce/gce_loadbalancer_internal.go @@ -0,0 +1,636 @@ +/* +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 gce + +import ( + "fmt" + "strconv" + "strings" + + "github.com/golang/glog" + compute "google.golang.org/api/compute/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/kubernetes/pkg/api/v1" + v1_service "k8s.io/kubernetes/pkg/api/v1/service" + "k8s.io/kubernetes/pkg/cloudprovider" +) + +const ( + allInstances = "ALL" +) + +type lbBalancingMode string + +func (gce *GCECloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) { + nm := types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace} + ports, protocol := getPortsAndProtocol(svc.Spec.Ports) + scheme := schemeInternal + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + shared := !v1_service.RequestsOnlyLocalTraffic(svc) + backendServiceName := makeBackendServiceName(loadBalancerName, clusterID, shared, scheme, protocol, svc.Spec.SessionAffinity) + backendServiceLink := gce.getBackendServiceLink(backendServiceName) + + // Ensure instance groups exist and nodes are assigned to groups + igName := makeInstanceGroupName(clusterID) + igLinks, err := gce.ensureInternalInstanceGroups(igName, nodes) + if err != nil { + return nil, err + } + + // Lock the sharedResourceLock to prevent any deletions of shared resources while assembling shared resources here + gce.sharedResourceLock.Lock() + defer gce.sharedResourceLock.Unlock() + + // Ensure health check for backend service is created + // By default, use the node health check endpoint + hcName := makeHealthCheckName(loadBalancerName, clusterID, shared) + hcPath, hcPort := GetNodesHealthCheckPath(), GetNodesHealthCheckPort() + if !shared { + // Service requires a special health check, retrieve the OnlyLocal port & path + hcPath, hcPort = v1_service.GetServiceHealthCheckPathPort(svc) + } + hc, err := gce.ensureInternalHealthCheck(hcName, nm, shared, hcPath, hcPort) + if err != nil { + return nil, err + } + + // Ensure firewall rules if necessary + if gce.OnXPN() { + glog.V(2).Infof("ensureInternalLoadBalancer: cluster is on a cross-project network (XPN) network project %v, compute project %v - skipping firewall creation", gce.networkProjectID, gce.projectID) + } else { + if err = gce.ensureInternalFirewalls(loadBalancerName, clusterID, nm, svc, strconv.Itoa(int(hcPort)), shared, nodes); err != nil { + return nil, err + } + } + + expectedFwdRule := &compute.ForwardingRule{ + Name: loadBalancerName, + Description: fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, nm.String()), + IPAddress: svc.Spec.LoadBalancerIP, + BackendService: backendServiceLink, + Ports: ports, + IPProtocol: string(protocol), + LoadBalancingScheme: string(scheme), + } + + // Specify subnetwork if network type is manual + if len(gce.subnetworkURL) > 0 { + expectedFwdRule.Subnetwork = gce.subnetworkURL + } else { + expectedFwdRule.Network = gce.networkURL + } + + // Delete the previous internal load balancer if necessary + fwdRuleDeleted, err := gce.clearExistingInternalLB(loadBalancerName, existingFwdRule, expectedFwdRule, backendServiceName) + if err != nil { + return nil, err + } + + bsDescription := makeBackendServiceDescription(nm, shared) + err = gce.ensureInternalBackendService(backendServiceName, bsDescription, svc.Spec.SessionAffinity, scheme, protocol, igLinks, hc.SelfLink) + if err != nil { + return nil, err + } + + // If we previously deleted the forwarding rule or it never existed, finally create it. + if fwdRuleDeleted || existingFwdRule == nil { + glog.V(2).Infof("ensureInternalLoadBalancer(%v(%v)): creating forwarding rule", loadBalancerName, svc.Name) + if err = gce.CreateRegionForwardingRule(expectedFwdRule, gce.region); err != nil { + return nil, err + } + } + + // Get the most recent forwarding rule for the new address. + existingFwdRule, err = gce.GetRegionForwardingRule(loadBalancerName, gce.region) + if err != nil { + return nil, err + } + + status := &v1.LoadBalancerStatus{} + status.Ingress = []v1.LoadBalancerIngress{{IP: existingFwdRule.IPAddress}} + return status, nil +} + +func (gce *GCECloud) clearExistingInternalLB(loadBalancerName string, existingFwdRule, expectedFwdRule *compute.ForwardingRule, expectedBSName string) (fwdRuleDeleted bool, err error) { + if existingFwdRule == nil { + return false, nil + } + + if !fwdRuleEqual(existingFwdRule, expectedFwdRule) { + glog.V(2).Infof("clearExistingInternalLB(%v: deleting existing forwarding rule with IP address %v", loadBalancerName, existingFwdRule.IPAddress) + if err = gce.DeleteRegionForwardingRule(loadBalancerName, gce.region); err != nil && !isNotFound(err) { + return false, err + } + fwdRuleDeleted = true + } + + existingBSName := getNameFromLink(existingFwdRule.BackendService) + bs, err := gce.GetRegionBackendService(existingBSName, gce.region) + if err == nil { + if bs.Name != expectedBSName { + glog.V(2).Infof("clearExistingInternalLB(%v): expected backend service %q does not match actual %q - deleting backend service & healthcheck & firewall", loadBalancerName, expectedBSName, bs.Name) + // Delete the backend service as well in case it's switching between shared, nonshared, tcp, udp. + var existingHCName string + if len(bs.HealthChecks) == 1 { + existingHCName = getNameFromLink(bs.HealthChecks[0]) + } + if err = gce.teardownInternalBackendResources(existingBSName, existingHCName); err != nil { + glog.Warningf("clearExistingInternalLB: could not delete old resources: %v", err) + } else { + glog.V(2).Infof("clearExistingInternalLB: done deleting old resources") + } + } + } else { + glog.Warningf("clearExistingInternalLB(%v): failed to retrieve existing backend service %v", loadBalancerName, existingBSName) + } + return fwdRuleDeleted, nil +} + +func (gce *GCECloud) updateInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, nodes []*v1.Node) error { + gce.sharedResourceLock.Lock() + defer gce.sharedResourceLock.Unlock() + + igName := makeInstanceGroupName(clusterID) + igLinks, err := gce.ensureInternalInstanceGroups(igName, nodes) + if err != nil { + return err + } + + // Generate the backend service name + _, protocol := getPortsAndProtocol(svc.Spec.Ports) + scheme := schemeInternal + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + shared := !v1_service.RequestsOnlyLocalTraffic(svc) + backendServiceName := makeBackendServiceName(loadBalancerName, clusterID, shared, scheme, protocol, svc.Spec.SessionAffinity) + // Ensure the backend service has the proper backend instance-group links + return gce.ensureInternalBackendServiceGroups(backendServiceName, igLinks) +} + +func (gce *GCECloud) ensureInternalLoadBalancerDeleted(clusterName, clusterID string, svc *v1.Service) error { + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + _, protocol := getPortsAndProtocol(svc.Spec.Ports) + scheme := schemeInternal + shared := !v1_service.RequestsOnlyLocalTraffic(svc) + var err error + + gce.sharedResourceLock.Lock() + defer gce.sharedResourceLock.Unlock() + + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting region internal forwarding rule", loadBalancerName) + if err = gce.DeleteRegionForwardingRule(loadBalancerName, gce.region); err != nil && !isNotFound(err) { + return err + } + + backendServiceName := makeBackendServiceName(loadBalancerName, clusterID, shared, scheme, protocol, svc.Spec.SessionAffinity) + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting region backend service %v", loadBalancerName, backendServiceName) + if err = gce.DeleteRegionBackendService(backendServiceName, gce.region); err != nil && !isNotFoundOrInUse(err) { + return err + } + + // Only delete the health check & health check firewall if they aren't being used by another LB. If we get + // an ResourceInuseBy error, then we can skip deleting the firewall. + hcInUse := false + hcName := makeHealthCheckName(loadBalancerName, clusterID, shared) + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting health check %v", loadBalancerName, hcName) + if err = gce.DeleteHealthCheck(hcName); err != nil && !isNotFoundOrInUse(err) { + return err + } else if isInUsedByError(err) { + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): healthcheck %v still in use", loadBalancerName, hcName) + hcInUse = true + } + + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting firewall for traffic", loadBalancerName) + if err = gce.DeleteFirewall(loadBalancerName); err != nil { + return err + } + + if hcInUse { + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): skipping firewall for healthcheck", loadBalancerName) + } else { + glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting firewall for healthcheck", loadBalancerName) + fwHCName := makeHealthCheckFirewallkName(loadBalancerName, clusterID, shared) + if err = gce.DeleteFirewall(fwHCName); err != nil && !isInUsedByError(err) { + return err + } + } + + // Try deleting instance groups - expect ResourceInuse error if needed by other LBs + igName := makeInstanceGroupName(clusterID) + if err = gce.ensureInternalInstanceGroupsDeleted(igName); err != nil && !isInUsedByError(err) { + return err + } + + return nil +} + +func (gce *GCECloud) teardownInternalBackendResources(bsName, hcName string) error { + if err := gce.DeleteRegionBackendService(bsName, gce.region); err != nil { + if isNotFound(err) { + glog.V(2).Infof("backend service already deleted: %v, err: %v", bsName, err) + } else if err != nil && isInUsedByError(err) { + glog.V(2).Infof("backend service in use: %v, err: %v", bsName, err) + } else { + return fmt.Errorf("failed to delete backend service: %v, err: %v", bsName, err) + } + } + + if hcName == "" { + return nil + } + hcInUse := false + if err := gce.DeleteHealthCheck(hcName); err != nil { + if isNotFound(err) { + glog.V(2).Infof("health check already deleted: %v, err: %v", hcName, err) + } else if err != nil && isInUsedByError(err) { + hcInUse = true + glog.V(2).Infof("health check in use: %v, err: %v", hcName, err) + } else { + return fmt.Errorf("failed to delete health check: %v, err: %v", hcName, err) + } + } + + hcFirewallName := makeHealthCheckFirewallkNameFromHC(hcName) + if hcInUse { + glog.V(2).Infof("skipping deletion of health check firewall: %v", hcFirewallName) + return nil + } + + if err := gce.DeleteFirewall(hcFirewallName); err != nil && !isNotFound(err) { + return fmt.Errorf("failed to delete health check firewall: %v, err: %v", hcFirewallName, err) + } + return nil +} + +func (gce *GCECloud) ensureInternalFirewall(fwName, fwDesc string, sourceRanges []string, ports []string, protocol v1.Protocol, nodes []*v1.Node) error { + glog.V(2).Infof("ensureInternalFirewall(%v): checking existing firewall", fwName) + targetTags, err := gce.GetNodeTags(nodeNames(nodes)) + if err != nil { + return err + } + + existingFirewall, err := gce.GetFirewall(fwName) + if err != nil && !isNotFound(err) { + return err + } + + expectedFirewall := &compute.Firewall{ + Name: fwName, + Description: fwDesc, + Network: gce.networkURL, + SourceRanges: sourceRanges, + TargetTags: targetTags, + Allowed: []*compute.FirewallAllowed{ + { + IPProtocol: strings.ToLower(string(protocol)), + Ports: ports, + }, + }, + } + + if existingFirewall == nil { + glog.V(2).Infof("ensureInternalFirewall(%v): creating firewall", fwName) + return gce.CreateFirewall(expectedFirewall) + } + + if firewallRuleEqual(expectedFirewall, existingFirewall) { + return nil + } + + glog.V(2).Infof("ensureInternalFirewall(%v): updating firewall", fwName) + return gce.UpdateFirewall(expectedFirewall) +} + +func (gce *GCECloud) ensureInternalFirewalls(loadBalancerName, clusterID string, nm types.NamespacedName, svc *v1.Service, healthCheckPort string, shared bool, nodes []*v1.Node) error { + // First firewall is for ingress traffic + fwDesc := makeFirewallDescription(nm.String(), svc.Spec.LoadBalancerIP) + ports, protocol := getPortsAndProtocol(svc.Spec.Ports) + sourceRanges, err := v1_service.GetLoadBalancerSourceRanges(svc) + if err != nil { + return err + } + err = gce.ensureInternalFirewall(loadBalancerName, fwDesc, sourceRanges.StringSlice(), ports, protocol, nodes) + if err != nil { + return err + } + + // Second firewall is for health checking nodes / services + fwHCName := makeHealthCheckFirewallkName(loadBalancerName, clusterID, shared) + hcSrcRanges := LoadBalancerSrcRanges() + return gce.ensureInternalFirewall(fwHCName, "", hcSrcRanges, []string{healthCheckPort}, v1.ProtocolTCP, nodes) +} + +func (gce *GCECloud) ensureInternalHealthCheck(name string, svcName types.NamespacedName, shared bool, path string, port int32) (*compute.HealthCheck, error) { + glog.V(2).Infof("ensureInternalHealthCheck(%v, %v, %v): checking existing health check", name, path, port) + expectedHC := newInternalLBHealthCheck(name, svcName, shared, path, port) + + hc, err := gce.GetHealthCheck(name) + if err != nil && !isNotFound(err) { + return nil, err + } + + if hc == nil { + glog.V(2).Infof("ensureInternalHealthCheck: did not find health check %v, creating one with port %v path %v", name, port, path) + if err = gce.CreateHealthCheck(expectedHC); err != nil { + return nil, err + } + hc, err = gce.GetHealthCheck(name) + if err != nil { + glog.Errorf("Failed to get http health check %v", err) + return nil, err + } + glog.V(2).Infof("ensureInternalHealthCheck: created health check %v", name) + return hc, nil + } + + if healthChecksEqual(expectedHC, hc) { + return hc, nil + } + + glog.V(2).Infof("ensureInternalHealthCheck: health check %v exists but parameters have drifted - updating...", name) + if err := gce.UpdateHealthCheck(expectedHC); err != nil { + glog.Warningf("Failed to reconcile http health check %v parameters", name) + return nil, err + } + glog.V(2).Infof("ensureInternalHealthCheck: corrected health check %v parameters successful", name) + return hc, nil +} + +func (gce *GCECloud) ensureInternalInstanceGroup(name, zone string, nodes []*v1.Node) (string, error) { + glog.V(2).Infof("ensureInternalInstanceGroup(%v, %v): checking group that it contains %v nodes", name, zone, len(nodes)) + ig, err := gce.GetInstanceGroup(name, zone) + if err != nil && !isNotFound(err) { + return "", err + } + + kubeNodes := sets.NewString() + for _, n := range nodes { + kubeNodes.Insert(n.Name) + } + + gceNodes := sets.NewString() + if ig == nil { + glog.V(2).Infof("ensureInternalInstanceGroup(%v, %v): creating instance group", name, zone) + ig, err = gce.CreateInstanceGroup(name, zone) + if err != nil { + return "", err + } + } else { + instances, err := gce.ListInstancesInInstanceGroup(name, zone, allInstances) + if err != nil { + return "", err + } + + for _, ins := range instances.Items { + parts := strings.Split(ins.Instance, "/") + gceNodes.Insert(parts[len(parts)-1]) + } + } + + removeNodes := gceNodes.Difference(kubeNodes).List() + addNodes := kubeNodes.Difference(gceNodes).List() + + if len(removeNodes) != 0 { + glog.V(2).Infof("ensureInternalInstanceGroup(%v, %v): removing nodes: %v", name, zone, removeNodes) + instanceRefs := gce.ToInstanceReferences(zone, removeNodes) + // Possible we'll receive 404's here if the instance was deleted before getting to this point. + if err = gce.RemoveInstancesFromInstanceGroup(name, zone, instanceRefs); err != nil && !isNotFound(err) { + return "", err + } + } + + if len(addNodes) != 0 { + glog.V(2).Infof("ensureInternalInstanceGroup(%v, %v): adding nodes: %v", name, zone, addNodes) + instanceRefs := gce.ToInstanceReferences(zone, addNodes) + if err = gce.AddInstancesToInstanceGroup(name, zone, instanceRefs); err != nil { + return "", err + } + } + + return ig.SelfLink, nil +} + +// ensureInternalInstanceGroups generates an unmanaged instance group for every zone +// where a K8s node exists. It also ensures that each node belongs to an instance group +func (gce *GCECloud) ensureInternalInstanceGroups(name string, nodes []*v1.Node) ([]string, error) { + zonedNodes := splitNodesByZone(nodes) + glog.V(2).Infof("ensureInternalInstanceGroups(%v): %d nodes over %d zones in region %v", name, len(nodes), len(zonedNodes), gce.region) + var igLinks []string + for zone, nodes := range zonedNodes { + igLink, err := gce.ensureInternalInstanceGroup(name, zone, nodes) + if err != nil { + return []string{}, err + } + igLinks = append(igLinks, igLink) + } + + return igLinks, nil +} + +func (gce *GCECloud) ensureInternalInstanceGroupsDeleted(name string) error { + // List of nodes isn't available here - fetch all zones in region and try deleting this cluster's ig + zones, err := gce.ListZonesInRegion(gce.region) + if err != nil { + return err + } + + glog.V(2).Infof("ensureInternalInstanceGroupsDeleted(%v): deleting instance group in all %d zones", name, len(zones)) + for _, z := range zones { + if err := gce.DeleteInstanceGroup(name, z.Name); err != nil && !isNotFound(err) { + return err + } + } + return nil +} + +func (gce *GCECloud) ensureInternalBackendService(name, description string, affinityType v1.ServiceAffinity, scheme lbScheme, protocol v1.Protocol, igLinks []string, hcLink string) error { + glog.V(2).Infof("ensureInternalBackendService(%v, %v, %v): checking existing backend service with %d groups", name, scheme, protocol, len(igLinks)) + bs, err := gce.GetRegionBackendService(name, gce.region) + if err != nil && !isNotFound(err) { + return err + } + + backends := backendsFromGroupLinks(igLinks) + expectedBS := &compute.BackendService{ + Name: name, + Protocol: string(protocol), + Description: description, + HealthChecks: []string{hcLink}, + Backends: backends, + SessionAffinity: translateAffinityType(affinityType), + LoadBalancingScheme: string(scheme), + } + + // Create backend service if none was found + if bs == nil { + glog.V(2).Infof("ensureInternalBackendService: creating backend service %v", name) + err := gce.CreateRegionBackendService(expectedBS, gce.region) + if err != nil { + return err + } + glog.V(2).Infof("ensureInternalBackendService: created backend service %v successfully", name) + return nil + } + // Check existing backend service + existingIGLinks := sets.NewString() + for _, be := range bs.Backends { + existingIGLinks.Insert(be.Group) + } + + if backendSvcEqual(expectedBS, bs) { + return nil + } + + glog.V(2).Infof("ensureInternalBackendService: updating backend service %v", name) + if err := gce.UpdateRegionBackendService(expectedBS, gce.region); err != nil { + return err + } + glog.V(2).Infof("ensureInternalBackendService: updated backend service %v successfully", name) + return nil +} + +func (gce *GCECloud) ensureInternalBackendServiceGroups(name string, igLinks []string) error { + glog.V(2).Infof("ensureInternalBackendServiceGroups(%v): checking existing backend service's groups", name) + bs, err := gce.GetRegionBackendService(name, gce.region) + if err != nil { + return err + } + + backends := backendsFromGroupLinks(igLinks) + if backendsListEqual(bs.Backends, backends) { + return nil + } + + glog.V(2).Infof("ensureInternalBackendServiceGroups: updating backend service %v", name) + bs.Backends = backends + if err := gce.UpdateRegionBackendService(bs, gce.region); err != nil { + return err + } + glog.V(2).Infof("ensureInternalBackendServiceGroups: updated backend service %v successfully", name) + return nil +} + +func backendsFromGroupLinks(igLinks []string) []*compute.Backend { + var backends []*compute.Backend + for _, igLink := range igLinks { + backends = append(backends, &compute.Backend{ + Group: igLink, + }) + } + return backends +} + +func newInternalLBHealthCheck(name string, svcName types.NamespacedName, shared bool, path string, port int32) *compute.HealthCheck { + httpSettings := compute.HTTPHealthCheck{ + Port: int64(port), + RequestPath: path, + } + desc := "" + if !shared { + desc = makeHealthCheckDescription(svcName.String()) + } + return &compute.HealthCheck{ + Name: name, + CheckIntervalSec: gceHcCheckIntervalSeconds, + TimeoutSec: gceHcTimeoutSeconds, + HealthyThreshold: gceHcHealthyThreshold, + UnhealthyThreshold: gceHcUnhealthyThreshold, + HttpHealthCheck: &httpSettings, + Type: "HTTP", + Description: desc, + } +} + +func firewallRuleEqual(a, b *compute.Firewall) bool { + return a.Description == b.Description && + len(a.Allowed) == 1 && len(a.Allowed) == len(b.Allowed) && + a.Allowed[0].IPProtocol == b.Allowed[0].IPProtocol && + equalStringSets(a.Allowed[0].Ports, b.Allowed[0].Ports) && + equalStringSets(a.SourceRanges, b.SourceRanges) && + equalStringSets(a.TargetTags, b.TargetTags) +} + +func healthChecksEqual(a, b *compute.HealthCheck) bool { + return a.HttpHealthCheck != nil && b.HttpHealthCheck != nil && + a.HttpHealthCheck.Port == b.HttpHealthCheck.Port && + a.HttpHealthCheck.RequestPath == b.HttpHealthCheck.RequestPath && + a.Description == b.Description && + a.CheckIntervalSec == b.CheckIntervalSec && + a.TimeoutSec == b.TimeoutSec && + a.UnhealthyThreshold == b.UnhealthyThreshold && + a.HealthyThreshold == b.HealthyThreshold +} + +// backendsListEqual asserts that backend lists are equal by instance group link only +func backendsListEqual(a, b []*compute.Backend) bool { + if len(a) != len(b) { + return false + } + if len(a) == 0 { + return true + } + + aSet := sets.NewString() + for _, v := range a { + aSet.Insert(v.Group) + } + bSet := sets.NewString() + for _, v := range b { + bSet.Insert(v.Group) + } + + return aSet.Equal(bSet) +} + +func backendSvcEqual(a, b *compute.BackendService) bool { + return a.Protocol == b.Protocol && + a.Description == b.Description && + a.SessionAffinity == b.SessionAffinity && + a.LoadBalancingScheme == b.LoadBalancingScheme && + equalStringSets(a.HealthChecks, b.HealthChecks) && + backendsListEqual(a.Backends, b.Backends) +} + +func fwdRuleEqual(a, b *compute.ForwardingRule) bool { + return (a.IPAddress == "" || b.IPAddress == "" || a.IPAddress == b.IPAddress) && + a.IPProtocol == b.IPProtocol && + a.LoadBalancingScheme == b.LoadBalancingScheme && + equalStringSets(a.Ports, b.Ports) && + a.BackendService == b.BackendService +} + +func getPortsAndProtocol(svcPorts []v1.ServicePort) (ports []string, protocol v1.Protocol) { + if len(svcPorts) == 0 { + return []string{}, v1.ProtocolUDP + } + + // GCP doesn't support multiple protocols for a single load balancer + protocol = svcPorts[0].Protocol + for _, p := range svcPorts { + ports = append(ports, strconv.Itoa(int(p.Port))) + } + return ports, protocol +} + +func (gce *GCECloud) getBackendServiceLink(name string) string { + return fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/regions/%s/backendServices/%s", gce.projectID, gce.region, name) +} + +func getNameFromLink(link string) string { + fields := strings.Split(link, "/") + return fields[len(fields)-1] +} diff --git a/pkg/cloudprovider/providers/gce/gce_loadbalancer_naming.go b/pkg/cloudprovider/providers/gce/gce_loadbalancer_naming.go new file mode 100644 index 00000000000..7279be63ffe --- /dev/null +++ b/pkg/cloudprovider/providers/gce/gce_loadbalancer_naming.go @@ -0,0 +1,103 @@ +/* +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 gce + +import ( + "fmt" + "strings" + + "k8s.io/apimachinery/pkg/types" + "k8s.io/kubernetes/pkg/api/v1" +) + +// Internal Load Balancer + +// Instance groups remain legacy named to stay consistent with ingress +func makeInstanceGroupName(clusterID string) string { + return fmt.Sprintf("k8s-ig--%s", clusterID) +} + +func makeBackendServiceName(loadBalancerName, clusterID string, shared bool, scheme lbScheme, protocol v1.Protocol, svcAffinity v1.ServiceAffinity) string { + if shared { + affinity := "" + switch svcAffinity { + case v1.ServiceAffinityClientIP: + affinity = "clientip" + default: + affinity = "noaffinity" + } + + return fmt.Sprintf("k8s-%s-%s-%s-%s", clusterID, strings.ToLower(string(scheme)), strings.ToLower(string(protocol)), affinity) + } + return loadBalancerName +} + +func makeHealthCheckName(loadBalancerName, clusterID string, shared bool) string { + if shared { + return fmt.Sprintf("k8s-%s-node", clusterID) + } + + return loadBalancerName +} + +func makeHealthCheckFirewallkNameFromHC(healthCheckName string) string { + return healthCheckName + "-hc" +} + +func makeHealthCheckFirewallkName(loadBalancerName, clusterID string, shared bool) string { + if shared { + return fmt.Sprintf("k8s-%s-node-hc", clusterID) + } + return loadBalancerName + "-hc" +} + +func makeBackendServiceDescription(nm types.NamespacedName, shared bool) string { + if shared { + return "" + } + return fmt.Sprintf(`{"kubernetes.io/service-name":"%s"`, nm.String()) +} + +// External Load Balancer + +// makeNodesHealthCheckName returns name of the health check resource used by +// the GCE load balancers (l4) for performing health checks on nodes. +func makeNodesHealthCheckName(clusterID string) string { + return fmt.Sprintf("k8s-%v-node", clusterID) +} + +func makeHealthCheckDescription(serviceName string) string { + return fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, serviceName) +} + +// MakeHealthCheckFirewallName returns the firewall name used by the GCE load +// balancers (l4) for performing health checks. +func MakeHealthCheckFirewallName(clusterID, hcName string, isNodesHealthCheck bool) string { + if isNodesHealthCheck { + return makeNodesHealthCheckName(clusterID) + "-http-hc" + } + return "k8s-" + hcName + "-http-hc" +} + +func makeFirewallName(name string) string { + return fmt.Sprintf("k8s-fw-%s", name) +} + +func makeFirewallDescription(serviceName, ipAddress string) string { + return fmt.Sprintf(`{"kubernetes.io/service-name":"%s", "kubernetes.io/service-ip":"%s"}`, + serviceName, ipAddress) +} From 05aaef3edce16fb37401b10f34b269398fab7bdc Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Thu, 1 Jun 2017 15:25:42 -0700 Subject: [PATCH 6/8] Hook external & internal lb together --- .../providers/gce/gce_annotations.go | 51 +++++ .../providers/gce/gce_loadbalancer.go | 205 ++++++++++++++++++ 2 files changed, 256 insertions(+) create mode 100644 pkg/cloudprovider/providers/gce/gce_annotations.go create mode 100644 pkg/cloudprovider/providers/gce/gce_loadbalancer.go diff --git a/pkg/cloudprovider/providers/gce/gce_annotations.go b/pkg/cloudprovider/providers/gce/gce_annotations.go new file mode 100644 index 00000000000..98c72577a40 --- /dev/null +++ b/pkg/cloudprovider/providers/gce/gce_annotations.go @@ -0,0 +1,51 @@ +/* +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 gce + +import "k8s.io/kubernetes/pkg/api/v1" + +type LoadBalancerType string + +const ( + // ServiceAnnotationLoadBalancerType is the annotation used on a service with type LoadBalancer + // dictates what specific kind of GCP LB should be assembled. + // Currently, only "internal" is supported. + ServiceAnnotationLoadBalancerType = "cloud.google.com/load-balancer-type" + + LBTypeInternal LoadBalancerType = "internal" +) + +// GetLoadBalancerAnnotationType returns the type of GCP load balancer which should be assembled. +func GetLoadBalancerAnnotationType(service *v1.Service) (LoadBalancerType, bool) { + v := LoadBalancerType("") + if service.Spec.Type != v1.ServiceTypeLoadBalancer { + return v, false + } + + l, ok := service.Annotations[ServiceAnnotationLoadBalancerType] + v = LoadBalancerType(l) + if !ok { + return v, false + } + + switch v { + case LBTypeInternal: + return v, true + default: + return v, false + } +} diff --git a/pkg/cloudprovider/providers/gce/gce_loadbalancer.go b/pkg/cloudprovider/providers/gce/gce_loadbalancer.go new file mode 100644 index 00000000000..b449d62b64a --- /dev/null +++ b/pkg/cloudprovider/providers/gce/gce_loadbalancer.go @@ -0,0 +1,205 @@ +/* +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 gce + +import ( + "flag" + "fmt" + "net" + "strings" + "time" + + "github.com/golang/glog" + + "k8s.io/kubernetes/pkg/api/v1" + "k8s.io/kubernetes/pkg/cloudprovider" + netsets "k8s.io/kubernetes/pkg/util/net/sets" +) + +type cidrs struct { + ipn netsets.IPNet + isSet bool +} + +var ( + lbSrcRngsFlag cidrs +) + +func newLoadBalancerMetricContext(request, region string) *metricContext { + return &metricContext{ + start: time.Now(), + attributes: []string{"loadbalancer_" + request, region, unusedMetricLabel}, + } +} + +type lbScheme string + +const ( + schemeExternal lbScheme = "EXTERNAL" + schemeInternal lbScheme = "INTERNAL" +) + +func init() { + var err error + // LB L7 proxies and all L3/4/7 health checkers have client addresses within these known CIDRs. + lbSrcRngsFlag.ipn, err = netsets.ParseIPNets([]string{"130.211.0.0/22", "35.191.0.0/16", "209.85.152.0/22", "209.85.204.0/22"}...) + if err != nil { + panic("Incorrect default GCE L7 source ranges") + } + + flag.Var(&lbSrcRngsFlag, "cloud-provider-gce-lb-src-cidrs", "CIDRS opened in GCE firewall for LB traffic proxy & health checks") +} + +// String is the method to format the flag's value, part of the flag.Value interface. +func (c *cidrs) String() string { + return strings.Join(c.ipn.StringSlice(), ",") +} + +// Set supports a value of CSV or the flag repeated multiple times +func (c *cidrs) Set(value string) error { + // On first Set(), clear the original defaults + if !c.isSet { + c.isSet = true + c.ipn = make(netsets.IPNet) + } else { + return fmt.Errorf("GCE LB CIDRS have already been set") + } + + for _, cidr := range strings.Split(value, ",") { + _, ipnet, err := net.ParseCIDR(cidr) + if err != nil { + return err + } + + c.ipn.Insert(ipnet) + } + return nil +} + +// LoadBalancerSrcRanges contains the ranges of ips used by the GCE load balancers (l4 & L7) +// for proxying client requests and performing health checks. +func LoadBalancerSrcRanges() []string { + return lbSrcRngsFlag.ipn.StringSlice() +} + +// GetLoadBalancer is an implementation of LoadBalancer.GetLoadBalancer +func (gce *GCECloud) GetLoadBalancer(clusterName string, svc *v1.Service) (*v1.LoadBalancerStatus, bool, error) { + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + fwd, err := gce.GetRegionForwardingRule(loadBalancerName, gce.region) + if err == nil { + status := &v1.LoadBalancerStatus{} + status.Ingress = []v1.LoadBalancerIngress{{IP: fwd.IPAddress}} + + return status, true, nil + } + return nil, false, ignoreNotFound(err) +} + +// EnsureLoadBalancer is an implementation of LoadBalancer.EnsureLoadBalancer. +func (gce *GCECloud) EnsureLoadBalancer(clusterName string, svc *v1.Service, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) { + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + desiredScheme := getSvcScheme(svc) + clusterID, err := gce.ClusterID.GetID() + if err != nil { + return nil, err + } + + glog.V(4).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v): ensure %v loadbalancer", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, desiredScheme) + + existingFwdRule, err := gce.GetRegionForwardingRule(loadBalancerName, gce.region) + if err != nil && !isNotFound(err) { + return nil, err + } + + if existingFwdRule != nil { + existingScheme := lbScheme(strings.ToUpper(existingFwdRule.LoadBalancingScheme)) + + // If the loadbalancer type changes between INTERNAL and EXTERNAL, the old load balancer should be deleted. + if existingScheme != desiredScheme { + glog.V(4).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v): deleting existing %v loadbalancer", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, existingScheme) + switch existingScheme { + case schemeInternal: + err = gce.ensureInternalLoadBalancerDeleted(clusterName, clusterID, svc) + default: + err = gce.ensureExternalLoadBalancerDeleted(clusterName, svc) + } + glog.V(4).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v): done deleting existing %v loadbalancer. err: %v", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, existingScheme, err) + if err != nil { + return nil, err + } + } + } + + var status *v1.LoadBalancerStatus + switch desiredScheme { + case schemeInternal: + status, err = gce.ensureInternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes) + default: + status, err = gce.ensureExternalLoadBalancer(clusterName, svc, existingFwdRule, nodes) + } + glog.V(4).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v): done ensuring loadbalancer. err: %v", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, err) + return status, err +} + +// UpdateLoadBalancer is an implementation of LoadBalancer.UpdateLoadBalancer. +func (gce *GCECloud) UpdateLoadBalancer(clusterName string, svc *v1.Service, nodes []*v1.Node) error { + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + scheme := getSvcScheme(svc) + clusterID, err := gce.ClusterID.GetID() + if err != nil { + return err + } + + glog.V(4).Infof("UpdateLoadBalancer(%v, %v, %v, %v, %v): updating with %d nodes", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, len(nodes)) + + switch scheme { + case schemeInternal: + err = gce.updateInternalLoadBalancer(clusterName, clusterID, svc, nodes) + default: + err = gce.updateExternalLoadBalancer(clusterName, svc, nodes) + } + glog.V(4).Infof("UpdateLoadBalancer(%v, %v, %v, %v, %v): done updating. err: %v", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, err) + return err +} + +// EnsureLoadBalancerDeleted is an implementation of LoadBalancer.EnsureLoadBalancerDeleted. +func (gce *GCECloud) EnsureLoadBalancerDeleted(clusterName string, svc *v1.Service) error { + loadBalancerName := cloudprovider.GetLoadBalancerName(svc) + scheme := getSvcScheme(svc) + clusterID, err := gce.ClusterID.GetID() + if err != nil { + return err + } + + glog.V(4).Infof("EnsureLoadBalancerDeleted(%v, %v, %v, %v, %v): deleting loadbalancer", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region) + + switch scheme { + case schemeInternal: + err = gce.ensureInternalLoadBalancerDeleted(clusterName, clusterID, svc) + default: + err = gce.ensureExternalLoadBalancerDeleted(clusterName, svc) + } + glog.V(4).Infof("EnsureLoadBalancerDeleted(%v, %v, %v, %v, %v): done deleting loadbalancer. err: %v", clusterName, svc.Namespace, svc.Name, loadBalancerName, gce.region, err) + return err +} + +func getSvcScheme(svc *v1.Service) lbScheme { + if typ, ok := GetLoadBalancerAnnotationType(svc); ok && typ == LBTypeInternal { + return schemeInternal + } + return schemeExternal +} From 7248c61ea54da8a05afe1e1bc874e43e9186debc Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Thu, 1 Jun 2017 15:26:05 -0700 Subject: [PATCH 7/8] Update test utilities & build file --- pkg/cloudprovider/providers/gce/BUILD | 7 ++++++- test/e2e/framework/ingress_utils.go | 5 +++-- test/e2e/framework/util.go | 11 +++-------- 3 files changed, 12 insertions(+), 11 deletions(-) diff --git a/pkg/cloudprovider/providers/gce/BUILD b/pkg/cloudprovider/providers/gce/BUILD index 2d407afb860..2155b28cde1 100644 --- a/pkg/cloudprovider/providers/gce/BUILD +++ b/pkg/cloudprovider/providers/gce/BUILD @@ -13,6 +13,8 @@ go_library( srcs = [ "doc.go", "gce.go", + "gce_addresses.go", + "gce_annotations.go", "gce_backendservice.go", "gce_cert.go", "gce_clusterid.go", @@ -24,9 +26,12 @@ go_library( "gce_instancegroup.go", "gce_instances.go", "gce_loadbalancer.go", + "gce_loadbalancer_external.go", + "gce_loadbalancer_internal.go", + "gce_loadbalancer_naming.go", "gce_op.go", "gce_routes.go", - "gce_staticip.go", + "gce_targetpool.go", "gce_targetproxy.go", "gce_urlmap.go", "gce_util.go", diff --git a/test/e2e/framework/ingress_utils.go b/test/e2e/framework/ingress_utils.go index 4b9912bcb9a..6d6969cd8d0 100644 --- a/test/e2e/framework/ingress_utils.go +++ b/test/e2e/framework/ingress_utils.go @@ -717,9 +717,10 @@ func (cont *GCEIngressController) Init() { // invoking deleteStaticIPs. func (cont *GCEIngressController) CreateStaticIP(name string) string { gceCloud := cont.Cloud.Provider.(*gcecloud.GCECloud) - ip, err := gceCloud.ReserveGlobalStaticIP(name, "") + addr := &compute.Address{Name: name} + ip, err := gceCloud.ReserveGlobalAddress(addr) if err != nil { - if delErr := gceCloud.DeleteGlobalStaticIP(name); delErr != nil { + if delErr := gceCloud.DeleteGlobalAddress(name); delErr != nil { if cont.isHTTPErrorCode(delErr, http.StatusNotFound) { Logf("Static ip with name %v was not allocated, nothing to delete", name) } else { diff --git a/test/e2e/framework/util.go b/test/e2e/framework/util.go index 0f31c013ed9..37cf069d130 100644 --- a/test/e2e/framework/util.go +++ b/test/e2e/framework/util.go @@ -5167,20 +5167,15 @@ func CleanupGCEResources(loadBalancerName string) (retErr error) { !IsGoogleAPIHTTPErrorCode(err, http.StatusNotFound) { retErr = err } - if err := gceCloud.DeleteForwardingRule(loadBalancerName); err != nil && + if err := gceCloud.DeleteRegionForwardingRule(loadBalancerName, gceCloud.Region()); err != nil && !IsGoogleAPIHTTPErrorCode(err, http.StatusNotFound) { retErr = fmt.Errorf("%v\n%v", retErr, err) } - if err := gceCloud.DeleteGlobalStaticIP(loadBalancerName); err != nil && + if err := gceCloud.DeleteRegionAddress(loadBalancerName, gceCloud.Region()); err != nil && !IsGoogleAPIHTTPErrorCode(err, http.StatusNotFound) { retErr = fmt.Errorf("%v\n%v", retErr, err) } - // This function shells out to gcloud, so we can't compare for NotFound errors. - // TODO: Invoke cloudprovider method directly instead. - if err := DeleteGCEStaticIP(loadBalancerName); err != nil { - Logf("%v", err) - } var hcNames []string hc, getErr := gceCloud.GetHttpHealthCheck(loadBalancerName) if getErr != nil && !IsGoogleAPIHTTPErrorCode(getErr, http.StatusNotFound) { @@ -5190,7 +5185,7 @@ func CleanupGCEResources(loadBalancerName string) (retErr error) { if hc != nil { hcNames = append(hcNames, hc.Name) } - if err := gceCloud.DeleteTargetPool(loadBalancerName, hcNames...); err != nil && + if err := gceCloud.DeleteExternalTargetPoolAndChecks(loadBalancerName, gceCloud.Region(), hcNames...); err != nil && !IsGoogleAPIHTTPErrorCode(err, http.StatusNotFound) { retErr = fmt.Errorf("%v\n%v", retErr, err) } From 025f178b7e5ec7217fa11ee6a4d20d9251c3e8e3 Mon Sep 17 00:00:00 2001 From: Nick Sardo Date: Sun, 4 Jun 2017 10:26:33 -0700 Subject: [PATCH 8/8] Use new kubelet apis pkg for labels --- pkg/cloudprovider/providers/gce/gce_instances.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/cloudprovider/providers/gce/gce_instances.go b/pkg/cloudprovider/providers/gce/gce_instances.go index a2f6125be13..3afd3f2969f 100644 --- a/pkg/cloudprovider/providers/gce/gce_instances.go +++ b/pkg/cloudprovider/providers/gce/gce_instances.go @@ -28,12 +28,12 @@ import ( computealpha "google.golang.org/api/compute/v0.beta" compute "google.golang.org/api/compute/v1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/wait" "k8s.io/kubernetes/pkg/api/v1" "k8s.io/kubernetes/pkg/cloudprovider" + kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis" ) const ( @@ -59,7 +59,7 @@ func splitNodesByZone(nodes []*v1.Node) map[string][]*v1.Node { } func getZone(n *v1.Node) string { - zone, ok := n.Labels[metav1.LabelZoneFailureDomain] + zone, ok := n.Labels[kubeletapis.LabelZoneFailureDomain] if !ok { return defaultZone }