Add vpc_id column to emr_clusters.go

client/mocks/builders_test.go

@@ -529,18 +529,39 @@ func buildEcrRepositoriesMock(t *testing.T, ctrl *gomock.Controller) client.Serv
 
 func buildEmrClusters(t *testing.T, ctrl *gomock.Controller) client.Services {
 	m := mocks.NewMockEmrClient(ctrl)
+	ec2m := mocks.NewMockEc2Client(ctrl)
 	l := emrTypes.ClusterSummary{}
 	err := faker.FakeData(&l)
 	if err != nil {
 		t.Fatal(err)
 	}
 
+	c := emrTypes.Cluster{}
+	err = faker.FakeDataSkipFields(&c, []string{"Configurations", "InstanceCollectionType", "RepoUpgradeOnBoot", "ScaleDownBehavior"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	s := ec2.DescribeSubnetsOutput{}
+	err = faker.FakeData(&s)
+	if err != nil {
+		t.Fatal(err)
+	}
+
 	m.EXPECT().ListClusters(gomock.Any(), gomock.Any(), gomock.Any()).Return(
 		&emr.ListClustersOutput{
 			Clusters: []emrTypes.ClusterSummary{l},
 		}, nil)
+
+	m.EXPECT().DescribeCluster(gomock.Any(), gomock.Any(), gomock.Any()).Return(&emr.DescribeClusterOutput{
+		Cluster: &c,
+	}, nil)
+
+	ec2m.EXPECT().DescribeSubnets(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s, nil)
+
 	return client.Services{
 		EMR: m,
+		EC2: ec2m,
 	}
 }
 

client/services.go

@@ -206,6 +206,7 @@ type ElbV1Client interface {
 //go:generate mockgen -package=mocks -destination=./mocks/mock_emr.go . EmrClient
 type EmrClient interface {
 	ListClusters(ctx context.Context, params *emr.ListClustersInput, optFns ...func(*emr.Options)) (*emr.ListClustersOutput, error)
+	DescribeCluster(ctx context.Context, params *emr.DescribeClusterInput, optFns ...func(*emr.Options)) (*emr.DescribeClusterOutput, error)
 }
 
 //go:generate mockgen -package=mocks -destination=./mocks/mock_fsx.go . FsxClient

docs/tables/aws_emr_clusters.md

@@ -17,3 +17,4 @@ The summary description of the cluster.
 |status_timeline_creation_date_time|timestamp without time zone|The creation date and time of the cluster.|
 |status_timeline_end_date_time|timestamp without time zone|The date and time when the cluster was terminated.|
 |status_timeline_ready_date_time|timestamp without time zone|The date and time when the cluster was ready to run steps.|
+|vpc_id|text|The cluster vpc id.|

go.mod

@@ -48,7 +48,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/wafv2 v1.5.1
 	github.com/aws/smithy-go v1.8.0
 	github.com/cloudquery/cq-provider-sdk v0.5.0
-	github.com/cloudquery/faker/v3 v3.7.4
+	github.com/cloudquery/faker/v3 v3.7.5
 	github.com/gocarina/gocsv v0.0.0-20210516172204-ca9e8a8ddea8
 	github.com/golang/mock v1.6.0
 	github.com/hashicorp/go-hclog v1.0.0

go.sum

@@ -285,8 +285,9 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58/go.mod h1:EOBUe0h4xcZ5GoxqC5SDxFQ8gwyZPKQoEzownBlhI80=
 github.com/cloudquery/cq-provider-sdk v0.5.0 h1:oSae/a4zMYLD7EIBzJs1ZzJFq+R/cY3djqHeWt15nXc=
 github.com/cloudquery/cq-provider-sdk v0.5.0/go.mod h1:R9dd0bLv9GM4KH+ckxRpvkzSYoWAE3Z0me2v2w5uCJA=
-github.com/cloudquery/faker/v3 v3.7.4 h1:cCcU3r0yHpS0gqKj9rRKAGS0/hY33fBxbqCNFtDD4ec=
 github.com/cloudquery/faker/v3 v3.7.4/go.mod h1:1b8WVG9Gh0T2hVo1a8dWeXfu0AhqSB6J/mmJaesqOeo=
+github.com/cloudquery/faker/v3 v3.7.5 h1:G7ANdEEcm8TvAAjIwNWSLrYK36CFCiSlrCqOTGCccL0=
+github.com/cloudquery/faker/v3 v3.7.5/go.mod h1:1b8WVG9Gh0T2hVo1a8dWeXfu0AhqSB6J/mmJaesqOeo=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=

resources/emr_clusters.go

@@ -2,9 +2,12 @@ package resources
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/ec2"
 	"github.com/aws/aws-sdk-go-v2/service/emr"
+	"github.com/aws/aws-sdk-go-v2/service/emr/types"
 	"github.com/cloudquery/cq-provider-aws/client"
 	"github.com/cloudquery/cq-provider-sdk/provider/schema"
 )
@@ -94,6 +97,12 @@ func EmrClusters() *schema.Table {
 				Type:        schema.TypeTimestamp,
 				Resolver:    schema.PathResolver("Status.Timeline.ReadyDateTime"),
 			},
+			{
+				Name:        "vpc_id",
+				Description: "The cluster vpc id.",
+				Type:        schema.TypeString,
+				Resolver:    resolveEmrClustersVpcId,
+			},
 		},
 	}
 }
@@ -120,3 +129,37 @@ func fetchEmrClusters(ctx context.Context, meta schema.ClientMeta, parent *schem
 	}
 	return nil
 }
+
+func resolveEmrClustersVpcId(ctx context.Context, meta schema.ClientMeta, resource *schema.Resource, _ schema.Column) error {
+	c := meta.(*client.Client)
+	svc := c.Services().EMR
+	ec2Svc := c.Services().EC2
+
+	cluster := resource.Item.(types.ClusterSummary)
+
+	output, err := svc.DescribeCluster(ctx, &emr.DescribeClusterInput{ClusterId: cluster.Id})
+	if err != nil {
+		return err
+	}
+
+	if output.Cluster == nil || output.Cluster.Ec2InstanceAttributes == nil && output.Cluster.Ec2InstanceAttributes.Ec2SubnetId == nil {
+		return nil
+	}
+
+	subnetId := *output.Cluster.Ec2InstanceAttributes.Ec2SubnetId
+	subnetsOutput, err := ec2Svc.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{
+		SubnetIds: []string{subnetId},
+	})
+
+	// We can fetch cluster even if it's already terminated and not connects to vpc.
+	// If we can't fetch the subnet then we return nil
+	if err != nil {
+		return nil
+	}
+
+	subnets := subnetsOutput.Subnets
+	if len(subnets) != 1 {
+		return fmt.Errorf("expected only one subnet but got %d", len(subnets))
+	}
+	return resource.Set("vpc_id", subnets[0].VpcId)
+}

resources/migrations/5_v0.5.17.down.sql

@@ -48,3 +48,8 @@ ALTER TABLE IF EXISTS "aws_elbv2_load_balancer_availability_zone_addresses" ADD
 --aws_apigateway_domain_name_base_path_mappings
 ALTER TABLE IF EXISTS "aws_apigateway_domain_name_base_path_mappings" DROP CONSTRAINT IF EXISTS "aws_apigateway_domain_name_base_path_mappings_pk";
 ALTER TABLE IF EXISTS "aws_apigateway_domain_name_base_path_mappings" ADD CONSTRAINT "aws_apigateway_domain_name_base_path_mappings_pk" UNIQUE ("domain_name_cq_id", "rest_api_id");
+
+--aws_emr_clusters_vpc_id
+ALTER TABLE IF EXISTS "aws_emr_clusters"
+DROP
+COLUMN "vpc_id";

resources/migrations/5_v0.5.17.up.sql

@@ -42,5 +42,9 @@ ALTER TABLE IF EXISTS "aws_elbv2_load_balancer_availability_zone_addresses" ADD
 ALTER TABLE IF EXISTS "aws_apigateway_domain_name_base_path_mappings" DROP CONSTRAINT IF EXISTS "aws_apigateway_domain_name_base_path_mappings_pk";
 ALTER TABLE IF EXISTS "aws_apigateway_domain_name_base_path_mappings" ADD CONSTRAINT "aws_apigateway_domain_name_base_path_mappings_pk" UNIQUE ("domain_name_cq_id", "cq_id");
 
+--aws_emr_clusters_vpc_id
+ALTER TABLE IF EXISTS "aws_emr_clusters"
+ADD
+COLUMN "vpc_id";