Merge pull request #626 from kubernetes-sigs/session-caching

Implement caching at the AWS SDK session

Author: k8s-ci-robot

Gopkg.lock

@@ -28,8 +28,10 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/ingress/metric/collectors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/ticketmaster/aws-sdk-go-cache/cache"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -86,18 +88,21 @@ func main() {
 
 	conf.Client = kubeClient
 
+	cc := cache.NewConfig(5 * time.Minute)
+
 	reg := prometheus.NewRegistry()
 
+	reg.MustRegister(cc.NewCacheCollector(collectors.PrometheusNamespace))
 	reg.MustRegister(prometheus.NewGoCollector())
-	reg.MustRegister(prometheus.NewProcessCollector(os.Getpid(), ""))
+	reg.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
 
 	mc, err := metric.NewCollector(reg)
 	if err != nil {
 		glog.Fatalf("Error creating prometheus collectos:  %v", err)
 	}
 	mc.Start()
 
-	c := controller.NewALBController(conf, mc)
+	c := controller.NewALBController(conf, mc, cc)
 	go handleSigterm(c, func(code int) {
 		os.Exit(code)
 	})

cmd/main.go

@@ -13,11 +13,9 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/alb/tg"
-	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/aws/albcache"
 	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/aws/albelbv2"
 	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/ingress/annotations/loadbalancer"
 	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/ingress/controller/store"
-	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/ingress/metric"
 	"github.com/kubernetes-sigs/aws-alb-ingress-controller/pkg/util/log"
 	"github.com/kubernetes-sigs/aws-alb-ingress-controller/pkg/util/types"
 	util "github.com/kubernetes-sigs/aws-alb-ingress-controller/pkg/util/types"
@@ -43,8 +41,6 @@ func init() {
 	albelbv2.ELBV2svc = albelbv2.NewDummy()
 	albec2.EC2svc = &mocks.EC2API{}
 
-	albcache.NewCache(metric.DummyCollector{})
-
 	rOpts1 = &ReconcileOptions{
 		TargetGroups:    tg.TargetGroups{tg.DummyTG("tg1", "service")},
 		LoadBalancerArn: nil,

internal/alb/ls/listener_test.go

@@ -5,7 +5,6 @@ import (
 	"os"
 	"sort"
 	"strings"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/request"
@@ -21,7 +20,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/ec2/ec2iface"
 
-	"github.com/kubernetes-sigs/aws-alb-ingress-controller/internal/aws/albcache"
 	util "github.com/kubernetes-sigs/aws-alb-ingress-controller/pkg/util/types"
 )
 
@@ -34,11 +32,6 @@ const (
 
 	tagNameSubnetInternalELB = "kubernetes.io/role/internal-elb"
 	tagNameSubnetPublicELB   = "kubernetes.io/role/elb"
-
-	GetSecurityGroupsCacheTTL = time.Minute * 60
-	GetSubnetsCacheTTL        = time.Minute * 60
-
-	IsNodeHealthyCacheTTL = time.Minute * 5
 )
 
 // EC2svc is the singleton points to our aws EC2 api
@@ -87,8 +80,6 @@ type EC2 struct {
 }
 
 // EC2MData is our extension to AWS's ec2metadata.EC2Metadata
-// cache is not required for this struct as we only use it to lookup
-// instance metadata when the cache for the EC2 struct is expired.
 type EC2MData struct {
 	*ec2metadata.EC2Metadata
 }
@@ -113,23 +104,6 @@ func (e *EC2) GetSubnets(names []*string) (subnets []*string, err error) {
 		return
 	}
 
-	cacheName := "EC2.GetSubnets"
-	var queryNames []*string
-
-	for _, n := range names {
-		item := albcache.Get(cacheName, *n)
-
-		if item != nil {
-			subnets = append(subnets, item.Value().(*string))
-		} else {
-			queryNames = append(queryNames, n)
-		}
-	}
-
-	if len(queryNames) == 0 {
-		return
-	}
-
 	in := &ec2.DescribeSubnetsInput{Filters: []*ec2.Filter{
 		{
 			Name:   aws.String("tag:Name"),
@@ -147,9 +121,8 @@ func (e *EC2) GetSubnets(names []*string) (subnets []*string, err error) {
 	}
 
 	for _, subnet := range describeSubnetsOutput.Subnets {
-		value, ok := util.EC2Tags(subnet.Tags).Get("Name")
+		_, ok := util.EC2Tags(subnet.Tags).Get("Name")
 		if ok {
-			albcache.Set(cacheName, value, subnet.SubnetId, GetSubnetsCacheTTL)
 			subnets = append(subnets, subnet.SubnetId)
 		}
 	}
@@ -162,27 +135,10 @@ func (e *EC2) GetSecurityGroups(names []*string) (sgs []*string, err error) {
 		return
 	}
 
-	cacheName := "EC2.GetSecurityGroups"
-	var queryNames []*string
-
-	for _, n := range names {
-		item := albcache.Get(cacheName, *n)
-
-		if item != nil {
-			sgs = append(sgs, item.Value().(*string))
-		} else {
-			queryNames = append(queryNames, n)
-		}
-	}
-
-	if len(queryNames) == 0 {
-		return
-	}
-
 	in := &ec2.DescribeSecurityGroupsInput{Filters: []*ec2.Filter{
 		{
 			Name:   aws.String("tag:Name"),
-			Values: queryNames,
+			Values: names,
 		},
 		{
 			Name:   aws.String("vpc-id"),
@@ -196,8 +152,6 @@ func (e *EC2) GetSecurityGroups(names []*string) (sgs []*string, err error) {
 	}
 
 	for _, sg := range describeSecurityGroupsOutput.SecurityGroups {
-		name, _ := util.EC2Tags(sg.Tags).Get("Name")
-		albcache.Set(cacheName, name, sg.GroupId, GetSecurityGroupsCacheTTL)
 		sgs = append(sgs, sg.GroupId)
 	}
 
@@ -304,27 +258,12 @@ func (e *EC2) GetVPCID() (*string, error) {
 		return &v, nil
 	}
 
-	// If previously looked up (and not expired) the VpcId will be stored in the cache under the
-	// key 'vpc'.
-	cacheName := "EC2.GetVPCID"
-	item := albcache.Get(cacheName, "")
-
-	// cache hit: return (pointer of) VpcId value
-	if item != nil {
-		vpc = item.Value().(*string)
-		return vpc, nil
-	}
-
-	// cache miss: begin lookup of VpcId based on current EC2 instance
-	// retrieve identity of current running instance
 	identityDoc, err := EC2Metadatasvc.GetInstanceIdentityDocument()
 	if err != nil {
 		return nil, err
 	}
 
 	// capture instance ID for lookup in DescribeInstances
-	// don't bother caching this value as it should never be re-retrieved unless
-	// the cache for the VpcId (looked up below) expires.
 	descInstancesInput := &ec2.DescribeInstancesInput{
 		InstanceIds: []*string{aws.String(identityDoc.InstanceID)},
 	}
@@ -342,21 +281,10 @@ func (e *EC2) GetVPCID() (*string, error) {
 	}
 
 	vpc = descInstancesOutput.Reservations[0].Instances[0].VpcId
-	// cache the retrieved VpcId for next call
-	albcache.Set(cacheName, "", vpc, time.Minute*60)
 	return vpc, nil
 }
 
 func (e *EC2) GetVPC(id *string) (*ec2.Vpc, error) {
-	cacheName := "EC2.GetVPCID"
-	item := albcache.Get(cacheName, *id)
-
-	// cache hit: return (pointer of) VpcId value
-	if item != nil {
-		vpc := item.Value().(*ec2.Vpc)
-		return vpc, nil
-	}
-
 	o, err := e.DescribeVpcs(&ec2.DescribeVpcsInput{
 		VpcIds: []*string{id},
 	})
@@ -367,7 +295,6 @@ func (e *EC2) GetVPC(id *string) (*ec2.Vpc, error) {
 		return nil, fmt.Errorf("Invalid amount of VPCs %d returned for %s", len(o.Vpcs), *id)
 	}
 
-	albcache.Set(cacheName, *id, o.Vpcs[0], time.Minute*60)
 	return o.Vpcs[0], nil
 }
 
@@ -410,8 +337,6 @@ func ClusterSubnets(scheme *string) (util.Subnets, error) {
 	var out util.AWSStringSlice
 	var key string
 
-	cacheName := "ClusterSubnets"
-
 	if *scheme == elbv2.LoadBalancerSchemeEnumInternal {
 		key = tagNameSubnetInternalELB
 	} else if *scheme == elbv2.LoadBalancerSchemeEnumInternetFacing {
@@ -431,15 +356,7 @@ func ClusterSubnets(scheme *string) (util.Subnets, error) {
 			if *tag.Key == key {
 				p := strings.Split(arn, "/")
 				subnetID := &p[len(p)-1]
-				item := albcache.Get(cacheName, *subnetID)
-				if item != nil {
-					if subnetIsUsable(item.Value().(*ec2.Subnet), useableSubnets) {
-						useableSubnets = append(useableSubnets, item.Value().(*ec2.Subnet))
-						out = append(out, item.Value().(*ec2.Subnet).SubnetId)
-					}
-				} else {
-					filterValues = append(filterValues, subnetID)
-				}
+				filterValues = append(filterValues, subnetID)
 			}
 		}
 	}
@@ -466,7 +383,6 @@ func ClusterSubnets(scheme *string) (util.Subnets, error) {
 		if subnetIsUsable(subnet, useableSubnets) {
 			useableSubnets = append(useableSubnets, subnet)
 			out = append(out, subnet.SubnetId)
-			albcache.Set(cacheName, *subnet.SubnetId, subnet, time.Minute*60)
 		}
 	}
 
@@ -498,13 +414,6 @@ func subnetIsUsable(new *ec2.Subnet, existing []*ec2.Subnet) bool {
 
 // IsNodeHealthy returns true if the node is ready
 func (e *EC2) IsNodeHealthy(instanceid string) (bool, error) {
-	cacheName := "ec2.IsNodeHealthy"
-	item := albcache.Get(cacheName, instanceid)
-
-	if item != nil {
-		return item.Value().(bool), nil
-	}
-
 	in := &ec2.DescribeInstanceStatusInput{
 		InstanceIds: []*string{aws.String(instanceid)},
 	}
@@ -518,10 +427,8 @@ func (e *EC2) IsNodeHealthy(instanceid string) (bool, error) {
 			continue
 		}
 		if *instanceStatus.InstanceState.Code == 16 { // running
-			albcache.Set(cacheName, instanceid, true, IsNodeHealthyCacheTTL)
 			return true, nil
 		}
-		albcache.Set(cacheName, instanceid, false, IsNodeHealthyCacheTTL)
 		return false, nil
 	}
 

internal/aws/albcache/cache.go

@@ -30,11 +30,6 @@ func NewDummy() *Dummy {
 	return d
 }
 
-// CacheDelete ...
-func (d *Dummy) CacheDelete(string, string) {
-	return
-}
-
 // ClusterLoadBalancers ...
 func (d *Dummy) ClusterLoadBalancers() ([]*elbv2.LoadBalancer, error) {
 	return nil, nil