diff --git a/lib/cloud_controller.rb b/lib/cloud_controller.rb
index 2ded100bf04..5e235d86c14 100644
--- a/lib/cloud_controller.rb
+++ b/lib/cloud_controller.rb
@@ -115,6 +115,7 @@ module VCAP::CloudController; end
 require 'cloud_controller/user_audit_info'
 
 require 'cloud_controller/errands/rotate_database_key'
+require 'cloud_controller/errands/check_database_key'
 
 require 'services'
 
diff --git a/lib/cloud_controller/errands/check_database_key.rb b/lib/cloud_controller/errands/check_database_key.rb
new file mode 100644
index 00000000000..257edee17f9
--- /dev/null
+++ b/lib/cloud_controller/errands/check_database_key.rb
@@ -0,0 +1,42 @@
+# rubocop:disable Rails/Exit
+module VCAP::CloudController
+  class CheckDatabaseKey
+    class << self
+      def perform
+        @logger = Steno.logger('cc.check_database_key')
+
+        if Encryptor.current_encryption_key_label.blank?
+          logger.error('No current encryption key label is set')
+          exit 1
+        end
+
+        rows_needing_rotation = count_rows_needing_rotation
+        if rows_needing_rotation > 0
+          logger.info("#{rows_needing_rotation} row(s) need re-encryption with the current key")
+          exit 2
+        end
+
+        logger.info('All rows are encrypted with the current key')
+        exit 0
+      rescue StandardError => e
+        logger.error("Unexpected error: #{e.class}: #{e.message}")
+        exit 1
+      end
+
+      private
+
+      attr_accessor :logger
+
+      def count_rows_needing_rotation
+        current_key_label = Encryptor.current_encryption_key_label
+        Encryptor.encrypted_classes.sum do |klass|
+          klass.constantize.
+            exclude(encryption_key_label: current_key_label).
+            or(encryption_key_label: nil).
+            count
+        end
+      end
+    end
+  end
+end
+# rubocop:enable Rails/Exit
diff --git a/lib/tasks/rotate_database_key.rake b/lib/tasks/rotate_database_key.rake
index ee2c798b547..ad1cf6f185d 100644
--- a/lib/tasks/rotate_database_key.rake
+++ b/lib/tasks/rotate_database_key.rake
@@ -6,4 +6,12 @@ namespace :rotate_cc_database_key do
     BoshErrandEnvironment.new(RakeConfig.config).setup_environment
     VCAP::CloudController::RotateDatabaseKey.perform
   end
+
+  desc 'Check if database rows need re-encryption with the current key'
+  task check: :environment do
+    require 'cloud_controller/errands/check_database_key'
+    RakeConfig.context = :rotate_database_key
+    BoshErrandEnvironment.new(RakeConfig.config).setup_environment
+    VCAP::CloudController::CheckDatabaseKey.perform
+  end
 end
diff --git a/spec/unit/lib/cloud_controller/errands/check_database_key_spec.rb b/spec/unit/lib/cloud_controller/errands/check_database_key_spec.rb
new file mode 100644
index 00000000000..ebc9b7ae4fc
--- /dev/null
+++ b/spec/unit/lib/cloud_controller/errands/check_database_key_spec.rb
@@ -0,0 +1,89 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  RSpec.describe CheckDatabaseKey do
+    describe '#perform' do
+      let(:logger) { instance_double(Steno::Logger, info: nil, error: nil) }
+
+      before do
+        allow(Steno).to receive(:logger).and_return(logger)
+        allow(Encryptor).to receive_messages(
+          encrypted_classes: ['VCAP::CloudController::AppModel', 'VCAP::CloudController::TaskModel'],
+          current_encryption_key_label: 'current'
+        )
+      end
+
+      context 'when no current encryption key label is set' do
+        before do
+          allow(Encryptor).to receive(:current_encryption_key_label).and_return(nil)
+        end
+
+        it 'exits with 1' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit) { |e| expect(e.status).to eq(1) }
+        end
+
+        it 'logs an error' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit)
+          expect(logger).to have_received(:error).with('No current encryption key label is set')
+        end
+      end
+
+      context 'when all rows are encrypted with the current key' do
+        before do
+          allow(AppModel).to receive(:exclude).and_return(double(or: double(count: 0)))
+          allow(TaskModel).to receive(:exclude).and_return(double(or: double(count: 0)))
+        end
+
+        it 'exits with 0' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit) { |e| expect(e.status).to eq(0) }
+        end
+
+        it 'logs that all rows are current' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit)
+          expect(logger).to have_received(:info).with('All rows are encrypted with the current key')
+        end
+      end
+
+      context 'when rows need re-encryption' do
+        before do
+          allow(Encryptor).to receive(:current_encryption_key_label).and_return('old')
+          allow(Encryptor).to receive(:database_encryption_keys).and_return({ old: 'old-key', current: 'current-key' })
+
+          app = create(:app_model)
+          app.environment_variables = { 'key' => 'value' }
+          app.save(validate: false)
+
+          task = create(:task_model)
+          task.environment_variables = { 'key' => 'value' }
+          task.save(validate: false)
+
+          allow(Encryptor).to receive(:current_encryption_key_label).and_return('current')
+        end
+
+        it 'exits with 2' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit) { |e| expect(e.status).to eq(2) }
+        end
+
+        it 'logs the total number of rows needing re-encryption' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit)
+          expect(logger).to have_received(:info).with(a_string_matching(/\d+ row\(s\) need re-encryption with the current key/))
+        end
+      end
+
+      context 'when an unexpected error occurs' do
+        before do
+          allow(AppModel).to receive(:exclude).and_raise(StandardError.new('db connection failed'))
+        end
+
+        it 'exits with 1' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit) { |e| expect(e.status).to eq(1) }
+        end
+
+        it 'logs the error' do
+          expect { CheckDatabaseKey.perform }.to raise_error(SystemExit)
+          expect(logger).to have_received(:error).with('Unexpected error: StandardError: db connection failed')
+        end
+      end
+    end
+  end
+end