Merge pull request #48 from justinsb/upup_elb

upup: simple ELB support

upup/pkg/fi/cloudup/awstasks/load_balancer.go

@@ -0,0 +1,231 @@
+package awstasks
+
+import (
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/elb"
+	"github.com/golang/glog"
+	"k8s.io/kube-deploy/upup/pkg/fi"
+	"k8s.io/kube-deploy/upup/pkg/fi/cloudup/awsup"
+	"strconv"
+)
+
+//go:generate fitask -type=LoadBalancer
+type LoadBalancer struct {
+	Name *string
+
+	ID *string
+
+	DNSName      *string
+	HostedZoneId *string
+
+	Subnets        []*Subnet
+	SecurityGroups []*SecurityGroup
+
+	Listeners map[string]*LoadBalancerListener
+}
+
+type LoadBalancerListener struct {
+	InstancePort int
+}
+
+func (e *LoadBalancerListener) mapToAWS(loadBalancerPort int64) *elb.Listener {
+	return &elb.Listener{
+		LoadBalancerPort: aws.Int64(loadBalancerPort),
+
+		Protocol: aws.String("TCP"),
+
+		InstanceProtocol: aws.String("TCP"),
+		InstancePort:     aws.Int64(int64(e.InstancePort)),
+	}
+}
+
+var _ fi.HasDependencies = &LoadBalancerListener{}
+
+func (e *LoadBalancerListener) GetDependencies(tasks map[string]fi.Task) []fi.Task {
+	return nil
+}
+
+func findELB(cloud *awsup.AWSCloud, name string) (*elb.LoadBalancerDescription, error) {
+	request := &elb.DescribeLoadBalancersInput{
+		LoadBalancerNames: []*string{&name},
+	}
+
+	var found []*elb.LoadBalancerDescription
+	err := cloud.ELB.DescribeLoadBalancersPages(request, func(p *elb.DescribeLoadBalancersOutput, lastPage bool) (shouldContinue bool) {
+		for _, lb := range p.LoadBalancerDescriptions {
+			if aws.StringValue(lb.LoadBalancerName) == name {
+				found = append(found, lb)
+			} else {
+				glog.Warningf("Got ELB with unexpected name")
+			}
+		}
+
+		return true
+	})
+
+	if err != nil {
+		if awsError, ok := err.(awserr.Error); ok {
+			if awsError.Code() == "LoadBalancerNotFound" {
+				return nil, nil
+			}
+		}
+
+		return nil, fmt.Errorf("error listing ELBs: %v", err)
+	}
+
+	if len(found) == 0 {
+		return nil, nil
+	}
+
+	if len(found) != 1 {
+		return nil, fmt.Errorf("Found multiple ELBs with name %q", name)
+	}
+
+	return found[0], nil
+}
+
+func (e *LoadBalancer) Find(c *fi.Context) (*LoadBalancer, error) {
+	cloud := c.Cloud.(*awsup.AWSCloud)
+
+	lb, err := findELB(cloud, fi.StringValue(e.Name))
+	if err != nil {
+		return nil, err
+	}
+	if lb == nil {
+		return nil, nil
+	}
+
+	actual := &LoadBalancer{}
+	actual.Name = e.Name
+	actual.ID = e.DNSName
+	actual.DNSName = lb.DNSName
+	actual.HostedZoneId = lb.CanonicalHostedZoneNameID
+	for _, subnet := range lb.Subnets {
+		actual.Subnets = append(actual.Subnets, &Subnet{ID: subnet})
+	}
+
+	for _, sg := range lb.SecurityGroups {
+		actual.SecurityGroups = append(actual.SecurityGroups, &SecurityGroup{ID: sg})
+	}
+
+	actual.Listeners = make(map[string]*LoadBalancerListener)
+
+	for _, ld := range lb.ListenerDescriptions {
+		l := ld.Listener
+		loadBalancerPort := strconv.FormatInt(aws.Int64Value(l.LoadBalancerPort), 10)
+
+		actualListener := &LoadBalancerListener{}
+		actualListener.InstancePort = int(aws.Int64Value(l.InstancePort))
+		actual.Listeners[loadBalancerPort] = actualListener
+	}
+
+	// Avoid spurious mismatches
+	if subnetSlicesEqualIgnoreOrder(actual.Subnets, e.Subnets) {
+		actual.Subnets = e.Subnets
+	}
+
+	if e.DNSName == nil {
+		e.DNSName = actual.DNSName
+	}
+	if e.HostedZoneId == nil {
+		e.HostedZoneId = actual.HostedZoneId
+	}
+
+	return actual, nil
+}
+
+func (e *LoadBalancer) Run(c *fi.Context) error {
+	return fi.DefaultDeltaRunMethod(e, c)
+}
+
+func (s *LoadBalancer) CheckChanges(a, e, changes *LoadBalancer) error {
+	if a == nil {
+		if fi.StringValue(e.Name) == "" {
+			return fi.RequiredField("Name")
+		}
+		if len(e.SecurityGroups) == 0 {
+			return fi.RequiredField("SecurityGroups")
+		}
+		if len(e.Subnets) == 0 {
+			return fi.RequiredField("Subnets")
+		}
+	}
+	return nil
+}
+
+func (_ *LoadBalancer) RenderAWS(t *awsup.AWSAPITarget, a, e, changes *LoadBalancer) error {
+	if a == nil {
+		request := &elb.CreateLoadBalancerInput{}
+		request.LoadBalancerName = e.Name
+
+		for _, subnet := range e.Subnets {
+			request.Subnets = append(request.Subnets, subnet.ID)
+		}
+
+		for _, sg := range e.SecurityGroups {
+			request.SecurityGroups = append(request.SecurityGroups, sg.ID)
+		}
+
+		request.Listeners = []*elb.Listener{}
+
+		for loadBalancerPort, listener := range e.Listeners {
+			loadBalancerPortInt, err := strconv.ParseInt(loadBalancerPort, 10, 64)
+			if err != nil {
+				return fmt.Errorf("error parsing load balancer listener port: %q", loadBalancerPort)
+			}
+			awsListener := listener.mapToAWS(loadBalancerPortInt)
+			request.Listeners = append(request.Listeners, awsListener)
+		}
+
+		glog.V(2).Infof("Creating ELB with Name:%q", *e.Name)
+
+		response, err := t.Cloud.ELB.CreateLoadBalancer(request)
+		if err != nil {
+			return fmt.Errorf("error creating ELB: %v", err)
+		}
+
+		e.DNSName = response.DNSName
+		e.ID = response.DNSName
+
+		lb, err := findELB(t.Cloud, *e.Name)
+		if err != nil {
+			return err
+		}
+		if lb == nil {
+			// TODO: Retry?  Is this async
+			return fmt.Errorf("Unable to find newly created ELB")
+		}
+
+		e.HostedZoneId = lb.CanonicalHostedZoneNameID
+	} else {
+		if changes.Subnets != nil {
+			return fmt.Errorf("subnet changes on LoadBalancer not yet implemented")
+		}
+
+		if changes.Listeners != nil {
+			request := &elb.CreateLoadBalancerListenersInput{}
+			request.LoadBalancerName = e.Name
+
+			for loadBalancerPort, listener := range changes.Listeners {
+				loadBalancerPortInt, err := strconv.ParseInt(loadBalancerPort, 10, 64)
+				if err != nil {
+					return fmt.Errorf("error parsing load balancer listener port: %q", loadBalancerPort)
+				}
+				awsListener := listener.mapToAWS(loadBalancerPortInt)
+				request.Listeners = append(request.Listeners, awsListener)
+			}
+
+			glog.V(2).Infof("Creating LoadBalancer listeners")
+
+			_, err := t.Cloud.ELB.CreateLoadBalancerListeners(request)
+			if err != nil {
+				return fmt.Errorf("error creating LoadBalancerListeners: %v", err)
+			}
+		}
+	}
+
+	return t.AddAWSTags(*e.ID, t.Cloud.BuildTags(e.Name, nil))
+}

upup/pkg/fi/cloudup/awstasks/load_balancer_attachment.go

@@ -0,0 +1,78 @@
+package awstasks
+
+import (
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/autoscaling"
+	"github.com/golang/glog"
+	"k8s.io/kube-deploy/upup/pkg/fi"
+	"k8s.io/kube-deploy/upup/pkg/fi/cloudup/awsup"
+)
+
+type LoadBalancerAttachment struct {
+	LoadBalancer     *LoadBalancer
+	AutoscalingGroup *AutoscalingGroup
+}
+
+func (e *LoadBalancerAttachment) String() string {
+	return fi.TaskAsString(e)
+}
+
+func (e *LoadBalancerAttachment) Find(c *fi.Context) (*LoadBalancerAttachment, error) {
+	cloud := c.Cloud.(*awsup.AWSCloud)
+
+	if e.AutoscalingGroup != nil {
+		g, err := findAutoscalingGroup(cloud, *e.AutoscalingGroup.Name)
+		if err != nil {
+			return nil, err
+		}
+		if g == nil {
+			return nil, nil
+		}
+
+		for _, name := range g.LoadBalancerNames {
+			if aws.StringValue(name) != *e.AutoscalingGroup.Name {
+				continue
+			}
+
+			actual := &LoadBalancerAttachment{}
+			actual.LoadBalancer = e.LoadBalancer
+			actual.AutoscalingGroup = e.AutoscalingGroup
+			return actual, nil
+		}
+	}
+
+	return nil, nil
+}
+
+func (e *LoadBalancerAttachment) Run(c *fi.Context) error {
+	return fi.DefaultDeltaRunMethod(e, c)
+}
+
+func (s *LoadBalancerAttachment) CheckChanges(a, e, changes *LoadBalancerAttachment) error {
+	if a == nil {
+		if e.LoadBalancer == nil {
+			return fi.RequiredField("LoadBalancer")
+		}
+		if e.AutoscalingGroup == nil {
+			return fi.RequiredField("AutoscalingGroup")
+		}
+	}
+	return nil
+}
+
+func (_ *LoadBalancerAttachment) RenderAWS(t *awsup.AWSAPITarget, a, e, changes *LoadBalancerAttachment) error {
+	request := &autoscaling.AttachLoadBalancersInput{}
+	request.AutoScalingGroupName = e.AutoscalingGroup.Name
+	request.LoadBalancerNames = []*string{e.LoadBalancer.Name}
+
+	glog.V(2).Infof("Attaching autoscaling group %q to ELB %q", *e.AutoscalingGroup.Name, *e.LoadBalancer.Name)
+
+	_, err := t.Cloud.Autoscaling.AttachLoadBalancers(request)
+	if err != nil {
+		return fmt.Errorf("error attaching autoscaling group to ELB: %v", err)
+	}
+
+	return nil
+}

upup/pkg/fi/cloudup/awsup/aws_apitarget.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/golang/glog"
 	"k8s.io/kube-deploy/upup/pkg/fi"
 	"time"
@@ -42,16 +41,9 @@ func (t *AWSAPITarget) AddAWSTags(id string, expected map[string]string) error {
 	}
 
 	if len(missing) != 0 {
-		request := &ec2.CreateTagsInput{}
-		request.Resources = []*string{&id}
-		for k, v := range missing {
-			request.Tags = append(request.Tags, &ec2.Tag{
-				Key:   aws.String(k),
-				Value: aws.String(v),
-			})
-		}
+		glog.V(4).Infof("adding tags to %q: %v", id, missing)
 
-		_, err := t.Cloud.EC2.CreateTags(request)
+		err := t.Cloud.CreateTags(id, missing)
 		if err != nil {
 			return fmt.Errorf("error adding tags to resource %q: %v", id, err)
 		}
@@ -60,6 +52,32 @@ func (t *AWSAPITarget) AddAWSTags(id string, expected map[string]string) error {
 	return nil
 }
 
+func (t *AWSAPITarget) AddELBTags(loadBalancerName string, expected map[string]string) error {
+	actual, err := t.Cloud.GetELBTags(loadBalancerName)
+	if err != nil {
+		return fmt.Errorf("unexpected error fetching tags for resource: %v", err)
+	}
+
+	missing := map[string]string{}
+	for k, v := range expected {
+		actualValue, found := actual[k]
+		if found && actualValue == v {
+			continue
+		}
+		missing[k] = v
+	}
+
+	if len(missing) != 0 {
+		glog.V(4).Infof("adding tags to %q: %v", loadBalancerName, missing)
+		err := t.Cloud.CreateELBTags(loadBalancerName, missing)
+		if err != nil {
+			return fmt.Errorf("error adding tags to ELB %q: %v", loadBalancerName, err)
+		}
+	}
+
+	return nil
+}
+
 func (t *AWSAPITarget) WaitForInstanceRunning(instanceID string) error {
 	attempt := 0
 	for {

upup/pkg/fi/cloudup/awsup/aws_cloud.go

@@ -163,7 +163,62 @@ func (c *AWSCloud) CreateTags(resourceId string, tags map[string]string) error {
 	}
 }
 
-func (c *AWSCloud) BuildTags(name *string) map[string]string {
+func (c *AWSCloud) GetELBTags(loadBalancerName string) (map[string]string, error) {
+	tags := map[string]string{}
+
+	request := &elb.DescribeTagsInput{
+		LoadBalancerNames: []*string{&loadBalancerName},
+	}
+
+	attempt := 0
+	for {
+		attempt++
+
+		response, err := c.ELB.DescribeTags(request)
+		if err != nil {
+			return nil, fmt.Errorf("error listing tags on %v: %v", loadBalancerName, err)
+		}
+
+		for _, tagset := range response.TagDescriptions {
+			for _, tag := range tagset.Tags {
+				tags[aws.StringValue(tag.Key)] = aws.StringValue(tag.Value)
+			}
+		}
+
+		return tags, nil
+	}
+}
+
+// CreateELBTags will add tags to the specified loadBalancer, retrying up to MaxCreateTagsAttempts times if it hits an eventual-consistency type error
+func (c *AWSCloud) CreateELBTags(loadBalancerName string, tags map[string]string) error {
+	if len(tags) == 0 {
+		return nil
+	}
+
+	elbTags := []*elb.Tag{}
+	for k, v := range tags {
+		elbTags = append(elbTags, &elb.Tag{Key: aws.String(k), Value: aws.String(v)})
+	}
+
+	attempt := 0
+	for {
+		attempt++
+
+		request := &elb.AddTagsInput{
+			Tags:              elbTags,
+			LoadBalancerNames: []*string{&loadBalancerName},
+		}
+
+		_, err := c.ELB.AddTags(request)
+		if err != nil {
+			return fmt.Errorf("error creating tags on %v: %v", loadBalancerName, err)
+		}
+
+		return nil
+	}
+}
+
+func (c *AWSCloud) BuildTags(name *string, itemTags map[string]string) map[string]string {
 	tags := make(map[string]string)
 	if name != nil {
 		tags["Name"] = *name
@@ -173,6 +228,9 @@ func (c *AWSCloud) BuildTags(name *string) map[string]string {
 	for k, v := range c.tags {
 		tags[k] = v
 	}
+	for k, v := range itemTags {
+		tags[k] = v
+	}
 	return tags
 }