Merge pull request #1 from paybyphone/dynamodb_resources

Addition of the DyanmoDB resource, can add add tables, and also add global secondary indexes to existing tables

.kitchen.cloud.yml

@@ -143,3 +143,11 @@ suites:
   - name: iam_role
     run_list:
     - recipe[aws_test::iam_role]
+
+  - name: kinesis_stream
+    run_list:
+    - recipe[aws_test::kinesis_stream]
+
+  - name: dynamodb_table
+    run_list:
+    - recipe[aws_test::dynamodb_table]

README.md

@@ -572,6 +572,100 @@ Attribute parameters are:
    in the CloudFormation user guide.
 
 
+## aws_dynamodb_table
+
+Use this resource to create and delete DynamoDB tables. This includes the ability
+to add global secondary indexes to existing tables.
+
+```
+aws_dynamodb_table 'example-table' do
+  action :create
+  attribute_definitions [
+    { attribute_name: 'Id', attribute_type: 'N' },
+    { attribute_name: 'Foo', attribute_type: 'N' },
+    { attribute_name: 'Bar', attribute_type: 'N' },
+    { attribute_name: 'Baz', attribute_type: 'S' }
+  ]
+  key_schema [
+    { attribute_name: 'Id', key_type: 'HASH' },
+    { attribute_name: 'Foo', key_type: 'RANGE' }
+  ]
+  local_secondary_indexes [
+    {
+      index_name: 'BarIndex',
+      key_schema: [
+        {
+          attribute_name: 'Id',
+          key_type: 'HASH'
+        },
+        {
+          attribute_name: 'Bar',
+          key_type: 'RANGE'
+        }
+      ],
+      projection: {
+        projection_type: 'ALL'
+      }
+    }
+  ]
+  global_secondary_indexes [
+    {
+      index_name: 'BazIndex',
+      key_schema: [{
+        attribute_name: 'Baz',
+        key_type: 'HASH'
+      }],
+      projection: {
+        projection_type: 'ALL'
+      },
+      provisioned_throughput: {
+        read_capacity_units: 1,
+        write_capacity_units: 1
+      }
+    }
+  ]
+  provisioned_throughput ({
+    read_capacity_units: 1,
+    write_capacity_units: 1
+  })
+  stream_specification ({
+    stream_enabled: true,
+    stream_view_type: 'KEYS_ONLY'
+  })
+end
+```
+
+Actions:
+
+* `create`: Creates the table. Will update the following if the table exists:
+ * `global_secondary_indexes`: Will remove non-existent indexes, add new ones,
+   and update throughput for existing ones. All attributes need to be present
+   in `attribute_definitions`. No effect if the resource is omitted.
+ * `stream_specification`: Will update as shown. No effect is the resource is
+   omitted.
+ * `provisioned_throughput`: Will update as shown.
+* `delete`: Deletes the index.
+
+Attributes:
+
+ * `attribute_definitions`: Required. Attributes to create for the table.
+   Mainly this is used to specify attributes that are used in keys, as otherwise
+   one can add any attribute they want to a DynamoDB table.
+ * `key_schema`: Required. Used to create the primary key for the table.
+   Attributes need to be present in `attribute_definitions`.
+ * `local_secondary_indexes`: Used to create any local secondary indexes for the
+   table. Attributes need to be present in `attribute_definitions`.
+ * `global_secondary_indexes`: Used to create any global secondary indexes. Can
+   be done to an existing table. Attributes need to be present in
+   `attribute_definitions`.
+ * `provisioned_throughput`: Define the throughput for this table.
+ * `stream_specification`: Specify if there should be a stream for this table.
+
+Several of the attributes shown here take parameters as shown in the
+[AWS Ruby SDK Documentation](http://docs.aws.amazon.com/sdkforruby/api/Aws/DynamoDB/Client.html#create_table-instance_method).
+Also, the [AWS DynamoDB Documentation](http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Introduction.html)
+may be of further help as well.
+
 ## aws_kinesis_stream
 
 Use this resource to create and delete Kinesis streams. Note that this resource

libraries/dynamodb.rb

@@ -0,0 +1,13 @@
+require File.join(File.dirname(__FILE__), 'ec2')
+
+module Opscode
+  module Aws
+    module DynamoDB
+      include Opscode::Aws::Ec2
+
+      def dynamodb
+        @dynamodb ||= create_aws_interface(::Aws::DynamoDB::Client)
+      end
+    end
+  end
+end

providers/dynamodb_table.rb

@@ -0,0 +1,159 @@
+include Opscode::Aws::DynamoDB
+
+def whyrun_supported?
+  true
+end
+
+# check to see if the table exists
+def table_exists?
+  resp = dynamodb.describe_table(table_name: new_resource.table_name)
+  if resp.length > 0
+    true
+  else
+    false
+  end
+rescue ::Aws::DynamoDB::Errors::ResourceNotFoundException
+  false
+end
+
+# check to see if throughput (on table itself) has changed
+# NOTE: old_throughput needs to be a value from API, and
+# new_throughput needs to be from resource
+def throughput_changed?(old_throughput, new_throughput)
+  if old_throughput.read_capacity_units != new_throughput[:read_capacity_units] ||
+     old_throughput.write_capacity_units != new_throughput[:write_capacity_units]
+    true
+  else
+    false
+  end
+end
+
+# check to see if table stream spec has changed
+def stream_spec_changed?
+  resp = dynamodb.describe_table(table_name: new_resource.table_name)
+  if resp.table.stream_specification
+    if resp.table.stream_specification.stream_enabled != new_resource.stream_specification[:stream_enabled] ||
+       resp.table.stream_specification.stream_view_type != new_resource.stream_specification[:stream_view_type]
+      true
+    else
+      false
+    end
+  elsif new_resource.stream_specification
+    new_resource.stream_specification[:stream_enabled]
+  else
+    false
+  end
+end
+
+# assembles list of updates for the global secondary index
+def gsi_changes
+  resp = dynamodb.describe_table(table_name: new_resource.table_name)
+  global_secondary_index_updates = []
+  # only run if indexes are defined in resource
+  if new_resource.global_secondary_indexes
+    if resp.table.global_secondary_indexes
+      existing_indexes = resp.table.global_secondary_indexes
+    else
+      existing_indexes = []
+    end
+    existing_indexes.each do |gsi|
+      index = new_resource.global_secondary_indexes.index { |x| x[:index_name] == gsi.index_name }
+      if index
+        # found
+        if throughput_changed?(gsi.provisioned_throughput, new_resource.global_secondary_indexes[index][:provisioned_throughput])
+          global_secondary_index_updates.push(
+            update: {
+              index_name: gsi.index_name,
+              provisioned_throughput: new_resource.global_secondary_indexes[index][:provisioned_throughput]
+            }
+          )
+        end
+      else
+        # not found - delete
+        global_secondary_index_updates.push(delete: { index_name: gsi.index_name })
+      end
+    end
+    # reverse check to see if anything needs to be created
+    new_resource.global_secondary_indexes.each do |gsi|
+      unless existing_indexes.index { |x| x.index_name == gsi[:index_name] }
+        global_secondary_index_updates.push(create: gsi)
+      end
+    end
+  end
+  global_secondary_index_updates
+end
+
+# waits for a table to become ready (and throws exception if it times out)
+def wait_for_table
+  res = ::Aws::DynamoDB::Resource.new(client: dynamodb)
+  table = res.table(new_resource.table_name)
+  before_wait_hook = lambda do |attempts, response|
+    Chef::Log.debug("waiting for table to become active - attempt #{attempts}")
+  end
+  table.wait_until(before_wait: before_wait_hook, max_attempts: 30) { |waiter| waiter.table_status == 'ACTIVE' }
+end
+
+action :create do
+  if table_exists?
+    # Keys, and local secondary indexes are ignored on update. Attributes are
+    # through when we update global secondary indexes.
+    # update throughput
+    if throughput_changed?(dynamodb.describe_table(table_name: new_resource.table_name).table.provisioned_throughput, new_resource.provisioned_throughput)
+      converge_by("change throughput on DynamoDB table #{new_resource.table_name}") do
+        # wait for table to become ready (if it is not)
+        wait_for_table
+        dynamodb.update_table(
+          table_name: new_resource.table_name,
+          provisioned_throughput: new_resource.provisioned_throughput
+        )
+      end
+    end
+    # update stream spec
+    if stream_spec_changed?
+      converge_by("change stream spec on DynamoDB table #{new_resource.table_name}") do
+        # wait for table to become ready (if it is not)
+        wait_for_table
+        dynamodb.update_table(
+          table_name: new_resource.table_name,
+          stream_specification: new_resource.stream_specification
+        )
+      end
+    end
+    # get list of changes to global secondary indexes
+    global_secondary_index_updates = gsi_changes
+    Chef::Log.debug("gsi_changes dump: #{gsi_changes}")
+    # update existing indexes
+    [:update, :delete, :create].each do |op|
+      (global_secondary_index_updates.select { |update| update.keys.include?(op) }).each do |index|
+        converge_by("update global secondary index #{index[op][:index_name]} on table #{new_resource.table_name}") do
+          wait_for_table
+          dynamodb.update_table(
+            table_name: new_resource.table_name,
+            attribute_definitions: new_resource.attribute_definitions,
+            global_secondary_index_updates: [index]
+          )
+        end
+      end
+    end
+  else
+    converge_by("create DynamoDB table #{new_resource.table_name}") do
+      dynamodb.create_table(
+        table_name: new_resource.table_name,
+        attribute_definitions: new_resource.attribute_definitions,
+        key_schema: new_resource.key_schema,
+        local_secondary_indexes: new_resource.local_secondary_indexes,
+        global_secondary_indexes: new_resource.global_secondary_indexes,
+        provisioned_throughput: new_resource.provisioned_throughput,
+        stream_specification: new_resource.stream_specification
+      )
+    end
+  end
+end
+
+action :delete do
+  if table_exists?
+    converge_by("delete DynamoDB table #{new_resource.table_name}") do
+      dynamodb.delete_table(table_name: new_resource.table_name)
+    end
+  end
+end

resources/dynamodb_table.rb

@@ -0,0 +1,71 @@
+require 'aws-sdk'
+
+default_action :create
+actions :create, :delete
+
+attribute :table_name, kind_of: String, name_attribute: true
+attribute :attribute_definitions, kind_of: Array, required: true, callbacks: {
+  'should contain valid Aws::DynamoDB::Types::AttributeDefinition types' => lambda do |attrs|
+    attrs.each do |attr|
+      return false unless Chef::Resource::AwsDynamodbTable.valid_attr?(::Aws::DynamoDB::Types::AttributeDefinition, attr)
+    end
+    true
+  end
+}
+
+attribute :key_schema, kind_of: Array, required: true, callbacks: {
+  'should contain valid Aws::DynamoDB::Types::KeySchemaElement types' => lambda do |attrs|
+    attrs.each do |attr|
+      return false unless Chef::Resource::AwsDynamodbTable.valid_attr?(::Aws::DynamoDB::Types::KeySchemaElement, attr)
+    end
+    true
+  end
+}
+
+attribute :local_secondary_indexes, kind_of: Array, default: nil, callbacks: {
+  'should contain valid Aws::DynamoDB::Types::LocalSecondaryIndex types' => lambda do |attrs|
+    attrs.each do |attr|
+      return false unless Chef::Resource::AwsDynamodbTable.valid_attr?(::Aws::DynamoDB::Types::LocalSecondaryIndex, attr)
+    end
+    true
+  end
+}
+
+attribute :global_secondary_indexes, kind_of: Array, default: nil, callbacks: {
+  'should contain valid Aws::DynamoDB::Types::GlobalSecondaryIndex types' => lambda do |attrs|
+    attrs.each do |attr|
+      return false unless Chef::Resource::AwsDynamodbTable.valid_attr?(::Aws::DynamoDB::Types::GlobalSecondaryIndex, attr)
+    end
+    true
+  end
+}
+
+attribute :provisioned_throughput, kind_of: Hash, required: true, callbacks: {
+  'should contain valid Aws::DynamoDB::Types::ProvisionedThroughput types' => lambda do |attr|
+    Chef::Resource::AwsDynamodbTable.valid_attr?(::Aws::DynamoDB::Types::ProvisionedThroughput, attr)
+  end
+}
+
+attribute :stream_specification, kind_of: Hash, default: nil, callbacks: {
+  'should contain valid Aws::DynamoDB::Types::StreamSpecification types' => lambda do |attr|
+    Chef::Resource::AwsDynamodbTable.valid_attr?(::Aws::DynamoDB::Types::StreamSpecification, attr)
+  end
+}
+# AWS common attributes
+attribute :region, kind_of: String, default: nil
+attribute :aws_access_key, kind_of: String, default: nil
+attribute :aws_secret_access_key, kind_of: String, default: nil
+attribute :aws_session_token, kind_of: String, default: nil
+
+private
+
+def self.valid_attr?(attribute_class, attribute_value)
+  attr_obj = attribute_class.new(attribute_value)
+  if attr_obj.is_a?(attribute_class)
+    true
+  else
+    false
+  end
+rescue NameError
+  false
+end

test/fixtures/cookbooks/aws_test/recipes/dynamodb_table.rb

@@ -0,0 +1,30 @@
+aws_dynamodb_table 'kitchen-test-table' do
+  action :create
+  attribute_definitions [
+    { attribute_name: 'Id', attribute_type: 'N' },
+    { attribute_name: 'Foo', attribute_type: 'S' },
+  ]
+  key_schema [
+    { attribute_name: 'Id', key_type: 'HASH' }
+  ]
+  global_secondary_indexes [
+    {
+      index_name: 'FooIndex',
+      key_schema: [{
+        attribute_name: 'Foo',
+        key_type: 'HASH'
+      }],
+      projection: {
+        projection_type: 'ALL'
+      },
+      provisioned_throughput: {
+        read_capacity_units: 1,
+        write_capacity_units: 1
+      }
+    }
+  ]
+  provisioned_throughput ({
+    read_capacity_units: 1,
+    write_capacity_units: 1
+  })
+end