From c260536429aa89f8ab362a95deb3ee4496ac395d Mon Sep 17 00:00:00 2001 From: Martynas Pumputis Date: Tue, 7 Jun 2016 13:48:34 +0100 Subject: [PATCH] Add AWSVPC Tracker The tracker is responsible for managing the AWSVPC route tables which is used to implement the underlay routing based on VPC. --- .gitmodules | 3 + ipam/tracker/awsvpc.go | 286 +++++++++++++++++++++++++++++++ ipam/tracker/awsvpc_test.go | 52 ++++++ net/address/address.go | 16 ++ net/address/address_test.go | 6 + vendor/github.com/aws/aws-sdk-go | 1 + 6 files changed, 364 insertions(+) create mode 100644 ipam/tracker/awsvpc.go create mode 100644 ipam/tracker/awsvpc_test.go create mode 160000 vendor/github.com/aws/aws-sdk-go diff --git a/.gitmodules b/.gitmodules index 8636191c..b13e74dc 100644 --- a/.gitmodules +++ b/.gitmodules @@ -73,3 +73,6 @@ [submodule "vendor/github.com/j-keck/arping"] path = vendor/github.com/j-keck/arping url = https://github.com/j-keck/arping +[submodule "vendor/github.com/aws/aws-sdk-go"] + path = vendor/github.com/aws/aws-sdk-go + url = https://github.com/aws/aws-sdk-go diff --git a/ipam/tracker/awsvpc.go b/ipam/tracker/awsvpc.go new file mode 100644 index 00000000..1089b615 --- /dev/null +++ b/ipam/tracker/awsvpc.go @@ -0,0 +1,286 @@ +package tracker + +// The AWSVPC tracker tracks the IPAM ring changes and accordingly updates +// the AWS VPC route table and, if the announced change is local +// (denoted by local=true) to the host, the host route table. +// +// During the initialization, the tracker detects AWS VPC route table id which +// is associated with the host's subnet on the AWS network. If such a table does +// not exist, the default VPC route table is used. +// +// When a host A donates a range to a host B, the necessary route table +// updates (removal) happen on the host A first, and afterwards on +// the host B (installation). +// +// NB: there is a hard limit for 50 routes within any VPC route table +// (practically, it is 48, because one route is used by the AWS Internet GW and +// one by the AWS host subnet), therefore it is suggested to avoid +// an excessive fragmentation within the IPAM ring which might happen due to +// the claim operations or uneven distribution of containers across the hosts. + +import ( + "fmt" + "net" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/ec2metadata" + "github.com/aws/aws-sdk-go/aws/session" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/vishvananda/netlink" + + "github.com/weaveworks/weave/common" + wnet "github.com/weaveworks/weave/net" + "github.com/weaveworks/weave/net/address" +) + +type AWSVPCTracker struct { + ec2 *ec2.EC2 + instanceID string // EC2 Instance ID + routeTableID string // VPC Route Table ID + linkIndex int // The weave bridge link index +} + +// NewAWSVPCTracker creates and initialises AWS VPC based tracker. +func NewAWSVPCTracker() (*AWSVPCTracker, error) { + var ( + err error + session = session.New() + t = &AWSVPCTracker{} + ) + + // Detect region and instance id + meta := ec2metadata.New(session) + t.instanceID, err = meta.GetMetadata("instance-id") + if err != nil { + return nil, fmt.Errorf("cannot detect instance-id: %s", err) + } + region, err := meta.Region() + if err != nil { + return nil, fmt.Errorf("cannot detect region: %s", err) + } + + t.ec2 = ec2.New(session, aws.NewConfig().WithRegion(region)) + + routeTableID, err := t.detectRouteTableID() + if err != nil { + return nil, err + } + t.routeTableID = *routeTableID + + // Detect Weave bridge link index + link, err := netlink.LinkByName(wnet.WeaveBridgeName) + if err != nil { + return nil, fmt.Errorf("cannot find \"%s\" interface: %s", wnet.WeaveBridgeName, err) + } + t.linkIndex = link.Attrs().Index + + t.infof("AWSVPC has been initialized on %s instance for %s route table at %s region", + t.instanceID, t.routeTableID, region) + + return t, nil +} + +// HandleUpdate method updates the AWS VPC and the host route tables. +func (t *AWSVPCTracker) HandleUpdate(prevRanges, currRanges []address.Range, local bool) error { + t.debugf("replacing %q by %q; local(%t)", prevRanges, currRanges, local) + + prev, curr := removeCommon(address.NewCIDRs(merge(prevRanges)), address.NewCIDRs(merge(currRanges))) + + // It might make sense to do the removal first and then add entries + // because of the 50 routes limit. However, in such case a container might + // not be reachable for a short period of time which is not a desired behavior. + + // Add new entries + for _, cidr := range curr { + cidrStr := cidr.String() + t.debugf("adding route %s to %s", cidrStr, t.instanceID) + _, err := t.createVPCRoute(cidrStr) + if err != nil { + return fmt.Errorf("createVPCRoutes failed: %s", err) + } + if local { + err = t.createHostRoute(cidrStr) + if err != nil { + return fmt.Errorf("createHostRoute failed: %s", err) + } + } + } + + // Remove obsolete entries + for _, cidr := range prev { + cidrStr := cidr.String() + t.debugf("removing %s route", cidrStr) + _, err := t.deleteVPCRoute(cidrStr) + if err != nil { + return fmt.Errorf("deleteVPCRoute failed: %s", err) + } + if local { + err = t.deleteHostRoute(cidrStr) + if err != nil { + return fmt.Errorf("deleteHostRoute failed: %s", err) + } + } + } + + return nil +} + +func (t *AWSVPCTracker) createVPCRoute(cidr string) (*ec2.CreateRouteOutput, error) { + route := &ec2.CreateRouteInput{ + RouteTableId: &t.routeTableID, + InstanceId: &t.instanceID, + DestinationCidrBlock: &cidr, + } + return t.ec2.CreateRoute(route) +} + +func (t *AWSVPCTracker) createHostRoute(cidr string) error { + dst, err := parseCIDR(cidr) + if err != nil { + return err + } + route := &netlink.Route{ + LinkIndex: t.linkIndex, + Dst: dst, + Scope: netlink.SCOPE_LINK, + } + return netlink.RouteAdd(route) +} + +func (t *AWSVPCTracker) deleteVPCRoute(cidr string) (*ec2.DeleteRouteOutput, error) { + route := &ec2.DeleteRouteInput{ + RouteTableId: &t.routeTableID, + DestinationCidrBlock: &cidr, + } + return t.ec2.DeleteRoute(route) +} + +func (t *AWSVPCTracker) deleteHostRoute(cidr string) error { + dst, err := parseCIDR(cidr) + if err != nil { + return err + } + route := &netlink.Route{ + LinkIndex: t.linkIndex, + Dst: dst, + Scope: netlink.SCOPE_LINK, + } + return netlink.RouteDel(route) +} + +// detectRouteTableID detects AWS VPC Route Table ID of the given tracker instance. +func (t *AWSVPCTracker) detectRouteTableID() (*string, error) { + instancesParams := &ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String(t.instanceID)}, + } + instancesResp, err := t.ec2.DescribeInstances(instancesParams) + if err != nil { + return nil, fmt.Errorf("DescribeInstances failed: %s", err) + } + if len(instancesResp.Reservations) == 0 || + len(instancesResp.Reservations[0].Instances) == 0 { + return nil, fmt.Errorf("cannot find %s instance within reservations", t.instanceID) + } + vpcID := instancesResp.Reservations[0].Instances[0].VpcId + subnetID := instancesResp.Reservations[0].Instances[0].SubnetId + + // First try to find a routing table associated with the subnet of the instance + tablesParams := &ec2.DescribeRouteTablesInput{ + Filters: []*ec2.Filter{ + { + Name: aws.String("association.subnet-id"), + Values: []*string{subnetID}, + }, + }, + } + tablesResp, err := t.ec2.DescribeRouteTables(tablesParams) + if err != nil { + return nil, fmt.Errorf("DescribeRouteTables failed: %s", err) + } + if len(tablesResp.RouteTables) != 0 { + return tablesResp.RouteTables[0].RouteTableId, nil + } + // Fallback to the default routing table + tablesParams = &ec2.DescribeRouteTablesInput{ + Filters: []*ec2.Filter{ + { + Name: aws.String("association.main"), + Values: []*string{aws.String("true")}, + }, + { + Name: aws.String("vpc-id"), + Values: []*string{vpcID}, + }, + }, + } + tablesResp, err = t.ec2.DescribeRouteTables(tablesParams) + if err != nil { + return nil, fmt.Errorf("DescribeRouteTables failed: %s", err) + } + if len(tablesResp.RouteTables) != 0 { + return tablesResp.RouteTables[0].RouteTableId, nil + } + + return nil, fmt.Errorf("cannot find routetable for %s instance", t.instanceID) +} + +func (t *AWSVPCTracker) debugf(fmt string, args ...interface{}) { + common.Log.Debugf("[tracker] "+fmt, args...) +} + +func (t *AWSVPCTracker) infof(fmt string, args ...interface{}) { + common.Log.Infof("[tracker] "+fmt, args...) +} + +// Helpers + +// merge merges adjacent range entries. +// The given slice has to be sorted in increasing order. +func merge(r []address.Range) []address.Range { + var merged []address.Range + + for i := range r { + if prev := len(merged) - 1; prev >= 0 && merged[prev].End == r[i].Start { + merged[prev].End = r[i].End + } else { + merged = append(merged, r[i]) + } + } + + return merged +} + +// removeCommon filters out CIDR ranges which are contained in both a and b slices. +// Both slices have to be sorted in increasing order. +func removeCommon(a, b []address.CIDR) (newA, newB []address.CIDR) { + i, j := 0, 0 + + for i < len(a) && j < len(b) { + switch { + case a[i].Start() < b[j].Start() || a[i].End() < b[j].End(): + newA = append(newA, a[i]) + i++ + case a[i].Start() > b[j].Start() || a[i].End() > b[j].End(): + newB = append(newB, b[j]) + j++ + default: + i++ + j++ + } + + } + newA = append(newA, a[i:]...) + newB = append(newB, b[j:]...) + + return +} + +func parseCIDR(cidr string) (*net.IPNet, error) { + ip, ipnet, err := net.ParseCIDR(cidr) + if err != nil { + return nil, err + } + ipnet.IP = ip + + return ipnet, nil +} diff --git a/ipam/tracker/awsvpc_test.go b/ipam/tracker/awsvpc_test.go new file mode 100644 index 00000000..b99500a1 --- /dev/null +++ b/ipam/tracker/awsvpc_test.go @@ -0,0 +1,52 @@ +package tracker + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/weaveworks/weave/net/address" +) + +var ( + r0to127 = cidr("10.0.0.0", "10.0.0.127") + r128to255 = cidr("10.0.0.128", "10.0.0.255") + r0to255 = cidr("10.0.0.0", "10.0.0.255") + r1dot0to255 = cidr("10.0.1.0", "10.0.1.255") + r2dot0to255 = cidr("10.0.2.0", "10.0.2.255") +) + +func TestRemoveCommon(t *testing.T) { + a := []address.CIDR{r0to127, r1dot0to255} + b := []address.CIDR{r1dot0to255, r2dot0to255} + newA, newB := removeCommon(a, b) + require.Equal(t, []address.CIDR{r0to127}, newA) + require.Equal(t, []address.CIDR{r2dot0to255}, newB) +} + +func TestMerge(t *testing.T) { + ranges := []address.Range{ + r0to127.Range(), + r128to255.Range(), + r2dot0to255.Range(), + } + require.Equal(t, []address.Range{r0to255.Range(), r2dot0to255.Range()}, merge(ranges)) +} + +// Helper + +// TODO(mp) DRY with helpers of other tests. + +func ip(s string) address.Address { + addr, _ := address.ParseIP(s) + return addr +} + +// [start; end] +func cidr(start, end string) address.CIDR { + c := address.Range{Start: ip(start), End: ip(end) + 1}.CIDRs() + if len(c) != 1 { + panic("invalid cidr") + } + return c[0] +} diff --git a/net/address/address.go b/net/address/address.go index 6252adcd..d5fb8e00 100644 --- a/net/address/address.go +++ b/net/address/address.go @@ -125,6 +125,22 @@ func ParseCIDR(s string) (CIDR, error) { } } +func NewCIDRs(ranges []Range) (cidrs []CIDR) { + for _, r := range ranges { + cidrs = append(cidrs, r.CIDRs()...) + } + return cidrs +} + +func (cidr CIDR) Start() Address { + return cidr.Addr +} + +// cidr = [Start; End) +func (cidr CIDR) End() Address { + return cidr.Range().End +} + func (cidr CIDR) IsSubnet() bool { mask := cidr.Size() - 1 return Offset(cidr.Addr)&mask == 0 diff --git a/net/address/address_test.go b/net/address/address_test.go index 326a5b2e..8718fdb6 100644 --- a/net/address/address_test.go +++ b/net/address/address_test.go @@ -61,3 +61,9 @@ func TestCIDRs(t *testing.T) { []CIDR{cidr("10.0.0.1/32"), cidr("10.0.0.2/31"), cidr("10.0.0.4/30"), cidr("10.0.0.8/32")}, r.CIDRs()) } + +func TestCIDRStartAndEnd(t *testing.T) { + cidr, _ := ParseCIDR("10.0.0.0/24") + require.Equal(t, ip("10.0.0.0"), cidr.Start(), "") + require.Equal(t, ip("10.0.1.0"), cidr.End(), "") +} diff --git a/vendor/github.com/aws/aws-sdk-go b/vendor/github.com/aws/aws-sdk-go new file mode 160000 index 00000000..e4be72e5 --- /dev/null +++ b/vendor/github.com/aws/aws-sdk-go @@ -0,0 +1 @@ +Subproject commit e4be72e584418dc0ff2af22ae247095143a7f258