diff --git a/.github/workflows/test_on_push.yaml b/.github/workflows/test_on_push.yaml
index bcf6d348c..71fa0da09 100644
--- a/.github/workflows/test_on_push.yaml
+++ b/.github/workflows/test_on_push.yaml
@@ -13,13 +13,19 @@ jobs:
       matrix:
         # We need 1.10.6 here to check that module works with
         # old Tarantool versions that don't have "tuple-keydef"/"tuple-merger" support.
-        tarantool-version: ["1.10.6", "1.10", "2.2", "2.3", "2.4", "2.5", "2.6", "2.7"]
+        tarantool-version: ["1.10.6", "1.10", "2.2", "2.3", "2.4", "2.5", "2.6", "2.7", "2.8"]
+        metrics-version: [""]
         remove-merger: [false]
         include:
           - tarantool-version: "2.7"
             remove-merger: true
+          - tarantool-version: "2.8"
+            metrics-version: "0.1.8"
+          - tarantool-version: "2.8"
+            metrics-version: "0.9.0"
           - tarantool-version: "2.8"
             coveralls: true
+            metrics-version: "0.12.0"
       fail-fast: false
     runs-on: [ubuntu-latest]
     steps:
@@ -47,6 +53,10 @@ jobs:
           tarantool --version
           ./deps.sh
 
+      - name: Install metrics
+        if: matrix.metrics-version != ''
+        run: tarantoolctl rocks install metrics ${{ matrix.metrics-version }}
+
       - name: Remove external merger if needed
         if: ${{ matrix.remove-merger }}
         run: rm .rocks/lib/tarantool/tuple/merger.so
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0a3d9325f..403ddf61c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 * Statistics for CRUD operations on router (#224).
+* Integrate CRUD statistics with `metrics` (#224).
 
 ### Changed
 
diff --git a/README.md b/README.md
index c523fbdc6..a46fa362e 100644
--- a/README.md
+++ b/README.md
@@ -603,9 +603,23 @@ crud.disable_stats()
 -- Enable statistics collect and recreates all collectors.
 crud.enable_stats()
 ```
-While statistics collection should not affect performance
-in a noticeable way, you may disable it if you want to
-prioritize performance.
+
+If [`metrics`](https://github.com/tarantool/metrics) found,
+you can use metrics collectors to store statistics
+instead of local collectors.
+It is required to use version `0.9.0` or greater,
+otherwise local collectors will be used.
+```
+-- Use metrics collectors.
+crud.enable_stats({ driver = 'metrics' })
+```
+By default, local collectors (`{ driver = 'local' }`)
+are used. Metrics collectors are much sophisticated and
+would show execution time quantile with aging for calls.
+Be wary that computing quantiles may affect overall
+performance under high load. Using local
+collectors or disabling stats is an option if
+you want to prioritize performance.
 
 Enabling stats on non-router instances is meaningless.
 
@@ -631,9 +645,34 @@ crud.stats()['insert']
 Each section contains different collectors for success calls
 and error (both error throw and `nil, err`) returns. `count`
 is total requests count since instance start or stats restart.
-`latency` is average time of requests execution,
+`latency` is 0.99 quantile of request execution time if `metrics`
+driver used, otherwise `latency` is total average.
 `time` is total time of requests execution.
 
+In `metrics` registry statistics are stored as `tnt_crud_stats` metrics
+with `operation` and `status` label_pairs.
+```
+metrics:collect()
+---
+- - label_pairs:
+      status: ok
+      operation: insert
+    value: 221411
+    metric_name: tnt_crud_stats_count
+  - label_pairs:
+      status: ok
+      operation: insert
+    value: 10.49834896344692
+    metric_name: tnt_crud_stats_sum
+  - label_pairs:
+      status: ok
+      operation: insert
+      quantile: 0.99
+    value: 0.00023606420935973
+    metric_name: tnt_crud_stats
+...
+```
+
 Additionally, `select` section contains `details` collectors.
 ```lua
 crud.stats()['select']['details']
@@ -647,7 +686,10 @@ crud.stats()['select']['details']
 (including those not executed successfully). `tuples_fetched`
 is a count of tuples fetched from storages during execution,
 `tuples_lookup` is a count of tuples looked up on storages
-while collecting response for call.
+while collecting response for call. In `metrics` registry they
+are stored as `tnt_crud_map_reduces`, `tnt_crud_tuples_fetched`
+and `tnt_crud_tuples_lookup` metrics with
+`{ operation = 'select' }` label_pairs.
 
 ## Cartridge roles
 
diff --git a/crud/stats/metrics_registry.lua b/crud/stats/metrics_registry.lua
new file mode 100644
index 000000000..75b24a34b
--- /dev/null
+++ b/crud/stats/metrics_registry.lua
@@ -0,0 +1,232 @@
+local is_package, metrics = pcall(require, 'metrics')
+
+local label = require('crud.stats.label')
+local dev_checks = require('crud.common.dev_checks')
+local registry_common = require('crud.stats.registry_common')
+
+local registry = {}
+local _registry = {}
+
+local metric_name = {
+    -- Summary collector for all operations.
+    op = 'tnt_crud_stats',
+    -- `*_count` and `*_sum` are automatically created
+    -- by summary collector.
+    op_count = 'tnt_crud_stats_count',
+    op_sum = 'tnt_crud_stats_sum',
+
+    -- Counter collectors for select/pairs details.
+    tuples_fetched = 'tnt_crud_tuples_fetched',
+    tuples_lookup = 'tnt_crud_tuples_lookup',
+    map_reduces = 'tnt_crud_map_reduces',
+}
+
+local LATENCY_QUANTILE = 0.99
+
+-- Raising quantile tolerance (1e-2) may result in crucial
+-- performance drops.
+local DEFAULT_QUANTILES = {
+    [LATENCY_QUANTILE] = 1e-2,
+}
+
+local DEFAULT_SUMMARY_PARAMS = {
+    age_buckets_count = 2,
+    max_age_time = 60,
+}
+
+--- Check if application supports metrics rock for registry
+--
+--  `metrics >= 0.9.0` is required to use summary with
+--  age buckets. `metrics >= 0.5.0, < 0.9.0` is unsupported
+--  due to quantile overflow bug
+--  (https://github.com/tarantool/metrics/issues/235).
+--
+-- @function is_supported
+--
+-- @treturn boolean Returns true if `metrics >= 0.9.0` found, false otherwise.
+--
+function registry.is_supported()
+    if is_package == false then
+        return false
+    end
+
+    -- Only metrics >= 0.9.0 supported.
+    local is_summary, summary = pcall(require, 'metrics.collectors.summary')
+    if is_summary == false or summary.rotate_age_buckets == nil then
+        return false
+    end
+
+    return true
+end
+
+
+--- Initialize collectors in global metrics registry
+--
+-- @function init
+--
+-- @treturn boolean Returns true.
+--
+function registry.init()
+    _registry[metric_name.op] = metrics.summary(
+        metric_name.op,
+        'CRUD router calls statistics',
+        DEFAULT_QUANTILES,
+        DEFAULT_SUMMARY_PARAMS)
+
+    _registry[metric_name.tuples_fetched] = metrics.counter(
+        metric_name.tuples_fetched,
+        'Tuples fetched from CRUD storages during select/pairs')
+
+    _registry[metric_name.tuples_lookup] = metrics.counter(
+        metric_name.tuples_lookup,
+        'Tuples looked up on CRUD storages while collecting response during select/pairs')
+
+    _registry[metric_name.map_reduces] = metrics.counter(
+        metric_name.map_reduces,
+        'Map reduces planned during CRUD select/pairs')
+
+    return true
+end
+
+--- Unregister collectors in global metrics registry
+--
+-- @function destroy
+--
+-- @treturn boolean Returns true.
+--
+function registry.destroy()
+    for _, c in pairs(_registry) do
+        metrics.registry:unregister(c)
+    end
+
+    _registry = {}
+    return true
+end
+
+--- Get copy of global metrics registry
+--
+-- @function get
+--
+-- @treturn table Returns copy of metrics registry.
+function registry.get()
+    local stats = {}
+
+    -- Fill empty collectors with zero values.
+    for _, op_label in pairs(label) do
+        stats[op_label] = registry_common.build_collector(op_label)
+    end
+
+    for _, obs in ipairs(_registry[metric_name.op]:collect()) do
+        local operation = obs.label_pairs.operation
+        local status = obs.label_pairs.status
+        if obs.metric_name == metric_name.op then
+            if obs.label_pairs.quantile == LATENCY_QUANTILE then
+                stats[operation][status].latency = obs.value
+            end
+        elseif obs.metric_name == metric_name.op_sum then
+            stats[operation][status].time = obs.value
+        elseif obs.metric_name == metric_name.op_count then
+            stats[operation][status].count = obs.value
+        end
+    end
+
+    local _, obs_tuples_fetched = next(_registry[metric_name.tuples_fetched]:collect())
+    if obs_tuples_fetched ~= nil then
+        stats[label.SELECT].details.tuples_fetched = obs_tuples_fetched.value
+    end
+
+    local _, obs_tuples_lookup = next(_registry[metric_name.tuples_lookup]:collect())
+    if obs_tuples_lookup ~= nil then
+        stats[label.SELECT].details.tuples_lookup = obs_tuples_lookup.value
+    end
+
+    local _, obs_map_reduces = next(_registry[metric_name.map_reduces]:collect())
+    if obs_map_reduces ~= nil then
+        stats[label.SELECT].details.map_reduces = obs_map_reduces.value
+    end
+
+    return stats
+end
+
+--- Increase requests count and update latency info
+--
+-- @function observe
+--
+-- @tparam string op_label
+--  Label of registry collectos.
+--  Use `require('crud.common.const').OP` to pick one.
+--
+-- @tparam boolean success
+--  true if no errors on execution, false otherwise.
+--
+-- @tparam number latency
+--  Time of call execution.
+--
+-- @treturn boolean Returns true.
+--
+
+local total = 0
+
+function registry.observe(op_label, success, latency)
+    dev_checks('string', 'boolean', 'number')
+
+    local label_pairs = { operation = op_label }
+    if success == true then
+        label_pairs.status = 'ok'
+    else
+        label_pairs.status = 'error'
+    end
+
+    local clock = require('clock')
+    local start = clock.monotonic()
+    _registry[metric_name.op]:observe(latency, label_pairs)
+    local diff = clock.monotonic() - start
+    -- require('log').error("latency: %f", latency)
+    -- require('log').error("diff: %f", diff)
+    total = total + diff
+    -- require('log').error("total: %f", total)
+
+    return true
+end
+
+--- Increase statistics of storage select/pairs calls
+--
+-- @function observe_fetch
+--
+-- @tparam number tuples_fetched
+--  Count of tuples fetched during storage call.
+--
+-- @tparam number tuples_lookup
+--  Count of tuples looked up on storages while collecting response.
+--
+-- @treturn boolean Returns true.
+--
+function registry.observe_fetch(tuples_fetched, tuples_lookup)
+    dev_checks('number', 'number')
+
+    local label_pairs = { operation = label.SELECT }
+
+    _registry[metric_name.tuples_fetched]:inc(tuples_fetched, label_pairs)
+    _registry[metric_name.tuples_lookup]:inc(tuples_lookup, label_pairs)
+    return true
+end
+
+--- Increase statistics of planned map reduces during select/pairs
+--
+-- @function observe_map_reduces
+--
+-- @tparam number count
+--  Count of map reduces planned.
+--
+-- @treturn boolean Returns true.
+--
+function registry.observe_map_reduces(count)
+    dev_checks('number')
+
+    local label_pairs = { operation = label.SELECT }
+
+    _registry[metric_name.map_reduces]:inc(count, label_pairs)
+    return true
+end
+
+return registry
diff --git a/crud/stats/module.lua b/crud/stats/module.lua
index 2d251e478..ba3deb7db 100644
--- a/crud/stats/module.lua
+++ b/crud/stats/module.lua
@@ -1,11 +1,23 @@
 local clock = require('clock')
+local checks = require('checks')
 local dev_checks = require('crud.common.dev_checks')
+local errors = require('errors')
 local utils = require('crud.common.utils')
 
-local stats_registry = require('crud.stats.local_registry')
+local StatsError = errors.new_class('StatsError', {capture_stack = false})
 
 local stats = {}
-local _is_enabled = false
+
+local stats_registry
+local local_registry = require('crud.stats.local_registry')
+local metrics_registry = require('crud.stats.metrics_registry')
+
+local drivers = {
+    ['local'] = local_registry,
+}
+if metrics_registry.is_supported() then
+    drivers['metrics'] = metrics_registry
+end
 
 --- Check if statistics module if enabled
 --
@@ -14,22 +26,43 @@ local _is_enabled = false
 -- @treturn[1] boolean Returns true or false.
 --
 function stats.is_enabled()
-    return _is_enabled
+    return stats_registry ~= nil
 end
 
 --- Initializes statistics registry and enables callbacks and wrappers
 --
 -- @function enable
 --
+-- @tparam table opts
+--
+--  @tfield string driver
+--   'local' or 'metrics' (default: 'local').
+--   If 'local', stores statistics in local registry (some Lua tables)
+--   and computes latency as overall average. 'metrics' requires
+--   `metrics >= 0.9.0` installed and stores statistics in
+--   global metrics registry (integrated with exporters)
+--   and computes latency as 0.99 quantile with aging. `metrics`
+--   driver may drastically affect performance on high load.
+--
 -- @treturn[1] boolean Returns true.
 --
-function stats.enable()
-    if _is_enabled then
+function stats.enable(opts)
+    checks({ driver = '?string' })
+
+    opts = opts or {}
+    opts.driver = opts.driver or 'local'
+
+    StatsError:assert(drivers[opts.driver] ~= nil, 'Unsupported driver: %s', opts.driver)
+
+    if stats_registry == drivers[opts.driver] then
         return true
     end
 
+    -- Disable old driver registry, if another one was requested.
+    stats.disable()
+
+    stats_registry = drivers[opts.driver]
     stats_registry.init()
-    _is_enabled = true
 
     return true
 end
@@ -41,8 +74,10 @@ end
 -- @treturn[1] boolean Returns true.
 --
 function stats.disable()
-    stats_registry.destroy()
-    _is_enabled = false
+    if stats.is_enabled() then
+        stats_registry.destroy()
+        stats_registry = nil
+    end
 
     return true
 end
@@ -55,7 +90,11 @@ end
 --  If statistics disabled, returns {}.
 --
 function stats.get()
-    return stats_registry.get()
+    if stats_registry ~= nil then
+        return stats_registry.get()
+    end
+
+    return {}
 end
 
 --- Wrap CRUD operation call to collect statistics
diff --git a/test/integration/stats_test.lua b/test/integration/stats_test.lua
index f6c0ad4a2..4352265fd 100644
--- a/test/integration/stats_test.lua
+++ b/test/integration/stats_test.lua
@@ -5,9 +5,16 @@ local t = require('luatest')
 
 local helpers = require('test.helper')
 
-local g = t.group('stats_integration')
+local pgroup = t.group('stats_integration', {
+    { driver = 'local' },
+    { driver = 'metrics' },
+})
 
-g.before_all(function(g)
+local pgroup_metrics = t.group('stats_metrics_integration', {
+    { driver = 'metrics' },
+})
+
+local before_all = function(g)
     g.cluster = helpers.Cluster:new({
         datadir = fio.tempdir(),
         server_command = helpers.entrypoint('srv_select'),
@@ -16,13 +23,33 @@ g.before_all(function(g)
     })
 
     g.cluster:start()
-end)
 
-g.after_all(function(g) helpers.stop_cluster(g.cluster) end)
+    local router = g.cluster:server('router').net_box
+    if g.params.driver == 'metrics' then
+        local is_metrics_supported = router:eval([[
+            return require('crud.stats.metrics_registry').is_supported()
+        ]])
+        t.skip_if(is_metrics_supported == false, 'Metrics registry is unsupported')
+    end
+
+    -- Enable required driver.
+    router:eval("require('crud').enable_stats(...)", { g.params })
+end
 
-g.before_each(function(g)
+local before_each = function(g)
     helpers.truncate_space_on_cluster(g.cluster, 'customers')
-end)
+end
+
+local after_all = function(g) helpers.stop_cluster(g.cluster) end
+
+pgroup.before_all(before_all)
+pgroup_metrics.before_all(before_all)
+
+pgroup.after_all(after_all)
+pgroup_metrics.after_all(after_all)
+
+pgroup.before_each(before_each)
+pgroup_metrics.before_each(before_each)
 
 local simple_operation_cases = {
     insert = {
@@ -200,10 +227,10 @@ for name, case in pairs(simple_operation_cases) do
     local test_name = ('test_%s_stats'):format(name)
 
     if case.prepare ~= nil then
-        g.before_test(test_name, case.prepare)
+        pgroup.before_test(test_name, case.prepare)
     end
 
-    g[test_name] = function(g)
+    pgroup[test_name] = function(g)
         local router = g.cluster:server('router').net_box
 
         -- Collect stats on active servers before call.
@@ -380,11 +407,11 @@ end
 for name, case in pairs(select_cases) do
     local test_name = ('test_%s_stats'):format(name)
 
-    g.before_test(test_name, function(g)
+    pgroup.before_test(test_name, function(g)
         prepare_select_data(g)
     end)
 
-    g[test_name] = function(g)
+    pgroup[test_name] = function(g)
         local router = g.cluster:server('router').net_box
         local op_label = 'select'
 
@@ -456,3 +483,253 @@ for name, case in pairs(select_cases) do
             'Expected count of map reduces planned')
     end
 end
+
+local function generate_stats(g)
+    local router = g.cluster:server('router').net_box
+    -- Generate non-null stats for all operations.
+    for _, case in pairs(simple_operation_cases) do
+        if case.prepare ~= nil then
+            case.prepare(g)
+        end
+        local _, err = router:call(case.func, case.args)
+        if case.expect_error ~= true then
+            t.assert_equals(err, nil)
+        end
+    end
+
+    prepare_select_data(g)
+    for _, case in pairs(select_cases) do
+        local _, err = router:eval(case.eval, { case.conditions })
+        if case.expect_error ~= true then
+            t.assert_equals(err, nil)
+        end
+    end
+end
+
+-- https://github.com/tarantool/metrics/blob/fc5a67072340b12f983f09b7d383aca9e2f10cf1/test/utils.lua#L22-L31
+local function find_obs(metric_name, label_pairs, observations)
+    for _, obs in pairs(observations) do
+        local same_label_pairs = pcall(t.assert_equals, obs.label_pairs, label_pairs)
+        if obs.metric_name == metric_name and same_label_pairs then
+            return obs
+        end
+    end
+    t.assert_items_include(
+        observations,
+        { metric_name = metric_name, label_pairs = label_pairs },
+        'Missing observation')
+end
+
+-- https://github.com/tarantool/metrics/blob/fc5a67072340b12f983f09b7d383aca9e2f10cf1/test/utils.lua#L55-L63
+local function find_metric(metric_name, metrics_data)
+    local m = {}
+    for _, v in ipairs(metrics_data) do
+        if v.metric_name == metric_name then
+            table.insert(m, v)
+        end
+    end
+    return #m > 0 and m or nil
+end
+
+local function get_unique_label_values(metrics_data, label_key)
+    local label_values_map = {}
+    for _, v in ipairs(metrics_data) do
+        local label_pairs = v.label_pairs or {}
+        if label_pairs[label_key] ~= nil then
+            label_values_map[label_pairs[label_key]] = true
+        end
+    end
+
+    local label_values = {}
+    for k, _ in pairs(label_values_map) do
+        table.insert(label_values, k)
+    end
+
+    return label_values
+end
+
+local function validate_stats(metrics)
+    local stats = find_metric('tnt_crud_stats', metrics)
+    t.assert_type(stats, 'table', '`tnt_crud_stats` summary metrics found')
+
+    local stats_count = find_metric('tnt_crud_stats_count', metrics)
+    t.assert_type(stats_count, 'table', '`tnt_crud_stats` summary metrics found')
+
+    local stats_sum = find_metric('tnt_crud_stats_sum', metrics)
+    t.assert_type(stats_sum, 'table', '`tnt_crud_stats` summary metrics found')
+
+
+    local expected_operations = { 'insert', 'get', 'replace', 'update',
+        'upsert', 'delete', 'select', 'truncate', 'len', 'borders' }
+
+    t.assert_items_equals(
+        get_unique_label_values(stats, 'operation'),
+        expected_operations,
+        'Metrics are labelled with operation')
+
+    t.assert_items_equals(
+        get_unique_label_values(stats_count, 'operation'),
+        expected_operations,
+        'Metrics are labelled with operation')
+
+    t.assert_items_equals(
+        get_unique_label_values(stats_sum, 'operation'),
+        expected_operations,
+        'Metrics are labelled with operation')
+
+
+    local expected_statuses = { 'ok', 'error' }
+
+    t.assert_items_equals(
+        get_unique_label_values(stats, 'status'),
+        expected_statuses,
+        'Metrics are labelled with status')
+
+    t.assert_items_equals(
+        get_unique_label_values(stats_count, 'status'),
+        expected_statuses,
+        'Metrics are labelled with operation')
+
+    t.assert_items_equals(
+        get_unique_label_values(stats_sum, 'status'),
+        expected_statuses,
+        'Metrics are labelled with operation')
+
+
+    local tuples_fetched = find_metric('tnt_crud_tuples_fetched', metrics)
+    t.assert_type(tuples_fetched, 'table', '`tnt_crud_tuples_fetched` metrics found')
+
+    t.assert_items_equals(
+        get_unique_label_values(tuples_fetched, 'operation'),
+        { 'select' },
+        'Metrics are labelled with operation')
+
+
+    local tuples_lookup = find_metric('tnt_crud_tuples_lookup', metrics)
+    t.assert_type(tuples_lookup, 'table', '`tnt_crud_tuples_lookup` metrics found')
+
+    t.assert_items_equals(
+        get_unique_label_values(tuples_lookup, 'operation'),
+        { 'select' },
+        'Metrics are labelled with operation')
+
+
+    local map_reduces = find_metric('tnt_crud_map_reduces', metrics)
+    t.assert_type(map_reduces, 'table', '`tnt_crud_map_reduces` metrics found')
+
+    t.assert_items_equals(
+        get_unique_label_values(map_reduces, 'operation'),
+        { 'select' },
+        'Metrics are labelled with operation')
+end
+
+pgroup_metrics.before_test(
+    'test_stats_stored_in_global_metrics_registry',
+    function(g) generate_stats(g) end
+)
+
+pgroup_metrics.test_stats_stored_in_global_metrics_registry = function(g)
+    local router = g.cluster:server('router').net_box
+    local metrics = router:eval("return require('metrics').collect()")
+    validate_stats(metrics)
+end
+
+pgroup_metrics.before_test(
+    'test_stats_in_metrics_registry_after_switch_to_metrics_driver',
+    function(g)
+        local router = g.cluster:server('router').net_box
+        router:eval("return require('crud').disable_stats()")
+    end
+)
+
+pgroup_metrics.test_stats_in_metrics_registry_after_switch_to_metrics_driver = function(g)
+    local router = g.cluster:server('router').net_box
+
+    -- Enable local driver.
+    router:eval("return require('crud').enable_stats({ driver = 'local' })")
+    -- Switch to metrics driver.
+    router:eval("return require('crud').enable_stats({ driver = 'metrics' })")
+
+    generate_stats(g)
+    local metrics = router:eval("return require('metrics').collect()")
+    validate_stats(metrics)
+end
+
+pgroup_metrics.before_test(
+    'test_metrics_updated_per_call',
+    function(g) generate_stats(g) end
+)
+
+pgroup_metrics.test_metrics_updated_per_call = function(g)
+    local router = g.cluster:server('router').net_box
+
+    local metrics_before = router:eval("return require('metrics').collect()")
+    local count_before = find_obs('tnt_crud_stats_count',
+        { operation = 'select', status = 'ok' }, metrics_before)
+    local time_before = find_obs('tnt_crud_stats_sum',
+        { operation = 'select', status = 'ok' }, metrics_before)
+    local tuples_lookup_before = find_obs('tnt_crud_tuples_lookup',
+        { operation = 'select' }, metrics_before)
+    local tuples_fetched_before = find_obs('tnt_crud_tuples_fetched',
+        { operation = 'select' }, metrics_before)
+    local map_reduces_before = find_obs('tnt_crud_map_reduces',
+        { operation = 'select' }, metrics_before)
+
+    local case = select_cases['select_by_secondary_index']
+    local _, err = router:eval(case.eval, { case.conditions })
+    t.assert_equals(err, nil)
+
+    local metrics_after = router:eval("return require('metrics').collect()")
+    local count_after = find_obs('tnt_crud_stats_count',
+        { operation = 'select', status = 'ok' }, metrics_after)
+    local time_after = find_obs('tnt_crud_stats_sum',
+        { operation = 'select', status = 'ok' }, metrics_after)
+    local tuples_lookup_after = find_obs('tnt_crud_tuples_lookup',
+        { operation = 'select' }, metrics_after)
+    local tuples_fetched_after = find_obs('tnt_crud_tuples_fetched',
+        { operation = 'select' }, metrics_after)
+    local map_reduces_after = find_obs('tnt_crud_map_reduces',
+        { operation = 'select' }, metrics_after)
+
+    t.assert_equals(count_after.value - count_before.value, 1,
+        '`select` metrics count increased')
+    t.assert_ge(time_after.value - time_before.value, 0,
+        '`select` total time increased')
+    t.assert_ge(tuples_lookup_after.value - tuples_lookup_before.value, case.tuples_lookup,
+        '`select` tuples lookup expected change')
+    t.assert_ge(tuples_fetched_after.value - tuples_fetched_before.value, case.tuples_lookup,
+        '`select` tuples feched expected change')
+    t.assert_ge(map_reduces_after.value - map_reduces_before.value, case.tuples_lookup,
+        '`select` map reduces expected change')
+end
+
+pgroup_metrics.before_test('test_metrics_collectors_destroyed_if_stats_disabled', function(g)
+    generate_stats(g)
+end)
+
+pgroup_metrics.test_metrics_collectors_destroyed_if_stats_disabled = function(g)
+    local router = g.cluster:server('router').net_box
+
+    local resp = router:eval("return require('crud').disable_stats()")
+    t.assert_equals(resp, true, 'Stats disabled on router')
+
+    local metrics = router:eval("return require('metrics').collect()")
+
+    local stats = find_metric('tnt_crud_stats', metrics)
+    t.assert_equals(stats, nil, '`tnt_crud_stats` summary metrics not found')
+
+    local stats_count = find_metric('tnt_crud_stats_count', metrics)
+    t.assert_equals(stats_count, nil, '`tnt_crud_stats` summary metrics not found')
+
+    local stats_sum = find_metric('tnt_crud_stats_sum', metrics)
+    t.assert_equals(stats_sum, nil, '`tnt_crud_stats` summary metrics not found')
+
+    local tuples_fetched = find_metric('tnt_crud_tuples_fetched', metrics)
+    t.assert_equals(tuples_fetched, nil, '`tnt_crud_tuples_fetched` metrics not found')
+
+    local tuples_lookup = find_metric('tnt_crud_tuples_lookup', metrics)
+    t.assert_equals(tuples_lookup, nil, '`tnt_crud_tuples_lookup` metrics not found')
+
+    local map_reduces = find_metric('tnt_crud_map_reduces', metrics)
+    t.assert_equals(map_reduces, nil, '`tnt_crud_map_reduces` metrics not found')
+end
diff --git a/test/unit/stats_test.lua b/test/unit/stats_test.lua
index 2277a416c..c28a42560 100644
--- a/test/unit/stats_test.lua
+++ b/test/unit/stats_test.lua
@@ -1,19 +1,33 @@
 local clock = require('clock')
 local fiber = require('fiber')
 local stats_module = require('crud.stats.module')
+local stats_metrics_registry = require('crud.stats.metrics_registry')
 
 local t = require('luatest')
-local g = t.group('stats_unit')
+local pgroup = t.group('stats_unit', {
+    { driver = 'local' },
+    { driver = 'metrics' },
+})
 
-g.before_each(function()
-    stats_module.enable()
+local group_driver = t.group('stats_driver_unit')
+
+pgroup.before_each(function(g)
+    if g.params.driver == 'metrics' then
+        local is_metrics_supported = stats_metrics_registry.is_supported()
+        t.skip_if(is_metrics_supported == false, 'Metrics registry is unsupported')
+    end
+    stats_module.enable(g.params)
 end)
 
-g.after_each(function()
+pgroup.after_each(function()
     stats_module.disable()
 end)
 
-g.test_get_returns_expected_format = function()
+group_driver.after_each(function()
+    stats_module.disable()
+end)
+
+pgroup.test_get_returns_expected_format = function()
     local stats = stats_module.get()
 
     t.assert_type(stats, 'table')
@@ -114,7 +128,7 @@ local measure_cases = {
 }
 
 for name, case in pairs(measure_cases) do
-    g[('test_%s'):format(name)] = function()
+    pgroup[('test_%s'):format(name)] = function()
         local stats_before = stats_module.get()
 
         local op_label = stats_module.label.INSERT
@@ -192,7 +206,7 @@ local preserve_throw_cases = {
 
 for name_head, disable_case in pairs(disable_stats_cases) do
     for name_tail, return_case in pairs(preserve_return_cases) do
-        g[('test_%s%s'):format(name_head, name_tail)] = function()
+        pgroup[('test_%s%s'):format(name_head, name_tail)] = function()
             local op_label = stats_module.label.INSERT
 
             disable_case.before_wrap()
@@ -205,7 +219,7 @@ for name_head, disable_case in pairs(disable_stats_cases) do
         end
     end
 
-    g[('test_%pairs_wrapper_preserves_return_values'):format(name_head)] = function()
+    pgroup[('test_%pairs_wrapper_preserves_return_values'):format(name_head)] = function()
         local input = { a = 'a', b = 'b' }
 
         local op_label = stats_module.label.INSERT
@@ -223,7 +237,7 @@ for name_head, disable_case in pairs(disable_stats_cases) do
     end
 
     for name_tail, throw_case in pairs(preserve_throw_cases) do
-        g[('test_%s%s'):format(name_head, name_tail)] = function()
+        pgroup[('test_%s%s'):format(name_head, name_tail)] = function()
             local op_label = stats_module.label.INSERT
 
             disable_case.before_wrap()
@@ -235,7 +249,7 @@ for name_head, disable_case in pairs(disable_stats_cases) do
     end
 end
 
-g.test_stats_is_empty_after_disable = function()
+pgroup.test_stats_is_empty_after_disable = function()
     stats_module.disable()
 
     stats_module.wrap(return_true, stats_module.label.INSERT)()
@@ -244,7 +258,7 @@ g.test_stats_is_empty_after_disable = function()
     t.assert_equals(stats, {})
 end
 
-g.test_stats_reenable_does_not_reset_stats = function()
+pgroup.test_stats_reenable_with_same_driver_does_not_reset_stats = function(g)
     -- Prepare non-default stats
     local op_label = stats_module.label.INSERT
     stats_module.wrap(return_true, op_label)()
@@ -252,14 +266,14 @@ g.test_stats_reenable_does_not_reset_stats = function()
     local stats_before = stats_module.get()
     t.assert_equals(stats_before[op_label].ok.count, 1, 'Non-zero stats prepared')
 
-    stats_module.enable()
+    stats_module.enable(g.params)
 
     local stats_after = stats_module.get()
 
     t.assert_equals(stats_after, stats_before, 'Stats have not been reset')
 end
 
-g.test_stats_fetch_callback = function()
+pgroup.test_stats_fetch_callback = function()
     local stats_before = stats_module.get()
 
     local storage_cursor_stats = { tuples_fetched = 5, tuples_lookup = 25 }
@@ -276,7 +290,7 @@ g.test_stats_fetch_callback = function()
     t.assert_equals(tuples_lookup_diff, 25, 'tuples_lookup is inremented by expected value')
 end
 
-g.test_disable_stats_before_fetch_callback_get_do_not_break_call = function()
+pgroup.test_disable_stats_before_fetch_callback_get_do_not_break_call = function()
     stats_module.disable()
 
     local storage_cursor_stats = { tuples_fetched = 5, tuples_lookup = 25 }
@@ -285,7 +299,7 @@ g.test_disable_stats_before_fetch_callback_get_do_not_break_call = function()
     t.success('No unexpected errors')
 end
 
-g.test_disable_stats_after_fetch_callback_get_do_not_break_call = function()
+pgroup.test_disable_stats_after_fetch_callback_get_do_not_break_call = function()
     local callback = stats_module.get_fetch_callback()
     stats_module.disable()
 
@@ -295,7 +309,7 @@ g.test_disable_stats_after_fetch_callback_get_do_not_break_call = function()
     t.success('No unexpected errors')
 end
 
-g.test_stats_map_reduce_increment = function()
+pgroup.test_stats_map_reduce_increment = function()
     local stats_before = stats_module.get()
 
     stats_module.inc_map_reduce_count()
@@ -309,10 +323,55 @@ g.test_stats_map_reduce_increment = function()
     t.assert_equals(map_reduces_diff, 1, 'map_reduces is inremented by 1')
 end
 
-g.test_disable_stats_do_not_break_inc_map_reduce_count_call = function()
+pgroup.test_disable_stats_do_not_break_inc_map_reduce_count_call = function()
     stats_module.disable()
 
     stats_module.inc_map_reduce_count()
 
     t.success('No unexpected errors')
 end
+
+group_driver.before_test(
+    'test_stats_reenable_with_different_driver_reset_stats',
+    function()
+        local is_metrics_supported = stats_metrics_registry.is_supported()
+        t.skip_if(is_metrics_supported == false, 'Metrics registry is unsupported')
+    end
+)
+
+group_driver.test_stats_reenable_with_different_driver_reset_stats = function()
+    stats_module.enable({ driver = 'metrics' })
+
+    -- Prepare non-default stats
+    local op_label = stats_module.label.INSERT
+    stats_module.wrap(return_true, op_label)()
+
+    local stats_before = stats_module.get()
+    t.assert_equals(stats_before[op_label].ok.count, 1, 'Non-zero stats prepared')
+
+    stats_module.enable({ driver = 'local' })
+
+    local stats_after = stats_module.get()
+
+    t.assert_equals(stats_after[op_label].ok.count, 0, 'Stats have been reset')
+end
+
+group_driver.test_unknown_driver_throws_error = function()
+    t.assert_error_msg_contains(
+        'Unsupported driver: unknown',
+        stats_module.enable, { driver = 'unknown' })
+end
+
+group_driver.before_test(
+    'test_stats_enable_with_metrics_throws_error_if_unsupported',
+    function()
+        local is_metrics_supported = stats_metrics_registry.is_supported()
+        t.skip_if(is_metrics_supported == true, 'Metrics registry is supported')
+    end
+)
+
+group_driver.test_stats_enable_with_metrics_throws_error_if_unsupported = function()
+    t.assert_error_msg_contains(
+        'Unsupported driver: metrics',
+        stats_module.enable, { driver = 'metrics' })
+end