diff --git a/snowflake/assets/configuration/spec.yaml b/snowflake/assets/configuration/spec.yaml
index 2e439e828b18e..9c34cc767456c 100644
--- a/snowflake/assets/configuration/spec.yaml
+++ b/snowflake/assets/configuration/spec.yaml
@@ -165,3 +165,5 @@ files:
           min_collection_interval.display_default: 3600
           min_collection_interval.value.example: 3600
           min_collection_interval.enabled: 3600
+          disable_generic_tags.hidden: False
+          disable_generic_tags.enabled: True
diff --git a/snowflake/datadog_checks/snowflake/check.py b/snowflake/datadog_checks/snowflake/check.py
index f33c6634268f0..f2f3c56afc5d7 100644
--- a/snowflake/datadog_checks/snowflake/check.py
+++ b/snowflake/datadog_checks/snowflake/check.py
@@ -5,7 +5,7 @@
 
 import snowflake.connector as sf
 
-from datadog_checks.base import AgentCheck, ConfigurationError
+from datadog_checks.base import AgentCheck, ConfigurationError, to_native_string
 from datadog_checks.base.utils.db import QueryManager
 
 from . import queries
@@ -156,3 +156,21 @@ def _collect_version(self):
         else:
             if version:
                 self.set_metadata('version', version)
+
+    # override
+    def _normalize_tags_type(self, tags, device_name=None, metric_name=None):
+        if self.disable_generic_tags:
+            return super(SnowflakeCheck, self)._normalize_tags_type(tags, device_name, metric_name)
+
+        # If disable_generic_tags is not enabled, for each generic tag we emmit both the generic and the non generic
+        # version to ease transition.
+        normalized_tags = []
+        for tag in tags:
+            if tag is not None:
+                try:
+                    tag = to_native_string(tag)
+                except UnicodeError:
+                    self.log.warning('Encoding error with tag `%s` for metric `%s`, ignoring tag', tag, metric_name)
+                    continue
+                normalized_tags.extend(list({tag, self.degeneralise_tag(tag)}))
+        return normalized_tags
diff --git a/snowflake/datadog_checks/snowflake/data/conf.yaml.example b/snowflake/datadog_checks/snowflake/data/conf.yaml.example
index 3c9a5921550d0..d055431750c5f 100644
--- a/snowflake/datadog_checks/snowflake/data/conf.yaml.example
+++ b/snowflake/datadog_checks/snowflake/data/conf.yaml.example
@@ -218,3 +218,9 @@ instances:
     ## This is useful for cluster-level checks.
     #
     # empty_default_hostname: false
+
+    ## @param disable_generic_tags - boolean - optional - default: false
+    ## Generic tags such as `cluster` will be replaced by <integration_name>_cluster to avoid
+    ## getting mixed with other integraton tags.
+    #
+    disable_generic_tags: true
diff --git a/snowflake/setup.py b/snowflake/setup.py
index 9c3d0a872cb4d..883046bf10244 100644
--- a/snowflake/setup.py
+++ b/snowflake/setup.py
@@ -27,7 +27,7 @@ def get_dependencies():
         return f.readlines()
 
 
-CHECKS_BASE_REQ = 'datadog-checks-base>=20.1.0'
+CHECKS_BASE_REQ = 'datadog-checks-base>=21.2.0'
 
 
 setup(
diff --git a/snowflake/tests/common.py b/snowflake/tests/common.py
index 2524b56102920..91f2f7a2a8c63 100644
--- a/snowflake/tests/common.py
+++ b/snowflake/tests/common.py
@@ -15,6 +15,7 @@
     'database': 'SNOWFLAKE',
     'schema': 'ACCOUNT_USAGE',
     'role': 'ACCOUNTADMIN',
+    'disable_generic_tags': True,
 }
 OAUTH_INSTANCE = {
     'user': 'testuser',
@@ -24,6 +25,7 @@
     'role': 'ACCOUNTADMIN',
     'authenticator': 'oauth',
     'token': 'testtoken',
+    'disable_generic_tags': True,
 }
 
 EXPECTED_TAGS = ['account:test_acct.us-central1.gcp']
diff --git a/snowflake/tests/test_e2e.py b/snowflake/tests/test_e2e.py
index cc69fa4708b99..37122f2d5263d 100644
--- a/snowflake/tests/test_e2e.py
+++ b/snowflake/tests/test_e2e.py
@@ -31,11 +31,11 @@ def test_mock_data(dd_agent_check, datadog_agent, instance):
 
     aggregator.assert_metric(
         'snowflake.billing.cloud_service.sum',
-        tags=EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'service:COMPUTE_WH'],
+        tags=EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'snowflake_service:COMPUTE_WH'],
     )
     aggregator.assert_metric(
         'snowflake.billing.cloud_service.avg',
-        tags=EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'service:COMPUTE_WH'],
+        tags=EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'snowflake_service:COMPUTE_WH'],
     )
     aggregator.assert_metric('snowflake.billing.total_credit.sum')
     aggregator.assert_metric('snowflake.billing.total_credit.avg')
diff --git a/snowflake/tests/test_snowflake.py b/snowflake/tests/test_snowflake.py
index 140ffdf46a3bf..4e8e7698f193d 100644
--- a/snowflake/tests/test_snowflake.py
+++ b/snowflake/tests/test_snowflake.py
@@ -68,7 +68,7 @@ def test_credit_usage_metrics(dd_run_check, aggregator, instance):
             Decimal('1.066997500000'),
         )
     ]
-    expected_tags = EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'service:COMPUTE_WH']
+    expected_tags = EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'snowflake_service:COMPUTE_WH']
     with mock.patch('datadog_checks.snowflake.SnowflakeCheck.execute_query_raw', return_value=expected_credit_usage):
         check = SnowflakeCheck(CHECK_NAME, {}, [instance])
         check._conn = mock.MagicMock()
diff --git a/snowflake/tests/test_unit.py b/snowflake/tests/test_unit.py
index 550cec264b8ea..0011edc98d053 100644
--- a/snowflake/tests/test_unit.py
+++ b/snowflake/tests/test_unit.py
@@ -8,7 +8,7 @@
 
 from datadog_checks.snowflake import SnowflakeCheck, queries
 
-from .common import CHECK_NAME
+from .common import CHECK_NAME, EXPECTED_TAGS
 
 PROXY_CONFIG = {'http': 'http_host', 'https': 'https_host', 'no_proxy': 'uri1,uri2;uri3,uri4'}
 INVALID_PROXY = {'http': 'unused', 'https': 'unused', 'no_proxy': 'unused'}
@@ -204,3 +204,21 @@ def test_metric_group_exceptions(instance):
         check.log.warning.assert_called_once_with(
             "Invalid metric_groups found in snowflake conf.yaml: fake.metric.group"
         )
+
+
+def test_emit_generic_and_non_generic_tags_by_default(instance):
+    instance = copy.deepcopy(instance)
+    instance['disable_generic_tags'] = False
+    check = SnowflakeCheck(CHECK_NAME, {}, [instance])
+    tags = EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'service:COMPUTE_WH']
+    normalised_tags = tags + ['snowflake_service:COMPUTE_WH']
+    assert set(normalised_tags) == set(check._normalize_tags_type(tags))
+
+
+def test_emit_non_generic_tags_when_disabled(instance):
+    instance = copy.deepcopy(instance)
+    instance['disable_generic_tags'] = True
+    check = SnowflakeCheck(CHECK_NAME, {}, [instance])
+    tags = EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'service:COMPUTE_WH']
+    normalised_tags = EXPECTED_TAGS + ['service_type:WAREHOUSE_METERING', 'snowflake_service:COMPUTE_WH']
+    assert set(normalised_tags) == set(check._normalize_tags_type(tags))
diff --git a/snowflake/tox.ini b/snowflake/tox.ini
index c8bfe4b3f12b4..deb24c0ca5a33 100644
--- a/snowflake/tox.ini
+++ b/snowflake/tox.ini
@@ -26,5 +26,3 @@ passenv =
 commands =
     pip install -r requirements.in
     pytest -v {posargs}
-setenv =
-    DDEV_SKIP_GENERIC_TAGS_CHECK=true