provider/aws: AWS Kinesis Stream support

builtin/providers/aws/config.go

@@ -14,11 +14,12 @@ import (
  "github.com/awslabs/aws-sdk-go/service/elasticache"
  "github.com/awslabs/aws-sdk-go/service/elb"
  "github.com/awslabs/aws-sdk-go/service/iam"
+ "github.com/awslabs/aws-sdk-go/service/kinesis"
  "github.com/awslabs/aws-sdk-go/service/rds"
  "github.com/awslabs/aws-sdk-go/service/route53"
  "github.com/awslabs/aws-sdk-go/service/s3"
- "github.com/awslabs/aws-sdk-go/service/sqs"
  "github.com/awslabs/aws-sdk-go/service/sns"
+ "github.com/awslabs/aws-sdk-go/service/sqs"
 )
 
 type Config struct {
@@ -43,6 +44,7 @@ type AWSClient struct {
  region string
  rdsconn *rds.RDS
  iamconn *iam.IAM
+ kinesisconn *kinesis.Kinesis
  elasticacheconn *elasticache.ElastiCache
 }
 
@@ -100,6 +102,9 @@ func (c *Config) Client() (interface{}, error) {
  log.Println("[INFO] Initializing IAM Connection")
  client.iamconn = iam.New(awsConfig)
 
+ log.Println("[INFO] Initializing Kinesis Connection")
+ client.kinesisconn = kinesis.New(awsConfig)
+
  err := c.ValidateAccountId(client.iamconn)
  if err != nil {
  errs = append(errs, err)

builtin/providers/aws/provider.go

@@ -109,6 +109,7 @@ func Provider() terraform.ResourceProvider {
  "aws_instance": resourceAwsInstance(),
  "aws_internet_gateway": resourceAwsInternetGateway(),
  "aws_key_pair": resourceAwsKeyPair(),
+ "aws_kinesis_stream": resourceAwsKinesisStream(),
  "aws_launch_configuration": resourceAwsLaunchConfiguration(),
  "aws_lb_cookie_stickiness_policy": resourceAwsLBCookieStickinessPolicy(),
  "aws_main_route_table_association": resourceAwsMainRouteTableAssociation(),

builtin/providers/aws/resource_aws_kinesis_stream.go

@@ -0,0 +1,156 @@
+package aws
+
+import (
+ "fmt"
+ "time"
+
+ "github.com/awslabs/aws-sdk-go/aws"
+ "github.com/awslabs/aws-sdk-go/aws/awserr"
+ "github.com/awslabs/aws-sdk-go/service/kinesis"
+ "github.com/hashicorp/terraform/helper/resource"
+ "github.com/hashicorp/terraform/helper/schema"
+)
+
+func resourceAwsKinesisStream() *schema.Resource {
+ return &schema.Resource{
+ Create: resourceAwsKinesisStreamCreate,
+ Read: resourceAwsKinesisStreamRead,
+ Delete: resourceAwsKinesisStreamDelete,
+
+ Schema: map[string]*schema.Schema{
+ "name": &schema.Schema{
+ Type: schema.TypeString,
+ Required: true,
+ ForceNew: true,
+ },
+
+ "shard_count": &schema.Schema{
+ Type: schema.TypeInt,
+ Required: true,
+ ForceNew: true,
+ },
+
+ "arn": &schema.Schema{
+ Type: schema.TypeString,
+ Optional: true,
+ Computed: true,
+ },
+ },
+ }
+}
+
+func resourceAwsKinesisStreamCreate(d *schema.ResourceData, meta interface{}) error {
+ conn := meta.(*AWSClient).kinesisconn
+ sn := d.Get("name").(string)
+ createOpts := &kinesis.CreateStreamInput{
+ ShardCount: aws.Long(int64(d.Get("shard_count").(int))),
+ StreamName: aws.String(sn),
+ }
+
+ _, err := conn.CreateStream(createOpts)
+ if err != nil {
+ if awsErr, ok := err.(awserr.Error); ok {
+ return fmt.Errorf("[WARN] Error creating Kinesis Stream: \"%s\", code: \"%s\"", awsErr.Message(), awsErr.Code())
+ }
+ return err
+ }
+
+ stateConf := &resource.StateChangeConf{
+ Pending: []string{"CREATING"},
+ Target: "ACTIVE",
+ Refresh: streamStateRefreshFunc(conn, sn),
+ Timeout: 5 * time.Minute,
+ Delay: 10 * time.Second,
+ MinTimeout: 3 * time.Second,
+ }
+
+ streamRaw, err := stateConf.WaitForState()
+ if err != nil {
+ return fmt.Errorf(
+ "Error waiting for Kinesis Stream (%s) to become active: %s",
+ sn, err)
+ }
+
+ s := streamRaw.(*kinesis.StreamDescription)
+ d.SetId(*s.StreamARN)
+ d.Set("arn", s.StreamARN)
+
+ return nil
+}
+
+func resourceAwsKinesisStreamRead(d *schema.ResourceData, meta interface{}) error {
+ conn := meta.(*AWSClient).kinesisconn
+ describeOpts := &kinesis.DescribeStreamInput{
+ StreamName: aws.String(d.Get("name").(string)),
+ Limit: aws.Long(1),
+ }
+ resp, err := conn.DescribeStream(describeOpts)
+ if err != nil {
+ if awsErr, ok := err.(awserr.Error); ok {
+ if awsErr.Code() == "ResourceNotFoundException" {
+ d.SetId("")
+ return nil
+ }
+ return fmt.Errorf("[WARN] Error reading Kinesis Stream: \"%s\", code: \"%s\"", awsErr.Message(), awsErr.Code())
+ }
+ return err
+ }
+
+ s := resp.StreamDescription
+ d.Set("arn", *s.StreamARN)
+ d.Set("shard_count", len(s.Shards))
+
+ return nil
+}
+
+func resourceAwsKinesisStreamDelete(d *schema.ResourceData, meta interface{}) error {
+ conn := meta.(*AWSClient).kinesisconn
+ sn := d.Get("name").(string)
+ _, err := conn.DeleteStream(&kinesis.DeleteStreamInput{
+ StreamName: aws.String(sn),
+ })
+
+ if err != nil {
+ return err
+ }
+
+ stateConf := &resource.StateChangeConf{
+ Pending: []string{"DELETING"},
+ Target: "DESTROYED",
+ Refresh: streamStateRefreshFunc(conn, sn),
+ Timeout: 5 * time.Minute,
+ Delay: 10 * time.Second,
+ MinTimeout: 3 * time.Second,
+ }
+
+ _, err = stateConf.WaitForState()
+ if err != nil {
+ return fmt.Errorf(
+ "Error waiting for Stream (%s) to be destroyed: %s",
+ sn, err)
+ }
+
+ d.SetId("")
+ return nil
+}
+
+func streamStateRefreshFunc(conn *kinesis.Kinesis, sn string) resource.StateRefreshFunc {
+ return func() (interface{}, string, error) {
+ describeOpts := &kinesis.DescribeStreamInput{
+ StreamName: aws.String(sn),
+ Limit: aws.Long(1),
+ }
+ resp, err := conn.DescribeStream(describeOpts)
+ 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 resp.StreamDescription, *resp.StreamDescription.StreamStatus, nil
+ }
+}

builtin/providers/aws/resource_aws_kinesis_stream_test.go

@@ -0,0 +1,108 @@
+package aws
+
+import (
+ "fmt"
+ "math/rand"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/awslabs/aws-sdk-go/aws"
+ "github.com/awslabs/aws-sdk-go/service/kinesis"
+ "github.com/hashicorp/terraform/helper/resource"
+ "github.com/hashicorp/terraform/terraform"
+)
+
+func TestAccKinesisStream_basic(t *testing.T) {
+ var stream kinesis.StreamDescription
+
+ resource.Test(t, resource.TestCase{
+ PreCheck: func() { testAccPreCheck(t) },
+ Providers: testAccProviders,
+ CheckDestroy: testAccCheckKinesisStreamDestroy,
+ Steps: []resource.TestStep{
+ resource.TestStep{
+ Config: testAccKinesisStreamConfig,
+ Check: resource.ComposeTestCheckFunc(
+ testAccCheckKinesisStreamExists("aws_kinesis_stream.test_stream", &stream),
+ testAccCheckAWSKinesisStreamAttributes(&stream),
+ ),
+ },
+ },
+ })
+}
+
+func testAccCheckKinesisStreamExists(n string, stream *kinesis.StreamDescription) resource.TestCheckFunc {
+ return func(s *terraform.State) error {
+ rs, ok := s.RootModule().Resources[n]
+ if !ok {
+ return fmt.Errorf("Not found: %s", n)
+ }
+
+ if rs.Primary.ID == "" {
+ return fmt.Errorf("No Kinesis ID is set")
+ }
+
+ conn := testAccProvider.Meta().(*AWSClient).kinesisconn
+ describeOpts := &kinesis.DescribeStreamInput{
+ StreamName: aws.String(rs.Primary.Attributes["name"]),
+ Limit: aws.Long(1),
+ }
+ resp, err := conn.DescribeStream(describeOpts)
+ if err != nil {
+ return err
+ }
+
+ *stream = *resp.StreamDescription
+
+ return nil
+ }
+}
+
+func testAccCheckAWSKinesisStreamAttributes(stream *kinesis.StreamDescription) resource.TestCheckFunc {
+ return func(s *terraform.State) error {
+ if !strings.HasPrefix(*stream.StreamName, "terraform-kinesis-test") {
+ return fmt.Errorf("Bad Stream name: %s", *stream.StreamName)
+ }
+ for _, rs := range s.RootModule().Resources {
+ if rs.Type != "aws_kinesis_stream" {
+ continue
+ }
+ if *stream.StreamARN != rs.Primary.Attributes["arn"] {
+ return fmt.Errorf("Bad Stream ARN\n\t expected: %s\n\tgot: %s\n", rs.Primary.Attributes["arn"], *stream.StreamARN)
+ }
+ }
+ return nil
+ }
+}
+
+func testAccCheckKinesisStreamDestroy(s *terraform.State) error {
+ for _, rs := range s.RootModule().Resources {
+ if rs.Type != "aws_kinesis_stream" {
+ continue
+ }
+ conn := testAccProvider.Meta().(*AWSClient).kinesisconn
+ describeOpts := &kinesis.DescribeStreamInput{
+ StreamName: aws.String(rs.Primary.Attributes["name"]),
+ Limit: aws.Long(1),
+ }
+ resp, err := conn.DescribeStream(describeOpts)
+ if err == nil {
+ if resp.StreamDescription != nil && *resp.StreamDescription.StreamStatus != "DELETING" {
+ return fmt.Errorf("Error: Stream still exists")
+ }
+ }
+
+ return nil
+
+ }
+
+ return nil
+}
+
+var testAccKinesisStreamConfig = fmt.Sprintf(`
+resource "aws_kinesis_stream" "test_stream" {
+ name = "terraform-kinesis-test-%d"
+ shard_count = 1
+}
+`, rand.New(rand.NewSource(time.Now().UnixNano())).Int())

website/source/docs/providers/aws/r/kinesis_stream.html.markdown

@@ -0,0 +1,44 @@
+---
+layout: "aws"
+page_title: "AWS: aws_kinesis_stream"
+sidebar_current: "docs-aws-resource-kinesis-stream"
+description: |-
+ Provides a AWS Kinesis Stream
+---
+
+# aws\_kinesis\_stream
+
+Provides a Kinesis Stream resource. Amazon Kinesis is a managed service that 
+scales elastically for real-time processing of streaming big data.
+
+For more details, see the [Amazon Kinesis Documentation][1].
+
+## Example Usage
+
+```
+resource "aws_kinesis_stream" "test_stream" {
+ name = "terraform-kinesis-test"
+ shard_count = 1
+}
+```
+
+## Argument Reference
+
+The following arguments are supported:
+
+* `name` - (Required) A name to identify the stream. This is unique to the 
+AWS account and region the Stream is created in.
+* `shard_count` – (Required) The number of shards that the stream will use.
+Amazon has guidlines for specifying the Stream size that should be referenced 
+when creating a Kinesis stream. See [Amazon Kinesis Streams][2] for more.
+
+## Attributes Reference
+
+* `id` - The unique Stream id
+* `name` - The unique Stream name (same as `id`)
+* `shard_count` - The count of Shards for this Stream
+* `arn` - The Amazon Resource Name (ARN) specifying the Stream
+
+
+[1]: http://aws.amazon.com/documentation/kinesis/
+[2]: http://docs.aws.amazon.com/kinesis/latest/dev/amazon-kinesis-streams.html

website/source/layouts/aws.erb

@@ -101,6 +101,10 @@
  <a href="/docs/providers/aws/r/iam_user_policy.html">aws_iam_user_policy</a>
  </li>
 
+ <li<%= sidebar_current("docs-aws-resource-kinesis-stream") %>>
+ <a href="/docs/providers/aws/r/kinesis_stream.html">aws_kinesis_stream</a>
+ </li>
+
  <li<%= sidebar_current("docs-aws-resource-instance") %>>
  <a href="/docs/providers/aws/r/instance.html">aws_instance</a>
  </li>