diff --git a/lib/cloud_controller/metrics/request_metrics.rb b/lib/cloud_controller/metrics/request_metrics.rb
index 2d3e597a272..889082ed973 100644
--- a/lib/cloud_controller/metrics/request_metrics.rb
+++ b/lib/cloud_controller/metrics/request_metrics.rb
@@ -3,33 +3,21 @@
 module VCAP::CloudController
   module Metrics
     class RequestMetrics
-      def initialize(statsd=CloudController::DependencyLocator.instance.statsd_client, prometheus_updater=CloudController::DependencyLocator.instance.prometheus_updater)
-        @counter = 0
-        @statsd = statsd
+      def initialize(statsd_updater=CloudController::DependencyLocator.instance.statsd_updater, prometheus_updater=CloudController::DependencyLocator.instance.prometheus_updater)
+        @statsd_updater = statsd_updater
         @prometheus_updater = prometheus_updater
       end
 
       def start_request
-        @counter += 1
-        @statsd.gauge('cc.requests.outstanding.gauge', @counter)
-        @statsd.increment 'cc.requests.outstanding'
-
+        @statsd_updater.start_request
         @prometheus_updater.increment_gauge_metric(:cc_requests_outstanding_total)
       end
 
       def complete_request(status)
-        http_status_code = "#{status.to_s[0]}XX"
-        http_status_metric = "cc.http_status.#{http_status_code}"
-        @counter -= 1
-        @statsd.gauge('cc.requests.outstanding.gauge', @counter)
-        @statsd.batch do |batch|
-          batch.decrement 'cc.requests.outstanding'
-          batch.increment 'cc.requests.completed'
-          batch.increment http_status_metric
-        end
+        @statsd_updater.complete_request(status)
 
-        @prometheus_updater.decrement_gauge_metric(:cc_requests_outstanding_total)
         @prometheus_updater.increment_counter_metric(:cc_requests_completed_total)
+        @prometheus_updater.decrement_gauge_metric(:cc_requests_outstanding_total)
       end
     end
   end
diff --git a/lib/cloud_controller/metrics/statsd_updater.rb b/lib/cloud_controller/metrics/statsd_updater.rb
index b6daefce901..700ff227631 100644
--- a/lib/cloud_controller/metrics/statsd_updater.rb
+++ b/lib/cloud_controller/metrics/statsd_updater.rb
@@ -2,8 +2,9 @@
 
 module VCAP::CloudController::Metrics
   class StatsdUpdater
-    def initialize(statsd=CloudController::DependencyLocator.instance.statsd_client)
+    def initialize(statsd=CloudController::DependencyLocator.instance.statsd_client, redis_connection_pool_size: nil)
       @statsd = statsd
+      @redis_connection_pool_size = redis_connection_pool_size
     end
 
     def update_deploying_count(deploying_count)
@@ -93,10 +94,72 @@ def report_staging_failure_metrics(duration_ns)
       @statsd.timing('cc.staging.failed_duration', nanoseconds_to_milliseconds(duration_ns))
     end
 
+    def start_request
+      @statsd.increment('cc.requests.outstanding')
+      @statsd.gauge('cc.requests.outstanding.gauge', store.increment_request_outstanding_gauge)
+    end
+
+    def complete_request(status)
+      http_status_code = "#{status.to_s[0]}XX"
+      http_status_metric = "cc.http_status.#{http_status_code}"
+      @statsd.gauge('cc.requests.outstanding.gauge', store.decrement_request_outstanding_gauge)
+      @statsd.batch do |batch|
+        batch.decrement 'cc.requests.outstanding'
+        batch.increment 'cc.requests.completed'
+        batch.increment http_status_metric
+      end
+    end
+
     private
 
     def nanoseconds_to_milliseconds(time_ns)
       (time_ns / 1e6).to_i
     end
+
+    def store
+      return @store if defined?(@store)
+
+      redis_socket = VCAP::CloudController::Config.config.get(:redis, :socket)
+      @store = redis_socket.nil? ? InMemoryStore.new : RedisStore.new(redis_socket, @redis_connection_pool_size)
+    end
+
+    class InMemoryStore
+      def initialize
+        @mutex = Mutex.new
+        @counter = 0
+      end
+
+      def increment_request_outstanding_gauge
+        @mutex.synchronize do
+          @counter += 1
+        end
+      end
+
+      def decrement_request_outstanding_gauge
+        @mutex.synchronize do
+          @counter -= 1
+        end
+      end
+    end
+
+    class RedisStore
+      def initialize(socket, connection_pool_size)
+        connection_pool_size ||= VCAP::CloudController::Config.config.get(:puma, :max_threads) || 1
+        @redis = ConnectionPool::Wrapper.new(size: connection_pool_size) do
+          Redis.new(timeout: 1, path: socket)
+        end
+        @prefix = 'metrics'
+        @request_outstanding_gauge_key = "#{@prefix}:cc.requests.outstanding.gauge"
+        @redis.set(@request_outstanding_gauge_key, 0)
+      end
+
+      def increment_request_outstanding_gauge
+        @redis.incr(@request_outstanding_gauge_key)
+      end
+
+      def decrement_request_outstanding_gauge
+        @redis.decr(@request_outstanding_gauge_key)
+      end
+    end
   end
 end
diff --git a/lib/cloud_controller/runner.rb b/lib/cloud_controller/runner.rb
index 99b4f3ebbc0..8c4be34ab0c 100644
--- a/lib/cloud_controller/runner.rb
+++ b/lib/cloud_controller/runner.rb
@@ -33,7 +33,7 @@ def initialize(argv)
 
       request_logs = VCAP::CloudController::Logs::RequestLogs.new(Steno.logger('cc.api'))
 
-      request_metrics = VCAP::CloudController::Metrics::RequestMetrics.new(CloudController::DependencyLocator.instance.statsd_client,
+      request_metrics = VCAP::CloudController::Metrics::RequestMetrics.new(CloudController::DependencyLocator.instance.statsd_updater,
                                                                            CloudController::DependencyLocator.instance.prometheus_updater)
       builder = RackAppBuilder.new
       app     = builder.build(@config, request_metrics, request_logs)
diff --git a/spec/unit/lib/cloud_controller/metrics/request_metrics_spec.rb b/spec/unit/lib/cloud_controller/metrics/request_metrics_spec.rb
index 8d5bf7dd220..3266e2f34f4 100644
--- a/spec/unit/lib/cloud_controller/metrics/request_metrics_spec.rb
+++ b/spec/unit/lib/cloud_controller/metrics/request_metrics_spec.rb
@@ -3,55 +3,37 @@
 
 module VCAP::CloudController::Metrics
   RSpec.describe RequestMetrics do
-    let(:statsd_client) { double(:statsd_client) }
+    let(:statsd_updater) { double(:statsd_updater) }
     let(:prometheus_client) { double(:prometheus_client) }
-    let(:request_metrics) { RequestMetrics.new(statsd_client, prometheus_client) }
+    let(:request_metrics) { RequestMetrics.new(statsd_updater, prometheus_client) }
 
     before do
       allow(prometheus_client).to receive(:update_gauge_metric)
       allow(prometheus_client).to receive(:decrement_gauge_metric)
       allow(prometheus_client).to receive(:increment_gauge_metric)
       allow(prometheus_client).to receive(:increment_counter_metric)
+      allow(statsd_updater).to receive(:start_request)
+      allow(statsd_updater).to receive(:complete_request)
     end
 
     describe '#start_request' do
-      before do
-        allow(statsd_client).to receive(:increment)
-        allow(statsd_client).to receive(:gauge)
-      end
-
       it 'increments outstanding requests for statsd' do
         request_metrics.start_request
-
-        expect(statsd_client).to have_received(:gauge).with('cc.requests.outstanding.gauge', 1)
-        expect(statsd_client).to have_received(:increment).with('cc.requests.outstanding')
+        expect(statsd_updater).to have_received(:start_request)
       end
 
       it 'increments outstanding requests for prometheus' do
         request_metrics.start_request
-
         expect(prometheus_client).to have_received(:increment_gauge_metric).with(:cc_requests_outstanding_total)
       end
     end
 
     describe '#complete_request' do
-      let(:batch) { double(:batch) }
       let(:status) { 204 }
 
-      before do
-        allow(statsd_client).to receive(:batch).and_yield(batch)
-        allow(statsd_client).to receive(:gauge)
-        allow(batch).to receive(:increment)
-        allow(batch).to receive(:decrement)
-      end
-
       it 'increments completed, decrements outstanding, increments status for statsd' do
         request_metrics.complete_request(status)
-
-        expect(statsd_client).to have_received(:gauge).with('cc.requests.outstanding.gauge', -1)
-        expect(batch).to have_received(:decrement).with('cc.requests.outstanding')
-        expect(batch).to have_received(:increment).with('cc.requests.completed')
-        expect(batch).to have_received(:increment).with('cc.http_status.2XX')
+        expect(statsd_updater).to have_received(:complete_request).with(status)
       end
 
       it 'increments completed and decrements outstanding for prometheus' do
@@ -60,17 +42,6 @@ module VCAP::CloudController::Metrics
         expect(prometheus_client).to have_received(:decrement_gauge_metric).with(:cc_requests_outstanding_total)
         expect(prometheus_client).to have_received(:increment_counter_metric).with(:cc_requests_completed_total)
       end
-
-      it 'normalizes http status codes in statsd' do
-        request_metrics.complete_request(200)
-        expect(batch).to have_received(:increment).with('cc.http_status.2XX')
-
-        request_metrics.complete_request(300)
-        expect(batch).to have_received(:increment).with('cc.http_status.3XX')
-
-        request_metrics.complete_request(400)
-        expect(batch).to have_received(:increment).with('cc.http_status.4XX')
-      end
     end
   end
 end
diff --git a/spec/unit/lib/cloud_controller/metrics/statsd_updater_spec.rb b/spec/unit/lib/cloud_controller/metrics/statsd_updater_spec.rb
index 6523a4dcccd..027ec04c712 100644
--- a/spec/unit/lib/cloud_controller/metrics/statsd_updater_spec.rb
+++ b/spec/unit/lib/cloud_controller/metrics/statsd_updater_spec.rb
@@ -5,6 +5,7 @@ module VCAP::CloudController::Metrics
   RSpec.describe StatsdUpdater do
     let(:updater) { StatsdUpdater.new(statsd_client) }
     let(:statsd_client) { Statsd.new('localhost', 9999) }
+    let(:store) { double(:store) }
 
     describe '#update_deploying_count' do
       before do
@@ -281,5 +282,175 @@ module VCAP::CloudController::Metrics
         expect(statsd_client).to have_received(:timing).with('cc.staging.failed_duration', duration_ms)
       end
     end
+
+    describe '#start_request' do
+      before do
+        allow(statsd_client).to receive(:increment)
+        allow(statsd_client).to receive(:gauge)
+        allow(updater).to receive(:store).and_return(store)
+        allow(store).to receive(:increment_request_outstanding_gauge).and_return(4)
+      end
+
+      it 'increments outstanding requests for statsd' do
+        updater.start_request
+
+        expect(store).to have_received(:increment_request_outstanding_gauge)
+        expect(statsd_client).to have_received(:gauge).with('cc.requests.outstanding.gauge', 4)
+        expect(statsd_client).to have_received(:increment).with('cc.requests.outstanding')
+      end
+    end
+
+    describe '#complete_request' do
+      let(:batch) { double(:batch) }
+      let(:status) { 204 }
+
+      before do
+        allow(statsd_client).to receive(:batch).and_yield(batch)
+        allow(statsd_client).to receive(:gauge)
+        allow(batch).to receive(:increment)
+        allow(batch).to receive(:decrement)
+        allow(updater).to receive(:store).and_return(store)
+        allow(store).to receive(:decrement_request_outstanding_gauge).and_return(5)
+      end
+
+      it 'increments completed, decrements outstanding, increments status for statsd' do
+        updater.complete_request(status)
+
+        expect(store).to have_received(:decrement_request_outstanding_gauge)
+        expect(statsd_client).to have_received(:gauge).with('cc.requests.outstanding.gauge', 5)
+        expect(batch).to have_received(:decrement).with('cc.requests.outstanding')
+        expect(batch).to have_received(:increment).with('cc.requests.completed')
+        expect(batch).to have_received(:increment).with('cc.http_status.2XX')
+      end
+
+      it 'normalizes http status codes in statsd' do
+        updater.complete_request(200)
+        expect(batch).to have_received(:increment).with('cc.http_status.2XX')
+
+        updater.complete_request(300)
+        expect(batch).to have_received(:increment).with('cc.http_status.3XX')
+
+        updater.complete_request(400)
+        expect(batch).to have_received(:increment).with('cc.http_status.4XX')
+      end
+    end
+
+    describe '#store' do
+      let(:config) { double(:config) }
+
+      before do
+        allow(config).to receive(:get).with(:cc, :enable_statsd_metrics).and_return(true)
+        allow(VCAP::CloudController::Config).to receive(:config).and_return(config)
+      end
+
+      context 'when redis socket is not configured' do
+        before do
+          allow(config).to receive(:get).with(:redis, :redis).and_return(nil)
+        end
+
+        it 'returns an instance of InMemoryStore' do
+          store = updater.send(:store)
+          expect(store).to be_an_instance_of(StatsdUpdater::InMemoryStore)
+        end
+      end
+
+      context 'when redis socket is configured' do
+        let(:redis_socket) { 'redis.sock' }
+
+        before do
+          allow(config).to receive(:get).with(:redis, :socket).and_return(redis_socket)
+          allow(config).to receive(:get).with(:puma, :max_threads).and_return(nil)
+        end
+
+        it 'returns an instance of RedisStore' do
+          expect(ConnectionPool::Wrapper).to receive(:new).with(size: 1).and_call_original
+          store = updater.send(:store)
+          expect(store).to be_an_instance_of(StatsdUpdater::RedisStore)
+        end
+
+        context 'when puma max threads is set' do
+          let(:pool_size) { 10 }
+
+          before do
+            allow(config).to receive(:get).with(:puma, :max_threads).and_return(pool_size)
+          end
+
+          it 'passes the connection pool size to RedisStore' do
+            expect(ConnectionPool::Wrapper).to receive(:new).with(size: pool_size).and_call_original
+            updater.send(:store)
+          end
+        end
+      end
+    end
+
+    describe StatsdUpdater::InMemoryStore do
+      let(:store) { StatsdUpdater::InMemoryStore.new }
+
+      it 'increments the counter' do
+        expect(store.increment_request_outstanding_gauge).to eq(1)
+        expect(store.increment_request_outstanding_gauge).to eq(2)
+        expect(store.increment_request_outstanding_gauge).to eq(3)
+      end
+
+      it 'decrements the counter' do
+        expect(store.decrement_request_outstanding_gauge).to eq(-1)
+        expect(store.decrement_request_outstanding_gauge).to eq(-2)
+        expect(store.decrement_request_outstanding_gauge).to eq(-3)
+      end
+    end
+
+    describe StatsdUpdater::RedisStore do
+      let(:redis_socket) { 'redis.sock' }
+      let(:connection_pool_size) { 5 }
+      let(:redis_store) { StatsdUpdater::RedisStore.new(redis_socket, connection_pool_size) }
+      let(:redis) { instance_double(Redis) }
+      let(:config) { double(:config) }
+
+      before do
+        allow(VCAP::CloudController::Config).to receive(:config).and_return(config)
+        allow(ConnectionPool::Wrapper).to receive(:new).and_return(redis)
+        allow(redis).to receive(:set)
+      end
+
+      describe 'initialization' do
+        it 'clears cc.requests.outstanding.gauge' do
+          expect(redis).to receive(:set).with('metrics:cc.requests.outstanding.gauge', 0)
+          StatsdUpdater::RedisStore.new(redis_socket, connection_pool_size)
+        end
+
+        it 'configures a Redis connection pool with specified size' do
+          expect(ConnectionPool::Wrapper).to receive(:new).with(size: connection_pool_size).and_call_original
+          StatsdUpdater::RedisStore.new(redis_socket, connection_pool_size)
+        end
+
+        context 'when the connection pool size is not provided' do
+          it 'uses a default connection pool size of 1' do
+            expect(ConnectionPool::Wrapper).to receive(:new).with(size: 1).and_call_original
+            StatsdUpdater::RedisStore.new(redis_socket, nil)
+          end
+
+          context 'when puma max threads is set' do
+            before do
+              allow(config).to receive(:get).with(:puma, :max_threads).and_return(10)
+            end
+
+            it 'uses puma max threads' do
+              expect(ConnectionPool::Wrapper).to receive(:new).with(size: 10).and_call_original
+              StatsdUpdater::RedisStore.new(redis_socket, nil)
+            end
+          end
+        end
+      end
+
+      it 'increments the gauge in Redis' do
+        allow(redis).to receive(:incr).with('metrics:cc.requests.outstanding.gauge').and_return(1)
+        expect(redis_store.increment_request_outstanding_gauge).to eq(1)
+      end
+
+      it 'decrements the gauge in Redis' do
+        allow(redis).to receive(:decr).with('metrics:cc.requests.outstanding.gauge').and_return(-1)
+        expect(redis_store.decrement_request_outstanding_gauge).to eq(-1)
+      end
+    end
   end
 end