Basic support of SQS. #32

CHANGELOG.md

@@ -7,6 +7,7 @@ As model/relations for resources may evolve, if you have any issues with models
 - [#6](https://github.com/wallix/awless/issues/6): Create Linux installer shell script: `curl https://raw.githubusercontent.com/wallix/awless/master/getawless.sh | bash`
 - Better help menus by splitting one-liner template commands from general commands
 - [#32](https://github.com/wallix/awless/issues/32): Basic support of [SNS](https://aws.amazon.com/sns/) (CRUD for topics and subscriptions)
+- [#32](https://github.com/wallix/awless/issues/32): Basic support of [SQS](https://aws.amazon.com/sqs/) (CRUD for queues)
 - Run template: better dialog and remove noisy info
 - Template validation: notify on unexpected params; check names unicity against local graph
 - Log contextual error instead of hard failure when user has no rights to sync a service

cloud/aws/api.go

@@ -23,10 +23,12 @@ import (
 	"sync"
 
 	awssdk "github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/endpoints"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/iam"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/aws/aws-sdk-go/service/sqs"
 	"github.com/aws/aws-sdk-go/service/sts"
 	"github.com/aws/aws-sdk-go/service/sts/stsiface"
 	"github.com/wallix/awless/graph"
@@ -155,6 +157,8 @@ func (s *Access) fetch_all_user_graph() (*graph.Graph, []*iam.UserDetail, error)
 	return g, userDetails, nil
 }
 
+// STORAGE
+
 func (s *Storage) fetch_all_bucket_graph() (*graph.Graph, []*s3.Bucket, error) {
 	g := graph.NewGraph()
 	var buckets []*s3.Bucket
@@ -293,3 +297,53 @@ func (s *Storage) foreach_bucket_parallel(f func(b *s3.Bucket) error) error {
 
 	return nil
 }
+
+// QUEUE
+
+func (s *Queue) fetch_all_queue_graph() (*graph.Graph, []*string, error) {
+	g := graph.NewGraph()
+	var cloudResources []*string
+	out, err := s.ListQueues(&sqs.ListQueuesInput{})
+	if err != nil {
+		return nil, cloudResources, err
+	}
+	errc := make(chan error)
+	var wg sync.WaitGroup
+
+	for _, output := range out.QueueUrls {
+		cloudResources = append(cloudResources, output)
+		wg.Add(1)
+		go func(url *string) {
+			defer wg.Done()
+			res := graph.InitResource(awssdk.StringValue(url), graph.Queue)
+			res.Properties["Id"] = awssdk.StringValue(url)
+			attrs, err := s.GetQueueAttributes(&sqs.GetQueueAttributesInput{AttributeNames: []*string{awssdk.String("All")}, QueueUrl: url})
+			if e, ok := err.(awserr.RequestFailure); ok && (e.Code() == sqs.ErrCodeQueueDoesNotExist || e.Code() == sqs.ErrCodeQueueDeletedRecently) {
+				return
+			}
+			if err != nil {
+				errc <- err
+				return
+			}
+			for k, v := range attrs.Attributes {
+				res.Properties[k] = awssdk.StringValue(v)
+			}
+			g.AddResource(res)
+		}(output)
+
+	}
+
+	go func() {
+		wg.Wait()
+		close(errc)
+	}()
+
+	for err := range errc {
+		if err != nil {
+			return g, cloudResources, err
+		}
+	}
+
+	return g, cloudResources, nil
+
+}

cloud/aws/definitions/definitions.go

@@ -38,40 +38,47 @@ var Services = []service{
 		Name: "infra",
 		Api:  "ec2",
 		Fetchers: []fetcher{
-			{ResourceType: graph.Instance.String(), AWSType: "Instance", ApiMethod: "DescribeInstancesPages", Input: "DescribeInstancesInput{}", Output: "DescribeInstancesOutput", OutputsExtractor: "Instances", OutputsContainers: "Reservations", Multipage: true},
-			{ResourceType: graph.Subnet.String(), AWSType: "Subnet", ApiMethod: "DescribeSubnets", Input: "DescribeSubnetsInput{}", Output: "DescribeSubnetsOutput", OutputsExtractor: "Subnets"},
-			{ResourceType: graph.Vpc.String(), AWSType: "Vpc", ApiMethod: "DescribeVpcs", Input: "DescribeVpcsInput{}", Output: "DescribeVpcsOutput", OutputsExtractor: "Vpcs"},
-			{ResourceType: graph.Keypair.String(), AWSType: "KeyPairInfo", ApiMethod: "DescribeKeyPairs", Input: "DescribeKeyPairsInput{}", Output: "DescribeKeyPairsOutput", OutputsExtractor: "KeyPairs"},
-			{ResourceType: graph.SecurityGroup.String(), AWSType: "SecurityGroup", ApiMethod: "DescribeSecurityGroups", Input: "DescribeSecurityGroupsInput{}", Output: "DescribeSecurityGroupsOutput", OutputsExtractor: "SecurityGroups"},
-			{ResourceType: graph.Volume.String(), AWSType: "Volume", ApiMethod: "DescribeVolumesPages", Input: "DescribeVolumesInput{}", Output: "DescribeVolumesOutput", OutputsExtractor: "Volumes", Multipage: true},
-			{ResourceType: graph.InternetGateway.String(), AWSType: "InternetGateway", ApiMethod: "DescribeInternetGateways", Input: "DescribeInternetGatewaysInput{}", Output: "DescribeInternetGatewaysOutput", OutputsExtractor: "InternetGateways"},
-			{ResourceType: graph.RouteTable.String(), AWSType: "RouteTable", ApiMethod: "DescribeRouteTables", Input: "DescribeRouteTablesInput{}", Output: "DescribeRouteTablesOutput", OutputsExtractor: "RouteTables"},
+			{ResourceType: graph.Instance.String(), AWSType: "ec2.Instance", ApiMethod: "DescribeInstancesPages", Input: "DescribeInstancesInput{}", Output: "DescribeInstancesOutput", OutputsExtractor: "Instances", OutputsContainers: "Reservations", Multipage: true},
+			{ResourceType: graph.Subnet.String(), AWSType: "ec2.Subnet", ApiMethod: "DescribeSubnets", Input: "DescribeSubnetsInput{}", Output: "DescribeSubnetsOutput", OutputsExtractor: "Subnets"},
+			{ResourceType: graph.Vpc.String(), AWSType: "ec2.Vpc", ApiMethod: "DescribeVpcs", Input: "DescribeVpcsInput{}", Output: "DescribeVpcsOutput", OutputsExtractor: "Vpcs"},
+			{ResourceType: graph.Keypair.String(), AWSType: "ec2.KeyPairInfo", ApiMethod: "DescribeKeyPairs", Input: "DescribeKeyPairsInput{}", Output: "DescribeKeyPairsOutput", OutputsExtractor: "KeyPairs"},
+			{ResourceType: graph.SecurityGroup.String(), AWSType: "ec2.SecurityGroup", ApiMethod: "DescribeSecurityGroups", Input: "DescribeSecurityGroupsInput{}", Output: "DescribeSecurityGroupsOutput", OutputsExtractor: "SecurityGroups"},
+			{ResourceType: graph.Volume.String(), AWSType: "ec2.Volume", ApiMethod: "DescribeVolumesPages", Input: "DescribeVolumesInput{}", Output: "DescribeVolumesOutput", OutputsExtractor: "Volumes", Multipage: true},
+			{ResourceType: graph.InternetGateway.String(), AWSType: "ec2.InternetGateway", ApiMethod: "DescribeInternetGateways", Input: "DescribeInternetGatewaysInput{}", Output: "DescribeInternetGatewaysOutput", OutputsExtractor: "InternetGateways"},
+			{ResourceType: graph.RouteTable.String(), AWSType: "ec2.RouteTable", ApiMethod: "DescribeRouteTables", Input: "DescribeRouteTablesInput{}", Output: "DescribeRouteTablesOutput", OutputsExtractor: "RouteTables"},
 		},
 	},
 	{
 		Name: "access",
 		Api:  "iam",
 		Fetchers: []fetcher{
-			{ResourceType: graph.User.String(), AWSType: "UserDetail", ManualFetcher: true},
-			{ResourceType: graph.Group.String(), AWSType: "GroupDetail", ApiMethod: "GetAccountAuthorizationDetails", Input: "GetAccountAuthorizationDetailsInput{Filter: []*string{awssdk.String(iam.EntityTypeGroup)}}", Output: "GetAccountAuthorizationDetailsOutput", OutputsExtractor: "GroupDetailList"},
-			{ResourceType: graph.Role.String(), AWSType: "RoleDetail", ApiMethod: "GetAccountAuthorizationDetails", Input: "GetAccountAuthorizationDetailsInput{Filter: []*string{awssdk.String(iam.EntityTypeRole)}}", Output: "GetAccountAuthorizationDetailsOutput", OutputsExtractor: "RoleDetailList"},
-			{ResourceType: graph.Policy.String(), AWSType: "Policy", ApiMethod: "ListPolicies", Input: "ListPoliciesInput{OnlyAttached: awssdk.Bool(true)}", Output: "ListPoliciesOutput", OutputsExtractor: "Policies"},
+			{ResourceType: graph.User.String(), AWSType: "iam.UserDetail", ManualFetcher: true},
+			{ResourceType: graph.Group.String(), AWSType: "iam.GroupDetail", ApiMethod: "GetAccountAuthorizationDetails", Input: "GetAccountAuthorizationDetailsInput{Filter: []*string{awssdk.String(iam.EntityTypeGroup)}}", Output: "GetAccountAuthorizationDetailsOutput", OutputsExtractor: "GroupDetailList"},
+			{ResourceType: graph.Role.String(), AWSType: "iam.RoleDetail", ApiMethod: "GetAccountAuthorizationDetails", Input: "GetAccountAuthorizationDetailsInput{Filter: []*string{awssdk.String(iam.EntityTypeRole)}}", Output: "GetAccountAuthorizationDetailsOutput", OutputsExtractor: "RoleDetailList"},
+			{ResourceType: graph.Policy.String(), AWSType: "iam.Policy", ApiMethod: "ListPolicies", Input: "ListPoliciesInput{OnlyAttached: awssdk.Bool(true)}", Output: "ListPoliciesOutput", OutputsExtractor: "Policies"},
 		},
 	},
 	{
 		Name: "storage",
 		Api:  "s3",
 		Fetchers: []fetcher{
-			{ResourceType: graph.Bucket.String(), AWSType: "Bucket", ManualFetcher: true},
-			{ResourceType: graph.Object.String(), AWSType: "Object", ManualFetcher: true},
+			{ResourceType: graph.Bucket.String(), AWSType: "s3.Bucket", ManualFetcher: true},
+			{ResourceType: graph.Object.String(), AWSType: "s3.Object", ManualFetcher: true},
 		},
 	},
 	{
 		Name: "notification",
 		Api:  "sns",
 		Fetchers: []fetcher{
-			{ResourceType: graph.Subscription.String(), AWSType: "Subscription", ApiMethod: "ListSubscriptionsPages", Input: "ListSubscriptionsInput{}", Output: "ListSubscriptionsOutput", OutputsExtractor: "Subscriptions", Multipage: true},
-			{ResourceType: graph.Topic.String(), AWSType: "Topic", ApiMethod: "ListTopicsPages", Input: "ListTopicsInput{}", Output: "ListTopicsOutput", OutputsExtractor: "Topics", Multipage: true},
+			{ResourceType: graph.Subscription.String(), AWSType: "sns.Subscription", ApiMethod: "ListSubscriptionsPages", Input: "ListSubscriptionsInput{}", Output: "ListSubscriptionsOutput", OutputsExtractor: "Subscriptions", Multipage: true},
+			{ResourceType: graph.Topic.String(), AWSType: "sns.Topic", ApiMethod: "ListTopicsPages", Input: "ListTopicsInput{}", Output: "ListTopicsOutput", OutputsExtractor: "Topics", Multipage: true},
+		},
+	},
+	{
+		Name: "queue",
+		Api:  "sqs",
+		Fetchers: []fetcher{
+			{ResourceType: graph.Queue.String(), AWSType: "string", ManualFetcher: true},
 		},
 	},
 }

cloud/aws/gen_api.go

@@ -35,6 +35,8 @@ import (
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
 	"github.com/aws/aws-sdk-go/service/sns"
 	"github.com/aws/aws-sdk-go/service/sns/snsiface"
+	"github.com/aws/aws-sdk-go/service/sqs"
+	"github.com/aws/aws-sdk-go/service/sqs/sqsiface"
 	"github.com/wallix/awless/cloud"
 	"github.com/wallix/awless/graph"
 )
@@ -44,6 +46,7 @@ func init() {
 	ServiceNames = append(ServiceNames, "access")
 	ServiceNames = append(ServiceNames, "storage")
 	ServiceNames = append(ServiceNames, "notification")
+	ServiceNames = append(ServiceNames, "queue")
 }
 
 var ServiceNames = []string{}
@@ -73,13 +76,17 @@ var ResourceTypesPerAPI = map[string][]string{
 		"subscription",
 		"topic",
 	},
+	"sqs": {
+		"queue",
+	},
 }
 
 var ServicePerAPI = map[string]string{
 	"ec2": "infra",
 	"iam": "access",
 	"s3":  "storage",
 	"sns": "notification",
+	"sqs": "queue",
 }
 
 var ServicePerResourceType = map[string]string{
@@ -99,6 +106,7 @@ var ServicePerResourceType = map[string]string{
 	"storageobject":   "storage",
 	"subscription":    "notification",
 	"topic":           "notification",
+	"queue":           "queue",
 }
 
 type Infra struct {
@@ -1199,3 +1207,116 @@ func (s *Notification) fetch_all_topic_graph() (*graph.Graph, []*sns.Topic, erro
 
 	return g, cloudResources, badResErr
 }
+
+type Queue struct {
+	once   oncer
+	region string
+	sqsiface.SQSAPI
+}
+
+func NewQueue(sess *session.Session) *Queue {
+	region := awssdk.StringValue(sess.Config.Region)
+	return &Queue{SQSAPI: sqs.New(sess), region: region}
+}
+
+func (s *Queue) Name() string {
+	return "queue"
+}
+
+func (s *Queue) Provider() string {
+	return "aws"
+}
+
+func (s *Queue) ProviderAPI() string {
+	return "sqs"
+}
+
+func (s *Queue) ProviderRunnableAPI() interface{} {
+	return s.SQSAPI
+}
+
+func (s *Queue) ResourceTypes() (all []string) {
+	all = append(all, "queue")
+	return
+}
+
+func (s *Queue) FetchResources() (*graph.Graph, error) {
+	g := graph.NewGraph()
+	regionN := graph.InitResource(s.region, graph.Region)
+	g.AddResource(regionN)
+	var queueList []*string
+
+	errc := make(chan error)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		var resGraph *graph.Graph
+		var err error
+		resGraph, queueList, err = s.fetch_all_queue_graph()
+		if err != nil {
+			errc <- err
+			return
+		}
+		g.AddGraph(resGraph)
+	}()
+
+	go func() {
+		wg.Wait()
+		close(errc)
+	}()
+
+	for err := range errc {
+		switch ee := err.(type) {
+		case awserr.RequestFailure:
+			switch ee.Message() {
+			case "Access Denied":
+				return g, cloud.ErrFetchAccessDenied
+			default:
+				return g, ee
+			}
+		case nil:
+			continue
+		default:
+			return g, ee
+		}
+	}
+
+	errc = make(chan error)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for _, r := range queueList {
+			for _, fn := range addParentsFns["queue"] {
+				err := fn(g, r)
+				if err != nil {
+					errc <- err
+					return
+				}
+			}
+		}
+	}()
+
+	go func() {
+		wg.Wait()
+		close(errc)
+	}()
+
+	for err := range errc {
+		if err != nil {
+			return g, err
+		}
+	}
+
+	return g, nil
+}
+
+func (s *Queue) FetchByType(t string) (*graph.Graph, error) {
+	switch t {
+	case "queue":
+		graph, _, err := s.fetch_all_queue_graph()
+		return graph, err
+	default:
+		return nil, fmt.Errorf("aws queue: unsupported fetch for type %s", t)
+	}
+}

cloud/aws/init.go

@@ -26,7 +26,7 @@ import (
 )
 
 var (
-	AccessService, InfraService, StorageService, NotificationService cloud.Service
+	AccessService, InfraService, StorageService, NotificationService, QueueService cloud.Service
 
 	SecuAPI Security
 )
@@ -62,11 +62,13 @@ func InitServices(region, profile string) error {
 	StorageService = NewStorage(sess)
 	SecuAPI = NewSecu(sess)
 	NotificationService = NewNotification(sess)
+	QueueService = NewQueue(sess)
 
 	cloud.ServiceRegistry[InfraService.Name()] = InfraService
 	cloud.ServiceRegistry[AccessService.Name()] = AccessService
 	cloud.ServiceRegistry[StorageService.Name()] = StorageService
 	cloud.ServiceRegistry[NotificationService.Name()] = NotificationService
+	cloud.ServiceRegistry[QueueService.Name()] = QueueService
 
 	return nil
 }

cloud/aws/model.go

@@ -138,7 +138,7 @@ var awsResourcesDef = map[graph.ResourceType]map[string]*propertyTransform{
 		"Size":         {name: "Size", transform: extractValueFn},
 		"Class":        {name: "StorageClass", transform: extractValueFn},
 	},
-	//SNS
+	//Notification
 	graph.Subscription: {
 		"Id":              {name: "Endpoint", transform: extractValueFn},
 		"Endpoint":        {name: "Endpoint", transform: extractValueFn},
@@ -151,4 +151,6 @@ var awsResourcesDef = map[graph.ResourceType]map[string]*propertyTransform{
 		"Id":       {name: "TopicArn", transform: extractValueFn},
 		"TopicArn": {name: "TopicArn", transform: extractValueFn},
 	},
+	//Queue
+	graph.Queue: {}, //Manually set
 }

cloud/generators/aws.go

@@ -163,7 +163,7 @@ func (s *{{ Title $service.Name }}) FetchResources() (*graph.Graph, error) {
 	g.AddResource(regionN)
 
 	{{- range $index, $fetcher := $service.Fetchers }}
-  var {{ $fetcher.ResourceType }}List []*{{ $service.Api }}.{{ $fetcher.AWSType }}
+  var {{ $fetcher.ResourceType }}List []*{{ $fetcher.AWSType }}
   {{- end }}
 
 	errc := make(chan error)
@@ -250,9 +250,9 @@ func (s *{{ Title $service.Name }}) FetchByType(t string) (*graph.Graph, error)
 
 {{ range $index, $fetcher := $service.Fetchers }}
 {{- if not $fetcher.ManualFetcher }}
-func (s *{{ Title $service.Name }}) fetch_all_{{ $fetcher.ResourceType }}_graph() (*graph.Graph, []*{{ $service.Api }}.{{ $fetcher.AWSType }}, error) {
+func (s *{{ Title $service.Name }}) fetch_all_{{ $fetcher.ResourceType }}_graph() (*graph.Graph, []*{{ $fetcher.AWSType }}, error) {
   g := graph.NewGraph()
-	var cloudResources []*{{ $service.Api }}.{{ $fetcher.AWSType }}
+	var cloudResources []*{{ $fetcher.AWSType }}
 	{{- if $fetcher.Multipage }}
 	var badResErr error
 	err := s.{{ $fetcher.ApiMethod }}(&{{ $service.Api }}.{{ $fetcher.Input }},

commands/inspect.go

@@ -70,7 +70,9 @@ var inspectCmd = &cobra.Command{
 			}
 
 			graphPerService, err := sync.DefaultSyncer.Sync(services...)
-			logger.Error(err)
+			if err != nil {
+				logger.Error(err)
+			}
 
 			for _, g := range graphPerService {
 				graphs = append(graphs, g)