Merge pull request #449 from kubernetes-sigs/pod-targets

Support routing directly to pods

Author: bigkraig

docs/ingress-resources.md

@@ -62,6 +62,7 @@ alb.ingress.kubernetes.io/healthcheck-timeout-seconds
 alb.ingress.kubernetes.io/healthy-threshold-count
 alb.ingress.kubernetes.io/unhealthy-threshold-count
 alb.ingress.kubernetes.io/listen-ports
+alb.ingress.kubernetes.io/target-type
 alb.ingress.kubernetes.io/security-groups
 alb.ingress.kubernetes.io/subnets
 alb.ingress.kubernetes.io/success-codes
@@ -96,6 +97,8 @@ Optional annotations are:
 
 - **listen-ports**: Defines the ports the ALB will expose. It defaults to `[{"HTTP": 80}]` unless a certificate ARN is defined, then it is `[{"HTTPS": 443}]`. Uses a format as follows '[{"HTTP":8080,"HTTPS": 443}]'.
 
+- **target-type**: Defines if the EC2 instance ID or the pod IP are used in the managed Target Groups. Defaults to `instance`. Valid options are `instance` and `pod`. With `instance` the Target Group targets are `<ec2 instance id>:<node port>`, for `pod` the targets are `<pod ip>:<pod port>`. When using the pod IP, it will route from all availabilty zones. `pod` is to be used when the pod network is routable and can be reached by the ALB.
+
 - **security-groups**: [Security groups](http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/VPC_SecurityGroups.html) that should be applied to the ALB instance. These can be referenced by security group IDs or the name tag associated with each security group. Example ID values are `sg-723a380a,sg-a6181ede,sg-a5181edd`. Example tag values are `appSG, webSG`. When the annotation is not present, the controller will create a security group with appropriate ports allowing access to `0.0.0.0/0` and attached to the ALB. It will also create a security group for instances that allows all TCP traffic when the source is the security group created for the ALB.
 
 - **subnets**: The subnets where the ALB instance should be deployed. Must include 2 subnets, each in a different [availability zone](http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html). These can be referenced by subnet IDs or the name tag associated with the subnet. Example values for subnet IDs are `subnet-a4f0098e,subnet-457ed533,subnet-95c904cd`. Example values for name tags are: `webSubnet,appSubnet`. If subnets are not specified the ALB controller will attempt to detect qualified subnets. This qualification is done by locating subnets that match the following criteria.
@@ -133,6 +136,7 @@ alb.ingress.kubernetes.io/healthcheck-protocol
 alb.ingress.kubernetes.io/healthcheck-timeout-seconds
 alb.ingress.kubernetes.io/healthy-threshold-count
 alb.ingress.kubernetes.io/unhealthy-threshold-count
+alb.ingress.kubernetes.io/target-type
 alb.ingress.kubernetes.io/success-codes
 alb.ingress.kubernetes.io/target-group-attributes
 ```

examples/2048/2048-ingress.yaml

@@ -1,21 +1,19 @@
 apiVersion: extensions/v1beta1
 kind: Ingress
 metadata:
-  name: "nginx-ingress"
+  name: "2048-ingress"
   namespace: "2048-game"
   annotations:
     kubernetes.io/ingress.class: alb
     alb.ingress.kubernetes.io/scheme: internal
-    alb.ingress.kubernetes.io/subnets: subnet-1234
-    alb.ingress.kubernetes.io/security-groups: sg-1234
+    alb.ingress.kubernetes.io/target-type: pod
   labels:
-    app: 2048-nginx-ingress
+    app: 2048-ingress
 spec:
   rules:
-  - host: 2048.example.com
-    http:
+  - http:
       paths:
       - path: /
         backend:
           serviceName: "service-2048"
-          servicePort: 80
+          servicePort: 80

examples/alb-ingress-controller.yaml

@@ -94,3 +94,5 @@ spec:
       restartPolicy: Always
       securityContext: {}
       terminationGracePeriodSeconds: 30
+      serviceAccountName: alb-ingress
+      serviceAccount: alb-ingress

pkg/alb/lb/loadbalancer.go

@@ -40,7 +40,7 @@ type NewDesiredLoadBalancerOptions struct {
 	IngressRules          []extensions.IngressRule
 	GetServiceNodePort    func(string, int32) (*int64, error)
 	GetServiceAnnotations func(string, string) (*map[string]string, error)
-	GetNodes              func() util.AWSStringSlice
+	TargetsFunc           func(*string, string, string, *int64) albelbv2.TargetDescriptions
 }
 
 // NewDesiredLoadBalancer returns a new loadbalancer.LoadBalancer based on the opts provided.
@@ -125,7 +125,7 @@ func NewDesiredLoadBalancer(o *NewDesiredLoadBalancerOptions) (newLoadBalancer *
 		GetServiceNodePort:    o.GetServiceNodePort,
 		GetServiceAnnotations: o.GetServiceAnnotations,
 		AnnotationFactory:     o.AnnotationFactory,
-		GetNodes:              o.GetNodes,
+		TargetsFunc:           o.TargetsFunc,
 	})
 
 	if err != nil {
@@ -176,15 +176,6 @@ type NewCurrentLoadBalancerOptions struct {
 
 // NewCurrentLoadBalancer returns a new loadbalancer.LoadBalancer based on an elbv2.LoadBalancer.
 func NewCurrentLoadBalancer(o *NewCurrentLoadBalancerOptions) (newLoadBalancer *LoadBalancer, err error) {
-	lbTags := o.ResourceTags.LoadBalancers[*o.LoadBalancer.LoadBalancerArn]
-
-	namespace, ingressName, err := tagsFromLB(lbTags)
-	if err != nil {
-		return nil, fmt.Errorf("The LoadBalancer %s does not have the proper tags, can't import: %s", *o.LoadBalancer.LoadBalancerName, err.Error())
-	}
-
-	name := createLBName(namespace, ingressName, o.ALBNamePrefix)
-
 	attrs, err := albelbv2.ELBV2svc.DescribeLoadBalancerAttributesFiltered(o.LoadBalancer.LoadBalancerArn)
 	if err != nil {
 		return newLoadBalancer, fmt.Errorf("Failed to retrieve attributes from ELBV2 in AWS. Error: %s", err.Error())
@@ -239,8 +230,8 @@ func NewCurrentLoadBalancer(o *NewCurrentLoadBalancerOptions) (newLoadBalancer *
 	}
 
 	newLoadBalancer = &LoadBalancer{
-		id:         name,
-		tags:       tags{current: lbTags},
+		id:         *o.LoadBalancer.LoadBalancerName,
+		tags:       tags{current: o.ResourceTags.LoadBalancers[*o.LoadBalancer.LoadBalancerArn]},
 		lb:         lb{current: o.LoadBalancer},
 		logger:     o.Logger,
 		attributes: attributes{current: attrs},
@@ -335,6 +326,7 @@ func (l *LoadBalancer) Reconcile(rOpts *ReconcileOptions) []error {
 		IgnoreDeletes:     true,
 	}
 
+	// Creates target groups
 	tgs, err := l.targetgroups.Reconcile(tgsOpts)
 	if err != nil {
 		errors = append(errors, err)
@@ -353,12 +345,13 @@ func (l *LoadBalancer) Reconcile(rOpts *ReconcileOptions) []error {
 		l.listeners = ltnrs
 	}
 
-	// Decide: Is this still needed?
+	// Does not consider TG used for listener default action
 	for _, listener := range l.listeners {
 		unusedTGs := listener.GetRules().FindUnusedTGs(l.targetgroups)
 		unusedTGs.StripDesiredState()
 	}
 
+	// removes target groups
 	tgsOpts.IgnoreDeletes = false
 	tgs, err = l.targetgroups.Reconcile(tgsOpts)
 	if err != nil {

pkg/alb/ls/listener.go

@@ -110,6 +110,20 @@ func NewCurrentListener(o *NewCurrentListenerOptions) (*Listener, error) {
 // results in no action, the creation, the deletion, or the modification of an AWS listener to
 // satisfy the ingress's current state.
 func (l *Listener) Reconcile(rOpts *ReconcileOptions) error {
+	// If there is a desired listener, set some of the ARNs which are not available when we assemble the desired state
+	if l.ls.desired != nil {
+		l.ls.desired.LoadBalancerArn = rOpts.LoadBalancerArn
+
+		// Set the listener default action to the targetgroup from the default rule.
+		// Not good
+		if rOpts != nil {
+			defaultRule := l.rules.DefaultRule()
+			if defaultRule != nil {
+				l.ls.desired.DefaultActions[0].TargetGroupArn = defaultRule.TargetGroupArn(rOpts.TargetGroups)
+			}
+		}
+	}
+
 	switch {
 	case l.ls.desired == nil: // listener should be deleted
 		if l.ls.current == nil {
@@ -156,14 +170,6 @@ func (l *Listener) Reconcile(rOpts *ReconcileOptions) error {
 
 // Adds a Listener to an existing ALB in AWS. This Listener maps the ALB to an existing TargetGroup.
 func (l *Listener) create(rOpts *ReconcileOptions) error {
-	l.ls.desired.LoadBalancerArn = rOpts.LoadBalancerArn
-
-	// Set the listener default action to the targetgroup from the default rule.
-	defaultRule := l.rules.DefaultRule()
-	if defaultRule != nil {
-		l.ls.desired.DefaultActions[0].TargetGroupArn = defaultRule.TargetGroupArn(rOpts.TargetGroups)
-	}
-
 	// Attempt listener creation.
 	desired := l.ls.desired
 	in := &elbv2.CreateListenerInput{
@@ -172,12 +178,7 @@ func (l *Listener) create(rOpts *ReconcileOptions) error {
 		Protocol:        desired.Protocol,
 		Port:            desired.Port,
 		SslPolicy:       desired.SslPolicy,
-		DefaultActions: []*elbv2.Action{
-			{
-				Type:           desired.DefaultActions[0].Type,
-				TargetGroupArn: desired.DefaultActions[0].TargetGroupArn,
-			},
-		},
+		DefaultActions:  desired.DefaultActions,
 	}
 	o, err := albelbv2.ELBV2svc.CreateListener(in)
 	if err != nil {
@@ -191,11 +192,6 @@ func (l *Listener) create(rOpts *ReconcileOptions) error {
 
 // Modifies a listener
 func (l *Listener) modify(rOpts *ReconcileOptions) error {
-	if l.ls.current == nil {
-		// not a modify, a create
-		return l.create(rOpts)
-	}
-
 	desired := l.ls.desired
 	in := &elbv2.ModifyListenerInput{
 		ListenerArn:    l.ls.current.ListenerArn,
@@ -233,14 +229,6 @@ func (l *Listener) needsModification(rOpts *ReconcileOptions) bool {
 	lsc := l.ls.current
 	lsd := l.ls.desired
 
-	// Set the listener default action to the targetgroup from the default rule.
-	if rOpts != nil {
-		defaultRule := l.rules.DefaultRule()
-		if defaultRule != nil {
-			lsd.DefaultActions[0].TargetGroupArn = defaultRule.TargetGroupArn(rOpts.TargetGroups)
-		}
-	}
-
 	switch {
 	case lsc == nil && lsd == nil:
 		return false
@@ -281,3 +269,7 @@ func (l *Listener) stripCurrentState() {
 func (l *Listener) GetRules() rs.Rules {
 	return l.rules
 }
+
+func (l *Listener) DefaultActionArn() *string {
+	return l.ls.current.DefaultActions[0].TargetGroupArn
+}

pkg/alb/rs/rule.go

@@ -21,6 +21,7 @@ type NewDesiredRuleOptions struct {
 	IgnoreHostHeader bool
 	Path             string
 	SvcName          string
+	SvcPort          int32
 	Logger           *log.Logger
 }
 
@@ -60,31 +61,39 @@ func NewDesiredRule(o *NewDesiredRuleOptions) *Rule {
 	}
 
 	return &Rule{
-		svcname: svcname{desired: o.SvcName},
-		rs:      rs{desired: r},
-		logger:  o.Logger,
+		svc:    svc{desired: service{name: o.SvcName, port: o.SvcPort}},
+		rs:     rs{desired: r},
+		logger: o.Logger,
 	}
 }
 
 type NewCurrentRuleOptions struct {
 	SvcName string
+	SvcPort int32
 	Rule    *elbv2.Rule
 	Logger  *log.Logger
 }
 
 // NewCurrentRule creates a Rule from an elbv2.Rule
 func NewCurrentRule(o *NewCurrentRuleOptions) *Rule {
 	return &Rule{
-		svcname: svcname{current: o.SvcName},
-		rs:      rs{current: o.Rule},
-		logger:  o.Logger,
+		svc:    svc{current: service{name: o.SvcName, port: o.SvcPort}},
+		rs:     rs{current: o.Rule},
+		logger: o.Logger,
 	}
 }
 
 // Reconcile compares the current and desired state of this Rule instance. Comparison
 // results in no action, the creation, the deletion, or the modification of an AWS Rule to
 // satisfy the ingress's current state.
 func (r *Rule) Reconcile(rOpts *ReconcileOptions) error {
+	// If there is a desired rule, set some of the ARNs which are not available when we assemble the desired state
+	if r.rs.desired != nil {
+		for i := range r.rs.desired.Actions {
+			r.rs.desired.Actions[i].TargetGroupArn = r.TargetGroupArn(rOpts.TargetGroups)
+		}
+	}
+
 	switch {
 	case r.rs.desired == nil: // rule should be deleted
 		if r.rs.current == nil {
@@ -102,8 +111,10 @@ func (r *Rule) Reconcile(rOpts *ReconcileOptions) error {
 			log.Prettify(r.rs.current.Priority),
 			log.Prettify(r.rs.current.Conditions))
 
-	case *r.rs.desired.IsDefault: // rule is default (attached to listener), do nothing
-		// r.logger.Debugf("Found desired rule that is a default and is already created with its respective listener. Rule: %s", log.Prettify(r.rs.desired))
+	case *r.rs.desired.IsDefault:
+		// rule is default (attached to listener), do nothing
+		// Seems to happen automatically, if we try to change it we get an error:
+		// OperationNotPermitted: Default rule '<arn>' cannot be modified
 		r.rs.current = r.rs.desired
 
 	case r.rs.current == nil: // rule doesn't exist and should be created
@@ -121,23 +132,20 @@ func (r *Rule) Reconcile(rOpts *ReconcileOptions) error {
 			return err
 		}
 		rOpts.Eventf(api.EventTypeNormal, "MODIFY", "%s rule modified", *r.rs.current.Priority)
-
-	default:
-		// r.logger.Debugf("No rule modification required.")
 	}
 
 	return nil
 }
 
 func (r *Rule) TargetGroupArn(tgs tg.TargetGroups) *string {
-	i := tgs.LookupBySvc(r.svcname.desired)
+	i := tgs.LookupBySvc(r.svc.desired.name, r.svc.desired.port)
 	if i < 0 {
-		r.logger.Errorf("Failed to locate TargetGroup related to this service: %s", r.svcname.desired)
+		r.logger.Errorf("Failed to locate TargetGroup related to this service: %s:%d", r.svc.desired.name, r.svc.desired.port)
 		return nil
 	}
 	arn := tgs[i].CurrentARN()
 	if arn == nil {
-		r.logger.Errorf("Located TargetGroup but no known (current) state found: %s", r.svcname.desired)
+		r.logger.Errorf("Located TargetGroup but no known (current) state found: %s:%d", r.svc.desired.name, r.svc.desired.port)
 	}
 	return arn
 }
@@ -150,15 +158,13 @@ func (r *Rule) create(rOpts *ReconcileOptions) error {
 		Priority:    priority(r.rs.desired.Priority),
 	}
 
-	in.Actions[0].TargetGroupArn = r.TargetGroupArn(rOpts.TargetGroups)
-
 	o, err := albelbv2.ELBV2svc.CreateRule(in)
 	if err != nil {
 		rOpts.Eventf(api.EventTypeWarning, "ERROR", "Error creating %v rule: %s", *in.Priority, err.Error())
 		return fmt.Errorf("Failed Rule creation. Rule: %s | Error: %s", log.Prettify(r.rs.desired), err.Error())
 	}
 	r.rs.current = o.Rules[0]
-	r.svcname.current = r.svcname.desired
+	r.svc.current = r.svc.desired
 
 	return nil
 }
@@ -169,7 +175,6 @@ func (r *Rule) modify(rOpts *ReconcileOptions) error {
 		Conditions: r.rs.desired.Conditions,
 		RuleArn:    r.rs.current.RuleArn,
 	}
-	in.Actions[0].TargetGroupArn = r.TargetGroupArn(rOpts.TargetGroups)
 
 	o, err := albelbv2.ELBV2svc.ModifyRule(in)
 	if err != nil {
@@ -180,7 +185,7 @@ func (r *Rule) modify(rOpts *ReconcileOptions) error {
 	if len(o.Rules) > 0 {
 		r.rs.current = o.Rules[0]
 	}
-	r.svcname.current = r.svcname.desired
+	r.svc.current = r.svc.desired
 
 	return nil
 }
@@ -217,12 +222,17 @@ func (r *Rule) needsModification() bool {
 	case crs == nil:
 		r.logger.Debugf("Current is nil")
 		return true
-	// TODO: We need to sort these because they're causing false positives
+	case !util.DeepEqual(crs.Actions, drs.Actions):
+		r.logger.Debugf("Actions needs to be changed (%v != %v)", log.Prettify(crs.Actions), log.Prettify(drs.Actions))
+		return true
 	case !conditionsEqual(crs.Conditions, drs.Conditions):
 		r.logger.Debugf("Conditions needs to be changed (%v != %v)", log.Prettify(crs.Conditions), log.Prettify(drs.Conditions))
 		return true
-	case r.svcname.current != r.svcname.desired:
-		r.logger.Debugf("SvcName needs to be changed (%v != %v)", r.svcname.current, r.svcname.desired)
+	case r.svc.current.name != r.svc.desired.name:
+		r.logger.Debugf("SvcName needs to be changed (%v != %v)", r.svc.current.name, r.svc.desired.name)
+		return true
+	case r.svc.current.port != r.svc.desired.port && r.svc.current.port != 0: // Check against 0 because that is the default for legacy tags
+		r.logger.Debugf("SvcPort needs to be changed (%v != %v)", r.svc.current.port, r.svc.desired.port)
 		return true
 	}