diff --git a/internal/internal_test.go b/internal/internal_test.go
index c18991c2d..f6a1e2ac3 100644
--- a/internal/internal_test.go
+++ b/internal/internal_test.go
@@ -88,12 +88,14 @@ func TestCombinedOutputError(t *testing.T) {
 		t.Skip("'sleep' binary not available on OS, skipping.")
 	}
 	cmd := exec.Command(sleepbin, "foo")
-	expected, err := cmd.CombinedOutput()
+	expected, expectedErr := cmd.CombinedOutput()
 
 	cmd2 := exec.Command(sleepbin, "foo")
-	actual, err := CombinedOutputTimeout(cmd2, time.Second)
+	actual, actualErr := CombinedOutputTimeout(cmd2, time.Second)
 
-	assert.Error(t, err)
+	if expectedErr != nil {
+		assert.Error(t, actualErr)
+	}
 	assert.Equal(t, expected, actual)
 }
 
@@ -102,9 +104,14 @@ func TestRunError(t *testing.T) {
 		t.Skip("'sleep' binary not available on OS, skipping.")
 	}
 	cmd := exec.Command(sleepbin, "foo")
-	err := RunTimeout(cmd, time.Second)
+	expectedErr := cmd.Run()
 
-	assert.Error(t, err)
+	cmd2 := exec.Command(sleepbin, "foo")
+	actualErr := RunTimeout(cmd2, time.Second)
+
+	if expectedErr != nil {
+		assert.Error(t, actualErr)
+	}
 }
 
 func TestRandomSleep(t *testing.T) {
diff --git a/plugins/inputs/cloudwatch/README.md b/plugins/inputs/cloudwatch/README.md
index 4430e48fd..dfdb4aedc 100644
--- a/plugins/inputs/cloudwatch/README.md
+++ b/plugins/inputs/cloudwatch/README.md
@@ -64,11 +64,14 @@ Plugin Configuration utilizes [CloudWatch concepts](http://docs.aws.amazon.com/A
 - `names` must be valid CloudWatch [Metric](http://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/cloudwatch_concepts.html#Metric) names
 - `dimensions` must be valid CloudWatch [Dimension](http://docs.aws.amazon.com/AmazonCloudWatch/latest/DeveloperGuide/cloudwatch_concepts.html#Dimension) name/value pairs
 
-Omitting or specifying a value of `'*'` for a dimension value configures all available metrics that contain a dimension with the specified name
-to be retrieved. If specifying >1 dimension, then the metric must contain *all* the configured dimensions where the the value of the
-wildcard dimension is ignored.
+A wildcard (`*`) can be used in the dimension value as follows:
+- `value = '*' or value = ''` - includes metrics that have the specified dimension name and any value same as `Regexp('.*')`
+- `value = 'foo*'` - includes metrics that have the specified dimension name and any value that starts with foo same as `Regexp('foo.*')`
 
-Example:
+If dimension wildcards are used, the metric must contain *all* the configured dimensions where the the value
+matches the wildcard pattern.
+
+Example 1:
 ```
 [[inputs.cloudwatch.metrics]]
   names = ['Latency']
@@ -76,7 +79,7 @@ Example:
   ## Dimension filters for Metric (optional)
   [[inputs.cloudwatch.metrics.dimensions]]
     name = 'LoadBalancerName'
-    value = 'p-example'
+    value = 'foo-*'
 
   [[inputs.cloudwatch.metrics.dimensions]]
     name = 'AvailabilityZone'
@@ -84,18 +87,41 @@ Example:
 ```
 
 If the following ELBs are available:
-- name: `p-example`, availabilityZone: `us-east-1a`
-- name: `p-example`, availabilityZone: `us-east-1b`
-- name: `q-example`, availabilityZone: `us-east-1a`
-- name: `q-example`, availabilityZone: `us-east-1b`
+- LoadBalancerName: `foo-example1`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `foo-example1`, AvailabilityZone: `us-east-1b`
+- LoadBalancerName: `foo-example2`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `foo-example2`, AvailabilityZone: `us-east-1b`
+- LoadBalancerName: `bar-example`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `bar-example`, AvailabilityZone: `us-east-1b`
 
+Then 4 metrics will be output:
+- LoadBalancerName: `foo-example1`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `foo-example1`, AvailabilityZone: `us-east-1b`
+- LoadBalancerName: `foo-example2`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `foo-example2`, AvailabilityZone: `us-east-1b`
 
-Then 2 metrics will be output:
-- name: `p-example`, availabilityZone: `us-east-1a`
-- name: `p-example`, availabilityZone: `us-east-1b`
+Example 2:
+```
+[[inputs.cloudwatch.metrics]]
+  names = ['Latency']
+
+  ## Dimension filters for Metric (optional)
+  [[inputs.cloudwatch.metrics.dimensions]]
+    name = 'LoadBalancerName'
+    value = 'foo-example1'
+```
+
+If the following ELBs are available:
+- LoadBalancerName: `foo-example1`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `foo-example1`, AvailabilityZone: `us-east-1b`
+- LoadBalancerName: `foo-example2`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `foo-example2`, AvailabilityZone: `us-east-1b`
+- LoadBalancerName: `bar-example`, AvailabilityZone: `us-east-1a`
+- LoadBalancerName: `bar-example`, AvailabilityZone: `us-east-1b`
+
+Then 4 metrics will be output:
+- LoadBalancerName: `foo-example1` - aggregated across all availability zones.
 
-If the `AvailabilityZone` wildcard dimension was omitted, then a single metric (name: `p-example`)
-would be exported containing the aggregate values of the ELB across availability zones.
 
 #### Restrictions and Limitations
 - CloudWatch metrics are not available instantly via the CloudWatch API. You should adjust your collection `delay` to account for this lag in metrics availability based on your [monitoring subscription level](http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-cloudwatch-new.html)
diff --git a/plugins/inputs/cloudwatch/cloudwatch.go b/plugins/inputs/cloudwatch/cloudwatch.go
index ebc4147d8..d5f3d4f5f 100644
--- a/plugins/inputs/cloudwatch/cloudwatch.go
+++ b/plugins/inputs/cloudwatch/cloudwatch.go
@@ -2,6 +2,7 @@ package cloudwatch
 
 import (
 	"fmt"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
@@ -124,53 +125,9 @@ func (c *CloudWatch) Gather(acc telegraf.Accumulator) error {
 		c.initializeCloudWatch()
 	}
 
-	var metrics []*cloudwatch.Metric
-
-	// check for provided metric filter
-	if c.Metrics != nil {
-		metrics = []*cloudwatch.Metric{}
-		for _, m := range c.Metrics {
-			if !hasWilcard(m.Dimensions) {
-				dimensions := make([]*cloudwatch.Dimension, len(m.Dimensions))
-				for k, d := range m.Dimensions {
-					fmt.Printf("Dimension [%s]:[%s]\n", d.Name, d.Value)
-					dimensions[k] = &cloudwatch.Dimension{
-						Name:  aws.String(d.Name),
-						Value: aws.String(d.Value),
-					}
-				}
-				for _, name := range m.MetricNames {
-					metrics = append(metrics, &cloudwatch.Metric{
-						Namespace:  aws.String(c.Namespace),
-						MetricName: aws.String(name),
-						Dimensions: dimensions,
-					})
-				}
-			} else {
-				allMetrics, err := c.fetchNamespaceMetrics()
-				if err != nil {
-					return err
-				}
-				for _, name := range m.MetricNames {
-					for _, metric := range allMetrics {
-						if isSelected(metric, m.Dimensions) {
-							metrics = append(metrics, &cloudwatch.Metric{
-								Namespace:  aws.String(c.Namespace),
-								MetricName: aws.String(name),
-								Dimensions: metric.Dimensions,
-							})
-						}
-					}
-				}
-			}
-
-		}
-	} else {
-		var err error
-		metrics, err = c.fetchNamespaceMetrics()
-		if err != nil {
-			return err
-		}
+	metrics, err := selectMetrics(c)
+	if err != nil {
+		return err
 	}
 
 	metricCount := len(metrics)
@@ -226,6 +183,63 @@ func (c *CloudWatch) initializeCloudWatch() error {
 	return nil
 }
 
+/*
+ * Select metrics to gather
+ */
+func selectMetrics(c *CloudWatch) ([]*cloudwatch.Metric, error) {
+	var metrics []*cloudwatch.Metric
+
+	// check for provided metric filter
+	if c.Metrics == nil {
+		return c.fetchNamespaceMetrics()
+	}
+
+	metrics = []*cloudwatch.Metric{}
+	for _, m := range c.Metrics {
+		if hasWilcard(m.Dimensions) {
+			allMetrics, err := c.fetchNamespaceMetrics()
+			if err != nil {
+				return nil, err
+			}
+			for _, name := range m.MetricNames {
+				for _, metric := range allMetrics {
+					s, e := isSelected(metric, m.Dimensions)
+					if e != nil {
+						return nil, e
+					}
+					if s {
+						for _, d := range metric.Dimensions {
+							fmt.Printf("Dimension [%s]:[%s]\n", *d.Name, *d.Value)
+						}
+						metrics = append(metrics, &cloudwatch.Metric{
+							Namespace:  aws.String(c.Namespace),
+							MetricName: aws.String(name),
+							Dimensions: metric.Dimensions,
+						})
+					}
+				}
+			}
+		} else {
+			dimensions := make([]*cloudwatch.Dimension, len(m.Dimensions))
+			for k, d := range m.Dimensions {
+				fmt.Printf("Dimension [%s]:[%s]\n", d.Name, d.Value)
+				dimensions[k] = &cloudwatch.Dimension{
+					Name:  aws.String(d.Name),
+					Value: aws.String(d.Value),
+				}
+			}
+			for _, name := range m.MetricNames {
+				metrics = append(metrics, &cloudwatch.Metric{
+					Namespace:  aws.String(c.Namespace),
+					MetricName: aws.String(name),
+					Dimensions: dimensions,
+				})
+			}
+		}
+	}
+	return metrics, nil
+}
+
 /*
  * Fetch available metrics for given CloudWatch Namespace
  */
@@ -368,29 +382,34 @@ func (c *MetricCache) IsValid() bool {
 
 func hasWilcard(dimensions []*Dimension) bool {
 	for _, d := range dimensions {
-		if d.Value == "" || d.Value == "*" {
+		if d.Value == "" || strings.ContainsRune(d.Value, '*') {
 			return true
 		}
 	}
 	return false
 }
 
-func isSelected(metric *cloudwatch.Metric, dimensions []*Dimension) bool {
+func isSelected(metric *cloudwatch.Metric, dimensions []*Dimension) (bool, error) {
 	if len(metric.Dimensions) != len(dimensions) {
-		return false
+		return false, nil
 	}
 	for _, d := range dimensions {
+		if d.Value == "" {
+			d.Value = "*"
+		}
+		r, e := regexp.Compile(strings.Replace(d.Value, "*", ".*", -1))
+		if e != nil {
+			return false, e
+		}
 		selected := false
 		for _, d2 := range metric.Dimensions {
-			if d.Name == *d2.Name {
-				if d.Value == "" || d.Value == "*" || d.Value == *d2.Value {
-					selected = true
-				}
+			if d.Name == *d2.Name && r.MatchString(*d2.Value) {
+				selected = true
 			}
 		}
 		if !selected {
-			return false
+			return false, nil
 		}
 	}
-	return true
+	return true, nil
 }
diff --git a/plugins/inputs/cloudwatch/cloudwatch_test.go b/plugins/inputs/cloudwatch/cloudwatch_test.go
index 73fca9253..a6f88ccbd 100644
--- a/plugins/inputs/cloudwatch/cloudwatch_test.go
+++ b/plugins/inputs/cloudwatch/cloudwatch_test.go
@@ -11,9 +11,12 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-type mockCloudWatchClient struct{}
+type (
+	basicCloudWatchClient struct{}
+	regexCloudWatchClient struct{}
+)
 
-func (m *mockCloudWatchClient) ListMetrics(params *cloudwatch.ListMetricsInput) (*cloudwatch.ListMetricsOutput, error) {
+func (*basicCloudWatchClient) ListMetrics(params *cloudwatch.ListMetricsInput) (*cloudwatch.ListMetricsOutput, error) {
 	metric := &cloudwatch.Metric{
 		Namespace:  params.Namespace,
 		MetricName: aws.String("Latency"),
@@ -31,7 +34,7 @@ func (m *mockCloudWatchClient) ListMetrics(params *cloudwatch.ListMetricsInput)
 	return result, nil
 }
 
-func (m *mockCloudWatchClient) GetMetricStatistics(params *cloudwatch.GetMetricStatisticsInput) (*cloudwatch.GetMetricStatisticsOutput, error) {
+func (*basicCloudWatchClient) GetMetricStatistics(params *cloudwatch.GetMetricStatisticsInput) (*cloudwatch.GetMetricStatisticsOutput, error) {
 	dataPoint := &cloudwatch.Datapoint{
 		Timestamp:   params.EndTime,
 		Minimum:     aws.Float64(0.1),
@@ -48,7 +51,53 @@ func (m *mockCloudWatchClient) GetMetricStatistics(params *cloudwatch.GetMetricS
 	return result, nil
 }
 
-func TestGather(t *testing.T) {
+func tblMetric(params *cloudwatch.ListMetricsInput, d1 string, d2 string) *cloudwatch.Metric {
+	return &cloudwatch.Metric{
+		Namespace:  params.Namespace,
+		MetricName: aws.String("ConsumedReadCapacityUnits"),
+		Dimensions: []*cloudwatch.Dimension{
+			&cloudwatch.Dimension{
+				Name:  aws.String("TableName"),
+				Value: aws.String(d1),
+			},
+			&cloudwatch.Dimension{
+				Name:  aws.String("IndexName"),
+				Value: aws.String(d2),
+			},
+		},
+	}
+}
+
+func (*regexCloudWatchClient) ListMetrics(params *cloudwatch.ListMetricsInput) (*cloudwatch.ListMetricsOutput, error) {
+	metric1 := tblMetric(params, "foo-table1", "ix-foo-t1")
+	metric2 := tblMetric(params, "foo-table2", "ix-foo-t2")
+	metric3 := tblMetric(params, "bar-table1", "ix-bar-t1")
+	metric4 := tblMetric(params, "bar-table2", "ix-bar-t2")
+
+	result := &cloudwatch.ListMetricsOutput{
+		Metrics: []*cloudwatch.Metric{metric1, metric2, metric3, metric4},
+	}
+	return result, nil
+}
+
+func (*regexCloudWatchClient) GetMetricStatistics(params *cloudwatch.GetMetricStatisticsInput) (*cloudwatch.GetMetricStatisticsOutput, error) {
+	dataPoint := &cloudwatch.Datapoint{
+		Timestamp:   params.EndTime,
+		Minimum:     aws.Float64(0),
+		Maximum:     aws.Float64(10),
+		Average:     aws.Float64(4),
+		Sum:         aws.Float64(40),
+		SampleCount: aws.Float64(10),
+		Unit:        aws.String("Units"),
+	}
+	result := &cloudwatch.GetMetricStatisticsOutput{
+		Label:      aws.String("ConsumedReadCapacityUnits"),
+		Datapoints: []*cloudwatch.Datapoint{dataPoint},
+	}
+	return result, nil
+}
+
+func TestBasicGather(t *testing.T) {
 	duration, _ := time.ParseDuration("1m")
 	internalDuration := internal.Duration{
 		Duration: duration,
@@ -62,7 +111,7 @@ func TestGather(t *testing.T) {
 	}
 
 	var acc testutil.Accumulator
-	c.client = &mockCloudWatchClient{}
+	c.client = &basicCloudWatchClient{}
 
 	c.Gather(&acc)
 
@@ -83,6 +132,145 @@ func TestGather(t *testing.T) {
 
 }
 
+func TestSingleDimensionRegexGather(t *testing.T) {
+	duration, _ := time.ParseDuration("1m")
+	internalDuration := internal.Duration{
+		Duration: duration,
+	}
+	c := &CloudWatch{
+		Region:    "us-east-1",
+		Namespace: "AWS/DynamoDB",
+		Delay:     internalDuration,
+		Period:    internalDuration,
+		RateLimit: 10,
+		Metrics: []*Metric{
+			&Metric{
+				MetricNames: []string{"ConsumedReadCapacityUnits"},
+				Dimensions: []*Dimension{
+					&Dimension{Name: "TableName", Value: "foo*"},
+					&Dimension{Name: "IndexName", Value: ""},
+				},
+			},
+		},
+	}
+
+	var acc testutil.Accumulator
+	acc.SetDebug(true)
+	c.client = &regexCloudWatchClient{}
+
+	c.Gather(&acc)
+
+	fields := map[string]interface{}{}
+	fields["consumed_read_capacity_units_minimum"] = 0.
+	fields["consumed_read_capacity_units_maximum"] = 10.
+	fields["consumed_read_capacity_units_average"] = 4.
+	fields["consumed_read_capacity_units_sum"] = 40.
+	fields["consumed_read_capacity_units_sample_count"] = 10.
+
+	tags := map[string]string{}
+	tags["unit"] = "units"
+	tags["region"] = "us-east-1"
+	tags["table_name"] = "foo-table1"
+	tags["index_name"] = "ix-foo-t1"
+
+	assert.True(t, acc.HasMeasurement("cloudwatch_aws_dynamo_db"))
+	acc.AssertContainsTaggedFields(t, "cloudwatch_aws_dynamo_db", fields, tags)
+
+	tags["table_name"] = "foo-table2"
+	tags["index_name"] = "ix-foo-t2"
+	acc.AssertContainsTaggedFields(t, "cloudwatch_aws_dynamo_db", fields, tags)
+}
+
+func TestMultiDimensionRegexGather(t *testing.T) {
+	duration, _ := time.ParseDuration("1m")
+	internalDuration := internal.Duration{
+		Duration: duration,
+	}
+	c := &CloudWatch{
+		Region:    "us-east-1",
+		Namespace: "AWS/DynamoDB",
+		Delay:     internalDuration,
+		Period:    internalDuration,
+		RateLimit: 10,
+		Metrics: []*Metric{
+			&Metric{
+				MetricNames: []string{"ConsumedReadCapacityUnits"},
+				Dimensions: []*Dimension{
+					&Dimension{Name: "TableName", Value: "foo*"},
+					&Dimension{Name: "IndexName", Value: "*t2"},
+				},
+			},
+		},
+	}
+
+	var acc testutil.Accumulator
+	c.client = &regexCloudWatchClient{}
+
+	c.Gather(&acc)
+
+	fields := map[string]interface{}{}
+	fields["consumed_read_capacity_units_minimum"] = 0.
+	fields["consumed_read_capacity_units_maximum"] = 10.
+	fields["consumed_read_capacity_units_average"] = 4.
+	fields["consumed_read_capacity_units_sum"] = 40.
+	fields["consumed_read_capacity_units_sample_count"] = 10.
+
+	tags := map[string]string{}
+	tags["unit"] = "units"
+	tags["region"] = "us-east-1"
+	tags["table_name"] = "foo-table2"
+	tags["index_name"] = "ix-foo-t2"
+
+	assert.True(t, acc.HasMeasurement("cloudwatch_aws_dynamo_db"))
+	acc.AssertContainsTaggedFields(t, "cloudwatch_aws_dynamo_db", fields, tags)
+	assert.Equal(t, 1, acc.CountTaggedMeasurements("cloudwatch_aws_dynamo_db", tags))
+}
+
+func TestDimensionGather(t *testing.T) {
+	duration, _ := time.ParseDuration("1m")
+	internalDuration := internal.Duration{
+		Duration: duration,
+	}
+	c := &CloudWatch{
+		Region:    "us-east-1",
+		Namespace: "AWS/DynamoDB",
+		Delay:     internalDuration,
+		Period:    internalDuration,
+		RateLimit: 10,
+		Metrics: []*Metric{
+			&Metric{
+				MetricNames: []string{"ConsumedReadCapacityUnits"},
+				Dimensions: []*Dimension{
+					&Dimension{Name: "TableName", Value: "foo-table1"},
+					&Dimension{Name: "IndexName", Value: "ix-foo-t1"},
+				},
+			},
+		},
+	}
+
+	var acc testutil.Accumulator
+	c.client = &regexCloudWatchClient{}
+
+	c.Gather(&acc)
+
+	fields := map[string]interface{}{}
+	fields["consumed_read_capacity_units_minimum"] = 0.
+	fields["consumed_read_capacity_units_maximum"] = 10.
+	fields["consumed_read_capacity_units_average"] = 4.
+	fields["consumed_read_capacity_units_sum"] = 40.
+	fields["consumed_read_capacity_units_sample_count"] = 10.
+
+	tags := map[string]string{}
+	tags["unit"] = "units"
+	tags["region"] = "us-east-1"
+	tags["table_name"] = "foo-table1"
+	tags["index_name"] = "ix-foo-t1"
+
+	assert.True(t, acc.HasMeasurement("cloudwatch_aws_dynamo_db"))
+	acc.AssertContainsTaggedFields(t, "cloudwatch_aws_dynamo_db", fields, tags)
+	assert.Equal(t, 1, acc.CountTaggedMeasurements("cloudwatch_aws_dynamo_db", tags))
+}
+
 func TestGenerateStatisticsInputParams(t *testing.T) {
 	d := &cloudwatch.Dimension{
 		Name:  aws.String("LoadBalancerName"),
diff --git a/testutil/accumulator.go b/testutil/accumulator.go
index 99f9e3006..5e2faafef 100644
--- a/testutil/accumulator.go
+++ b/testutil/accumulator.go
@@ -185,6 +185,23 @@ func (a *Accumulator) AssertContainsTaggedFields(
 	assert.Fail(t, msg)
 }
 
+func (a *Accumulator) CountTaggedMeasurements(
+	measurement string,
+	tags map[string]string,
+) int {
+	a.Lock()
+	defer a.Unlock()
+
+	cnt := 0
+	for _, p := range a.Metrics {
+		if reflect.DeepEqual(tags, p.Tags) &&
+			p.Measurement == measurement {
+			cnt++
+		}
+	}
+	return cnt
+}
+
 func (a *Accumulator) AssertContainsFields(
 	t *testing.T,
 	measurement string,