Combine creating a volume and applying tags in one operation

The previous version forced us to create AWS IAM Policies that are too
permissive when dealing with volumes. That's because:

1. Volumes were created without tags that identifies the new resource as
managed by the cluster. So technically the resourse, at creation time,
is not owned by the cluster.

2. Tags were added to the volume making the resource now managed by the
cluster. The problem being that it could make ANY volume as managed by the
cluster. Thus allowing resources that aren't really part of the cluster,
or part of no cluster at all, to become a resource managed by the cluster.

By combining the operations we can both make the code simpler, since we
don't need to deal with deleting a volume in case we can't apply tags to
it, plus the security model gets a nice improvement.

pkg/cloudprovider/providers/aws/aws.go

@@ -2191,7 +2191,6 @@ func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (KubernetesVolumeID, er
 		return "", fmt.Errorf("invalid AWS VolumeType %q", volumeOptions.VolumeType)
 	}
 
-	// TODO: Should we tag this with the cluster id (so it gets deleted when the cluster does?)
 	request := &ec2.CreateVolumeInput{}
 	request.AvailabilityZone = aws.String(volumeOptions.AvailabilityZone)
 	request.Size = aws.Int64(int64(volumeOptions.CapacityGB))
@@ -2204,6 +2203,21 @@ func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (KubernetesVolumeID, er
 	if iops > 0 {
 		request.Iops = aws.Int64(iops)
 	}
+
+	tags := volumeOptions.Tags
+	tags = c.tagging.buildTags(ResourceLifecycleOwned, tags)
+
+	var tagList []*ec2.Tag
+	for k, v := range tags {
+		tagList = append(tagList, &ec2.Tag{
+			Key: aws.String(k), Value: aws.String(v),
+		})
+	}
+	request.TagSpecifications = append(request.TagSpecifications, &ec2.TagSpecification{
+		Tags:         tagList,
+		ResourceType: aws.String(ec2.ResourceTypeVolume),
+	})
+
 	response, err := c.ec2.CreateVolume(request)
 	if err != nil {
 		return "", err
@@ -2215,17 +2229,6 @@ func (c *Cloud) CreateDisk(volumeOptions *VolumeOptions) (KubernetesVolumeID, er
 	}
 	volumeName := KubernetesVolumeID("aws://" + aws.StringValue(response.AvailabilityZone) + "/" + string(awsID))
 
-	// apply tags
-	if err := c.tagging.createTags(c.ec2, string(awsID), ResourceLifecycleOwned, volumeOptions.Tags); err != nil {
-		// delete the volume and hope it succeeds
-		_, delerr := c.DeleteDisk(volumeName)
-		if delerr != nil {
-			// delete did not succeed, we have a stray volume!
-			return "", fmt.Errorf("error tagging volume %s, could not delete the volume: %q", volumeName, delerr)
-		}
-		return "", fmt.Errorf("error tagging volume %s: %q", volumeName, err)
-	}
-
 	// AWS has a bad habbit of reporting success when creating a volume with
 	// encryption keys that either don't exists or have wrong permissions.
 	// Such volume lives for couple of seconds and then it's silently deleted

pkg/cloudprovider/providers/aws/aws_test.go

@@ -68,6 +68,11 @@ func (m *MockedFakeEC2) DescribeSecurityGroups(request *ec2.DescribeSecurityGrou
 	return args.Get(0).([]*ec2.SecurityGroup), nil
 }
 
+func (m *MockedFakeEC2) CreateVolume(request *ec2.CreateVolumeInput) (*ec2.Volume, error) {
+	args := m.Called(request)
+	return args.Get(0).(*ec2.Volume), nil
+}
+
 type MockedFakeELB struct {
 	*FakeELB
 	mock.Mock
@@ -1393,6 +1398,38 @@ func TestFindSecurityGroupForInstanceMultipleTagged(t *testing.T) {
 	assert.Contains(t, err.Error(), "sg123(another_group)")
 }
 
+func TestCreateDisk(t *testing.T) {
+	awsServices := newMockedFakeAWSServices(TestClusterID)
+	c, _ := newAWSCloud(CloudConfig{}, awsServices)
+
+	volumeOptions := &VolumeOptions{
+		AvailabilityZone: "us-east-1a",
+		CapacityGB:       10,
+	}
+	request := &ec2.CreateVolumeInput{
+		AvailabilityZone: aws.String("us-east-1a"),
+		Encrypted:        aws.Bool(false),
+		VolumeType:       aws.String(DefaultVolumeType),
+		Size:             aws.Int64(10),
+		TagSpecifications: []*ec2.TagSpecification{
+			{ResourceType: aws.String(ec2.ResourceTypeVolume), Tags: []*ec2.Tag{
+				{Key: aws.String(TagNameKubernetesClusterLegacy), Value: aws.String(TestClusterID)},
+				{Key: aws.String(fmt.Sprintf("%s%s", TagNameKubernetesClusterPrefix, TestClusterID)), Value: aws.String(ResourceLifecycleOwned)},
+			}},
+		},
+	}
+	volume := &ec2.Volume{
+		AvailabilityZone: aws.String("us-east-1a"),
+		VolumeId:         aws.String("vol-volumeId0"),
+	}
+	awsServices.ec2.(*MockedFakeEC2).On("CreateVolume", request).Return(volume, nil)
+
+	volumeID, err := c.CreateDisk(volumeOptions)
+	assert.Nil(t, err, "Error creating disk: %v", err)
+	assert.Equal(t, volumeID, KubernetesVolumeID("aws://us-east-1a/vol-volumeId0"))
+	awsServices.ec2.(*MockedFakeEC2).AssertExpectations(t)
+}
+
 func newMockedFakeAWSServices(id string) *FakeAWSServices {
 	s := NewFakeAWSServices(id)
 	s.ec2 = &MockedFakeEC2{FakeEC2Impl: s.ec2.(*FakeEC2Impl)}