Adds Kinesis Stream Consumer resource and data source

aws/data_source_aws_kinesis_stream_consumer.go

@@ -0,0 +1,60 @@
+package aws
+
+import (
+	"github.com/hashicorp/terraform/helper/schema"
+)
+
+func dataSourceAwsKinesisStreamConsumer() *schema.Resource {
+	return &schema.Resource{
+		Read: dataSourceAwsKinesisStreamRead,
+
+		Schema: map[string]*schema.Schema{
+			"name": {
+				Type:     schema.TypeString,
+				Required: true,
+			},
+
+			"stream_arn": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"arn": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"creation_timestamp": {
+				Type:     schema.TypeInt,
+				Computed: true,
+			},
+
+			"status": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+		},
+	}
+}
+
+func dataSourceAwsKinesisStreamConsumerRead(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).kinesisconn
+	cn := d.Get("name").(string)
+
+	state, err := readKinesisStreamConsumerState(conn, cn)
+	if err != nil {
+		return err
+	}
+	d.SetId(state.arn)
+	d.Set("arn", state.arn)
+	d.Set("name", cn)
+	d.Set("stream_arn", state.streamArn)
+	d.Set("status", state.status)
+	d.Set("creation_timestamp", state.creationTimestamp)
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

aws/data_source_aws_kinesis_stream_consumer_test.go

@@ -0,0 +1,47 @@
+package aws
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/kinesis"
+	"github.com/hashicorp/terraform/helper/acctest"
+	"github.com/hashicorp/terraform/helper/resource"
+)
+
+func TestAccAWSKinesisStreamConsumerDataSource(t *testing.T) {
+	var stream kinesis.StreamDescription
+
+	cn := fmt.Sprintf("terraform-kinesis-stream-consumer-test-%d", acctest.RandInt())
+	config := fmt.Sprintf(testAccCheckAwsKinesisStreamConsumerDataSourceConfig, cn)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckKinesisStreamConsumerDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: config,
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckKinesisStreamExists("aws_kinesis_stream_consumer.test_stream_consumer", &stream),
+					resource.TestCheckResourceAttrSet("data.aws_kinesis_stream_consumer.test_stream_consumer", "arn"),
+					resource.TestCheckResourceAttrSet("data.aws_kinesis_stream_consumer.test_stream_consumer", "stream_arn"),
+					resource.TestCheckResourceAttr("data.aws_kinesis_stream_consumer.test_stream_consumer", "name", cn),
+					resource.TestCheckResourceAttr("data.aws_kinesis_stream_consumer.test_stream_consumer", "status", "ACTIVE"),
+					resource.TestCheckResourceAttrSet("data.aws_kinesis_stream_consumer.test_stream_consumer", "creation_timestamp"),
+				),
+			},
+		},
+	})
+}
+
+var testAccCheckAwsKinesisStreamConsumerDataSourceConfig = `
+resource "aws_kinesis_stream_consumer" "test_stream_consumer" {
+	name = "%s"
+	stream_arn = "${aws_kinesis_stream.stream.arn}"
+}
+
+data "aws_kinesis_stream_consumer" "test_stream_consumer" {
+	name = "${aws_kinesis_stream.test_stream_consumer.name}"
+}
+`

aws/provider.go

@@ -224,6 +224,7 @@ func Provider() terraform.ResourceProvider {
 			"aws_instances":                        dataSourceAwsInstances(),
 			"aws_ip_ranges":                        dataSourceAwsIPRanges(),
 			"aws_kinesis_stream":                   dataSourceAwsKinesisStream(),
+			"aws_kinesis_stream_consumer":          dataSourceAwsKinesisStreamConsumer(),
 			"aws_kms_alias":                        dataSourceAwsKmsAlias(),
 			"aws_kms_ciphertext":                   dataSourceAwsKmsCiphertext(),
 			"aws_kms_key":                          dataSourceAwsKmsKey(),
@@ -497,6 +498,7 @@ func Provider() terraform.ResourceProvider {
 			"aws_key_pair":                                     resourceAwsKeyPair(),
 			"aws_kinesis_firehose_delivery_stream":             resourceAwsKinesisFirehoseDeliveryStream(),
 			"aws_kinesis_stream":                               resourceAwsKinesisStream(),
+			"aws_kinesis_stream_consumer":                      resourceAwsKinesisStreamConsumer(),
 			"aws_kinesis_analytics_application":                resourceAwsKinesisAnalyticsApplication(),
 			"aws_kms_alias":                                    resourceAwsKmsAlias(),
 			"aws_kms_grant":                                    resourceAwsKmsGrant(),

aws/resource_aws_kinesis_stream_consumer.go

@@ -0,0 +1,184 @@
+package aws
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
+	"github.com/aws/aws-sdk-go/service/kinesis"
+	"github.com/hashicorp/terraform/helper/resource"
+	"github.com/hashicorp/terraform/helper/schema"
+)
+
+func resourceAwsKinesisStreamConsumer() *schema.Resource {
+	return &schema.Resource{
+		Create: resourceAwsKinesisStreamConsumerCreate,
+		Read:   resourceAwsKinesisStreamConsumerRead,
+		Delete: resourceAwsKinesisStreamConsumerDelete,
+		Importer: &schema.ResourceImporter{
+			State: resourceAwsKinesisStreamConsumerImport,
+		},
+
+		Timeouts: &schema.ResourceTimeout{
+			Create: schema.DefaultTimeout(5 * time.Minute),
+			Delete: schema.DefaultTimeout(120 * time.Minute),
+		},
+
+		Schema: map[string]*schema.Schema{
+			"name": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"stream_arn": {
+				Type:     schema.TypeString,
+				Required: true,
+			},
+
+			"arn": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+			},
+		},
+	}
+}
+
+func resourceAwsKinesisStreamConsumerImport(
+	d *schema.ResourceData, meta interface{}) ([]*schema.ResourceData, error) {
+	d.Set("name", d.Id())
+	return []*schema.ResourceData{d}, nil
+}
+
+func resourceAwsKinesisStreamConsumerCreate(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).kinesisconn
+	cn := d.Get("name").(string)
+	sa := d.Get("stream_arn").(string)
+	createOpts := &kinesis.RegisterStreamConsumerInput{
+		ConsumerName: aws.String(cn),
+		StreamARN:    aws.String(sa),
+	}
+
+	_, err := conn.RegisterStreamConsumer(createOpts)
+	if err != nil {
+		return fmt.Errorf("Unable to create stream consumer: %s", err)
+	}
+
+	// No error, wait for ACTIVE state
+	stateConf := &resource.StateChangeConf{
+		Pending:    []string{"CREATING"},
+		Target:     []string{"ACTIVE"},
+		Refresh:    streamConsumerStateRefreshFunc(conn, cn),
+		Timeout:    d.Timeout(schema.TimeoutCreate),
+		Delay:      10 * time.Second,
+		MinTimeout: 3 * time.Second,
+	}
+
+	streamRaw, err := stateConf.WaitForState()
+	if err != nil {
+		return fmt.Errorf(
+			"Error waiting for Kinesis Stream Consumer (%s) to become active: %s",
+			cn, err)
+	}
+
+	s := streamRaw.(*kinesisStreamConsumerState)
+
+	d.SetId(s.arn)
+	d.Set("arn", s.arn)
+
+	return resourceAwsKinesisStreamConsumerRead(d, meta)
+}
+
+func resourceAwsKinesisStreamConsumerRead(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).kinesisconn
+	cn := d.Get("name").(string)
+
+	state, err := readKinesisStreamConsumerState(conn, cn)
+	if err != nil {
+		if awsErr, ok := err.(awserr.Error); ok {
+			if awsErr.Code() == "ResourceNotFoundException" {
+				d.SetId("")
+				return nil
+			}
+			return fmt.Errorf("Error reading Kinesis Stream Consumer: \"%s\", code: \"%s\"", awsErr.Message(), awsErr.Code())
+		}
+		return err
+
+	}
+	d.SetId(state.arn)
+	d.Set("arn", state.arn)
+
+	return nil
+}
+
+func resourceAwsKinesisStreamConsumerDelete(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).kinesisconn
+	cn := d.Get("name").(string)
+
+	_, err := conn.DeregisterStreamConsumer(&kinesis.DeregisterStreamConsumerInput{
+		ConsumerName: aws.String(cn),
+	})
+	if err != nil {
+		return err
+	}
+
+	stateConf := &resource.StateChangeConf{
+		Pending:    []string{"DELETING"},
+		Target:     []string{"DESTROYED"},
+		Refresh:    streamConsumerStateRefreshFunc(conn, cn),
+		Timeout:    d.Timeout(schema.TimeoutDelete),
+		Delay:      10 * time.Second,
+		MinTimeout: 3 * time.Second,
+	}
+
+	_, err = stateConf.WaitForState()
+	if err != nil {
+		return fmt.Errorf(
+			"Error waiting for Stream Consumer (%s) to be destroyed: %s",
+			cn, err)
+	}
+
+	return nil
+}
+
+type kinesisStreamConsumerState struct {
+	arn               string
+	streamArn         string
+	creationTimestamp int64
+	status            string
+}
+
+func readKinesisStreamConsumerState(conn *kinesis.Kinesis, cn string) (*kinesisStreamConsumerState, error) {
+	describeOpts := &kinesis.DescribeStreamConsumerInput{
+		ConsumerName: aws.String(cn),
+	}
+
+	state := &kinesisStreamConsumerState{}
+	respDescribe, err := conn.DescribeStreamConsumer(describeOpts)
+
+	state.arn = aws.StringValue(respDescribe.ConsumerDescription.ConsumerARN)
+	state.streamArn = aws.StringValue(respDescribe.ConsumerDescription.StreamARN)
+	state.creationTimestamp = aws.TimeValue(respDescribe.ConsumerDescription.ConsumerCreationTimestamp).Unix()
+	state.status = aws.StringValue(respDescribe.ConsumerDescription.ConsumerStatus)
+
+	return state, err
+}
+
+func streamConsumerStateRefreshFunc(conn *kinesis.Kinesis, cn string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		state, err := readKinesisStreamConsumerState(conn, cn)
+		if err != nil {
+			if awsErr, ok := err.(awserr.Error); ok {
+				if awsErr.Code() == "ResourceNotFoundException" {
+					return 42, "DESTROYED", nil
+				}
+				return nil, awsErr.Code(), err
+			}
+			return nil, "failed", err
+		}
+
+		return state, state.status, nil
+	}
+}