diff --git a/newrelic/api/application.py b/newrelic/api/application.py
index 1ff425a7b0..e2e7be139f 100644
--- a/newrelic/api/application.py
+++ b/newrelic/api/application.py
@@ -142,6 +142,14 @@ def record_custom_metrics(self, metrics):
         if self.active and metrics:
             self._agent.record_custom_metrics(self._name, metrics)
 
+    def record_dimensional_metric(self, name, value, tags=None):
+        if self.active:
+            self._agent.record_dimensional_metric(self._name, name, value, tags)
+
+    def record_dimensional_metrics(self, metrics):
+        if self.active and metrics:
+            self._agent.record_dimensional_metrics(self._name, metrics)
+
     def record_custom_event(self, event_type, params):
         if self.active:
             self._agent.record_custom_event(self._name, event_type, params)
diff --git a/newrelic/api/transaction.py b/newrelic/api/transaction.py
index 4e2ddecb83..9afd49da16 100644
--- a/newrelic/api/transaction.py
+++ b/newrelic/api/transaction.py
@@ -68,7 +68,7 @@
 from newrelic.core.custom_event import create_custom_event
 from newrelic.core.log_event_node import LogEventNode
 from newrelic.core.stack_trace import exception_stack
-from newrelic.core.stats_engine import CustomMetrics, SampledDataSet
+from newrelic.core.stats_engine import CustomMetrics, DimensionalMetrics, SampledDataSet
 from newrelic.core.thread_utilization import utilization_tracker
 from newrelic.core.trace_cache import (
     TraceCacheActiveTraceError,
@@ -309,6 +309,7 @@ def __init__(self, application, enabled=None, source=None):
         self.synthetics_header = None
 
         self._custom_metrics = CustomMetrics()
+        self._dimensional_metrics = DimensionalMetrics()
 
         global_settings = application.global_settings
 
@@ -591,6 +592,7 @@ def __exit__(self, exc, value, tb):
             apdex_t=self.apdex,
             suppress_apdex=self.suppress_apdex,
             custom_metrics=self._custom_metrics,
+            dimensional_metrics=self._dimensional_metrics,
             guid=self.guid,
             cpu_time=self._cpu_user_time_value,
             suppress_transaction_trace=self.suppress_transaction_trace,
@@ -1597,6 +1599,16 @@ def record_custom_metrics(self, metrics):
         for name, value in metrics:
             self._custom_metrics.record_custom_metric(name, value)
 
+    def record_dimensional_metric(self, name, value, tags=None):
+        self._dimensional_metrics.record_dimensional_metric(name, value, tags)
+
+    def record_dimensional_metrics(self, metrics):
+        for metric in metrics:
+            name, value = metric[:2]
+            tags = metric[2] if len(metric) >= 3 else None
+
+            self._dimensional_metrics.record_dimensional_metric(name, value, tags)
+
     def record_custom_event(self, event_type, params):
         settings = self._settings
 
@@ -1908,6 +1920,44 @@ def record_custom_metrics(metrics, application=None):
         application.record_custom_metrics(metrics)
 
 
+def record_dimensional_metric(name, value, tags=None, application=None):
+    if application is None:
+        transaction = current_transaction()
+        if transaction:
+            transaction.record_dimensional_metric(name, value, tags)
+        else:
+            _logger.debug(
+                "record_dimensional_metric has been called but no "
+                "transaction was running. As a result, the following metric "
+                "has not been recorded. Name: %r Value: %r Tags: %r. To correct this "
+                "problem, supply an application object as a parameter to this "
+                "record_dimensional_metrics call.",
+                name,
+                value,
+                tags,
+            )
+    elif application.enabled:
+        application.record_dimensional_metric(name, value, tags)
+
+
+def record_dimensional_metrics(metrics, application=None):
+    if application is None:
+        transaction = current_transaction()
+        if transaction:
+            transaction.record_dimensional_metrics(metrics)
+        else:
+            _logger.debug(
+                "record_dimensional_metrics has been called but no "
+                "transaction was running. As a result, the following metrics "
+                "have not been recorded: %r. To correct this problem, "
+                "supply an application object as a parameter to this "
+                "record_dimensional_metric call.",
+                list(metrics),
+            )
+    elif application.enabled:
+        application.record_dimensional_metrics(metrics)
+
+
 def record_custom_event(event_type, params, application=None):
     """Record a custom event.
 
diff --git a/newrelic/common/metric_utils.py b/newrelic/common/metric_utils.py
new file mode 100644
index 0000000000..654f8051c4
--- /dev/null
+++ b/newrelic/common/metric_utils.py
@@ -0,0 +1,37 @@
+# Copyright 2010 New Relic, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+This module implements functions for creating a unique identity from a name and set of tags for use in dimensional metrics.
+"""
+
+from newrelic.core.attribute import process_user_attribute
+
+
+def create_metric_identity(name, tags=None):
+    if tags:
+        # Convert dicts to an iterable of tuples, other iterables should already be in this form
+        if isinstance(tags, dict):
+            tags = tags.items()  
+
+        # Apply attribute system sanitization.
+        # process_user_attribute returns (None, None) for results that fail sanitization.
+        # The filter removes these results fromt he iterable before creating the frozenset.
+        tags = frozenset(filter(lambda args: args[0] is not None, map(lambda args: process_user_attribute(*args), tags)))
+
+    # Set empty iterables after filtering to None 
+    if not tags and tags is not None:
+        tags = None
+
+    return (name, tags)
diff --git a/newrelic/core/agent.py b/newrelic/core/agent.py
index 8aab80d7e1..9d9aadab16 100644
--- a/newrelic/core/agent.py
+++ b/newrelic/core/agent.py
@@ -524,6 +524,33 @@ def record_custom_metrics(self, app_name, metrics):
 
         application.record_custom_metrics(metrics)
 
+    def record_dimensional_metric(self, app_name, name, value, tags=None):
+        """Records a basic metric for the named application. If there has
+        been no prior request to activate the application, the metric is
+        discarded.
+
+        """
+
+        application = self._applications.get(app_name, None)
+        if application is None or not application.active:
+            return
+
+        application.record_dimensional_metric(name, value, tags)
+
+    def record_dimensional_metrics(self, app_name, metrics):
+        """Records the metrics for the named application. If there has
+        been no prior request to activate the application, the metric is
+        discarded. The metrics should be an iterable yielding tuples
+        consisting of the name and value.
+
+        """
+
+        application = self._applications.get(app_name, None)
+        if application is None or not application.active:
+            return
+
+        application.record_dimensional_metrics(metrics)
+
     def record_custom_event(self, app_name, event_type, params):
         application = self._applications.get(app_name, None)
         if application is None or not application.active:
diff --git a/newrelic/core/application.py b/newrelic/core/application.py
index 2e7985d18a..82cdf8a9a0 100644
--- a/newrelic/core/application.py
+++ b/newrelic/core/application.py
@@ -510,6 +510,9 @@ def connect_to_data_collector(self, activate_agent):
         with self._stats_custom_lock:
             self._stats_custom_engine.reset_stats(configuration)
 
+        with self._stats_lock:
+            self._stats_engine.reset_stats(configuration)
+
         # Record an initial start time for the reporting period and
         # clear record of last transaction processed.
 
@@ -860,6 +863,50 @@ def record_custom_metrics(self, metrics):
                 self._global_events_account += 1
                 self._stats_custom_engine.record_custom_metric(name, value)
 
+    def record_dimensional_metric(self, name, value, tags=None):
+        """Record a dimensional metric against the application independent
+        of a specific transaction.
+
+        NOTE that this will require locking of the stats engine for
+        dimensional metrics and so under heavy use will have performance
+        issues. It is better to record the dimensional metric against an
+        active transaction as they will then be aggregated at the end of
+        the transaction when all other metrics are aggregated and so no
+        additional locking will be required.
+
+        """
+
+        if not self._active_session:
+            return
+
+        with self._stats_lock:
+            self._global_events_account += 1
+            self._stats_engine.record_dimensional_metric(name, value, tags)
+
+    def record_dimensional_metrics(self, metrics):
+        """Record a set of dimensional metrics against the application
+        independent of a specific transaction.
+
+        NOTE that this will require locking of the stats engine for
+        dimensional metrics and so under heavy use will have performance
+        issues. It is better to record the dimensional metric against an
+        active transaction as they will then be aggregated at the end of
+        the transaction when all other metrics are aggregated and so no
+        additional locking will be required.
+
+        """
+
+        if not self._active_session:
+            return
+
+        with self._stats_lock:
+            for metric in metrics:
+                name, value = metric[:2]
+                tags = metric[2] if len(metric) >= 3 else None
+
+                self._global_events_account += 1
+                self._stats_engine.record_dimensional_metric(name, value, tags)
+
     def record_custom_event(self, event_type, params):
         if not self._active_session:
             return
@@ -1452,11 +1499,14 @@ def harvest(self, shutdown=False, flexible=False):
                         _logger.debug("Normalizing metrics for harvest of %r.", self._app_name)
 
                         metric_data = stats.metric_data(metric_normalizer)
+                        dimensional_metric_data = stats.dimensional_metric_data(metric_normalizer)
 
                         _logger.debug("Sending metric data for harvest of %r.", self._app_name)
 
                         # Send metrics
                         self._active_session.send_metric_data(self._period_start, period_end, metric_data)
+                        if dimensional_metric_data:
+                            self._active_session.send_dimensional_metric_data(self._period_start, period_end, dimensional_metric_data)
 
                         _logger.debug("Done sending data for harvest of %r.", self._app_name)
 
diff --git a/newrelic/core/data_collector.py b/newrelic/core/data_collector.py
index bcfd3ab371..441f587dc8 100644
--- a/newrelic/core/data_collector.py
+++ b/newrelic/core/data_collector.py
@@ -35,6 +35,8 @@
 
 _logger = logging.getLogger(__name__)
 
+DIMENSIONAL_METRIC_DATA_TEMP = []  # TODO: REMOVE THIS
+
 
 class Session(object):
     PROTOCOL = AgentProtocol
@@ -143,6 +145,25 @@ def send_metric_data(self, start_time, end_time, metric_data):
         payload = (self.agent_run_id, start_time, end_time, metric_data)
         return self._protocol.send("metric_data", payload)
 
+    def send_dimensional_metric_data(self, start_time, end_time, metric_data):
+        """Called to submit dimensional metric data for specified period of time.
+        Time values are seconds since UNIX epoch as returned by the
+        time.time() function. The metric data should be iterable of
+        specific metrics.
+
+        NOTE: This data is sent not sent to the normal agent endpoints but is sent
+        to the MELT API endpoints to keep the entity separate. This is for use
+        with the machine learning integration only.
+        """
+
+        payload = (self.agent_run_id, start_time, end_time, metric_data)
+        # return self._protocol.send("metric_data", payload)
+
+        # TODO: REMOVE THIS. Replace with actual protocol.
+        DIMENSIONAL_METRIC_DATA_TEMP.append(payload)
+        _logger.debug("Dimensional Metrics: %r" % metric_data)
+        return 200  
+
     def send_log_events(self, sampling_info, log_event_data):
         """Called to submit sample set for log events."""
 
diff --git a/newrelic/core/stats_engine.py b/newrelic/core/stats_engine.py
index 17b0d99c12..9d59efd491 100644
--- a/newrelic/core/stats_engine.py
+++ b/newrelic/core/stats_engine.py
@@ -35,6 +35,7 @@
 from newrelic.api.settings import STRIP_EXCEPTION_MESSAGE
 from newrelic.api.time_trace import get_linking_metadata
 from newrelic.common.encoding_utils import json_encode
+from newrelic.common.metric_utils import create_metric_identity
 from newrelic.common.object_names import parse_exc_info
 from newrelic.common.streaming_utils import StreamBuffer
 from newrelic.core.attribute import (
@@ -180,6 +181,11 @@ def merge_custom_metric(self, value):
 
         self.merge_raw_time_metric(value)
 
+    def merge_dimensional_metric(self, value):
+        """Merge data value."""
+
+        self.merge_raw_time_metric(value)
+
 
 class CountStats(TimeStats):
     def merge_stats(self, other):
@@ -234,6 +240,35 @@ def reset_metric_stats(self):
         """
         self.__stats_table = {}
 
+class DimensionalMetrics(CustomMetrics):
+
+    """Extends CustomMetrics to allow a set of tags for metrics."""
+
+    def __contains__(self, key):
+        if not isinstance(key[1], frozenset):
+            # Convert tags dict to a frozen set for proper comparisons
+            key = create_metric_identity(*key)
+        return key in self.__stats_table
+
+    def record_dimensional_metric(self, name, value, tags=None):
+        """Record a single value metric, merging the data with any data
+        from prior value metrics with the same name.
+
+        """
+
+        key = create_metric_identity(name, tags)
+        self.record_custom_metric(key, value)
+
+class DimensionalStatsTable(dict):
+
+    """Extends dict to coerce a set of tags to a hashable identity."""
+
+    def __contains__(self, key):
+        if key[1] is not None and not isinstance(key[1], frozenset):
+            # Convert tags dict to a frozen set for proper comparisons
+            key = create_metric_identity(*key)
+        return super(DimensionalStatsTable, self).__contains__(key)
+
 
 class SlowSqlStats(list):
     def __init__(self):
@@ -433,6 +468,7 @@ class StatsEngine(object):
     def __init__(self):
         self.__settings = None
         self.__stats_table = {}
+        self.__dimensional_stats_table = DimensionalStatsTable()
         self._transaction_events = SampledDataSet()
         self._error_events = SampledDataSet()
         self._custom_events = SampledDataSet()
@@ -457,6 +493,10 @@ def settings(self):
     def stats_table(self):
         return self.__stats_table
 
+    @property
+    def dimensional_stats_table(self):
+        return self.__dimensional_stats_table
+
     @property
     def transaction_events(self):
         return self._transaction_events
@@ -499,7 +539,7 @@ def metrics_count(self):
 
         """
 
-        return len(self.__stats_table)
+        return len(self.__stats_table) + len(self.__dimensional_stats_table)
 
     def record_apdex_metric(self, metric):
         """Record a single apdex metric, merging the data with any data
@@ -887,6 +927,44 @@ def record_custom_metrics(self, metrics):
         for name, value in metrics:
             self.record_custom_metric(name, value)
 
+    def record_dimensional_metric(self, name, value, tags=None):
+        """Record a single value metric, merging the data with any data
+        from prior value metrics with the same name.
+
+        """
+        if isinstance(value, dict):
+            if len(value) == 1 and "count" in value:
+                new_stats = CountStats(call_count=value["count"])
+            else:
+                new_stats = TimeStats(*c2t(**value))
+        else:
+            new_stats = TimeStats(1, value, value, value, value, value**2)
+
+        key = create_metric_identity(name, tags)
+        stats = self.__dimensional_stats_table.get(key)
+        if stats is None:
+            self.__dimensional_stats_table[key] = new_stats
+        else:
+            stats.merge_stats(new_stats)
+
+        return key
+
+    def record_dimensional_metrics(self, metrics):
+        """Record the value metrics supplied by the iterable, merging
+        the data with any data from prior value metrics with the same
+        name.
+
+        """
+
+        if not self.__settings:
+            return
+
+        for metric in metrics:
+            name, value = metric[:2]
+            tags = metric[2] if len(metric) >= 3 else None
+
+            self.record_dimensional_metric(name, value, tags)
+
     def record_slow_sql_node(self, node):
         """Record a single sql metric, merging the data with any data
         from prior sql metrics for the same sql key.
@@ -997,6 +1075,8 @@ def record_transaction(self, transaction):
 
         self.merge_custom_metrics(transaction.custom_metrics.metrics())
 
+        self.merge_dimensional_metrics(transaction.dimensional_metrics.metrics())
+
         self.record_time_metrics(transaction.time_metrics(self))
 
         # Capture any errors if error collection is enabled.
@@ -1186,6 +1266,67 @@ def metric_data_count(self):
 
         return len(self.__stats_table)
 
+    def dimensional_metric_data(self, normalizer=None):
+        """Returns a list containing the low level metric data for
+        sending to the core application pertaining to the reporting
+        period. This consists of tuple pairs where first is dictionary
+        with name and scope keys with corresponding values, or integer
+        identifier if metric had an entry in dictionary mapping metric
+        (name, tags) as supplied from core application. The second is
+        the list of accumulated metric data, the list always being of
+        length 6.
+
+        """
+
+        if not self.__settings:
+            return []
+
+        result = []
+        normalized_stats = {}
+
+        # Metric Renaming and Re-Aggregation. After applying the metric
+        # renaming rules, the metrics are re-aggregated to collapse the
+        # metrics with same names after the renaming.
+
+        if self.__settings.debug.log_raw_metric_data:
+            _logger.info(
+                "Raw dimensional metric data for harvest of %r is %r.",
+                self.__settings.app_name,
+                list(six.iteritems(self.__dimensional_stats_table)),
+            )
+
+        if normalizer is not None:
+            for key, value in six.iteritems(self.__dimensional_stats_table):
+                key = (normalizer(key[0])[0], key[1])
+                stats = normalized_stats.get(key)
+                if stats is None:
+                    normalized_stats[key] = copy.copy(value)
+                else:
+                    stats.merge_stats(value)
+        else:
+            normalized_stats = self.__dimensional_stats_table
+
+        if self.__settings.debug.log_normalized_metric_data:
+            _logger.info(
+                "Normalized metric data for harvest of %r is %r.",
+                self.__settings.app_name,
+                list(six.iteritems(normalized_stats)),
+            )
+
+        for key, value in six.iteritems(normalized_stats):
+            key = dict(name=key[0], scope=key[1])
+            result.append((key, value))
+
+        return result
+
+    def dimensional_metric_data_count(self):
+        """Returns a count of the number of unique metrics."""
+
+        if not self.__settings:
+            return 0
+
+        return len(self.__dimensional_stats_table)
+
     def error_data(self):
         """Returns a to a list containing any errors collected during
         the reporting period.
@@ -1464,6 +1605,7 @@ def reset_stats(self, settings, reset_stream=False):
 
         self.__settings = settings
         self.__stats_table = {}
+        self.__dimensional_stats_table = {}
         self.__sql_stats_table = {}
         self.__slow_transaction = None
         self.__slow_transaction_map = {}
@@ -1491,6 +1633,7 @@ def reset_metric_stats(self):
         """
 
         self.__stats_table = {}
+        self.__dimensional_stats_table = {}
 
     def reset_transaction_events(self):
         """Resets the accumulated statistics back to initial state for
@@ -1827,6 +1970,24 @@ def merge_custom_metrics(self, metrics):
             else:
                 stats.merge_stats(other)
 
+    def merge_dimensional_metrics(self, metrics):
+        """
+        Merges in a set of dimensional metrics. The metrics should be
+        provide as an iterable where each item is a tuple of the metric
+        key and the accumulated stats for the metric. The metric key should
+        also be a tuple, containing a name and attribute filtered frozenset of tags.
+        """
+
+        if not self.__settings:
+            return
+
+        for key, other in metrics:
+            stats = self.__dimensional_stats_table.get(key)
+            if not stats:
+                self.__dimensional_stats_table[key] = other
+            else:
+                stats.merge_stats(other)
+
     def _snapshot(self):
         copy = object.__new__(StatsEngineSnapshot)
         copy.__dict__.update(self.__dict__)
diff --git a/newrelic/core/transaction_node.py b/newrelic/core/transaction_node.py
index 056d45a485..d63d7f9b65 100644
--- a/newrelic/core/transaction_node.py
+++ b/newrelic/core/transaction_node.py
@@ -65,6 +65,7 @@
         "apdex_t",
         "suppress_apdex",
         "custom_metrics",
+        "dimensional_metrics",
         "guid",
         "cpu_time",
         "suppress_transaction_trace",
diff --git a/tests/agent_features/test_dimensional_metrics.py b/tests/agent_features/test_dimensional_metrics.py
new file mode 100644
index 0000000000..82ddfad890
--- /dev/null
+++ b/tests/agent_features/test_dimensional_metrics.py
@@ -0,0 +1,106 @@
+# Copyright 2010 New Relic, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+
+from newrelic.api.application import application_instance
+from newrelic.api.background_task import background_task
+from newrelic.api.transaction import record_dimensional_metric, record_dimensional_metrics
+from newrelic.common.metric_utils import create_metric_identity
+
+from testing_support.fixtures import reset_core_stats_engine
+from testing_support.validators.validate_transaction_metrics import validate_transaction_metrics
+from testing_support.validators.validate_dimensional_metrics_outside_transaction import validate_dimensional_metrics_outside_transaction
+
+
+_test_tags_examples = [
+    (None, None),
+    ({}, None),
+    ([], None),
+    ({"str": "a"}, frozenset({("str", "a")})),
+    ({"int": 1}, frozenset({("int", 1)})),
+    ({"float": 1.0}, frozenset({("float", 1.0)})),
+    ({"bool": True}, frozenset({("bool", True)})),
+    ({"list": [1]}, frozenset({("list", "[1]")})),
+    ({"dict": {"subtag": 1}}, frozenset({("dict", "{'subtag': 1}")})),
+    ([("tags-as-list", 1)], frozenset({("tags-as-list", 1)})),
+]
+
+
+@pytest.mark.parametrize("tags,expected", _test_tags_examples)
+def test_create_metric_identity(tags, expected):
+    name = "Metric"
+    output_name, output_tags = create_metric_identity(name, tags=tags)
+    assert output_name == name, "Name does not match."
+    assert output_tags == expected, "Output tags do not match."
+
+
+@pytest.mark.parametrize("tags,expected", _test_tags_examples)
+def test_record_dimensional_metric_inside_transaction(tags, expected):
+    @validate_transaction_metrics("test_record_dimensional_metric_inside_transaction", background_task=True, dimensional_metrics=[
+        ("Metric", expected, 1),
+    ])
+    @background_task(name="test_record_dimensional_metric_inside_transaction")
+    def _test():
+        record_dimensional_metric("Metric", 1, tags=tags)
+
+    _test()
+
+
+@pytest.mark.parametrize("tags,expected", _test_tags_examples)
+@reset_core_stats_engine()
+def test_record_dimensional_metric_outside_transaction(tags, expected):
+    @validate_dimensional_metrics_outside_transaction([("Metric", expected, 1)])
+    def _test():
+        app = application_instance()
+        record_dimensional_metric("Metric", 1, tags=tags, application=app)
+
+    _test()
+
+
+@pytest.mark.parametrize("tags,expected", _test_tags_examples)
+def test_record_dimensional_metrics_inside_transaction(tags, expected):
+    @validate_transaction_metrics("test_record_dimensional_metrics_inside_transaction", background_task=True, dimensional_metrics=[("Metric/1", expected, 1), ("Metric/2", expected, 1)])
+    @background_task(name="test_record_dimensional_metrics_inside_transaction")
+    def _test():
+        record_dimensional_metrics([("Metric/1", 1, tags), ("Metric/2", 1, tags)])
+
+    _test()
+
+
+@pytest.mark.parametrize("tags,expected", _test_tags_examples)
+@reset_core_stats_engine()
+def test_record_dimensional_metrics_outside_transaction(tags, expected):
+    @validate_dimensional_metrics_outside_transaction([("Metric/1", expected, 1), ("Metric/2", expected, 1)])
+    def _test():
+        app = application_instance()
+        record_dimensional_metrics([("Metric/1", 1, tags), ("Metric/2", 1, tags)], application=app)
+
+    _test()
+
+
+def test_dimensional_metrics_different_tags():
+    @validate_transaction_metrics("test_dimensional_metrics_different_tags", background_task=True, dimensional_metrics=[
+        ("Metric", frozenset({("tag", 1)}), 1),
+        ("Metric", frozenset({("tag", 2)}), 2),
+    ])
+    @background_task(name="test_dimensional_metrics_different_tags")
+    def _test():
+        record_dimensional_metrics([
+            ("Metric", 1, {"tag": 1}),
+            ("Metric", 1, {"tag": 2}),
+        ])
+        record_dimensional_metric("Metric", 1, {"tag": 2})
+
+    _test()
diff --git a/tests/agent_unittests/test_harvest_loop.py b/tests/agent_unittests/test_harvest_loop.py
index 5f14b270ce..15b67a81e1 100644
--- a/tests/agent_unittests/test_harvest_loop.py
+++ b/tests/agent_unittests/test_harvest_loop.py
@@ -32,7 +32,7 @@
 from newrelic.core.function_node import FunctionNode
 from newrelic.core.log_event_node import LogEventNode
 from newrelic.core.root_node import RootNode
-from newrelic.core.stats_engine import CustomMetrics, SampledDataSet
+from newrelic.core.stats_engine import CustomMetrics, SampledDataSet, DimensionalMetrics
 from newrelic.core.transaction_node import TransactionNode
 from newrelic.network.exceptions import RetryDataForRequest
 
@@ -132,6 +132,7 @@ def transaction_node(request):
         apdex_t=0.5,
         suppress_apdex=False,
         custom_metrics=CustomMetrics(),
+        dimensional_metrics=DimensionalMetrics(),
         guid="4485b89db608aece",
         cpu_time=0.0,
         suppress_transaction_trace=False,
@@ -824,6 +825,7 @@ def test_flexible_events_harvested(allowlist_event):
     app._stats_engine.log_events.add(LogEventNode(1653609717, "WARNING", "A", {}))
     app._stats_engine.span_events.add("span event")
     app._stats_engine.record_custom_metric("CustomMetric/Int", 1)
+    app._stats_engine.record_dimensional_metric("DimensionalMetric/Int", 1, tags={"tag": "tag"})
 
     assert app._stats_engine.transaction_events.num_seen == 1
     assert app._stats_engine.error_events.num_seen == 1
@@ -831,6 +833,7 @@ def test_flexible_events_harvested(allowlist_event):
     assert app._stats_engine.log_events.num_seen == 1
     assert app._stats_engine.span_events.num_seen == 1
     assert app._stats_engine.record_custom_metric("CustomMetric/Int", 1)
+    assert app._stats_engine.record_dimensional_metric("DimensionalMetric/Int", 1, tags={"tag": "tag"})
 
     app.harvest(flexible=True)
 
@@ -850,7 +853,8 @@ def test_flexible_events_harvested(allowlist_event):
     assert app._stats_engine.span_events.num_seen == num_seen
 
     assert ("CustomMetric/Int", "") in app._stats_engine.stats_table
-    assert app._stats_engine.metrics_count() > 1
+    assert ("DimensionalMetric/Int", frozenset({("tag", "tag")})) in app._stats_engine.dimensional_stats_table
+    assert app._stats_engine.metrics_count() > 3
 
 
 @pytest.mark.parametrize(
diff --git a/tests/testing_support/fixtures.py b/tests/testing_support/fixtures.py
index 07de22cf0a..bf0e80a677 100644
--- a/tests/testing_support/fixtures.py
+++ b/tests/testing_support/fixtures.py
@@ -176,7 +176,10 @@ def wrap_shutdown_agent(wrapped, instance, args, kwargs):
 
     def wrap_record_custom_metric(wrapped, instance, args, kwargs):
         def _bind_params(name, value, *args, **kwargs):
-            return name
+            if isinstance(name, tuple):
+                return name[0]
+            else:
+                return name
 
         metric_name = _bind_params(*args, **kwargs)
         if (
diff --git a/tests/testing_support/validators/validate_dimensional_metrics_outside_transaction.py b/tests/testing_support/validators/validate_dimensional_metrics_outside_transaction.py
new file mode 100644
index 0000000000..7a3272bad7
--- /dev/null
+++ b/tests/testing_support/validators/validate_dimensional_metrics_outside_transaction.py
@@ -0,0 +1,93 @@
+# Copyright 2010 New Relic, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import copy
+
+from testing_support.fixtures import catch_background_exceptions
+from newrelic.common.object_wrapper import transient_function_wrapper, function_wrapper
+
+
+def validate_dimensional_metrics_outside_transaction(dimensional_metrics=None):
+    dimensional_metrics = dimensional_metrics or []
+
+    @function_wrapper
+    def _validate_wrapper(wrapped, instance, args, kwargs):
+
+        record_dimensional_metric_called = []
+        recorded_metrics = [None]
+
+        @transient_function_wrapper("newrelic.core.stats_engine", "StatsEngine.record_dimensional_metric")
+        @catch_background_exceptions
+        def _validate_dimensional_metrics_outside_transaction(wrapped, instance, args, kwargs):
+            record_dimensional_metric_called.append(True)
+            try:
+                result = wrapped(*args, **kwargs)
+            except:
+                raise
+            else:
+                metrics = instance.dimensional_stats_table
+                # Record a copy of the metric value so that the values aren't
+                # merged in the future
+                _metrics = {}
+                for k, v in metrics.items():
+                    _metrics[k] = copy.copy(v)
+                recorded_metrics[0] = _metrics
+
+            return result
+
+        def _validate(metrics, name, tags, count):
+            key = (name, tags)
+            metric = metrics.get(key)
+
+            def _metrics_table():
+                out = [""]
+                out.append("Expected: {0}: {1}".format(key, count))
+                for metric_key, metric_value in metrics.items():
+                    out.append("{0}: {1}".format(metric_key, metric_value[0]))
+                return "\n".join(out)
+
+            def _metric_details():
+                return "metric=%r, count=%r" % (key, metric.call_count)
+
+            if count is not None:
+                assert metric is not None, _metrics_table()
+                if count == "present":
+                    assert metric.call_count > 0, _metric_details()
+                else:
+                    assert metric.call_count == count, _metric_details()
+
+                assert metric.total_call_time >= 0, (key, metric)
+                assert metric.total_exclusive_call_time >= 0, (key, metric)
+                assert metric.min_call_time >= 0, (key, metric)
+                assert metric.sum_of_squares >= 0, (key, metric)
+
+            else:
+                assert metric is None, _metrics_table()
+
+        _new_wrapper = _validate_dimensional_metrics_outside_transaction(wrapped)
+        val = _new_wrapper(*args, **kwargs)
+        assert record_dimensional_metric_called
+        metrics = recorded_metrics[0]
+
+        record_dimensional_metric_called[:] = []
+        recorded_metrics[:] = []
+
+        for dimensional_metric, dimensional_tags, count in dimensional_metrics:
+            if isinstance(dimensional_tags, dict):
+                dimensional_tags = frozenset(dimensional_tags.items())
+            _validate(metrics, dimensional_metric, dimensional_tags, count)
+
+        return val
+
+    return _validate_wrapper
diff --git a/tests/testing_support/validators/validate_transaction_metrics.py b/tests/testing_support/validators/validate_transaction_metrics.py
index 7122b009aa..63c5b3551a 100644
--- a/tests/testing_support/validators/validate_transaction_metrics.py
+++ b/tests/testing_support/validators/validate_transaction_metrics.py
@@ -27,11 +27,13 @@ def validate_transaction_metrics(
     scoped_metrics=None,
     rollup_metrics=None,
     custom_metrics=None,
+    dimensional_metrics=None,
     index=-1,
 ):
     scoped_metrics = scoped_metrics or []
     rollup_metrics = rollup_metrics or []
     custom_metrics = custom_metrics or []
+    dimensional_metrics = dimensional_metrics or []
 
     if background_task:
         unscoped_metrics = [
@@ -56,6 +58,7 @@ def _validate_wrapper(wrapped, instance, args, kwargs):
 
         record_transaction_called = []
         recorded_metrics = []
+        recorded_dimensional_metrics = []
 
         @transient_function_wrapper("newrelic.core.stats_engine", "StatsEngine.record_transaction")
         @catch_background_exceptions
@@ -74,6 +77,14 @@ def _validate_transaction_metrics(wrapped, instance, args, kwargs):
                     _metrics[k] = copy.copy(v)
                 recorded_metrics.append(_metrics)
 
+                metrics = instance.dimensional_stats_table
+                # Record a copy of the metric value so that the values aren't
+                # merged in the future
+                _metrics = {}
+                for k, v in metrics.items():
+                    _metrics[k] = copy.copy(v)
+                recorded_dimensional_metrics.append(_metrics)
+
             return result
 
         def _validate(metrics, name, scope, count):
@@ -109,9 +120,11 @@ def _metric_details():
         val = _new_wrapper(*args, **kwargs)
         assert record_transaction_called
         metrics = recorded_metrics[index]
+        captured_dimensional_metrics = recorded_dimensional_metrics[index]
 
         record_transaction_called[:] = []
         recorded_metrics[:] = []
+        recorded_dimensional_metrics[:] = []
 
         for unscoped_metric in unscoped_metrics:
             _validate(metrics, unscoped_metric, "", 1)
@@ -125,6 +138,11 @@ def _metric_details():
         for custom_name, custom_count in custom_metrics:
             _validate(metrics, custom_name, "", custom_count)
 
+        for dimensional_name, dimensional_tags, dimensional_count in dimensional_metrics:
+            if isinstance(dimensional_tags, dict):
+                dimensional_tags = frozenset(dimensional_tags.items())
+            _validate(captured_dimensional_metrics, dimensional_name, dimensional_tags, dimensional_count)
+
         custom_metric_names = {name for name, _ in custom_metrics}
         for name, _ in metrics:
             if name not in custom_metric_names:
@@ -132,4 +150,4 @@ def _metric_details():
 
         return val
 
-    return _validate_wrapper
\ No newline at end of file
+    return _validate_wrapper
diff --git a/tox.ini b/tox.ini
index adc3206f80..480853b9e6 100644
--- a/tox.ini
+++ b/tox.ini
@@ -170,6 +170,9 @@ envlist =
     python-template_mako-{py27,py37,py310,py311}
 
 [testenv]
+basepython = 
+    py27: 2.7
+    pypy: pypy2.7
 deps =
     # Base Dependencies
     {py37,py38,py39,py310,py311,pypy37}: pytest==7.2.2