Merge pull request #8237 from srikiz/DO-AddLoadBalancer

[DigitalOcean] Add load balancer support for master HA

pkg/commands/BUILD.bazel

@@ -18,6 +18,7 @@ go_library(
         "//pkg/assets:go_default_library",
         "//pkg/client/simple:go_default_library",
         "//pkg/featureflag:go_default_library",
+        "//pkg/resources/digitalocean:go_default_library",
         "//upup/pkg/fi/cloudup:go_default_library",
         "//upup/pkg/fi/cloudup/aliup:go_default_library",
         "//upup/pkg/fi/cloudup/awstasks:go_default_library",

pkg/commands/status_discovery.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"k8s.io/kops/pkg/apis/kops"
+	"k8s.io/kops/pkg/resources/digitalocean"
 	"k8s.io/kops/upup/pkg/fi/cloudup"
 	"k8s.io/kops/upup/pkg/fi/cloudup/aliup"
 	"k8s.io/kops/upup/pkg/fi/cloudup/awstasks"
@@ -77,6 +78,10 @@ func (s *CloudDiscoveryStatusStore) GetApiIngressStatus(cluster *kops.Cluster) (
 		return osCloud.GetApiIngressStatus(cluster)
 	}
 
+	if doCloud, ok := cloud.(*digitalocean.Cloud); ok {
+		return doCloud.GetApiIngressStatus(cluster)
+	}
+
 	return nil, fmt.Errorf("API Ingress Status not implemented for %T", cloud)
 }
 

pkg/model/domodel/BUILD.bazel

@@ -3,15 +3,19 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 go_library(
     name = "go_default_library",
     srcs = [
+        "api_loadbalancer.go",
         "context.go",
         "droplets.go",
     ],
     importpath = "k8s.io/kops/pkg/model/domodel",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/apis/kops:go_default_library",
+        "//pkg/dns:go_default_library",
         "//pkg/model:go_default_library",
         "//upup/pkg/fi:go_default_library",
         "//upup/pkg/fi/cloudup/do:go_default_library",
         "//upup/pkg/fi/cloudup/dotasks:go_default_library",
+        "//upup/pkg/fi/fitasks:go_default_library",
     ],
 )

pkg/model/domodel/api_loadbalancer.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2019 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 domodel
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"k8s.io/kops/pkg/apis/kops"
+	"k8s.io/kops/pkg/dns"
+	"k8s.io/kops/upup/pkg/fi"
+	"k8s.io/kops/upup/pkg/fi/cloudup/do"
+	"k8s.io/kops/upup/pkg/fi/cloudup/dotasks"
+	"k8s.io/kops/upup/pkg/fi/fitasks"
+)
+
+// APILoadBalancerModelBuilder builds a LoadBalancer for accessing the API
+type APILoadBalancerModelBuilder struct {
+	*DOModelContext
+	Lifecycle *fi.Lifecycle
+}
+
+var _ fi.ModelBuilder = &APILoadBalancerModelBuilder{}
+
+func (b *APILoadBalancerModelBuilder) Build(c *fi.ModelBuilderContext) error {
+	// Configuration where a load balancer fronts the API
+	if !b.UseLoadBalancerForAPI() {
+		return nil
+	}
+
+	lbSpec := b.Cluster.Spec.API.LoadBalancer
+	if lbSpec == nil {
+		// Skipping API LB creation; not requested in Spec
+		return nil
+	}
+
+	switch lbSpec.Type {
+	case kops.LoadBalancerTypeInternal:
+		// OK
+	case kops.LoadBalancerTypePublic:
+		// OK
+	default:
+		return fmt.Errorf("unhandled LoadBalancer type %q", lbSpec.Type)
+	}
+
+	clusterName := strings.Replace(b.ClusterName(), ".", "-", -1)
+	loadbalancerName := "api-" + clusterName
+	clusterMasterTag := do.TagKubernetesClusterMasterPrefix + ":" + clusterName
+
+	// Create LoadBalancer for API LB
+	loadbalancer := &dotasks.LoadBalancer{
+		Name:       fi.String(loadbalancerName),
+		Region:     fi.String(b.Cluster.Spec.Subnets[0].Region),
+		DropletTag: fi.String(clusterMasterTag),
+		Lifecycle:  b.Lifecycle,
+	}
+	c.AddTask(loadbalancer)
+
+	// Temporarily do not know the role of the following function
+	if dns.IsGossipHostname(b.Cluster.Name) || b.UsePrivateDNS() {
+		// Ensure the LB hostname is included in the TLS certificate,
+		// if we're not going to use an alias for it
+		// TODO: I don't love this technique for finding the task by name & modifying it
+		masterKeypairTask, found := c.Tasks["Keypair/master"]
+		if !found {
+			return errors.New("keypair/master task not found")
+		}
+		masterKeypair := masterKeypairTask.(*fitasks.Keypair)
+		masterKeypair.AlternateNameTasks = append(masterKeypair.AlternateNameTasks, loadbalancer)
+	}
+
+	return nil
+
+}

pkg/resources/digitalocean/cloud.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/digitalocean/godo"
 	"golang.org/x/oauth2"
@@ -128,7 +129,43 @@ func (c *Cloud) Droplets() godo.DropletsService {
 	return c.Client.Droplets
 }
 
+func (c *Cloud) LoadBalancers() godo.LoadBalancersService {
+	return c.Client.LoadBalancers
+}
+
 // FindVPCInfo is not implemented, it's only here to satisfy the fi.Cloud interface
 func (c *Cloud) FindVPCInfo(id string) (*fi.VPCInfo, error) {
 	return nil, errors.New("not implemented")
 }
+
+func (c *Cloud) GetApiIngressStatus(cluster *kops.Cluster) ([]kops.ApiIngressStatus, error) {
+	var ingresses []kops.ApiIngressStatus
+	if cluster.Spec.MasterPublicName != "" {
+		// Note that this must match Digital Ocean's lb name
+		klog.V(2).Infof("Querying DO to find Loadbalancers for API (%q)", cluster.Name)
+
+		loadBalancers, err := getAllLoadBalancers(c)
+		if err != nil {
+			return nil, fmt.Errorf("LoadBalancers.List returned error: %v", err)
+		}
+
+		lbName := "api-" + strings.Replace(cluster.Name, ".", "-", -1)
+
+		for _, lb := range loadBalancers {
+			if lb.Name == lbName {
+				klog.V(10).Infof("Matching LB name found for API (%q)", cluster.Name)
+
+				if lb.Status != "active" {
+					return nil, fmt.Errorf("load-balancer is not yet active (current status: %s)", lb.Status)
+				}
+
+				address := lb.IP
+				ingresses = append(ingresses, kops.ApiIngressStatus{IP: address})
+
+				return ingresses, nil
+			}
+		}
+	}
+
+	return nil, nil
+}

pkg/resources/digitalocean/resources.go

@@ -33,9 +33,10 @@ import (
 )
 
 const (
-	resourceTypeDroplet   = "droplet"
-	resourceTypeVolume    = "volume"
-	resourceTypeDNSRecord = "dns-record"
+	resourceTypeDroplet      = "droplet"
+	resourceTypeVolume       = "volume"
+	resourceTypeDNSRecord    = "dns-record"
+	resourceTypeLoadBalancer = "loadbalancer"
 )
 
 type listFn func(fi.Cloud, string) ([]*resources.Resource, error)
@@ -47,6 +48,7 @@ func ListResources(cloud *Cloud, clusterName string) (map[string]*resources.Reso
 		listVolumes,
 		listDroplets,
 		listDNS,
+		listLoadBalancers,
 	}
 
 	for _, fn := range listFunctions {
@@ -265,6 +267,67 @@ func getAllRecordsByDomain(cloud *Cloud, domain string) ([]godo.DomainRecord, er
 	return allRecords, nil
 }
 
+func listLoadBalancers(cloud fi.Cloud, clusterName string) ([]*resources.Resource, error) {
+	c := cloud.(*Cloud)
+	var resourceTrackers []*resources.Resource
+
+	clusterTag := "KubernetesCluster-Master:" + strings.Replace(clusterName, ".", "-", -1)
+
+	lbs, err := getAllLoadBalancers(c)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list lbs: %v", err)
+	}
+
+	for _, lb := range lbs {
+		if strings.Contains(lb.Tag, clusterTag) {
+			resourceTracker := &resources.Resource{
+				Name:    lb.Name,
+				ID:      lb.ID,
+				Type:    resourceTypeLoadBalancer,
+				Deleter: deleteLoadBalancer,
+				Obj:     lb,
+			}
+
+			var blocks []string
+			for _, dropletID := range lb.DropletIDs {
+				blocks = append(blocks, "droplet:"+strconv.Itoa(dropletID))
+			}
+
+			resourceTracker.Blocks = blocks
+			resourceTrackers = append(resourceTrackers, resourceTracker)
+		}
+	}
+
+	return resourceTrackers, nil
+}
+
+func getAllLoadBalancers(cloud *Cloud) ([]godo.LoadBalancer, error) {
+	allLoadBalancers := []godo.LoadBalancer{}
+
+	opt := &godo.ListOptions{}
+	for {
+		lbs, resp, err := cloud.LoadBalancers().List(context.TODO(), opt)
+		if err != nil {
+			return nil, err
+		}
+
+		allLoadBalancers = append(allLoadBalancers, lbs...)
+
+		if resp.Links == nil || resp.Links.IsLastPage() {
+			break
+		}
+
+		page, err := resp.Links.CurrentPage()
+		if err != nil {
+			return nil, err
+		}
+
+		opt.Page = page + 1
+	}
+
+	return allLoadBalancers, nil
+}
+
 func deleteDroplet(cloud fi.Cloud, t *resources.Resource) error {
 	c := cloud.(*Cloud)
 
@@ -315,6 +378,18 @@ func deleteRecord(cloud fi.Cloud, domain string, t *resources.Resource) error {
 	return nil
 }
 
+func deleteLoadBalancer(cloud fi.Cloud, t *resources.Resource) error {
+	c := cloud.(*Cloud)
+	lb := t.Obj.(godo.LoadBalancer)
+	_, err := c.Client.LoadBalancers.Delete(context.TODO(), lb.ID)
+
+	if err != nil {
+		return fmt.Errorf("failed to delete load balancer with name %s %v", lb.Name, err)
+	}
+
+	return nil
+}
+
 func waitForDetach(cloud *Cloud, action *godo.Action) error {
 	timeout := time.After(10 * time.Second)
 	ticker := time.NewTicker(500 * time.Millisecond)

upup/pkg/fi/cloudup/apply_cluster.go

@@ -396,8 +396,9 @@ func (c *ApplyClusterCmd) Run() error {
 			modelContext.SSHPublicKeys = sshPublicKeys
 
 			l.AddTypes(map[string]interface{}{
-				"volume":  &dotasks.Volume{},
-				"droplet": &dotasks.Droplet{},
+				"volume":       &dotasks.Volume{},
+				"droplet":      &dotasks.Droplet{},
+				"loadbalancer": &dotasks.LoadBalancer{},
 			})
 		}
 	case kops.CloudProviderAWS:
@@ -643,8 +644,12 @@ func (c *ApplyClusterCmd) Run() error {
 					&model.IAMModelBuilder{KopsModelContext: modelContext, Lifecycle: &securityLifecycle},
 				)
 			case kops.CloudProviderDO:
+				doModelContext := &domodel.DOModelContext{
+					KopsModelContext: modelContext,
+				}
 				l.Builders = append(l.Builders,
 					&model.MasterVolumeBuilder{KopsModelContext: modelContext, Lifecycle: &clusterLifecycle},
+					&domodel.APILoadBalancerModelBuilder{DOModelContext: doModelContext, Lifecycle: &securityLifecycle},
 				)
 
 			case kops.CloudProviderGCE:

upup/pkg/fi/cloudup/dotasks/BUILD.bazel

@@ -5,6 +5,8 @@ go_library(
     srcs = [
         "droplet.go",
         "droplet_fitask.go",
+        "loadbalancer.go",
+        "loadbalancer_fitask.go",
         "volume.go",
         "volume_fitask.go",
     ],