Update go dependencies

This commit is contained in:
Manuel de Brito Fontes 2017-07-16 15:30:35 -04:00 committed by Minhan Xia
parent 060e449056
commit 4fb61c73d1
1192 changed files with 185874 additions and 302749 deletions

View file

@ -16,9 +16,9 @@ go_library(
],
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",
"//pkg/controller:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
],
)

View file

@ -12,7 +12,6 @@ reviewers:
- vishh
- mikedanese
- liggitt
- bprashanth
- gmarek
- erictune
- davidopp
@ -34,7 +33,6 @@ reviewers:
- rootfs
- jszczepkowski
- markturansky
- vmarmol
- girishkalele
- satnam6502
- jdef

16
vendor/k8s.io/kubernetes/pkg/cloudprovider/README.md generated vendored Normal file
View file

@ -0,0 +1,16 @@
##### Deprecation Notice: This directory has entered maintenance mode and will not be accepting new providers. Cloud Providers in this directory will continue to be actively developed or maintained and supported at their current level of support as a longer-term solution evolves.
## Overview:
The mechanism for supporting cloud providers is currently in transition: the original method of implementing cloud provider-specific functionality within the main kubernetes tree (here) is no longer advised; however, the proposed solution is still in development.
#### Guidance for potential cloud providers:
* Support for cloud providers is currently in a state of flux. Background information on motivation and the proposal for improving is in the github [proposal](https://git.k8s.io/community/contributors/design-proposals/cloud-provider-refactoring.md).
* In support of this plan, a new cloud-controller-manager binary was added in 1.6. This was the first of several steps (see the proposal for more information).
* Attempts to contribute new cloud providers or (to a lesser extent) persistent volumes to the core repo will likely meet with some pushback from reviewers/approvers.
* It is understood that this is an unfortunate situation in which 'the old way is no longer supported but the new way is not ready yet', but the initial path is unsustainable, and contributors are encouraged to participate in the implementation of the proposed long-term solution, as there is risk that PRs for new cloud providers here will not be approved.
* Though the fully productized support envisioned in the proposal is still 2 - 3 releases out, the foundational work is underway, and a motivated cloud provider could accomplish the work in a forward-looking way. Contributors are encouraged to assist with the implementation of the design outlined in the proposal.
#### Some additional context on status / direction:
* 1.6 added a new cloud-controller-manager binary that may be used for testing the new out-of-core cloudprovider flow.
* Setting cloud-provider=external allows for creation of a separate controller-manager binary
* 1.7 adds [extensible admission control](https://git.k8s.io/community/contributors/design-proposals/admission_control_extension.md), further enabling topology customization.

View file

@ -21,14 +21,15 @@ import (
"fmt"
"strings"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/controller"
)
// Interface is an abstract, pluggable interface for cloud providers.
type Interface interface {
// Initialize provides the cloud with a kubernetes client builder
// Initialize provides the cloud with a kubernetes client builder and may spawn goroutines
// to perform housekeeping activities within the cloud provider.
Initialize(clientBuilder controller.ControllerClientBuilder)
// LoadBalancer returns a balancer interface. Also returns true if the interface is supported, false otherwise.
LoadBalancer() (LoadBalancer, bool)
@ -67,6 +68,7 @@ func GetLoadBalancerName(service *v1.Service) string {
return ret
}
// GetInstanceProviderID builds a ProviderID for a node in a cloud.
func GetInstanceProviderID(cloud Interface, nodeName types.NodeName) (string, error) {
instances, ok := cloud.Instances()
if !ok {
@ -148,6 +150,9 @@ type Route struct {
// DestinationCIDR is the CIDR format IP range that this routing rule
// applies to.
DestinationCIDR string
// Blackhole is set to true if this is a blackhole route
// The node controller will delete the route if it is in the managed range.
Blackhole bool
}
// Routes is an abstract, pluggable interface for advanced routing rules.

View file

@ -13,8 +13,11 @@ go_library(
srcs = [
"doc.go",
"gce.go",
"gce_addresses.go",
"gce_annotations.go",
"gce_backendservice.go",
"gce_cert.go",
"gce_clusterid.go",
"gce_clusters.go",
"gce_disks.go",
"gce_firewall.go",
@ -23,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",
@ -35,36 +41,57 @@ go_library(
],
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",
"//pkg/api/v1/service:go_default_library",
"//pkg/client/clientset_generated/clientset:go_default_library",
"//pkg/cloudprovider:go_default_library",
"//pkg/controller:go_default_library",
"//pkg/kubelet/apis:go_default_library",
"//pkg/master/ports:go_default_library",
"//pkg/util/net/sets:go_default_library",
"//pkg/util/version:go_default_library",
"//pkg/volume:go_default_library",
"//vendor/cloud.google.com/go/compute/metadata:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
"//vendor/github.com/prometheus/client_golang/prometheus:go_default_library",
"//vendor/golang.org/x/oauth2:go_default_library",
"//vendor/golang.org/x/oauth2/google:go_default_library",
"//vendor/google.golang.org/api/cloudkms/v1:go_default_library",
"//vendor/google.golang.org/api/compute/v0.alpha:go_default_library",
"//vendor/google.golang.org/api/compute/v0.beta:go_default_library",
"//vendor/google.golang.org/api/compute/v1:go_default_library",
"//vendor/google.golang.org/api/container/v1:go_default_library",
"//vendor/google.golang.org/api/googleapi:go_default_library",
"//vendor/gopkg.in/gcfg.v1:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/watch:go_default_library",
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
"//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
],
)
go_test(
name = "go_default_test",
srcs = ["gce_test.go"],
srcs = [
"gce_disks_test.go",
"gce_healthchecks_test.go",
"gce_test.go",
],
library = ":go_default_library",
tags = ["automanaged"],
deps = [
"//pkg/cloudprovider:go_default_library",
"//pkg/kubelet/apis:go_default_library",
"//vendor/google.golang.org/api/compute/v1:go_default_library",
"//vendor/google.golang.org/api/googleapi:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
],
)
filegroup(

View file

@ -2,3 +2,7 @@ approvers:
- saad-ali
- jingxu97
- bowei
- freehan
- nicksardo
- mrhohn
- dnardo

View file

@ -22,12 +22,14 @@ import (
"net/http"
"regexp"
"strings"
"sync"
"time"
gcfg "gopkg.in/gcfg.v1"
"cloud.google.com/go/compute/metadata"
"gopkg.in/gcfg.v1"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/util/flowcontrol"
"k8s.io/kubernetes/pkg/cloudprovider"
@ -36,6 +38,8 @@ import (
"github.com/golang/glog"
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
cloudkms "google.golang.org/api/cloudkms/v1"
computealpha "google.golang.org/api/compute/v0.alpha"
computebeta "google.golang.org/api/compute/v0.beta"
compute "google.golang.org/api/compute/v1"
container "google.golang.org/api/container/v1"
@ -72,22 +76,61 @@ const (
gceHcHealthyThreshold = int64(1)
// Defaults to 5 * 2 = 10 seconds before the LB will steer traffic away
gceHcUnhealthyThreshold = int64(5)
gceComputeAPIEndpoint = "https://www.googleapis.com/compute/v1/"
)
// 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
serviceBeta *computebeta.Service
serviceAlpha *computealpha.Service
containerService *container.Service
cloudkmsService *cloudkms.Service
clientBuilder controller.ControllerClientBuilder
projectID string
region string
localZone string // The zone in which we are running
managedZones []string // List of zones we are spanning (for multi-AZ clusters, primarily when running on master)
networkURL string
nodeTags []string // List of tags to use on firewall rules for load balancers
nodeInstancePrefix string // If non-"", an advisory prefix for all nodes in the cluster
subnetworkURL string
networkProjectID string
onXPN bool
nodeTags []string // List of tags to use on firewall rules for load balancers
lastComputedNodeTags []string // List of node tags calculated in GetHostTags()
lastKnownNodeNames sets.String // List of hostnames used to calculate lastComputedHostTags in GetHostTags(names)
computeNodeTagLock sync.Mutex // Lock for computing and setting node tags
nodeInstancePrefix string // If non-"", an advisory prefix for all nodes in the cluster
useMetadataServer bool
operationPollRateLimiter flowcontrol.RateLimiter
manager ServiceManager
// sharedResourceLock is used to serialize GCE operations that may mutate shared state to
// prevent inconsistencies. For example, load balancers manipulation methods will take the
// lock to prevent shared resources from being prematurely deleted while the operation is
// in progress.
sharedResourceLock sync.Mutex
}
type ServiceManager interface {
// Creates a new persistent disk on GCE with the given disk spec.
CreateDisk(project string, zone string, disk *compute.Disk) (*compute.Operation, error)
// Gets the persistent disk from GCE with the given diskName.
GetDisk(project string, zone string, diskName string) (*compute.Disk, error)
// Deletes the persistent disk from GCE with the given diskName.
DeleteDisk(project string, zone string, disk string) (*compute.Operation, error)
// Waits until GCE reports the given operation in the given zone as done.
WaitForZoneOp(op *compute.Operation, zone string, mc *metricContext) error
}
type GCEServiceManager struct {
gce *GCECloud
}
type Config struct {
@ -96,9 +139,12 @@ type Config struct {
TokenBody string `gcfg:"token-body"`
ProjectID string `gcfg:"project-id"`
NetworkName string `gcfg:"network-name"`
SubnetworkName string `gcfg:"subnetwork-name"`
NodeTags []string `gcfg:"node-tags"`
NodeInstancePrefix string `gcfg:"node-instance-prefix"`
Multizone bool `gcfg:"multizone"`
// Specifying ApiEndpoint will override the default GCE compute API endpoint.
ApiEndpoint string `gcfg:"api-endpoint"`
}
}
@ -115,8 +161,19 @@ func (g *GCECloud) GetComputeService() *compute.Service {
return g.service
}
// Raw access to the cloudkmsService of GCE cloud. Required for encryption of etcd using Google KMS.
func (g *GCECloud) GetKMSService() *cloudkms.Service {
return g.cloudkmsService
}
// Returns the ProjectID corresponding to the project this cloud is in.
func (g *GCECloud) GetProjectID() string {
return g.projectID
}
// newGCECloud creates a new instance of GCECloud.
func newGCECloud(config io.Reader) (*GCECloud, error) {
apiEndpoint := ""
projectID, zone, err := getProjectAndZone()
if err != nil {
return nil, err
@ -131,7 +188,8 @@ func newGCECloud(config io.Reader) (*GCECloud, error) {
if err != nil {
return nil, err
}
networkURL := gceNetworkURL(projectID, networkName)
networkURL := gceNetworkURL("", projectID, networkName)
subnetworkURL := ""
// By default, Kubernetes clusters only run against one zone
managedZones := []string{zone}
@ -146,15 +204,23 @@ func newGCECloud(config io.Reader) (*GCECloud, error) {
return nil, err
}
glog.Infof("Using GCE provider config %+v", cfg)
if cfg.Global.ApiEndpoint != "" {
apiEndpoint = cfg.Global.ApiEndpoint
}
if cfg.Global.ProjectID != "" {
projectID = cfg.Global.ProjectID
}
if cfg.Global.NetworkName != "" {
if strings.Contains(cfg.Global.NetworkName, "/") {
networkURL = cfg.Global.NetworkName
} else {
networkURL = gceNetworkURL(cfg.Global.ProjectID, cfg.Global.NetworkName)
}
if cfg.Global.NetworkName != "" && strings.Contains(cfg.Global.NetworkName, "/") {
networkURL = cfg.Global.NetworkName
} else {
networkURL = gceNetworkURL(apiEndpoint, projectID, networkName)
}
if cfg.Global.SubnetworkName != "" && strings.Contains(cfg.Global.SubnetworkName, "/") {
subnetworkURL = cfg.Global.SubnetworkName
} else {
subnetworkURL = gceSubnetworkURL(apiEndpoint, cfg.Global.ProjectID, region, cfg.Global.SubnetworkName)
}
if cfg.Global.TokenURL != "" {
tokenSource = NewAltTokenSource(cfg.Global.TokenURL, cfg.Global.TokenBody)
@ -166,46 +232,78 @@ func newGCECloud(config io.Reader) (*GCECloud, error) {
}
}
return CreateGCECloud(projectID, region, zone, managedZones, networkURL, nodeTags,
nodeInstancePrefix, tokenSource, true /* useMetadataServer */)
return CreateGCECloud(apiEndpoint, projectID, region, zone, managedZones, networkURL, subnetworkURL,
nodeTags, nodeInstancePrefix, tokenSource, true /* useMetadataServer */)
}
// Creates a GCECloud object using the specified parameters.
// If no networkUrl is specified, loads networkName via rest call.
// If no tokenSource is specified, uses oauth2.DefaultTokenSource.
// If managedZones is nil / empty all zones in the region will be managed.
func CreateGCECloud(projectID, region, zone string, managedZones []string, networkURL string, nodeTags []string,
func CreateGCECloud(apiEndpoint, projectID, region, zone string, managedZones []string, networkURL, subnetworkURL string, nodeTags []string,
nodeInstancePrefix string, tokenSource oauth2.TokenSource, useMetadataServer bool) (*GCECloud, error) {
client, err := newOauthClient(tokenSource)
if err != nil {
return nil, err
}
service, err := compute.New(client)
if err != nil {
return nil, err
}
client, err = newOauthClient(tokenSource)
if err != nil {
return nil, err
}
serviceBeta, err := computebeta.New(client)
if err != nil {
return nil, err
}
client, err = newOauthClient(tokenSource)
if err != nil {
return nil, err
}
serviceAlpha, err := computealpha.New(client)
if err != nil {
return nil, err
}
// Expect override api endpoint to always be v1 api and follows the same pattern as prod.
// Generate alpha and beta api endpoints based on override v1 api endpoint.
// For example,
// staging API endpoint: https://www.googleapis.com/compute/staging_v1/
if apiEndpoint != "" {
service.BasePath = fmt.Sprintf("%sprojects/", apiEndpoint)
serviceBeta.BasePath = fmt.Sprintf("%sprojects/", strings.Replace(apiEndpoint, "v1", "beta", 0))
serviceAlpha.BasePath = fmt.Sprintf("%sprojects/", strings.Replace(apiEndpoint, "v1", "alpha", 0))
}
containerService, err := container.New(client)
if err != nil {
return nil, err
}
cloudkmsService, err := cloudkms.New(client)
if err != nil {
return nil, err
}
if networkURL == "" {
networkName, err := getNetworkNameViaAPICall(service, projectID)
if err != nil {
return nil, err
}
networkURL = gceNetworkURL(projectID, networkName)
networkURL = gceNetworkURL(apiEndpoint, projectID, networkName)
}
networkProjectID, err := getProjectIDInURL(networkURL)
if err != nil {
return nil, err
}
onXPN := networkProjectID != projectID
if len(managedZones) == 0 {
managedZones, err = getZonesForRegion(service, projectID, region)
if err != nil {
@ -218,24 +316,35 @@ func CreateGCECloud(projectID, region, zone string, managedZones []string, netwo
operationPollRateLimiter := flowcontrol.NewTokenBucketRateLimiter(10, 100) // 10 qps, 100 bucket size.
return &GCECloud{
gce := &GCECloud{
service: service,
serviceBeta: serviceBeta,
containerService: containerService,
cloudkmsService: cloudkmsService,
projectID: projectID,
networkProjectID: networkProjectID,
onXPN: onXPN,
region: region,
localZone: zone,
managedZones: managedZones,
networkURL: networkURL,
subnetworkURL: subnetworkURL,
nodeTags: nodeTags,
nodeInstancePrefix: nodeInstancePrefix,
useMetadataServer: useMetadataServer,
operationPollRateLimiter: operationPollRateLimiter,
}, nil
}
gce.manager = &GCEServiceManager{gce}
return gce, nil
}
// Initialize passes a Kubernetes clientBuilder interface to the cloud provider
func (gce *GCECloud) Initialize(clientBuilder controller.ControllerClientBuilder) {}
// Initialize takes in a clientBuilder and spawns a goroutine for watching the clusterid configmap.
// This must be called before utilizing the funcs of gce.ClusterID
func (gce *GCECloud) Initialize(clientBuilder controller.ControllerClientBuilder) {
gce.clientBuilder = clientBuilder
go gce.watchClusterID()
}
// LoadBalancer returns an implementation of LoadBalancer for Google Compute Engine.
func (gce *GCECloud) LoadBalancer() (cloudprovider.LoadBalancer, bool) {
@ -266,6 +375,26 @@ func (gce *GCECloud) ProviderName() string {
return ProviderName
}
// Region returns the region
func (gce *GCECloud) Region() string {
return gce.region
}
// OnXPN returns true if the cluster is running on a cross project network (XPN)
func (gce *GCECloud) OnXPN() bool {
return gce.onXPN
}
// NetworkURL returns the network url
func (gce *GCECloud) NetworkURL() string {
return gce.networkURL
}
// SubnetworkURL returns the subnetwork url
func (gce *GCECloud) SubnetworkURL() string {
return gce.subnetworkURL
}
// Known-useless DNS search path.
var uselessDNSSearchRE = regexp.MustCompile(`^[0-9]+.google.internal.$`)
@ -283,8 +412,32 @@ func (gce *GCECloud) ScrubDNS(nameservers, searches []string) (nsOut, srchOut []
// GCECloud implements cloudprovider.Interface.
var _ cloudprovider.Interface = (*GCECloud)(nil)
func gceNetworkURL(project, network string) string {
return fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s", project, network)
func gceNetworkURL(apiEndpoint, project, network string) string {
if apiEndpoint == "" {
apiEndpoint = gceComputeAPIEndpoint
}
return apiEndpoint + strings.Join([]string{"projects", project, "global", "networks", network}, "/")
}
func gceSubnetworkURL(apiEndpoint, project, region, subnetwork string) string {
if apiEndpoint == "" {
apiEndpoint = gceComputeAPIEndpoint
}
return apiEndpoint + strings.Join([]string{"projects", project, "regions", region, "subnetworks", subnetwork}, "/")
}
// getProjectIDInURL parses typical full resource URLS and shorter URLS
// https://www.googleapis.com/compute/v1/projects/myproject/global/networks/mycustom
// projects/myproject/global/networks/mycustom
// All return "myproject"
func getProjectIDInURL(urlStr string) (string, error) {
fields := strings.Split(urlStr, "/")
for i, v := range fields {
if v == "projects" && i < len(fields)-1 {
return fields[i+1], nil
}
}
return "", fmt.Errorf("could not find project field in url: %v", urlStr)
}
func getNetworkNameViaMetadata() (string, error) {
@ -362,3 +515,31 @@ func newOauthClient(tokenSource oauth2.TokenSource) (*http.Client, error) {
return oauth2.NewClient(oauth2.NoContext, tokenSource), nil
}
func (manager *GCEServiceManager) CreateDisk(
project string,
zone string,
disk *compute.Disk) (*compute.Operation, error) {
return manager.gce.service.Disks.Insert(project, zone, disk).Do()
}
func (manager *GCEServiceManager) GetDisk(
project string,
zone string,
diskName string) (*compute.Disk, error) {
return manager.gce.service.Disks.Get(project, zone, diskName).Do()
}
func (manager *GCEServiceManager) DeleteDisk(
project string,
zone string,
diskName string) (*compute.Operation, error) {
return manager.gce.service.Disks.Delete(project, zone, diskName).Do()
}
func (manager *GCEServiceManager) WaitForZoneOp(op *compute.Operation, zone string, mc *metricContext) error {
return manager.gce.waitForZoneOp(op, zone, mc)
}

View file

@ -0,0 +1,87 @@
/*
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 newAddressMetricContext(request, region string) *metricContext {
return &metricContext{
start: time.Now(),
attributes: []string{"address_" + request, region, unusedMetricLabel},
}
}
// 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) ReserveGlobalAddress(addr *compute.Address) error {
mc := newAddressMetricContext("reserve", "")
op, err := gce.service.GlobalAddresses.Insert(gce.projectID, addr).Do()
if err != nil {
return mc.Observe(err)
}
return gce.waitForGlobalOp(op, mc)
}
// 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)
}
return gce.waitForGlobalOp(op, mc)
}
// 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) error {
mc := newAddressMetricContext("reserve", region)
op, err := gce.service.Addresses.Insert(gce.projectID, region, addr).Do()
if err != nil {
return mc.Observe(err)
}
return gce.waitForRegionOp(op, region, mc)
}
// 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)
}

View file

@ -0,0 +1,67 @@
/*
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/api/core/v1"
type LoadBalancerType string
const (
// ServiceAnnotationLoadBalancerType is annotated 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"
// ServiceAnnotationInternalBackendShare is annotated on a service with "true" when users
// want to share GCP Backend Services for a set of internal load balancers.
// ALPHA feature - this may be removed in a future release.
ServiceAnnotationILBBackendShare = "cloud.google.com/load-balancer-backend-share"
)
// 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
}
}
// GetLoadBalancerAnnotationBackendShare returns whether this service's backend service should be
// shared with other load balancers. Health checks and the healthcheck firewall will be shared regardless.
func GetLoadBalancerAnnotationBackendShare(service *v1.Service) bool {
l, exists := service.Annotations[ServiceAnnotationILBBackendShare]
if exists && l == "true" {
return true
}
return false
}

View file

@ -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},
}
}
// GetBackendService retrieves a backend by name.
func (gce *GCECloud) GetBackendService(name string) (*compute.BackendService, error) {
mc := newBackendServiceMetricContext("get")
// GetGlobalBackendService retrieves a backend by name.
func (gce *GCECloud) GetGlobalBackendService(name string) (*compute.BackendService, error) {
mc := newBackendServiceMetricContext("get", "")
v, err := gce.service.BackendServices.Get(gce.projectID, name).Do()
return v, mc.Observe(err)
}
// UpdateBackendService applies the given BackendService as an update to an existing service.
func (gce *GCECloud) UpdateBackendService(bg *compute.BackendService) error {
mc := newBackendServiceMetricContext("update")
// UpdateGlobalBackendService applies the given BackendService as an update to an existing service.
func (gce *GCECloud) UpdateGlobalBackendService(bg *compute.BackendService) error {
mc := newBackendServiceMetricContext("update", "")
op, err := gce.service.BackendServices.Update(gce.projectID, bg.Name, bg).Do()
if err != nil {
return mc.Observe(err)
@ -48,9 +48,9 @@ func (gce *GCECloud) UpdateBackendService(bg *compute.BackendService) error {
return gce.waitForGlobalOp(op, mc)
}
// DeleteBackendService deletes the given BackendService by name.
func (gce *GCECloud) DeleteBackendService(name string) error {
mc := newBackendServiceMetricContext("delete")
// DeleteGlobalBackendService deletes the given BackendService by name.
func (gce *GCECloud) DeleteGlobalBackendService(name string) error {
mc := newBackendServiceMetricContext("delete", "")
op, err := gce.service.BackendServices.Delete(gce.projectID, name).Do()
if err != nil {
if isHTTPErrorCode(err, http.StatusNotFound) {
@ -62,9 +62,9 @@ func (gce *GCECloud) DeleteBackendService(name string) error {
return gce.waitForGlobalOp(op, mc)
}
// CreateBackendService creates the given BackendService.
func (gce *GCECloud) CreateBackendService(bg *compute.BackendService) error {
mc := newBackendServiceMetricContext("create")
// CreateGlobalBackendService creates the given BackendService.
func (gce *GCECloud) CreateGlobalBackendService(bg *compute.BackendService) error {
mc := newBackendServiceMetricContext("create", "")
op, err := gce.service.BackendServices.Insert(gce.projectID, bg).Do()
if err != nil {
return mc.Observe(err)
@ -73,16 +73,81 @@ func (gce *GCECloud) CreateBackendService(bg *compute.BackendService) error {
return gce.waitForGlobalOp(op, mc)
}
// ListBackendServices lists all backend services in the project.
func (gce *GCECloud) ListBackendServices() (*compute.BackendServiceList, error) {
// ListGlobalBackendServices lists all backend services in the project.
func (gce *GCECloud) ListGlobalBackendServices() (*compute.BackendServiceList, error) {
mc := newBackendServiceMetricContext("list", "")
// TODO: use PageToken to list all not just the first 500
return gce.service.BackendServices.List(gce.projectID).Do()
v, err := gce.service.BackendServices.List(gce.projectID).Do()
return v, mc.Observe(err)
}
// GetHealth returns the health of the BackendService identified by the given
// GetGlobalBackendServiceHealth returns the health of the BackendService identified by the given
// name, in the given instanceGroup. The instanceGroupLink is the fully
// qualified self link of an instance group.
func (gce *GCECloud) GetHealth(name string, instanceGroupLink string) (*compute.BackendServiceGroupHealth, error) {
func (gce *GCECloud) GetGlobalBackendServiceHealth(name string, instanceGroupLink string) (*compute.BackendServiceGroupHealth, error) {
mc := newBackendServiceMetricContext("get_health", "")
groupRef := &compute.ResourceGroupReference{Group: instanceGroupLink}
return gce.service.BackendServices.GetHealth(gce.projectID, name, groupRef).Do()
v, err := gce.service.BackendServices.GetHealth(gce.projectID, name, groupRef).Do()
return v, mc.Observe(err)
}
// GetRegionBackendService retrieves a backend by name.
func (gce *GCECloud) GetRegionBackendService(name, region string) (*compute.BackendService, error) {
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, 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, region, mc)
}
// DeleteRegionBackendService deletes the given BackendService by name.
func (gce *GCECloud) DeleteRegionBackendService(name, region string) error {
mc := newBackendServiceMetricContext("delete", region)
op, err := gce.service.RegionBackendServices.Delete(gce.projectID, region, name).Do()
if err != nil {
if isHTTPErrorCode(err, http.StatusNotFound) {
return nil
}
return mc.Observe(err)
}
return gce.waitForRegionOp(op, region, mc)
}
// CreateRegionBackendService creates the given BackendService.
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, region, mc)
}
// ListRegionBackendServices lists all backend services in the project.
func (gce *GCECloud) ListRegionBackendServices(region string) (*compute.BackendServiceList, error) {
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)
}
// GetRegionalBackendServiceHealth returns the health of the BackendService identified by the given
// 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", region)
groupRef := &compute.ResourceGroupReference{Group: instanceGroupLink}
v, err := gce.service.RegionBackendServices.GetHealth(gce.projectID, region, name, groupRef).Do()
return v, mc.Observe(err)
}

View file

@ -0,0 +1,258 @@
/*
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 (
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"reflect"
"sync"
"time"
"github.com/golang/glog"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/cache"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
)
const (
// Key used to persist UIDs to configmaps.
UIDConfigMapName = "ingress-uid"
// Namespace which contains the above config map
UIDNamespace = metav1.NamespaceSystem
// Data keys for the specific ids
UIDCluster = "uid"
UIDProvider = "provider-uid"
UIDLengthBytes = 8
// Frequency of the updateFunc event handler being called
// This does not actually query the apiserver for current state - the local cache value is used.
updateFuncFrequency = 10 * time.Minute
)
type ClusterID struct {
idLock sync.RWMutex
client clientset.Interface
cfgMapKey string
store cache.Store
providerID *string
clusterID *string
}
// Continually watches for changes to the cluster id config map
func (gce *GCECloud) watchClusterID() {
gce.ClusterID = ClusterID{
cfgMapKey: fmt.Sprintf("%v/%v", UIDNamespace, UIDConfigMapName),
client: gce.clientBuilder.ClientOrDie("cloud-provider"),
}
mapEventHandler := cache.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
m, ok := obj.(*v1.ConfigMap)
if !ok || m == nil {
glog.Errorf("Expected v1.ConfigMap, item=%+v, typeIsOk=%v", obj, ok)
return
}
if m.Namespace != UIDNamespace ||
m.Name != UIDConfigMapName {
return
}
glog.V(4).Infof("Observed new configmap for clusteriD: %v, %v; setting local values", m.Name, m.Data)
gce.ClusterID.update(m)
},
UpdateFunc: func(old, cur interface{}) {
m, ok := cur.(*v1.ConfigMap)
if !ok || m == nil {
glog.Errorf("Expected v1.ConfigMap, item=%+v, typeIsOk=%v", cur, ok)
return
}
if m.Namespace != UIDNamespace ||
m.Name != UIDConfigMapName {
return
}
if reflect.DeepEqual(old, cur) {
return
}
glog.V(4).Infof("Observed updated configmap for clusteriD %v, %v; setting local values", m.Name, m.Data)
gce.ClusterID.update(m)
},
}
listerWatcher := cache.NewListWatchFromClient(gce.ClusterID.client.Core().RESTClient(), "configmaps", UIDNamespace, fields.Everything())
var controller cache.Controller
gce.ClusterID.store, controller = cache.NewInformer(newSingleObjectListerWatcher(listerWatcher, UIDConfigMapName), &v1.ConfigMap{}, updateFuncFrequency, mapEventHandler)
controller.Run(nil)
}
// GetID returns the id which is unique to this cluster
// if federated, return the provider id (unique to the cluster)
// if not federated, return the cluster id
func (ci *ClusterID) GetID() (string, error) {
if err := ci.getOrInitialize(); err != nil {
return "", err
}
ci.idLock.RLock()
defer ci.idLock.RUnlock()
if ci.clusterID == nil {
return "", errors.New("Could not retrieve cluster id")
}
// If provider ID is set, (Federation is enabled) use this field
if ci.providerID != nil {
return *ci.providerID, nil
}
// providerID is not set, use the cluster id
return *ci.clusterID, nil
}
// GetFederationId returns the id which could represent the entire Federation
// or just the cluster if not federated.
func (ci *ClusterID) GetFederationId() (string, bool, error) {
if err := ci.getOrInitialize(); err != nil {
return "", false, err
}
ci.idLock.RLock()
defer ci.idLock.RUnlock()
if ci.clusterID == nil {
return "", false, errors.New("Could not retrieve cluster id")
}
// If provider ID is not set, return false
if ci.providerID == nil || *ci.clusterID == *ci.providerID {
return "", false, nil
}
return *ci.clusterID, true, nil
}
// getOrInitialize either grabs the configmaps current value or defines the value
// and sets the configmap. This is for the case of the user calling GetClusterID()
// before the watch has begun.
func (ci *ClusterID) getOrInitialize() error {
if ci.store == nil {
return errors.New("GCECloud.ClusterID is not ready. Call Initialize() before using.")
}
if ci.clusterID != nil {
return nil
}
exists, err := ci.getConfigMap()
if err != nil {
return err
} else if exists {
return nil
}
// The configmap does not exist - let's try creating one.
newId, err := makeUID()
if err != nil {
return err
}
glog.V(4).Infof("Creating clusteriD: %v", newId)
cfg := &v1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
Name: UIDConfigMapName,
Namespace: UIDNamespace,
},
}
cfg.Data = map[string]string{
UIDCluster: newId,
UIDProvider: newId,
}
if _, err := ci.client.Core().ConfigMaps(UIDNamespace).Create(cfg); err != nil {
glog.Errorf("GCE cloud provider failed to create %v config map to store cluster id: %v", ci.cfgMapKey, err)
return err
}
glog.V(2).Infof("Created a config map containing clusteriD: %v", newId)
ci.update(cfg)
return nil
}
func (ci *ClusterID) getConfigMap() (bool, error) {
item, exists, err := ci.store.GetByKey(ci.cfgMapKey)
if err != nil {
return false, err
}
if !exists {
return false, nil
}
m, ok := item.(*v1.ConfigMap)
if !ok || m == nil {
err = fmt.Errorf("Expected v1.ConfigMap, item=%+v, typeIsOk=%v", item, ok)
glog.Error(err)
return false, err
}
ci.update(m)
return true, nil
}
func (ci *ClusterID) update(m *v1.ConfigMap) {
ci.idLock.Lock()
defer ci.idLock.Unlock()
if clusterID, exists := m.Data[UIDCluster]; exists {
ci.clusterID = &clusterID
}
if provId, exists := m.Data[UIDProvider]; exists {
ci.providerID = &provId
}
}
func makeUID() (string, error) {
b := make([]byte, UIDLengthBytes)
_, err := rand.Read(b)
if err != nil {
return "", err
}
return hex.EncodeToString(b), nil
}
func newSingleObjectListerWatcher(lw cache.ListerWatcher, objectName string) *singleObjListerWatcher {
return &singleObjListerWatcher{lw: lw, objectName: objectName}
}
type singleObjListerWatcher struct {
lw cache.ListerWatcher
objectName string
}
func (sow *singleObjListerWatcher) List(options metav1.ListOptions) (runtime.Object, error) {
options.FieldSelector = "metadata.name=" + sow.objectName
return sow.lw.List(options)
}
func (sow *singleObjListerWatcher) Watch(options metav1.ListOptions) (watch.Interface, error) {
options.FieldSelector = "metadata.name=" + sow.objectName
return sow.lw.Watch(options)
}

View file

@ -20,13 +20,12 @@ import (
"encoding/json"
"fmt"
"net/http"
"path"
"strings"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/kubernetes/pkg/cloudprovider"
kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
"k8s.io/kubernetes/pkg/volume"
"github.com/golang/glog"
@ -41,7 +40,7 @@ const (
DiskTypeStandard = "pd-standard"
diskTypeDefault = DiskTypeStandard
diskTypeUriTemplate = "https://www.googleapis.com/compute/v1/projects/%s/zones/%s/diskTypes/%s"
diskTypeUriTemplate = "%s/zones/%s/diskTypes/%s"
)
// Disks is interface for manipulation with GCE PDs.
@ -234,7 +233,12 @@ func (gce *GCECloud) CreateDisk(
default:
return fmt.Errorf("invalid GCE disk type %q", diskType)
}
diskTypeUri := fmt.Sprintf(diskTypeUriTemplate, gce.projectID, zone, diskType)
projectsApiEndpoint := gceComputeAPIEndpoint + "projects/"
if gce.service != nil {
projectsApiEndpoint = gce.service.BasePath
}
diskTypeUri := projectsApiEndpoint + fmt.Sprintf(diskTypeUriTemplate, gce.projectID, zone, diskType)
diskToCreate := &compute.Disk{
Name: name,
@ -244,12 +248,15 @@ func (gce *GCECloud) CreateDisk(
}
mc := newDiskMetricContext("create", zone)
createOp, err := gce.service.Disks.Insert(gce.projectID, zone, diskToCreate).Do()
if err != nil {
createOp, err := gce.manager.CreateDisk(gce.projectID, zone, diskToCreate)
if isGCEError(err, "alreadyExists") {
glog.Warningf("GCE PD %q already exists, reusing", name)
return nil
} else if err != nil {
return mc.Observe(err)
}
err = gce.waitForZoneOp(createOp, zone, mc)
err = gce.manager.WaitForZoneOp(createOp, zone, mc)
if isGCEError(err, "alreadyExists") {
glog.Warningf("GCE PD %q already exists, reusing", name)
return nil
@ -310,8 +317,8 @@ func (gce *GCECloud) GetAutoLabelsForPD(name string, zone string) (map[string]st
}
labels := make(map[string]string)
labels[metav1.LabelZoneFailureDomain] = zone
labels[metav1.LabelZoneRegion] = region
labels[kubeletapis.LabelZoneFailureDomain] = zone
labels[kubeletapis.LabelZoneRegion] = region
return labels, nil
}
@ -320,7 +327,7 @@ func (gce *GCECloud) GetAutoLabelsForPD(name string, zone string) (map[string]st
// If not found, returns (nil, nil)
func (gce *GCECloud) findDiskByName(diskName string, zone string) (*GCEDisk, error) {
mc := newDiskMetricContext("get", zone)
disk, err := gce.service.Disks.Get(gce.projectID, zone, diskName).Do()
disk, err := gce.manager.GetDisk(gce.projectID, zone, diskName)
if err == nil {
d := &GCEDisk{
Zone: lastComponent(disk.Zone),
@ -407,12 +414,12 @@ func (gce *GCECloud) doDeleteDisk(diskToDelete string) error {
mc := newDiskMetricContext("delete", disk.Zone)
deleteOp, err := gce.service.Disks.Delete(gce.projectID, disk.Zone, disk.Name).Do()
deleteOp, err := gce.manager.DeleteDisk(gce.projectID, disk.Zone, disk.Name)
if err != nil {
return mc.Observe(err)
}
return gce.waitForZoneOp(deleteOp, disk.Zone, mc)
return gce.manager.WaitForZoneOp(deleteOp, disk.Zone, mc)
}
// Converts a Disk resource to an AttachedDisk resource.
@ -421,9 +428,8 @@ func (gce *GCECloud) convertDiskToAttachedDisk(disk *GCEDisk, readWrite string)
DeviceName: disk.Name,
Kind: disk.Kind,
Mode: readWrite,
Source: "https://" + path.Join(
"www.googleapis.com/compute/v1/projects/",
gce.projectID, "zones", disk.Zone, "disks", disk.Name),
Source: gce.service.BasePath + strings.Join([]string{
gce.projectID, "zones", disk.Zone, "disks", disk.Name}, "/"),
Type: "PERSISTENT",
}
}

View file

@ -19,96 +19,51 @@ package gce
import (
"time"
"k8s.io/kubernetes/pkg/api/v1"
netsets "k8s.io/kubernetes/pkg/util/net/sets"
compute "google.golang.org/api/compute/v1"
)
func newFirewallMetricContext(request string, region string) *metricContext {
func newFirewallMetricContext(request string) *metricContext {
return &metricContext{
start: time.Now(),
attributes: []string{"firewall_" + request, region, unusedMetricLabel},
attributes: []string{"firewall_" + request, unusedMetricLabel, unusedMetricLabel},
}
}
// GetFirewall returns the Firewall by name.
func (gce *GCECloud) GetFirewall(name string) (*compute.Firewall, error) {
mc := newFirewallMetricContext("get", "")
mc := newFirewallMetricContext("get")
v, err := gce.service.Firewalls.Get(gce.projectID, name).Do()
return v, mc.Observe(err)
}
// CreateFirewall creates the given firewall rule.
func (gce *GCECloud) CreateFirewall(name, desc string, sourceRanges netsets.IPNet, ports []int64, hostNames []string) error {
region, err := GetGCERegion(gce.localZone)
// CreateFirewall creates the passed firewall
func (gce *GCECloud) CreateFirewall(f *compute.Firewall) error {
mc := newFirewallMetricContext("create")
op, err := gce.service.Firewalls.Insert(gce.projectID, f).Do()
if err != nil {
return err
return mc.Observe(err)
}
mc := newFirewallMetricContext("create", region)
// TODO: This completely breaks modularity in the cloudprovider but
// the methods shared with the TCPLoadBalancer take v1.ServicePorts.
svcPorts := []v1.ServicePort{}
// TODO: Currently the only consumer of this method is the GCE L7
// loadbalancer controller, which never needs a protocol other than
// TCP. We should pipe through a mapping of port:protocol and
// default to TCP if UDP ports are required. This means the method
// signature will change forcing downstream clients to refactor
// interfaces.
for _, p := range ports {
svcPorts = append(svcPorts, v1.ServicePort{Port: int32(p), Protocol: v1.ProtocolTCP})
}
hosts, err := gce.getInstancesByNames(hostNames)
if err != nil {
mc.Observe(err)
return err
}
return mc.Observe(gce.createFirewall(name, region, desc, sourceRanges, svcPorts, hosts))
return gce.waitForGlobalOp(op, mc)
}
// DeleteFirewall deletes the given firewall rule.
func (gce *GCECloud) DeleteFirewall(name string) error {
region, err := GetGCERegion(gce.localZone)
mc := newFirewallMetricContext("delete")
op, err := gce.service.Firewalls.Delete(gce.projectID, name).Do()
if err != nil {
return err
return mc.Observe(err)
}
mc := newFirewallMetricContext("delete", region)
return mc.Observe(gce.deleteFirewall(name, region))
return gce.waitForGlobalOp(op, mc)
}
// UpdateFirewall applies the given firewall rule as an update to an
// existing firewall rule with the same name.
func (gce *GCECloud) UpdateFirewall(name, desc string, sourceRanges netsets.IPNet, ports []int64, hostNames []string) error {
region, err := GetGCERegion(gce.localZone)
// UpdateFirewall applies the given firewall as an update to an existing service.
func (gce *GCECloud) UpdateFirewall(f *compute.Firewall) error {
mc := newFirewallMetricContext("update")
op, err := gce.service.Firewalls.Update(gce.projectID, f.Name, f).Do()
if err != nil {
return err
return mc.Observe(err)
}
mc := newFirewallMetricContext("update", region)
// TODO: This completely breaks modularity in the cloudprovider but
// the methods shared with the TCPLoadBalancer take v1.ServicePorts.
svcPorts := []v1.ServicePort{}
// TODO: Currently the only consumer of this method is the GCE L7
// loadbalancer controller, which never needs a protocol other than
// TCP. We should pipe through a mapping of port:protocol and
// default to TCP if UDP ports are required. This means the method
// signature will change, forcing downstream clients to refactor
// interfaces.
for _, p := range ports {
svcPorts = append(svcPorts, v1.ServicePort{Port: int32(p), Protocol: v1.ProtocolTCP})
}
hosts, err := gce.getInstancesByNames(hostNames)
if err != nil {
mc.Observe(err)
return err
}
return mc.Observe(gce.updateFirewall(name, region, desc, sourceRanges, svcPorts, hosts))
return gce.waitForGlobalOp(op, mc)
}

View file

@ -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)
}

View file

@ -19,9 +19,31 @@ package gce
import (
"time"
"k8s.io/api/core/v1"
"k8s.io/kubernetes/pkg/master/ports"
utilversion "k8s.io/kubernetes/pkg/util/version"
"github.com/golang/glog"
compute "google.golang.org/api/compute/v1"
)
const (
nodesHealthCheckPath = "/healthz"
lbNodesHealthCheckPort = ports.ProxyHealthzPort
)
var (
minNodesHealthCheckVersion *utilversion.Version
)
func init() {
if v, err := utilversion.ParseGeneric("1.7.0"); err != nil {
panic(err)
} else {
minNodesHealthCheckVersion = v
}
}
func newHealthcheckMetricContext(request string) *metricContext {
return &metricContext{
start: time.Now(),
@ -178,3 +200,37 @@ func (gce *GCECloud) ListHealthChecks() (*compute.HealthCheckList, error) {
v, err := gce.service.HealthChecks.List(gce.projectID).Do()
return v, mc.Observe(err)
}
// GetNodesHealthCheckPort returns the health check port used by the GCE load
// balancers (l4) for performing health checks on nodes.
func GetNodesHealthCheckPort() int32 {
return lbNodesHealthCheckPort
}
// GetNodesHealthCheckPath returns the health check path used by the GCE load
// balancers (l4) for performing health checks on nodes.
func GetNodesHealthCheckPath() string {
return nodesHealthCheckPath
}
// isAtLeastMinNodesHealthCheckVersion checks if a version is higher than
// `minNodesHealthCheckVersion`.
func isAtLeastMinNodesHealthCheckVersion(vstring string) bool {
version, err := utilversion.ParseGeneric(vstring)
if err != nil {
glog.Errorf("vstring (%s) is not a valid version string: %v", vstring, err)
return false
}
return version.AtLeast(minNodesHealthCheckVersion)
}
// supportsNodesHealthCheck returns false if anyone of the nodes has version
// lower than `minNodesHealthCheckVersion`.
func supportsNodesHealthCheck(nodes []*v1.Node) bool {
for _, node := range nodes {
if !isAtLeastMinNodesHealthCheckVersion(node.Status.NodeInfo.KubeProxyVersion) {
return false
}
}
return true
}

View file

@ -17,12 +17,8 @@ limitations under the License.
package gce
import (
"fmt"
"net/http"
"strings"
"time"
"github.com/golang/glog"
compute "google.golang.org/api/compute/v1"
)
@ -37,7 +33,6 @@ func newInstanceGroupMetricContext(request string, zone string) *metricContext {
// instances. It is the callers responsibility to add named ports.
func (gce *GCECloud) CreateInstanceGroup(name string, zone string) (*compute.InstanceGroup, error) {
mc := newInstanceGroupMetricContext("create", zone)
op, err := gce.service.InstanceGroups.Insert(
gce.projectID, zone, &compute.InstanceGroup{Name: name}).Do()
if err != nil {
@ -55,12 +50,10 @@ func (gce *GCECloud) CreateInstanceGroup(name string, zone string) (*compute.Ins
// DeleteInstanceGroup deletes an instance group.
func (gce *GCECloud) DeleteInstanceGroup(name string, zone string) error {
mc := newInstanceGroupMetricContext("delete", zone)
op, err := gce.service.InstanceGroups.Delete(
gce.projectID, zone, name).Do()
if err != nil {
mc.Observe(err)
return err
return mc.Observe(err)
}
return gce.waitForZoneOp(op, zone, mc)
@ -88,25 +81,19 @@ 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 {
mc.Observe(err)
return err
return mc.Observe(err)
}
return gce.waitForZoneOp(op, zone, mc)
@ -114,71 +101,35 @@ 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 {
if isHTTPErrorCode(err, http.StatusNotFound) {
mc.Observe(nil)
return nil
}
mc.Observe(err)
return err
return mc.Observe(err)
}
return gce.waitForZoneOp(op, zone, mc)
}
// AddPortToInstanceGroup adds a port to the given instance group.
func (gce *GCECloud) AddPortToInstanceGroup(ig *compute.InstanceGroup, port int64) (*compute.NamedPort, error) {
mc := newInstanceGroupMetricContext("add_port", ig.Zone)
for _, np := range ig.NamedPorts {
if np.Port == port {
glog.V(3).Infof("Instance group %v already has named port %+v", ig.Name, np)
return np, nil
}
}
glog.Infof("Adding port %v to instance group %v with %d ports", port, ig.Name, len(ig.NamedPorts))
namedPort := compute.NamedPort{Name: fmt.Sprintf("port%v", port), Port: port}
ig.NamedPorts = append(ig.NamedPorts, &namedPort)
// setNamedPorts is a zonal endpoint, meaning we invoke it by re-creating a URL like:
// {project}/zones/{zone}/instanceGroups/{instanceGroup}/setNamedPorts, so the "zone"
// parameter given to SetNamedPorts must not be the entire zone URL.
zoneURLParts := strings.Split(ig.Zone, "/")
zone := zoneURLParts[len(zoneURLParts)-1]
// SetNamedPortsOfInstanceGroup sets the list of named ports on a given instance group
func (gce *GCECloud) SetNamedPortsOfInstanceGroup(igName, zone string, namedPorts []*compute.NamedPort) error {
mc := newInstanceGroupMetricContext("set_namedports", zone)
op, err := gce.service.InstanceGroups.SetNamedPorts(
gce.projectID, zone, ig.Name,
&compute.InstanceGroupsSetNamedPortsRequest{
NamedPorts: ig.NamedPorts}).Do()
gce.projectID, zone, igName,
&compute.InstanceGroupsSetNamedPortsRequest{NamedPorts: namedPorts}).Do()
if err != nil {
mc.Observe(err)
return nil, err
return mc.Observe(err)
}
if err = gce.waitForZoneOp(op, zone, mc); err != nil {
return nil, err
}
return &namedPort, nil
return gce.waitForZoneOp(op, zone, mc)
}
// GetInstanceGroup returns an instance group by name.

View file

@ -25,14 +25,19 @@ import (
"cloud.google.com/go/compute/metadata"
"github.com/golang/glog"
computealpha "google.golang.org/api/compute/v0.beta"
computebeta "google.golang.org/api/compute/v0.beta"
compute "google.golang.org/api/compute/v1"
"k8s.io/api/core/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 (
defaultZone = ""
)
func newInstancesMetricContext(request, zone string) *metricContext {
@ -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[kubeletapis.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.service.BasePath, 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")
@ -270,7 +303,7 @@ func (gce *GCECloud) AliasRanges(nodeName types.NodeName) (cidrs []string, err e
return
}
var res *computealpha.Instance
var res *computebeta.Instance
res, err = gce.serviceBeta.Instances.Get(
gce.projectID, instance.Zone, instance.Name).Do()
if err != nil {
@ -443,3 +476,117 @@ func (gce *GCECloud) isCurrentInstance(instanceID string) bool {
return currentInstanceID == canonicalizeInstanceName(instanceID)
}
// ComputeHostTags grabs all tags from all instances being added to the pool.
// * The longest tag that is a prefix of the instance name is used
// * If any instance has no matching prefix tag, return error
// Invoking this method to get host tags is risky since it depends on the format
// of the host names in the cluster. Only use it as a fallback if gce.nodeTags
// is unspecified
func (gce *GCECloud) computeHostTags(hosts []*gceInstance) ([]string, error) {
// TODO: We could store the tags in gceInstance, so we could have already fetched it
hostNamesByZone := make(map[string]map[string]bool) // map of zones -> map of names -> bool (for easy lookup)
nodeInstancePrefix := gce.nodeInstancePrefix
for _, host := range hosts {
if !strings.HasPrefix(host.Name, gce.nodeInstancePrefix) {
glog.Warningf("instance '%s' does not conform to prefix '%s', ignoring filter", host, gce.nodeInstancePrefix)
nodeInstancePrefix = ""
}
z, ok := hostNamesByZone[host.Zone]
if !ok {
z = make(map[string]bool)
hostNamesByZone[host.Zone] = z
}
z[host.Name] = true
}
tags := sets.NewString()
for zone, hostNames := range hostNamesByZone {
pageToken := ""
page := 0
for ; page == 0 || (pageToken != "" && page < maxPages); page++ {
listCall := gce.service.Instances.List(gce.projectID, zone)
if nodeInstancePrefix != "" {
// Add the filter for hosts
listCall = listCall.Filter("name eq " + nodeInstancePrefix + ".*")
}
// Add the fields we want
// TODO(zmerlynn): Internal bug 29524655
// listCall = listCall.Fields("items(name,tags)")
if pageToken != "" {
listCall = listCall.PageToken(pageToken)
}
res, err := listCall.Do()
if err != nil {
return nil, err
}
pageToken = res.NextPageToken
for _, instance := range res.Items {
if !hostNames[instance.Name] {
continue
}
longest_tag := ""
for _, tag := range instance.Tags.Items {
if strings.HasPrefix(instance.Name, tag) && len(tag) > len(longest_tag) {
longest_tag = tag
}
}
if len(longest_tag) > 0 {
tags.Insert(longest_tag)
} else {
return nil, fmt.Errorf("Could not find any tag that is a prefix of instance name for instance %s", instance.Name)
}
}
}
if page >= maxPages {
glog.Errorf("computeHostTags exceeded maxPages=%d for Instances.List: truncating.", maxPages)
}
}
if len(tags) == 0 {
return nil, fmt.Errorf("No instances found")
}
return tags.List(), nil
}
// GetNodeTags will first try returning the list of tags specified in GCE cloud Configuration.
// If they weren't provided, it'll compute the host tags with the given hostnames. If the list
// of hostnames has not changed, a cached set of nodetags are returned.
func (gce *GCECloud) GetNodeTags(nodeNames []string) ([]string, error) {
// If nodeTags were specified through configuration, use them
if len(gce.nodeTags) > 0 {
return gce.nodeTags, nil
}
gce.computeNodeTagLock.Lock()
defer gce.computeNodeTagLock.Unlock()
// Early return if hosts have not changed
hosts := sets.NewString(nodeNames...)
if hosts.Equal(gce.lastKnownNodeNames) {
return gce.lastComputedNodeTags, nil
}
// Get GCE instance data by hostname
instances, err := gce.getInstancesByNames(nodeNames)
if err != nil {
return nil, err
}
// Determine list of host tags
tags, err := gce.computeHostTags(instances)
if err != nil {
return nil, err
}
// Save the list of tags
gce.lastKnownNodeNames = hosts
gce.lastComputedNodeTags = tags
return tags, nil
}

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,938 @@
/*
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"
"net/http"
"strconv"
"strings"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
utilerrors "k8s.io/apimachinery/pkg/util/errors"
"k8s.io/apimachinery/pkg/util/sets"
apiservice "k8s.io/kubernetes/pkg/api/v1/service"
"k8s.io/kubernetes/pkg/cloudprovider"
netsets "k8s.io/kubernetes/pkg/util/net/sets"
"github.com/golang/glog"
compute "google.golang.org/api/compute/v1"
)
// 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.
//
// 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) ensureExternalLoadBalancer(clusterName, clusterID 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")
}
hostNames := nodeNames(nodes)
supportsNodesHealthCheck := supportsNodesHealthCheck(nodes)
hosts, err := gce.getInstancesByNames(hostNames)
if err != nil {
return nil, err
}
loadBalancerName := cloudprovider.GetLoadBalancerName(apiService)
loadBalancerIP := apiService.Spec.LoadBalancerIP
ports := apiService.Spec.Ports
portStr := []string{}
for _, p := range apiService.Spec.Ports {
portStr = append(portStr, fmt.Sprintf("%s/%d", p.Protocol, p.Port))
}
affinityType := apiService.Spec.SessionAffinity
serviceName := types.NamespacedName{Namespace: apiService.Namespace, Name: apiService.Name}
glog.V(2).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v, %v, %v)",
loadBalancerName, gce.region, loadBalancerIP, portStr, hostNames, serviceName, apiService.Annotations)
// Check if the forwarding rule exists, and if so, what its IP is.
fwdRuleExists, fwdRuleNeedsUpdate, fwdRuleIP, err := gce.forwardingRuleNeedsUpdate(loadBalancerName, gce.region, loadBalancerIP, ports)
if err != nil {
return nil, err
}
if !fwdRuleExists {
glog.V(2).Infof("Forwarding rule %v for Service %v/%v doesn't exist",
loadBalancerName, apiService.Namespace, apiService.Name)
}
// Make sure we know which IP address will be used and have properly reserved
// it as static before moving forward with the rest of our operations.
//
// We use static IP addresses when updating a load balancer to ensure that we
// can replace the load balancer's other components without changing the
// address its service is reachable on. We do it this way rather than always
// keeping the static IP around even though this is more complicated because
// it makes it less likely that we'll run into quota issues. Only 7 static
// IP addresses are allowed per region by default.
//
// We could let an IP be allocated for us when the forwarding rule is created,
// but we need the IP to set up the firewall rule, and we want to keep the
// forwarding rule creation as the last thing that needs to be done in this
// function in order to maintain the invariant that "if the forwarding rule
// exists, the LB has been fully created".
ipAddress := ""
// Through this process we try to keep track of whether it is safe to
// release the IP that was allocated. If the user specifically asked for
// an IP, we assume they are managing it themselves. Otherwise, we will
// release the IP in case of early-terminating failure or upon successful
// creating of the LB.
// TODO(#36535): boil this logic down into a set of component functions
// and key the flag values off of errors returned.
isUserOwnedIP := false // if this is set, we never release the IP
isSafeToReleaseIP := false
defer func() {
if isUserOwnedIP {
return
}
if isSafeToReleaseIP {
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)
}
} else {
glog.Warningf("orphaning static IP %s during update of load balancer (%v(%v), %v): %v", ipAddress, loadBalancerName, serviceName, gce.region, err)
}
}()
if loadBalancerIP != "" {
// If a specific IP address has been requested, we have to respect the
// user's request and use that IP. If the forwarding rule was already using
// a different IP, it will be harmlessly abandoned because it was only an
// ephemeral IP (or it was a different static IP owned by the user, in which
// case we shouldn't delete it anyway).
if isStatic, err := gce.projectOwnsStaticIP(loadBalancerName, gce.region, loadBalancerIP); err != nil {
return nil, fmt.Errorf("failed to test if this GCE project owns the static IP %s: %v", loadBalancerIP, err)
} else if isStatic {
// The requested IP is a static IP, owned and managed by the user.
isUserOwnedIP = true
isSafeToReleaseIP = false
ipAddress = loadBalancerIP
glog.V(4).Infof("EnsureLoadBalancer(%v(%v)): using user-provided static IP %s", loadBalancerName, serviceName, ipAddress)
} else if loadBalancerIP == fwdRuleIP {
// The requested IP is not a static IP, but is currently assigned
// to this forwarding rule, so we can keep it.
isUserOwnedIP = false
isSafeToReleaseIP = true
ipAddress, _, err = gce.ensureStaticIP(loadBalancerName, serviceName.String(), gce.region, fwdRuleIP)
if err != nil {
return nil, fmt.Errorf("failed to ensure static IP %s: %v", fwdRuleIP, err)
}
glog.V(4).Infof("EnsureLoadBalancer(%v(%v)): using user-provided non-static IP %s", loadBalancerName, serviceName, ipAddress)
} else {
// The requested IP is not static and it is not assigned to the
// current forwarding rule. It might be attached to a different
// rule or it might not be part of this project at all. Either
// way, we can't use it.
return nil, fmt.Errorf("requested ip %s is neither static nor assigned to LB %s(%v): %v", loadBalancerIP, loadBalancerName, serviceName, err)
}
} else {
// The user did not request a specific IP.
isUserOwnedIP = false
// This will either allocate a new static IP if the forwarding rule didn't
// already have an IP, or it will promote the forwarding rule's current
// IP from ephemeral to static, or it will just get the IP if it is
// already static.
existed := false
ipAddress, existed, err = gce.ensureStaticIP(loadBalancerName, serviceName.String(), gce.region, fwdRuleIP)
if err != nil {
return nil, fmt.Errorf("failed to ensure static IP %s: %v", fwdRuleIP, err)
}
if existed {
// If the IP was not specifically requested by the user, but it
// already existed, it seems to be a failed update cycle. We can
// use this IP and try to run through the process again, but we
// should not release the IP unless it is explicitly flagged as OK.
isSafeToReleaseIP = false
glog.V(4).Infof("EnsureLoadBalancer(%v(%v)): adopting static IP %s", loadBalancerName, serviceName, ipAddress)
} else {
// For total clarity. The IP did not pre-exist and the user did
// not ask for a particular one, so we can release the IP in case
// of failure or success.
isSafeToReleaseIP = true
glog.V(4).Infof("EnsureLoadBalancer(%v(%v)): allocated static IP %s", loadBalancerName, serviceName, ipAddress)
}
}
// Deal with the firewall next. The reason we do this here rather than last
// is because the forwarding rule is used as the indicator that the load
// balancer is fully created - it's what getLoadBalancer checks for.
// Check if user specified the allow source range
sourceRanges, err := apiservice.GetLoadBalancerSourceRanges(apiService)
if err != nil {
return nil, err
}
firewallExists, firewallNeedsUpdate, err := gce.firewallNeedsUpdate(loadBalancerName, serviceName.String(), gce.region, ipAddress, ports, sourceRanges)
if err != nil {
return nil, err
}
if firewallNeedsUpdate {
desc := makeFirewallDescription(serviceName.String(), ipAddress)
// Unlike forwarding rules and target pools, firewalls can be updated
// without needing to be deleted and recreated.
if firewallExists {
glog.Infof("EnsureLoadBalancer(%v(%v)): updating firewall", loadBalancerName, serviceName)
if err := gce.updateFirewall(makeFirewallName(loadBalancerName), gce.region, desc, sourceRanges, ports, hosts); err != nil {
return nil, err
}
glog.Infof("EnsureLoadBalancer(%v(%v)): updated firewall", loadBalancerName, serviceName)
} else {
glog.Infof("EnsureLoadBalancer(%v(%v)): creating firewall", loadBalancerName, serviceName)
if err := gce.createFirewall(makeFirewallName(loadBalancerName), gce.region, desc, sourceRanges, ports, hosts); err != nil {
return nil, err
}
glog.Infof("EnsureLoadBalancer(%v(%v)): created firewall", loadBalancerName, serviceName)
}
}
tpExists, tpNeedsUpdate, err := gce.targetPoolNeedsUpdate(loadBalancerName, gce.region, affinityType)
if err != nil {
return nil, err
}
if !tpExists {
glog.Infof("Target pool %v for Service %v/%v doesn't exist", loadBalancerName, apiService.Namespace, apiService.Name)
}
// Check which health check needs to create and which health check needs to delete.
// Health check management is coupled with target pool operation to prevent leaking.
var hcToCreate, hcToDelete *compute.HttpHealthCheck
hcLocalTrafficExisting, err := gce.GetHttpHealthCheck(loadBalancerName)
if err != nil && !isHTTPErrorCode(err, http.StatusNotFound) {
return nil, fmt.Errorf("error checking HTTP health check %s: %v", loadBalancerName, err)
}
if path, healthCheckNodePort := apiservice.GetServiceHealthCheckPathPort(apiService); path != "" {
glog.V(4).Infof("service %v (%v) needs local traffic health checks on: %d%s)", apiService.Name, loadBalancerName, healthCheckNodePort, path)
if hcLocalTrafficExisting == nil {
// This logic exists to detect a transition for non-OnlyLocal to OnlyLocal service
// turn on the tpNeedsUpdate flag to delete/recreate fwdrule/tpool updating the
// 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())
}
tpNeedsUpdate = true
}
hcToCreate = makeHttpHealthCheck(loadBalancerName, path, healthCheckNodePort)
} else {
glog.V(4).Infof("Service %v needs nodes health checks.", apiService.Name)
if hcLocalTrafficExisting != nil {
// This logic exists to detect a transition from OnlyLocal to non-OnlyLocal service
// and turn on the tpNeedsUpdate flag to delete/recreate fwdrule/tpool updating the
// target pool to use nodes health check.
glog.V(2).Infof("Updating from local traffic health checks to nodes health checks for service %v LB %v", apiService.Name, loadBalancerName)
hcToDelete = hcLocalTrafficExisting
tpNeedsUpdate = true
}
if supportsNodesHealthCheck {
hcToCreate = makeHttpHealthCheck(makeNodesHealthCheckName(clusterID), GetNodesHealthCheckPath(), GetNodesHealthCheckPort())
}
}
// Now we get to some slightly more interesting logic.
// First, neither target pools nor forwarding rules can be updated in place -
// they have to be deleted and recreated.
// Second, forwarding rules are layered on top of target pools in that you
// can't delete a target pool that's currently in use by a forwarding rule.
// Thus, we have to tear down the forwarding rule if either it or the target
// pool needs to be updated.
if fwdRuleExists && (fwdRuleNeedsUpdate || tpNeedsUpdate) {
// Begin critical section. If we have to delete the forwarding rule,
// 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.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 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.DeleteExternalTargetPoolAndChecks(loadBalancerName, gce.region, clusterID, 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)
}
// Once we've deleted the resources (if necessary), build them back up (or for
// the first time if they're new).
if tpNeedsUpdate {
createInstances := hosts
if len(hosts) > maxTargetPoolCreateInstances {
createInstances = createInstances[:maxTargetPoolCreateInstances]
}
// Pass healthchecks to createTargetPool which needs them as health check links in the target pool
if err := gce.createTargetPool(loadBalancerName, serviceName.String(), ipAddress, gce.region, clusterID, createInstances, affinityType, hcToCreate); err != nil {
return nil, fmt.Errorf("failed to create target pool %s: %v", loadBalancerName, err)
}
if hcToCreate != nil {
glog.Infof("EnsureLoadBalancer(%v(%v)): created health checks %v for target pool", loadBalancerName, serviceName, hcToCreate.Name)
}
if len(hosts) <= maxTargetPoolCreateInstances {
glog.Infof("EnsureLoadBalancer(%v(%v)): created target pool", loadBalancerName, serviceName)
} else {
glog.Infof("EnsureLoadBalancer(%v(%v)): created initial target pool (now updating with %d hosts)", loadBalancerName, serviceName, len(hosts)-maxTargetPoolCreateInstances)
created := sets.NewString()
for _, host := range createInstances {
created.Insert(host.makeComparableHostPath())
}
if err := gce.updateTargetPool(loadBalancerName, created, hosts); err != nil {
return nil, fmt.Errorf("failed to update target pool %s: %v", loadBalancerName, err)
}
glog.Infof("EnsureLoadBalancer(%v(%v)): updated target pool (with %d hosts)", loadBalancerName, serviceName, len(hosts)-maxTargetPoolCreateInstances)
}
}
if tpNeedsUpdate || fwdRuleNeedsUpdate {
glog.Infof("EnsureLoadBalancer(%v(%v)): creating forwarding rule, IP %s", loadBalancerName, serviceName, ipAddress)
if err := gce.createForwardingRule(loadBalancerName, serviceName.String(), gce.region, ipAddress, ports); err != nil {
return nil, fmt.Errorf("failed to create forwarding rule %s: %v", loadBalancerName, err)
}
// End critical section. It is safe to release the static IP (which
// just demotes it to ephemeral) now that it is attached. In the case
// of a user-requested IP, the "is user-owned" flag will be set,
// preventing it from actually being released.
isSafeToReleaseIP = true
glog.Infof("EnsureLoadBalancer(%v(%v)): created forwarding rule, IP %s", loadBalancerName, serviceName, ipAddress)
}
status := &v1.LoadBalancerStatus{}
status.Ingress = []v1.LoadBalancerIngress{{IP: ipAddress}}
return status, nil
}
// 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
}
loadBalancerName := cloudprovider.GetLoadBalancerName(service)
pool, err := gce.service.TargetPools.Get(gce.projectID, gce.region, loadBalancerName).Do()
if err != nil {
return err
}
existing := sets.NewString()
for _, instance := range pool.Instances {
existing.Insert(hostURLToComparablePath(instance))
}
return gce.updateTargetPool(loadBalancerName, existing, hosts)
}
// ensureExternalLoadBalancerDeleted is the external implementation of LoadBalancer.EnsureLoadBalancerDeleted
func (gce *GCECloud) ensureExternalLoadBalancerDeleted(clusterName, clusterID string, service *v1.Service) error {
loadBalancerName := cloudprovider.GetLoadBalancerName(service)
var hcNames []string
if path, _ := apiservice.GetServiceHealthCheckPathPort(service); path != "" {
hcToDelete, err := gce.GetHttpHealthCheck(loadBalancerName)
if err != nil && !isHTTPErrorCode(err, http.StatusNotFound) {
glog.Infof("Failed to retrieve health check %v:%v", loadBalancerName, err)
return err
}
hcNames = append(hcNames, hcToDelete.Name)
} else {
// EnsureLoadBalancerDeleted() could be triggered by changing service from
// LoadBalancer type to others. In this case we have no idea whether it was
// using local traffic health check or nodes health check. Attempt to delete
// both to prevent leaking.
hcNames = append(hcNames, loadBalancerName)
hcNames = append(hcNames, makeNodesHealthCheckName(clusterID))
}
errs := utilerrors.AggregateGoroutines(
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 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 := ignoreNotFound(gce.DeleteRegionForwardingRule(loadBalancerName, gce.region)); err != nil {
return err
}
if err := gce.DeleteExternalTargetPoolAndChecks(loadBalancerName, gce.region, clusterID, hcNames...); err != nil {
return err
}
return nil
},
)
if errs != nil {
return utilerrors.Flatten(errs)
}
return nil
}
func (gce *GCECloud) DeleteExternalTargetPoolAndChecks(name, region, clusterID 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 err
}
// Deletion of health checks is allowed only after the TargetPool reference is deleted
for _, hcName := range hcNames {
if err := func() error {
// Check whether it is nodes health check, which has different name from the load-balancer.
isNodesHealthCheck := hcName != name
if isNodesHealthCheck {
// Lock to prevent deleting necessary nodes health check before it gets attached
// to target pool.
gce.sharedResourceLock.Lock()
defer gce.sharedResourceLock.Unlock()
}
glog.Infof("Deleting health check %v", hcName)
if err := gce.DeleteHttpHealthCheck(hcName); err != nil {
// Delete nodes health checks will fail if any other target pool is using it.
if isInUsedByError(err) {
glog.V(4).Infof("Health check %v is in used: %v.", hcName, err)
return nil
} else if !isHTTPErrorCode(err, http.StatusNotFound) {
glog.Warningf("Failed to delete health check %v: %v", hcName, err)
return err
}
// StatusNotFound could happen when:
// - This is the first attempt but we pass in a healthcheck that is already deleted
// to prevent leaking.
// - This is the first attempt but user manually deleted the heathcheck.
// - This is a retry and in previous round we failed to delete the healthcheck firewall
// after deleted the healthcheck.
// We continue to delete the healthcheck firewall to prevent leaking.
glog.V(4).Infof("Health check %v is already deleted.", hcName)
}
// If health check is deleted without error, it means no load-balancer is using it.
// So we should delete the health check firewall as well.
fwName := MakeHealthCheckFirewallName(clusterID, hcName, isNodesHealthCheck)
glog.Infof("Deleting firewall %v.", fwName)
if err := gce.DeleteFirewall(fwName); err != nil {
if isHTTPErrorCode(err, http.StatusNotFound) {
glog.V(4).Infof("Firewall %v is already deleted.", fwName)
return nil
}
return err
}
return nil
}(); err != nil {
return err
}
}
return nil
}
func (gce *GCECloud) createTargetPool(name, serviceName, ipAddress, region, clusterID string, hosts []*gceInstance, affinityType v1.ServiceAffinity, hc *compute.HttpHealthCheck) error {
// health check management is coupled with targetPools to prevent leaks. A
// target pool is the only thing that requires a health check, so we delete
// associated checks on teardown, and ensure checks on setup.
hcLinks := []string{}
if hc != nil {
// Check whether it is nodes health check, which has different name from the load-balancer.
isNodesHealthCheck := hc.Name != name
if isNodesHealthCheck {
// Lock to prevent necessary nodes health check / firewall gets deleted.
gce.sharedResourceLock.Lock()
defer gce.sharedResourceLock.Unlock()
}
if !gce.OnXPN() {
if err := gce.ensureHttpHealthCheckFirewall(serviceName, ipAddress, region, clusterID, 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 {
return fmt.Errorf("Failed to ensure health check for %v port %d path %v: %v", name, hc.Port, hc.RequestPath, err)
}
hcLinks = append(hcLinks, hc.SelfLink)
}
var instances []string
for _, host := range hosts {
instances = append(instances, makeHostURL(gce.service.BasePath, gce.projectID, host.Zone, host.Name))
}
glog.Infof("Creating targetpool %v with %d healthchecks", name, len(hcLinks))
pool := &compute.TargetPool{
Name: name,
Description: fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, serviceName),
Instances: instances,
SessionAffinity: translateAffinityType(affinityType),
HealthChecks: hcLinks,
}
if _, err := gce.CreateTargetPool(pool, region); err != nil && !isHTTPErrorCode(err, http.StatusConflict) {
return err
}
return nil
}
func (gce *GCECloud) updateTargetPool(loadBalancerName string, existing sets.String, hosts []*gceInstance) error {
var toAdd []*compute.InstanceReference
var toRemove []*compute.InstanceReference
for _, host := range hosts {
link := host.makeComparableHostPath()
if !existing.Has(link) {
toAdd = append(toAdd, &compute.InstanceReference{Instance: link})
}
existing.Delete(link)
}
for link := range existing {
toRemove = append(toRemove, &compute.InstanceReference{Instance: link})
}
if len(toAdd) > 0 {
if err := gce.AddInstancesToTargetPool(loadBalancerName, gce.region, toAdd); err != nil {
return err
}
}
if len(toRemove) > 0 {
if err := gce.RemoveInstancesFromTargetPool(loadBalancerName, gce.region, toRemove); err != nil {
return err
}
}
// 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.GetTargetPool(loadBalancerName, gce.region)
if err != nil {
return err
}
if len(updatedPool.Instances) != len(hosts) {
glog.Errorf("Unexpected number of instances (%d) in target pool %s after updating (expected %d). Instances in updated pool: %s",
len(updatedPool.Instances), loadBalancerName, len(hosts), strings.Join(updatedPool.Instances, ","))
return fmt.Errorf("Unexpected number of instances (%d) in target pool %s after update (expected %d)", len(updatedPool.Instances), loadBalancerName, len(hosts))
}
return nil
}
func (gce *GCECloud) targetPoolURL(name, region string) string {
return gce.service.BasePath + strings.Join([]string{gce.projectID, "regions", region, "targetPools", name}, "/")
}
func makeHttpHealthCheck(name, path string, port int32) *compute.HttpHealthCheck {
return &compute.HttpHealthCheck{
Name: name,
Port: int64(port),
RequestPath: path,
Host: "",
Description: makeHealthCheckDescription(name),
CheckIntervalSec: gceHcCheckIntervalSeconds,
TimeoutSec: gceHcTimeoutSeconds,
HealthyThreshold: gceHcHealthyThreshold,
UnhealthyThreshold: gceHcUnhealthyThreshold,
}
}
func (gce *GCECloud) ensureHttpHealthCheck(name, path string, port int32) (hc *compute.HttpHealthCheck, err error) {
newHC := makeHttpHealthCheck(name, path, port)
hc, err = gce.GetHttpHealthCheck(name)
if hc == nil || err != nil && isHTTPErrorCode(err, http.StatusNotFound) {
glog.Infof("Did not find health check %v, creating port %v path %v", name, port, path)
if err = gce.CreateHttpHealthCheck(newHC); err != nil {
return nil, err
}
hc, err = gce.GetHttpHealthCheck(name)
if err != nil {
glog.Errorf("Failed to get http health check %v", err)
return nil, err
}
glog.Infof("Created HTTP health check %v healthCheckNodePort: %d", name, port)
return hc, nil
}
// Validate health check fields
glog.V(4).Infof("Checking http health check params %s", name)
drift := hc.Port != int64(port) || hc.RequestPath != path || hc.Description != makeHealthCheckDescription(name)
drift = drift || hc.CheckIntervalSec != gceHcCheckIntervalSeconds || hc.TimeoutSec != gceHcTimeoutSeconds
drift = drift || hc.UnhealthyThreshold != gceHcUnhealthyThreshold || hc.HealthyThreshold != gceHcHealthyThreshold
if drift {
glog.Warningf("Health check %v exists but parameters have drifted - updating...", name)
if err := gce.UpdateHttpHealthCheck(newHC); err != nil {
glog.Warningf("Failed to reconcile http health check %v parameters", name)
return nil, err
}
glog.V(4).Infof("Corrected health check %v parameters successful", name)
}
return hc, nil
}
// Passing nil for requested IP is perfectly fine - it just means that no specific
// IP is being requested.
// Returns whether the forwarding rule exists, whether it needs to be updated,
// what its IP address is (if it exists), and any error we encountered.
func (gce *GCECloud) forwardingRuleNeedsUpdate(name, region string, loadBalancerIP string, ports []v1.ServicePort) (exists bool, needsUpdate bool, ipAddress string, err error) {
fwd, err := gce.service.ForwardingRules.Get(gce.projectID, region, name).Do()
if err != nil {
if isHTTPErrorCode(err, http.StatusNotFound) {
return false, true, "", nil
}
// Err on the side of caution in case of errors. Caller should notice the error and retry.
// We never want to end up recreating resources because gce api flaked.
return true, false, "", fmt.Errorf("error getting load balancer's forwarding rule: %v", err)
}
// If the user asks for a specific static ip through the Service spec,
// check that we're actually using it.
// TODO: we report loadbalancer IP through status, so we want to verify if
// that matches the forwarding rule as well.
if loadBalancerIP != "" && loadBalancerIP != fwd.IPAddress {
glog.Infof("LoadBalancer ip for forwarding rule %v was expected to be %v, but was actually %v", fwd.Name, fwd.IPAddress, loadBalancerIP)
return true, true, fwd.IPAddress, nil
}
portRange, err := loadBalancerPortRange(ports)
if err != nil {
// Err on the side of caution in case of errors. Caller should notice the error and retry.
// We never want to end up recreating resources because gce api flaked.
return true, false, "", err
}
if portRange != fwd.PortRange {
glog.Infof("LoadBalancer port range for forwarding rule %v was expected to be %v, but was actually %v", fwd.Name, fwd.PortRange, portRange)
return true, true, fwd.IPAddress, nil
}
// The service controller verified all the protocols match on the ports, just check the first one
if string(ports[0].Protocol) != fwd.IPProtocol {
glog.Infof("LoadBalancer protocol for forwarding rule %v was expected to be %v, but was actually %v", fwd.Name, fwd.IPProtocol, string(ports[0].Protocol))
return true, true, fwd.IPAddress, nil
}
return true, false, fwd.IPAddress, nil
}
// Doesn't check whether the hosts have changed, since host updating is handled
// separately.
func (gce *GCECloud) targetPoolNeedsUpdate(name, region string, affinityType v1.ServiceAffinity) (exists bool, needsUpdate bool, err error) {
tp, err := gce.service.TargetPools.Get(gce.projectID, region, name).Do()
if err != nil {
if isHTTPErrorCode(err, http.StatusNotFound) {
return false, true, nil
}
// Err on the side of caution in case of errors. Caller should notice the error and retry.
// We never want to end up recreating resources because gce api flaked.
return true, false, fmt.Errorf("error getting load balancer's target pool: %v", err)
}
// TODO: If the user modifies their Service's session affinity, it *should*
// reflect in the associated target pool. However, currently not setting the
// session affinity on a target pool defaults it to the empty string while
// not setting in on a Service defaults it to None. There is a lack of
// documentation around the default setting for the target pool, so if we
// find it's the undocumented empty string, don't blindly recreate the
// target pool (which results in downtime). Fix this when we have formally
// defined the defaults on either side.
if tp.SessionAffinity != "" && translateAffinityType(affinityType) != tp.SessionAffinity {
glog.Infof("LoadBalancer target pool %v changed affinity from %v to %v", name, tp.SessionAffinity, affinityType)
return true, true, nil
}
return true, false, nil
}
func (h *gceInstance) makeComparableHostPath() string {
return fmt.Sprintf("/zones/%s/instances/%s", h.Zone, h.Name)
}
func nodeNames(nodes []*v1.Node) []string {
ret := make([]string, len(nodes))
for i, node := range nodes {
ret[i] = node.Name
}
return ret
}
func makeHostURL(projectsApiEndpoint, projectID, zone, host string) string {
host = canonicalizeInstanceName(host)
return projectsApiEndpoint + strings.Join([]string{projectID, "zones", zone, "instances", host}, "/")
}
func hostURLToComparablePath(hostURL string) string {
idx := strings.Index(hostURL, "/zones/")
if idx < 0 {
return ""
}
return hostURL[idx:]
}
func loadBalancerPortRange(ports []v1.ServicePort) (string, error) {
if len(ports) == 0 {
return "", fmt.Errorf("no ports specified for GCE load balancer")
}
// The service controller verified all the protocols match on the ports, just check and use the first one
if ports[0].Protocol != v1.ProtocolTCP && ports[0].Protocol != v1.ProtocolUDP {
return "", fmt.Errorf("Invalid protocol %s, only TCP and UDP are supported", string(ports[0].Protocol))
}
minPort := int32(65536)
maxPort := int32(0)
for i := range ports {
if ports[i].Port < minPort {
minPort = ports[i].Port
}
if ports[i].Port > maxPort {
maxPort = ports[i].Port
}
}
return fmt.Sprintf("%d-%d", minPort, maxPort), nil
}
// translate from what K8s supports to what the cloud provider supports for session affinity.
func translateAffinityType(affinityType v1.ServiceAffinity) string {
switch affinityType {
case v1.ServiceAffinityClientIP:
return gceAffinityTypeClientIP
case v1.ServiceAffinityNone:
return gceAffinityTypeNone
default:
glog.Errorf("Unexpected affinity type: %v", affinityType)
return gceAffinityTypeNone
}
}
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) {
return false, true, nil
}
return false, false, fmt.Errorf("error getting load balancer's firewall: %v", err)
}
if fw.Description != makeFirewallDescription(serviceName, ipAddress) {
return true, true, nil
}
if len(fw.Allowed) != 1 || (fw.Allowed[0].IPProtocol != "tcp" && fw.Allowed[0].IPProtocol != "udp") {
return true, true, nil
}
// Make sure the allowed ports match.
allowedPorts := make([]string, len(ports))
for ix := range ports {
allowedPorts[ix] = strconv.Itoa(int(ports[ix].Port))
}
if !equalStringSets(allowedPorts, fw.Allowed[0].Ports) {
return true, true, nil
}
// The service controller already verified that the protocol matches on all ports, no need to check.
actualSourceRanges, err := netsets.ParseIPNets(fw.SourceRanges...)
if err != nil {
// This really shouldn't happen... GCE has returned something unexpected
glog.Warningf("Error parsing firewall SourceRanges: %v", fw.SourceRanges)
// We don't return the error, because we can hopefully recover from this by reconfiguring the firewall
return true, true, nil
}
if !sourceRanges.Equal(actualSourceRanges) {
return true, true, nil
}
return true, false, nil
}
func (gce *GCECloud) ensureHttpHealthCheckFirewall(serviceName, ipAddress, region, clusterID string, hosts []*gceInstance, hcName string, hcPort int32, isNodesHealthCheck bool) error {
// Prepare the firewall params for creating / checking.
desc := fmt.Sprintf(`{"kubernetes.io/cluster-id":"%s"}`, clusterID)
if !isNodesHealthCheck {
desc = makeFirewallDescription(serviceName, ipAddress)
}
sourceRanges := lbSrcRngsFlag.ipn
ports := []v1.ServicePort{{Protocol: "tcp", Port: hcPort}}
fwName := MakeHealthCheckFirewallName(clusterID, hcName, isNodesHealthCheck)
fw, err := gce.service.Firewalls.Get(gce.projectID, fwName).Do()
if err != nil {
if !isHTTPErrorCode(err, http.StatusNotFound) {
return fmt.Errorf("error getting firewall for health checks: %v", err)
}
glog.Infof("Creating firewall %v for health checks.", fwName)
if err := gce.createFirewall(fwName, region, desc, sourceRanges, ports, hosts); err != nil {
return err
}
glog.Infof("Created firewall %v for health checks.", fwName)
return nil
}
// Validate firewall fields.
if fw.Description != desc ||
len(fw.Allowed) != 1 ||
fw.Allowed[0].IPProtocol != string(ports[0].Protocol) ||
!equalStringSets(fw.Allowed[0].Ports, []string{string(ports[0].Port)}) ||
!equalStringSets(fw.SourceRanges, sourceRanges.StringSlice()) {
glog.Warningf("Firewall %v exists but parameters have drifted - updating...", fwName)
if err := gce.updateFirewall(fwName, region, desc, sourceRanges, ports, hosts); err != nil {
glog.Warningf("Failed to reconcile firewall %v parameters.", fwName)
return err
}
glog.V(4).Infof("Corrected firewall %v parameters successful", fwName)
}
return nil
}
func (gce *GCECloud) createForwardingRule(name, serviceName, region, ipAddress string, ports []v1.ServicePort) error {
portRange, err := loadBalancerPortRange(ports)
if err != nil {
return err
}
req := &compute.ForwardingRule{
Name: name,
Description: fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, serviceName),
IPAddress: ipAddress,
IPProtocol: string(ports[0].Protocol),
PortRange: portRange,
Target: gce.targetPoolURL(name, region),
}
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 {
firewall, err := gce.firewallObject(name, region, desc, sourceRanges, ports, hosts)
if err != nil {
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 {
firewall, err := gce.firewallObject(name, region, desc, sourceRanges, ports, hosts)
if err != nil {
return err
}
if err = gce.UpdateFirewall(firewall); err != nil && !isHTTPErrorCode(err, http.StatusConflict) {
return err
}
return nil
}
func (gce *GCECloud) firewallObject(name, region, desc string, sourceRanges netsets.IPNet, ports []v1.ServicePort, hosts []*gceInstance) (*compute.Firewall, error) {
allowedPorts := make([]string, len(ports))
for ix := range ports {
allowedPorts[ix] = strconv.Itoa(int(ports[ix].Port))
}
// If the node tags to be used for this cluster have been predefined in the
// provider config, just use them. Otherwise, invoke computeHostTags method to get the tags.
hostTags := gce.nodeTags
if len(hostTags) == 0 {
var err error
if hostTags, err = gce.computeHostTags(hosts); err != nil {
return nil, fmt.Errorf("No node tags supplied and also failed to parse the given lists of hosts for tags. Abort creating firewall rule.")
}
}
firewall := &compute.Firewall{
Name: name,
Description: desc,
Network: gce.networkURL,
SourceRanges: sourceRanges.StringSlice(),
TargetTags: hostTags,
Allowed: []*compute.FirewallAllowed{
{
// TODO: Make this more generic. Currently this method is only
// used to create firewall rules for loadbalancers, which have
// exactly one protocol, so we can never end up with a list of
// mixed TCP and UDP ports. It should be possible to use a
// single firewall rule for both a TCP and UDP lb.
IPProtocol: strings.ToLower(string(ports[0].Protocol)),
Ports: allowedPorts,
},
},
}
return firewall, nil
}
func (gce *GCECloud) projectOwnsStaticIP(name, region string, ipAddress string) (bool, error) {
pageToken := ""
page := 0
for ; page == 0 || (pageToken != "" && page < maxPages); page++ {
listCall := gce.service.Addresses.List(gce.projectID, region)
if pageToken != "" {
listCall = listCall.PageToken(pageToken)
}
addresses, err := listCall.Do()
if err != nil {
return false, fmt.Errorf("failed to list gce IP addresses: %v", err)
}
pageToken = addresses.NextPageToken
for _, addr := range addresses.Items {
if addr.Address == ipAddress {
// This project does own the address, so return success.
return true, nil
}
}
}
if page >= maxPages {
glog.Errorf("projectOwnsStaticIP exceeded maxPages=%d for Addresses.List; truncating.", maxPages)
}
return false, nil
}
func (gce *GCECloud) ensureStaticIP(name, serviceName, region, existingIP string) (ipAddress string, created bool, err error) {
// If the address doesn't exist, this will create it.
// If the existingIP exists but is ephemeral, this will promote it to static.
// If the address already exists, this will harmlessly return a StatusConflict
// and we'll grab the IP before returning.
existed := false
addressObj := &compute.Address{
Name: name,
Description: fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, serviceName),
}
if existingIP != "" {
addressObj.Address = existingIP
}
if err = gce.ReserveRegionAddress(addressObj, region); err != nil {
if !isHTTPErrorCode(err, http.StatusConflict) {
return "", false, fmt.Errorf("error creating gce static IP address: %v", err)
}
// StatusConflict == the IP exists already.
existed = true
}
addr, err := gce.GetRegionAddress(name, region)
if err != nil {
return "", false, fmt.Errorf("error getting static IP address: %v", err)
}
return addr.Address, existed, nil
}

View file

@ -0,0 +1,637 @@
/*
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/api/core/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
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)
sharedBackend := shareBackendService(svc)
backendServiceName := makeBackendServiceName(loadBalancerName, clusterID, sharedBackend, 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
}
// Get existing backend service (if exists)
var existingBackendService *compute.BackendService
if existingFwdRule != nil && existingFwdRule.BackendService != "" {
existingBSName := getNameFromLink(existingFwdRule.BackendService)
if existingBackendService, err = gce.GetRegionBackendService(existingBSName, gce.region); err != nil && !isNotFound(err) {
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 exists before creating the backend service. The health check is shared
// if externalTrafficPolicy=Cluster.
sharedHealthCheck := !v1_service.RequestsOnlyLocalTraffic(svc)
hcName := makeHealthCheckName(loadBalancerName, clusterID, sharedHealthCheck)
hcPath, hcPort := GetNodesHealthCheckPath(), GetNodesHealthCheckPort()
if !sharedHealthCheck {
// Service requires a special health check, retrieve the OnlyLocal port & path
hcPath, hcPort = v1_service.GetServiceHealthCheckPathPort(svc)
}
hc, err := gce.ensureInternalHealthCheck(hcName, nm, sharedHealthCheck, 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)), sharedHealthCheck, 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
}
fwdRuleDeleted := false
if existingFwdRule != nil && !fwdRuleEqual(existingFwdRule, expectedFwdRule) {
glog.V(2).Infof("ensureInternalLoadBalancer(%v): deleting existing forwarding rule with IP address %v", loadBalancerName, existingFwdRule.IPAddress)
if err = gce.DeleteRegionForwardingRule(loadBalancerName, gce.region); err != nil && !isNotFound(err) {
return nil, err
}
fwdRuleDeleted = true
}
bsDescription := makeBackendServiceDescription(nm, sharedBackend)
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): creating forwarding rule", loadBalancerName)
if err = gce.CreateRegionForwardingRule(expectedFwdRule, gce.region); err != nil {
return nil, err
}
}
// Delete the previous internal load balancer resources if necessary
if existingBackendService != nil {
gce.clearPreviousInternalResources(loadBalancerName, existingBackendService, backendServiceName, hcName)
}
// 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) clearPreviousInternalResources(loadBalancerName string, existingBackendService *compute.BackendService, expectedBSName, expectedHCName string) {
// If a new backend service was created, delete the old one.
if existingBackendService.Name != expectedBSName {
glog.V(2).Infof("clearPreviousInternalResources(%v): expected backend service %q does not match previous %q - deleting backend service", loadBalancerName, expectedBSName, existingBackendService.Name)
if err := gce.teardownInternalBackendService(existingBackendService.Name); err != nil && !isNotFound(err) {
glog.Warningf("clearPreviousInternalResources: could not delete old backend service: %v, err: %v", existingBackendService.Name, err)
}
}
// If a new health check was created, delete the old one.
if len(existingBackendService.HealthChecks) == 1 {
existingHCName := getNameFromLink(existingBackendService.HealthChecks[0])
if existingHCName != expectedHCName {
glog.V(2).Infof("clearPreviousInternalResources(%v): expected health check %q does not match previous %q - deleting health check", loadBalancerName, expectedHCName, existingHCName)
if err := gce.teardownInternalHealthCheckAndFirewall(existingHCName); err != nil {
glog.Warningf("clearPreviousInternalResources: could not delete existing healthcheck: %v, err: %v", existingHCName, err)
}
}
} else if len(existingBackendService.HealthChecks) > 1 {
glog.Warningf("clearPreviousInternalResources(%v): more than one health check on the backend service %v, %v", loadBalancerName, existingBackendService.Name, existingBackendService.HealthChecks)
}
}
// updateInternalLoadBalancer is called when the list of nodes has changed. Therefore, only the instance groups
// and possibly the backend service need to be updated.
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)
backendServiceName := makeBackendServiceName(loadBalancerName, clusterID, shareBackendService(svc), 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
sharedBackend := shareBackendService(svc)
sharedHealthCheck := !v1_service.RequestsOnlyLocalTraffic(svc)
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, sharedBackend, scheme, protocol, svc.Spec.SessionAffinity)
glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting region backend service %v", loadBalancerName, backendServiceName)
if err := gce.teardownInternalBackendService(backendServiceName); err != nil {
return err
}
glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting firewall for traffic", loadBalancerName)
if err := gce.DeleteFirewall(loadBalancerName); err != nil {
return err
}
hcName := makeHealthCheckName(loadBalancerName, clusterID, sharedHealthCheck)
glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting health check %v and its firewall", loadBalancerName, hcName)
if err := gce.teardownInternalHealthCheckAndFirewall(hcName); err != nil {
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) teardownInternalBackendService(bsName string) error {
if err := gce.DeleteRegionBackendService(bsName, gce.region); err != nil {
if isNotFound(err) {
glog.V(2).Infof("teardownInternalBackendService(%v): backend service already deleted. err: %v", bsName, err)
return nil
} else if isInUsedByError(err) {
glog.V(2).Infof("teardownInternalBackendService(%v): backend service in use.", bsName)
return nil
} else {
return fmt.Errorf("failed to delete backend service: %v, err: %v", bsName, err)
}
}
glog.V(2).Infof("teardownInternalBackendService(%v): backend service deleted", bsName)
return nil
}
func (gce *GCECloud) teardownInternalHealthCheckAndFirewall(hcName string) error {
if err := gce.DeleteHealthCheck(hcName); err != nil {
if isNotFound(err) {
glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check does not exist.", hcName)
// Purposely do not early return - double check the firewall does not exist
} else if isInUsedByError(err) {
glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check in use.", hcName)
return nil
} else {
return fmt.Errorf("failed to delete health check: %v, err: %v", hcName, err)
}
}
glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check deleted", hcName)
hcFirewallName := makeHealthCheckFirewallNameFromHC(hcName)
if err := gce.DeleteFirewall(hcFirewallName); err != nil && !isNotFound(err) {
return fmt.Errorf("failed to delete health check firewall: %v, err: %v", hcFirewallName, err)
}
glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check firewall deleted", hcFirewallName)
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, sharedHealthCheck 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 := makeHealthCheckFirewallName(loadBalancerName, clusterID, sharedHealthCheck)
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): attempting delete instance group in all %d zones", name, len(zones))
for _, z := range zones {
if err := gce.DeleteInstanceGroup(name, z.Name); err != nil && !isNotFoundOrInUse(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)
// Set fingerprint for optimistic locking
expectedBS.Fingerprint = bs.Fingerprint
if err := gce.UpdateRegionBackendService(expectedBS, gce.region); err != nil {
return err
}
glog.V(2).Infof("ensureInternalBackendService: updated backend service %v successfully", name)
return nil
}
// ensureInternalBackendServiceGroups updates backend services if their list of backend instance groups is incorrect.
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)
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 shareBackendService(svc *v1.Service) bool {
return GetLoadBalancerAnnotationBackendShare(svc) && !v1_service.RequestsOnlyLocalTraffic(svc)
}
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 gce.service.BasePath + strings.Join([]string{gce.projectID, "regions", gce.region, "backendServices", name}, "/")
}
func getNameFromLink(link string) string {
if link == "" {
return ""
}
fields := strings.Split(link, "/")
return fields[len(fields)-1]
}

View file

@ -0,0 +1,113 @@
/*
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 (
"crypto/sha1"
"encoding/hex"
"fmt"
"strings"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
)
// 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 {
hash := sha1.New()
// For every non-nil option, hash its value. Currently, only service affinity is relevant.
hash.Write([]byte(string(svcAffinity)))
hashed := hex.EncodeToString(hash.Sum(nil))
hashed = hashed[:16]
// k8s- 4
// {clusterid}- 17
// {scheme}- 9 (internal/external)
// {protocol}- 4 (tcp/udp)
// nmv1- 5 (naming convention version)
// {suffix} 16 (hash of settings)
// -----------------
// 55 characters used
return fmt.Sprintf("k8s-%s-%s-%s-nmv1-%s", clusterID, strings.ToLower(string(scheme)), strings.ToLower(string(protocol)), hashed)
}
return loadBalancerName
}
func makeHealthCheckName(loadBalancerName, clusterID string, shared bool) string {
if shared {
return fmt.Sprintf("k8s-%s-node", clusterID)
}
return loadBalancerName
}
func makeHealthCheckFirewallNameFromHC(healthCheckName string) string {
return healthCheckName + "-hc"
}
func makeHealthCheckFirewallName(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)
}

View file

@ -102,7 +102,7 @@ func (gce *GCECloud) CreateRoute(clusterName string, nameHint string, route *clo
}).Do()
if err != nil {
if isHTTPErrorCode(err, http.StatusConflict) {
glog.Info("Route %v already exists.")
glog.Infof("Route %v already exists.", routeName)
return nil
} else {
return mc.Observe(err)

View file

@ -1,66 +0,0 @@
/*
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 newStaticIPMetricContext(request string) *metricContext {
return &metricContext{
start: time.Now(),
attributes: []string{"staticip_" + request, unusedMetricLabel, unusedMetricLabel},
}
}
// ReserveGlobalStaticIP creates a global static IP.
// 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()
if err != nil {
return nil, mc.Observe(err)
}
if err := gce.waitForGlobalOp(op, mc); err != nil {
return nil, err
}
return gce.service.GlobalAddresses.Get(gce.projectID, name).Do()
}
// DeleteGlobalStaticIP deletes a global static IP by name.
func (gce *GCECloud) DeleteGlobalStaticIP(name string) error {
mc := newStaticIPMetricContext("delete")
op, err := gce.service.GlobalAddresses.Delete(gce.projectID, name).Do()
if err != nil {
return mc.Observe(err)
}
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")
v, err := gce.service.GlobalAddresses.Get(gce.projectID, name).Do()
return v, mc.Observe(err)
}

View file

@ -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)
}

View file

@ -19,10 +19,12 @@ package gce
import (
"errors"
"fmt"
"net/http"
"regexp"
"strings"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets"
"cloud.google.com/go/compute/metadata"
compute "google.golang.org/api/compute/v1"
@ -105,6 +107,14 @@ func isHTTPErrorCode(err error, code int) bool {
return ok && apiErr.Code == code
}
func isInUsedByError(err error) bool {
apiErr, ok := err.(*googleapi.Error)
if !ok || apiErr.Code != http.StatusBadRequest {
return false
}
return strings.Contains(apiErr.Message, "being used by")
}
// splitProviderID splits a provider's id into core components.
// A providerID is build out of '${ProviderName}://${project-id}/${zone}/${instance-name}'
// See cloudprovider.GetInstanceProviderID.
@ -115,3 +125,27 @@ func splitProviderID(providerID string) (project, zone, instance string, err err
}
return matches[1], matches[2], matches[3], nil
}
func equalStringSets(x, y []string) bool {
if len(x) != len(y) {
return false
}
xString := sets.NewString(x...)
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)
}

View file

@ -16,11 +16,42 @@ 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"
"strings"
)
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 gce.service.BasePath + strings.Join([]string{gce.projectID, "regions", region}, "/")
}