AWS: Filter by Cluster tag, rationalize EC2 abstraction

cluster/aws/util.sh

@@ -508,12 +508,13 @@ function kube-up {
 
   echo "Using VPC $VPC_ID"
 
-  SUBNET_ID=$($AWS_CMD describe-subnets | get_subnet_id $VPC_ID $ZONE)
+  SUBNET_ID=$($AWS_CMD describe-subnets --filters Name=tag:KubernetesCluster,Values=${CLUSTER_ID} | get_subnet_id $VPC_ID $ZONE)
   if [[ -z "$SUBNET_ID" ]]; then
     echo "Creating subnet."
     SUBNET_ID=$($AWS_CMD create-subnet --cidr-block $INTERNAL_IP_BASE.0/24 --vpc-id $VPC_ID --availability-zone ${ZONE} | json_val '["Subnet"]["SubnetId"]')
+    add-tag $SUBNET_ID KubernetesCluster ${CLUSTER_ID}
   else
-    EXISTING_CIDR=$($AWS_CMD describe-subnets | get_cidr $VPC_ID $ZONE)
+    EXISTING_CIDR=$($AWS_CMD describe-subnets --filters Name=tag:KubernetesCluster,Values=${CLUSTER_ID} | get_cidr $VPC_ID $ZONE)
     echo "Using existing CIDR $EXISTING_CIDR"
     INTERNAL_IP_BASE=${EXISTING_CIDR%.*}
     MASTER_INTERNAL_IP=${INTERNAL_IP_BASE}${MASTER_IP_SUFFIX}
@@ -863,6 +864,7 @@ function kube-down {
 
     subnet_ids=$($AWS_CMD --output text describe-subnets \
                           --filters Name=vpc-id,Values=${vpc_id} \
+                                    Name=tag:KubernetesCluster,Values=${CLUSTER_ID} \
                           --query Subnets[].SubnetId \
              | tr "\t" "\n")
     for subnet_id in ${subnet_ids}; do

pkg/cloudprovider/aws/aws_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/resource"
+	"github.com/golang/glog"
 )
 
 const TestClusterId = "clusterid.test"
@@ -234,23 +235,58 @@ type FakeEC2 struct {
 	aws *FakeAWSServices
 }
 
-func contains(haystack []string, needle string) bool {
+func contains(haystack []*string, needle string) bool {
 	for _, s := range haystack {
-		if needle == s {
+		// (deliberately panic if s == nil)
+		if needle == *s {
 			return true
 		}
 	}
 	return false
 }
 
-func (self *FakeEC2) Instances(instanceIds []string, filter *ec2InstanceFilter) (instances []*ec2.Instance, err error) {
+func instanceMatchesFilter(instance *ec2.Instance, filter *ec2.Filter) bool {
+	name := *filter.Name
+	if name == "private-dns-name" {
+		if instance.PrivateDNSName == nil {
+			return false
+		}
+		return contains(filter.Values, *instance.PrivateDNSName)
+	}
+	panic("Unknown filter name: " + name)
+}
+
+func (self *FakeEC2) DescribeInstances(request *ec2.DescribeInstancesInput) ([]*ec2.Instance, error) {
 	matches := []*ec2.Instance{}
 	for _, instance := range self.aws.instances {
-		if filter != nil && !filter.Matches(instance) {
-			continue
+		if request.InstanceIDs != nil {
+			if instance.InstanceID == nil {
+				glog.Warning("Instance with no instance id: ", instance)
+				continue
+			}
+
+			found := false
+			for _, instanceId := range request.InstanceIDs {
+				if *instanceId == *instance.InstanceID {
+					found = true
+					break
+				}
+			}
+			if !found {
+				continue
+			}
 		}
-		if instanceIds != nil && !contains(instanceIds, *instance.InstanceID) {
-			continue
+		if request.Filters != nil {
+			allMatch := true
+			for _, filter := range request.Filters {
+				if !instanceMatchesFilter(instance, filter) {
+					allMatch = false
+					break
+				}
+			}
+			if !allMatch {
+				continue
+			}
 		}
 		matches = append(matches, instance)
 	}
@@ -280,7 +316,7 @@ func (ec2 *FakeEC2) DetachVolume(request *ec2.DetachVolumeInput) (resp *ec2.Volu
 	panic("Not implemented")
 }
 
-func (ec2 *FakeEC2) Volumes(volumeIDs []string, filter *ec2.Filter) (resp *ec2.DescribeVolumesOutput, err error) {
+func (ec2 *FakeEC2) DescribeVolumes(request *ec2.DescribeVolumesInput) ([]*ec2.Volume, error) {
 	panic("Not implemented")
 }
 
@@ -292,7 +328,7 @@ func (ec2 *FakeEC2) DeleteVolume(volumeID string) (resp *ec2.DeleteVolumeOutput,
 	panic("Not implemented")
 }
 
-func (ec2 *FakeEC2) DescribeSecurityGroups(groupIds []string, filterName string, filterVpcId string) ([]*ec2.SecurityGroup, error) {
+func (ec2 *FakeEC2) DescribeSecurityGroups(request *ec2.DescribeSecurityGroupsInput) ([]*ec2.SecurityGroup, error) {
 	panic("Not implemented")
 }
 
@@ -304,11 +340,11 @@ func (ec2 *FakeEC2) AuthorizeSecurityGroupIngress(*ec2.AuthorizeSecurityGroupIng
 	panic("Not implemented")
 }
 
-func (ec2 *FakeEC2) DescribeVPCs(*ec2.DescribeVPCsInput) (*ec2.DescribeVPCsOutput, error) {
+func (ec2 *FakeEC2) DescribeVPCs(*ec2.DescribeVPCsInput) ([]*ec2.VPC, error) {
 	panic("Not implemented")
 }
 
-func (ec2 *FakeEC2) DescribeSubnets(*ec2.DescribeSubnetsInput) (*ec2.DescribeSubnetsOutput, error) {
+func (ec2 *FakeEC2) DescribeSubnets(*ec2.DescribeSubnetsInput) ([]*ec2.Subnet, error) {
 	panic("Not implemented")
 }