diff --git a/builtin/providers/aws/provider.go b/builtin/providers/aws/provider.go
index 0e54e7b501a4..5d8e3853785b 100644
--- a/builtin/providers/aws/provider.go
+++ b/builtin/providers/aws/provider.go
@@ -93,7 +93,7 @@ func Provider() terraform.ResourceProvider {
 			"aws_db_parameter_group":           resourceAwsDbParameterGroup(),
 			"aws_db_security_group":            resourceAwsDbSecurityGroup(),
 			"aws_db_subnet_group":              resourceAwsDbSubnetGroup(),
-			"aws_dynamodb_table":				resourceAwsDynamoDbTable(),
+			"aws_dynamodb_table":               resourceAwsDynamoDbTable(),
 			"aws_ebs_volume":                   resourceAwsEbsVolume(),
 			"aws_ecs_cluster":                  resourceAwsEcsCluster(),
 			"aws_ecs_service":                  resourceAwsEcsService(),
@@ -103,6 +103,7 @@ func Provider() terraform.ResourceProvider {
 			"aws_elasticache_security_group":   resourceAwsElasticacheSecurityGroup(),
 			"aws_elasticache_subnet_group":     resourceAwsElasticacheSubnetGroup(),
 			"aws_elb":                          resourceAwsElb(),
+			"aws_flow_log":                     resourceAwsFlowLog(),
 			"aws_iam_access_key":               resourceAwsIamAccessKey(),
 			"aws_iam_group_policy":             resourceAwsIamGroupPolicy(),
 			"aws_iam_group":                    resourceAwsIamGroup(),
diff --git a/builtin/providers/aws/resource_aws_flow_log.go b/builtin/providers/aws/resource_aws_flow_log.go
new file mode 100644
index 000000000000..39b3c75661fc
--- /dev/null
+++ b/builtin/providers/aws/resource_aws_flow_log.go
@@ -0,0 +1,155 @@
+package aws
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awsutil"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/hashicorp/terraform/helper/schema"
+)
+
+func resourceAwsFlowLog() *schema.Resource {
+	return &schema.Resource{
+		Create: resourceAwsLogFlowCreate,
+		Read:   resourceAwsLogFlowRead,
+		Delete: resourceAwsLogFlowDelete,
+
+		Schema: map[string]*schema.Schema{
+			"iam_role_arn": &schema.Schema{
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"log_group_name": &schema.Schema{
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"vpc_id": &schema.Schema{
+				Type:          schema.TypeString,
+				Optional:      true,
+				ForceNew:      true,
+				ConflictsWith: []string{"subnet_id", "eni_id"},
+			},
+
+			"subnet_id": &schema.Schema{
+				Type:          schema.TypeString,
+				Optional:      true,
+				ForceNew:      true,
+				ConflictsWith: []string{"eni_id", "vpc_id"},
+			},
+
+			"eni_id": &schema.Schema{
+				Type:          schema.TypeString,
+				Optional:      true,
+				ForceNew:      true,
+				ConflictsWith: []string{"subnet_id", "vpc_id"},
+			},
+
+			"traffic_type": &schema.Schema{
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+		},
+	}
+}
+
+func resourceAwsLogFlowCreate(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).ec2conn
+
+	types := []struct {
+		ID   string
+		Type string
+	}{
+		{ID: d.Get("vpc_id").(string), Type: "VPC"},
+		{ID: d.Get("subnet_id").(string), Type: "Subnet"},
+		{ID: d.Get("eni_id").(string), Type: "NetworkInterface"},
+	}
+
+	var resourceId string
+	var resourceType string
+	for _, t := range types {
+		if t.ID != "" {
+			resourceId = t.ID
+			resourceType = t.Type
+			break
+		}
+	}
+
+	if resourceId == "" || resourceType == "" {
+		return fmt.Errorf("Error: Flow Logs require either a VPC, Subnet, or ENI ID")
+	}
+
+	opts := &ec2.CreateFlowLogsInput{
+		DeliverLogsPermissionARN: aws.String(d.Get("iam_role_arn").(string)),
+		LogGroupName:             aws.String(d.Get("log_group_name").(string)),
+		ResourceIDs:              []*string{aws.String(resourceId)},
+		ResourceType:             aws.String(resourceType),
+		TrafficType:              aws.String(d.Get("traffic_type").(string)),
+	}
+
+	log.Printf(
+		"[DEBUG] Flow Log Create configuration: %s", awsutil.StringValue(opts))
+	resp, err := conn.CreateFlowLogs(opts)
+	if err != nil {
+		return fmt.Errorf("Error creating Flow Log for (%s), error: %s", resourceId, err)
+	}
+
+	if len(resp.FlowLogIDs) > 1 {
+		return fmt.Errorf("Error: multiple Flow Logs created for (%s)", resourceId)
+	}
+
+	d.SetId(*resp.FlowLogIDs[0])
+
+	return resourceAwsLogFlowRead(d, meta)
+}
+
+func resourceAwsLogFlowRead(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).ec2conn
+
+	opts := &ec2.DescribeFlowLogsInput{
+		FlowLogIDs: []*string{aws.String(d.Id())},
+	}
+
+	resp, err := conn.DescribeFlowLogs(opts)
+	if err != nil {
+		log.Printf("[WARN] Error describing Flow Logs for id (%s)", d.Id())
+		d.SetId("")
+		return nil
+	}
+
+	if len(resp.FlowLogs) == 0 {
+		log.Printf("[WARN] No Flow Logs found for id (%s)", d.Id())
+		d.SetId("")
+		return nil
+	}
+
+	fl := resp.FlowLogs[0]
+
+	d.Set("traffic_type", fl.TrafficType)
+	d.Set("log_group_name", fl.LogGroupName)
+	d.Set("iam_role_arn", fl.DeliverLogsPermissionARN)
+
+	return nil
+}
+
+func resourceAwsLogFlowDelete(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*AWSClient).ec2conn
+
+	log.Printf(
+		"[DEBUG] Flow Log Destroy: %s", d.Id())
+	_, err := conn.DeleteFlowLogs(&ec2.DeleteFlowLogsInput{
+		FlowLogIDs: []*string{aws.String(d.Id())},
+	})
+
+	if err != nil {
+		return fmt.Errorf("[WARN] Error deleting Flow Log with ID (%s), error: %s", d.Id(), err)
+	}
+
+	return nil
+}
diff --git a/builtin/providers/aws/resource_aws_flow_log_test.go b/builtin/providers/aws/resource_aws_flow_log_test.go
new file mode 100644
index 000000000000..b938a6fea787
--- /dev/null
+++ b/builtin/providers/aws/resource_aws_flow_log_test.go
@@ -0,0 +1,212 @@
+package aws
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/hashicorp/terraform/helper/resource"
+	"github.com/hashicorp/terraform/terraform"
+)
+
+func TestAccFlowLog_basic(t *testing.T) {
+	var flowLog ec2.FlowLog
+	lgn := os.Getenv("LOG_GROUP_NAME")
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckFlowLogDestroy,
+		Steps: []resource.TestStep{
+			resource.TestStep{
+				Config: fmt.Sprintf(testAccFlowLogConfig_basic, lgn),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckFlowLogExists("aws_flow_log.test_flow_log", &flowLog),
+					testAccCheckAWSFlowLogAttributes(&flowLog),
+				),
+			},
+		},
+	})
+}
+
+func TestAccFlowLog_subnet(t *testing.T) {
+	var flowLog ec2.FlowLog
+	lgn := os.Getenv("LOG_GROUP_NAME")
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckFlowLogDestroy,
+		Steps: []resource.TestStep{
+			resource.TestStep{
+				Config: fmt.Sprintf(testAccFlowLogConfig_subnet, lgn),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckFlowLogExists("aws_flow_log.test_flow_log_subnet", &flowLog),
+					testAccCheckAWSFlowLogAttributes(&flowLog),
+				),
+			},
+		},
+	})
+}
+
+func testAccCheckFlowLogExists(n string, flowLog *ec2.FlowLog) 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 Flow Log ID is set")
+		}
+
+		conn := testAccProvider.Meta().(*AWSClient).ec2conn
+		describeOpts := &ec2.DescribeFlowLogsInput{
+			FlowLogIDs: []*string{aws.String(rs.Primary.ID)},
+		}
+		resp, err := conn.DescribeFlowLogs(describeOpts)
+		if err != nil {
+			return err
+		}
+
+		if len(resp.FlowLogs) > 0 {
+			*flowLog = *resp.FlowLogs[0]
+			return nil
+		}
+		return fmt.Errorf("No Flow Logs found for id (%s)", rs.Primary.ID)
+	}
+}
+
+func testAccCheckAWSFlowLogAttributes(flowLog *ec2.FlowLog) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		if flowLog.FlowLogStatus != nil && *flowLog.FlowLogStatus == "ACTIVE" {
+			return nil
+		}
+		if flowLog.FlowLogStatus == nil {
+			return fmt.Errorf("Flow Log status is not ACTIVE, is nil")
+		} else {
+			return fmt.Errorf("Flow Log status is not ACTIVE, got: %s", *flowLog.FlowLogStatus)
+		}
+	}
+}
+
+func testAccCheckFlowLogDestroy(s *terraform.State) error {
+	for _, rs := range s.RootModule().Resources {
+		if rs.Type != "aws_flow_log" {
+			continue
+		}
+
+		return nil
+	}
+
+	return nil
+}
+
+var testAccFlowLogConfig_basic = `
+resource "aws_vpc" "default" {
+        cidr_block = "10.0.0.0/16"
+        tags {
+                Name = "tf-flow-log-test"
+        }
+}
+
+resource "aws_subnet" "test_subnet" {
+        vpc_id = "${aws_vpc.default.id}"
+        cidr_block = "10.0.1.0/24"
+
+        tags {
+                Name = "tf-flow-test"
+        }
+}
+
+resource "aws_iam_role" "test_role" {
+    name = "test_role"
+    assume_role_policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {
+        "Service": [
+          "ec2.amazonaws.com"
+        ]
+      },
+      "Action": [
+        "sts:AssumeRole"
+      ]
+    }
+  ]
+}
+EOF
+}
+
+resource "aws_flow_log" "test_flow_log" {
+        # log_group_name needs to exist before hand
+        # until we have a CloudWatch Log Group Resource
+        log_group_name = "tf-test-log-group"
+        iam_role_arn = "${aws_iam_role.test_role.arn}"
+        vpc_id = "${aws_vpc.default.id}"
+        traffic_type = "ALL"
+}
+
+resource "aws_flow_log" "test_flow_log_subnet" {
+        # log_group_name needs to exist before hand
+        # until we have a CloudWatch Log Group Resource
+        log_group_name = "%s"
+        iam_role_arn = "${aws_iam_role.test_role.arn}"
+        subnet_id = "${aws_subnet.test_subnet.id}"
+        traffic_type = "ALL"
+}
+`
+
+var testAccFlowLogConfig_subnet = `
+resource "aws_vpc" "default" {
+        cidr_block = "10.0.0.0/16"
+        tags {
+                Name = "tf-flow-log-test"
+        }
+}
+
+resource "aws_subnet" "test_subnet" {
+        vpc_id = "${aws_vpc.default.id}"
+        cidr_block = "10.0.1.0/24"
+
+        tags {
+                Name = "tf-flow-test"
+        }
+}
+
+resource "aws_iam_role" "test_role" {
+    name = "test_role"
+    assume_role_policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {
+        "Service": [
+          "ec2.amazonaws.com"
+        ]
+      },
+      "Action": [
+        "sts:AssumeRole"
+      ]
+    }
+  ]
+}
+EOF
+}
+
+resource "aws_flow_log" "test_flow_log_subnet" {
+        # log_group_name needs to exist before hand
+        # until we have a CloudWatch Log Group Resource
+        log_group_name = "%s"
+        iam_role_arn = "${aws_iam_role.test_role.arn}"
+        subnet_id = "${aws_subnet.test_subnet.id}"
+        traffic_type = "ALL"
+}
+`
diff --git a/website/source/docs/providers/aws/r/flow_log.html.markdown b/website/source/docs/providers/aws/r/flow_log.html.markdown
new file mode 100644
index 000000000000..0de997c3d331
--- /dev/null
+++ b/website/source/docs/providers/aws/r/flow_log.html.markdown
@@ -0,0 +1,84 @@
+---
+layout: "aws"
+page_title: "AWS: aws_flow_log"
+sidebar_current: "docs-aws-resource-flow-log"
+description: |-
+  Provides a VPC/Subnet/ENI Flow Log
+---
+
+# aws\_flow\_log
+
+Provides a VPC/Subnet/ENI Flow Log to capture IP traffic for a specific network
+interface, subnet, or VPC. Logs are sent to a CloudWatch Log Group.
+
+```
+resource "aws_flow_log" "test_flow_log" {
+  # log_group_name needs to exist before hand
+  # until we have a CloudWatch Log Group Resource
+  log_group_name = "tf-test-log-group"
+	iam_role_arn = "${aws_iam_role.test_role.arn}"
+  vpc_id = "${aws_vpc.default.id}"
+  traffic_type = "ALL"
+}
+
+resource "aws_iam_role" "test_role" {
+    name = "test_role"
+    assume_role_policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Sid": "",
+      "Effect": "Allow",
+      "Principal": {
+        "Service": "vpc-flow-logs.amazonaws.com"
+      },
+      "Action": "sts:AssumeRole"
+    }
+  ]
+} 
+EOF
+}
+
+resource "aws_iam_role_policy" "test_policy" {
+    name = "test_policy"
+    role = "${aws_iam_role.test_role.id}"
+    policy = <<EOF
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": [
+        "logs:CreateLogGroup",
+        "logs:CreateLogStream",
+        "logs:PutLogEvents",
+        "logs:DescribeLogGroups",
+        "logs:DescribeLogStreams"
+      ],
+      "Effect": "Allow",
+      "Resource": "*"
+    }
+  ]
+}   
+EOF
+}
+```
+
+## Argument Reference
+
+The following arguments are supported:
+
+* `log_group_name` - (Required) The name of the CloudWatch log group
+* `iam_role_arn` - (Required) The ARN for the IAM role that's used to post flow
+  logs to a CloudWatch Logs log group
+* `vpc_id` - (Optional) VPC ID to attach to
+* `subnet_id` - (Optional) Subnet ID to attach to
+* `eni_id` - (Optional) Elastic Network Interface ID to attach to
+* `traffic_type` - (Required) The type of traffic to capture. Valid values:
+  `ACCEPT`,`REJECT`, `ALL`
+
+## Attributes Reference
+
+The following attributes are exported:
+
+* `id` - The Flow Log ID
diff --git a/website/source/layouts/aws.erb b/website/source/layouts/aws.erb
index 801945dec969..5ca03ee54c1b 100644
--- a/website/source/layouts/aws.erb
+++ b/website/source/layouts/aws.erb
@@ -89,6 +89,10 @@
 							<a href="/docs/providers/aws/r/elb.html">aws_elb</a>
 						</li>
 
+                                                <li<%= sidebar_current("docs-aws-resource-flow-log") %>>
+                                                        <a href="/docs/providers/aws/r/flow_log.html">aws_flow_log</a>
+                                                </li>
+
 						<li<%= sidebar_current("docs-aws-resource-iam-access-key") %>>
 							<a href="/docs/providers/aws/r/iam_access_key.html">aws_iam_access_key</a>
 						</li>