Merge branch '339129-cleanup-policies-caching' into 'master'

Cleanup tags service: cache created_at values for tags that are kept

See merge request gitlab-org/gitlab!69459

app/services/container_expiration_policies/cleanup_service.rb

@@ -4,7 +4,7 @@ module ContainerExpirationPolicies
   class CleanupService
     attr_reader :repository
 
-    SERVICE_RESULT_FIELDS = %i[original_size before_truncate_size after_truncate_size before_delete_size deleted_size].freeze
+    SERVICE_RESULT_FIELDS = %i[original_size before_truncate_size after_truncate_size before_delete_size deleted_size cached_tags_count].freeze
 
     def initialize(repository)
       @repository = repository

app/services/projects/container_repository/cache_tags_created_at_service.rb

+# frozen_string_literal: true
+
+module Projects
+  module ContainerRepository
+    class CacheTagsCreatedAtService
+      def initialize(container_repository)
+        @container_repository = container_repository
+        @cached_tag_names = Set.new
+      end
+
+      def populate(tags)
+        return if tags.empty?
+
+        # This will load all tags in one Redis roundtrip
+        # the maximum number of tags is configurable and is set to 200 by default.
+        # https://gitlab.com/gitlab-org/gitlab/blob/master/doc/user/packages/container_registry/index.md#set-cleanup-limits-to-conserve-resources
+        keys = tags.map(&method(:cache_key))
+        cached_tags_count = 0
+
+        ::Gitlab::Redis::Cache.with do |redis|
+          tags.zip(redis.mget(keys)).each do |tag, created_at|
+            next unless created_at
+
+            tag.created_at = DateTime.rfc3339(created_at)
+            @cached_tag_names << tag.name
+            cached_tags_count += 1
+          end
+        end
+
+        cached_tags_count
+      end
+
+      def insert(tags, max_ttl_in_seconds)
+        return unless max_ttl_in_seconds
+        return if tags.empty?
+
+        # tags with nil created_at are not cacheable
+        # tags already cached don't need to be cached again
+        cacheable_tags = tags.select do |tag|
+          tag.created_at.present? && !tag.name.in?(@cached_tag_names)
+        end
+
+        return if cacheable_tags.empty?
+
+        now = Time.zone.now
+
+        ::Gitlab::Redis::Cache.with do |redis|
+          # we use a pipeline instead of a MSET because each tag has
+          # a specific ttl
+          redis.pipelined do
+            cacheable_tags.each do |tag|
+              created_at = tag.created_at
+              # ttl is the max_ttl_in_seconds reduced by the number
+              # of seconds that the tag has already existed
+              ttl = max_ttl_in_seconds - (now - created_at).seconds
+              ttl = ttl.to_i
+              redis.set(cache_key(tag), created_at.rfc3339, ex: ttl) if ttl > 0
+            end
+          end
+        end
+      end
+
+      private
+
+      def cache_key(tag)
+        "container_repository:{#{@container_repository.id}}:tag:#{tag.name}:created_at"
+      end
+    end
+  end
+end

app/services/projects/container_repository/cleanup_tags_service.rb

@@ -3,6 +3,8 @@
 module Projects
   module ContainerRepository
     class CleanupTagsService < BaseService
+      include ::Gitlab::Utils::StrongMemoize
+
       def execute(container_repository)
         return error('access denied') unless can_destroy?
         return error('invalid regex') unless valid_regex?
@@ -17,13 +19,16 @@ module Projects
         tags = truncate(tags)
         after_truncate_size = tags.size
 
-        tags = filter_keep_n(tags)
-        tags = filter_by_older_than(tags)
+        cached_tags_count = populate_tags_from_cache(container_repository, tags) || 0
+
+        tags = filter_keep_n(container_repository, tags)
+        tags = filter_by_older_than(container_repository, tags)
 
         delete_tags(container_repository, tags).tap do |result|
           result[:original_size] = original_size
           result[:before_truncate_size] = before_truncate_size
           result[:after_truncate_size] = after_truncate_size
+          result[:cached_tags_count] = cached_tags_count
           result[:before_delete_size] = tags.size
           result[:deleted_size] = result[:deleted]&.size
 
@@ -67,21 +72,26 @@ module Projects
         end
       end
 
-      def filter_keep_n(tags)
+      def filter_keep_n(container_repository, tags)
         return tags unless params['keep_n']
 
         tags = order_by_date(tags)
+        cache_tags(container_repository, tags.first(keep_n))
         tags.drop(keep_n)
       end
 
-      def filter_by_older_than(tags)
-        return tags unless params['older_than']
+      def filter_by_older_than(container_repository, tags)
+        return tags unless older_than
 
-        older_than = ChronicDuration.parse(params['older_than']).seconds.ago
+        older_than_timestamp = older_than_in_seconds.ago
 
-        tags.select do |tag|
-          tag.created_at && tag.created_at < older_than
+        tags, tags_to_keep = tags.partition do |tag|
+          tag.created_at && tag.created_at < older_than_timestamp
         end
+
+        cache_tags(container_repository, tags_to_keep)
+
+        tags
       end
 
       def can_destroy?
@@ -114,6 +124,28 @@ module Projects
         tags.sample(truncated_size)
       end
 
+      def populate_tags_from_cache(container_repository, tags)
+        cache(container_repository).populate(tags) if caching_enabled?(container_repository)
+      end
+
+      def cache_tags(container_repository, tags)
+        cache(container_repository).insert(tags, older_than_in_seconds) if caching_enabled?(container_repository)
+      end
+
+      def cache(container_repository)
+        # TODO Implement https://gitlab.com/gitlab-org/gitlab/-/issues/340277 to avoid passing
+        # the container repository parameter which is bad for a memoized function
+        strong_memoize(:cache) do
+          ::Projects::ContainerRepository::CacheTagsCreatedAtService.new(container_repository)
+        end
+      end
+
+      def caching_enabled?(container_repository)
+        params['container_expiration_policy'] &&
+          older_than.present? &&
+          Feature.enabled?(:container_registry_expiration_policies_caching, container_repository.project)
+      end
+
       def throttling_enabled?
         Feature.enabled?(:container_registry_expiration_policies_throttling)
       end
@@ -125,6 +157,16 @@ module Projects
       def keep_n
         params['keep_n'].to_i
       end
+
+      def older_than_in_seconds
+        strong_memoize(:older_than_in_seconds) do
+          ChronicDuration.parse(older_than).seconds
+        end
+      end
+
+      def older_than
+        params['older_than']
+      end
     end
   end
 end

app/workers/container_expiration_policies/cleanup_container_repository_worker.rb

@@ -21,6 +21,7 @@ module ContainerExpirationPolicies
       cleanup_tags_service_original_size
       cleanup_tags_service_before_truncate_size
       cleanup_tags_service_after_truncate_size
+      cleanup_tags_service_cached_tags_count
       cleanup_tags_service_before_delete_size
       cleanup_tags_service_deleted_size
     ].freeze
@@ -147,13 +148,27 @@ module ContainerExpirationPolicies
         log_extra_metadata_on_done(field, value)
       end
 
+      log_truncate(result)
+      log_cache_ratio(result)
+      log_extra_metadata_on_done(:running_jobs_count, running_jobs_count)
+    end
+
+    def log_cache_ratio(result)
+      tags_count = result.payload[:cleanup_tags_service_after_truncate_size]
+      cached_tags_count = result.payload[:cleanup_tags_service_cached_tags_count]
+
+      return unless tags_count && cached_tags_count && tags_count != 0
+
+      log_extra_metadata_on_done(:cleanup_tags_service_cache_hit_ratio, cached_tags_count / tags_count.to_f)
+    end
+
+    def log_truncate(result)
       before_truncate_size = result.payload[:cleanup_tags_service_before_truncate_size]
       after_truncate_size = result.payload[:cleanup_tags_service_after_truncate_size]
       truncated = before_truncate_size &&
                     after_truncate_size &&
                     before_truncate_size != after_truncate_size
       log_extra_metadata_on_done(:cleanup_tags_service_truncated, !!truncated)
-      log_extra_metadata_on_done(:running_jobs_count, running_jobs_count)
     end
 
     def policy

config/feature_flags/development/container_registry_expiration_policies_caching.yml

+---
+name: container_registry_expiration_policies_caching
+introduced_by_url:
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/340606
+milestone: '14.3'
+type: development
+group: group::package
+default_enabled: false

lib/container_registry/tag.rb

@@ -5,6 +5,7 @@ module ContainerRegistry
     include Gitlab::Utils::StrongMemoize
 
     attr_reader :repository, :name
+    attr_writer :created_at
 
     delegate :registry, :client, to: :repository
     delegate :revision, :short_revision, to: :config_blob, allow_nil: true
@@ -73,9 +74,10 @@ module ContainerRegistry
     end
 
     def created_at
+      return @created_at if @created_at
       return unless config
 
-      strong_memoize(:created_at) do
+      strong_memoize(:memoized_created_at) do
         DateTime.rfc3339(config['created'])
       rescue ArgumentError
         nil

spec/lib/container_registry/tag_spec.rb

@@ -60,6 +60,20 @@ RSpec.describe ContainerRegistry::Tag do
   end
 
   context 'manifest processing' do
+    shared_examples 'using the value manually set on created_at' do
+      let(:value) { 5.seconds.ago }
+
+      before do
+        tag.created_at = value
+      end
+
+      it 'does not use the config' do
+        expect(tag).not_to receive(:config)
+
+        expect(subject).to eq(value)
+      end
+    end
+
     context 'schema v1' do
       before do
         stub_request(:get, 'http://registry.gitlab/v2/group/test/manifests/tag')
@@ -93,6 +107,8 @@ RSpec.describe ContainerRegistry::Tag do
           subject { tag.created_at }
 
           it { is_expected.to be_nil }
+
+          it_behaves_like 'using the value manually set on created_at'
         end
       end
     end
@@ -117,6 +133,8 @@ RSpec.describe ContainerRegistry::Tag do
         subject { tag.created_at }
 
         it { is_expected.to be_nil }
+
+        it_behaves_like 'using the value manually set on created_at'
       end
     end
 
@@ -154,6 +172,8 @@ RSpec.describe ContainerRegistry::Tag do
             subject { tag.created_at }
 
             it { is_expected.not_to be_nil }
+
+            it_behaves_like 'using the value manually set on created_at'
           end
         end
 

spec/services/container_expiration_policies/cleanup_service_spec.rb

@@ -69,6 +69,7 @@ RSpec.describe ContainerExpirationPolicies::CleanupService do
             before_truncate_size: 800,
             after_truncate_size: 200,
             before_delete_size: 100,
+            cached_tags_count: 0,
             deleted_size: 100
           }
         end
@@ -86,6 +87,7 @@ RSpec.describe ContainerExpirationPolicies::CleanupService do
                 cleanup_tags_service_before_truncate_size: 800,
                 cleanup_tags_service_after_truncate_size: 200,
                 cleanup_tags_service_before_delete_size: 100,
+                cleanup_tags_service_cached_tags_count: 0,
                 cleanup_tags_service_deleted_size: 100
               )
             expect(ContainerRepository.waiting_for_cleanup.count).to eq(1)

spec/services/projects/container_repository/cache_tags_created_at_service_spec.rb

+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe ::Projects::ContainerRepository::CacheTagsCreatedAtService, :clean_gitlab_redis_cache do
+  let_it_be(:dummy_tag_class) { Struct.new(:name, :created_at) }
+  let_it_be(:repository) { create(:container_repository) }
+
+  let(:tags) { create_tags(5) }
+  let(:service) { described_class.new(repository) }
+
+  shared_examples 'not interacting with redis' do
+    it 'does not interact with redis' do
+      expect(::Gitlab::Redis::Cache).not_to receive(:with)
+
+      subject
+    end
+  end
+
+  describe '#populate' do
+    subject { service.populate(tags) }
+
+    context 'with tags' do
+      it 'gets values from redis' do
+        expect(::Gitlab::Redis::Cache).to receive(:with).and_call_original
+
+        expect(subject).to eq(0)
+
+        tags.each { |t| expect(t.created_at).to eq(nil) }
+      end
+
+      context 'with cached values' do
+        let(:cached_tags) { tags.first(2) }
+
+        before do
+          ::Gitlab::Redis::Cache.with do |redis|
+            cached_tags.each do |tag|
+              redis.set(cache_key(tag), rfc3339(10.days.ago))
+            end
+          end
+        end
+
+        it 'gets values from redis' do
+          expect(::Gitlab::Redis::Cache).to receive(:with).and_call_original
+
+          expect(subject).to eq(2)
+
+          cached_tags.each { |t| expect(t.created_at).not_to eq(nil) }
+          (tags - cached_tags).each { |t| expect(t.created_at).to eq(nil) }
+        end
+      end
+    end
+
+    context 'with no tags' do
+      let(:tags) { [] }
+
+      it_behaves_like 'not interacting with redis'
+    end
+  end
+
+  describe '#insert' do
+    let(:max_ttl) { 90.days }
+
+    subject { service.insert(tags, max_ttl) }
+
+    context 'with tags' do
+      let(:tag) { tags.first }
+      let(:ttl) { 90.days - 3.days }
+
+      before do
+        travel_to(Time.zone.local(2021, 9, 2, 12, 0, 0))
+
+        tag.created_at = DateTime.rfc3339(3.days.ago.rfc3339)
+      end
+
+      after do
+        travel_back
+      end
+
+      it 'inserts values in redis' do
+        ::Gitlab::Redis::Cache.with do |redis|
+          expect(redis)
+            .to receive(:set)
+                  .with(cache_key(tag), rfc3339(tag.created_at), ex: ttl.to_i)
+                  .and_call_original
+        end
+
+        subject
+      end
+
+      context 'with some of them already cached' do
+        let(:tag) { tags.first }
+
+        before do
+          ::Gitlab::Redis::Cache.with do |redis|
+            redis.set(cache_key(tag), rfc3339(10.days.ago))
+          end
+          service.populate(tags)
+        end
+
+        it_behaves_like 'not interacting with redis'
+      end
+    end
+
+    context 'with no tags' do
+      let(:tags) { [] }
+
+      it_behaves_like 'not interacting with redis'
+    end
+
+    context 'with no expires_in' do
+      let(:max_ttl) { nil }
+
+      it_behaves_like 'not interacting with redis'
+    end
+  end
+
+  def create_tags(size)
+    Array.new(size) do |i|
+      dummy_tag_class.new("Tag #{i}", nil)
+    end
+  end
+
+  def cache_key(tag)
+    "container_repository:{#{repository.id}}:tag:#{tag.name}:created_at"
+  end
+
+  def rfc3339(date_time)
+    # DateTime rfc3339 is different ActiveSupport::TimeWithZone rfc3339
+    # The caching will use DateTime rfc3339
+    DateTime.rfc3339(date_time.rfc3339).rfc3339
+  end
+end

spec/services/projects/container_repository/cleanup_tags_service_spec.rb

@@ -2,13 +2,13 @@
 
 require 'spec_helper'
 
-RSpec.describe Projects::ContainerRepository::CleanupTagsService do
+RSpec.describe Projects::ContainerRepository::CleanupTagsService, :clean_gitlab_redis_cache do
   using RSpec::Parameterized::TableSyntax
 
   let_it_be(:user) { create(:user) }
   let_it_be(:project, reload: true) { create(:project, :private) }
-  let_it_be(:repository) { create(:container_repository, :root, project: project) }
 
+  let(:repository) { create(:container_repository, :root, project: project) }
   let(:service) { described_class.new(project, user, params) }
   let(:tags) { %w[latest A Ba Bb C D E] }
 
@@ -41,289 +41,440 @@ RSpec.describe Projects::ContainerRepository::CleanupTagsService do
   describe '#execute' do
     subject { service.execute(repository) }
 
-    context 'when no params are specified' do
-      let(:params) { {} }
+    shared_examples 'reading and removing tags' do |caching_enabled: true|
+      context 'when no params are specified' do
+        let(:params) { {} }
 
-      it 'does not remove anything' do
-        expect_any_instance_of(Projects::ContainerRepository::DeleteTagsService)
-          .not_to receive(:execute)
+        it 'does not remove anything' do
+          expect_any_instance_of(Projects::ContainerRepository::DeleteTagsService)
+            .not_to receive(:execute)
+          expect_no_caching
 
-        is_expected.to eq(expected_service_response(before_truncate_size: 0, after_truncate_size: 0, before_delete_size: 0))
-      end
-    end
-
-    context 'when regex matching everything is specified' do
-      shared_examples 'removes all matches' do
-        it 'does remove all tags except latest' do
-          expect_delete(%w(A Ba Bb C D E))
-
-          is_expected.to eq(expected_service_response(deleted: %w(A Ba Bb C D E)))
+          is_expected.to eq(expected_service_response(before_truncate_size: 0, after_truncate_size: 0, before_delete_size: 0))
         end
       end
 
-      let(:params) do
-        { 'name_regex_delete' => '.*' }
-      end
+      context 'when regex matching everything is specified' do
+        shared_examples 'removes all matches' do
+          it 'does remove all tags except latest' do
+            expect_no_caching
 
-      it_behaves_like 'removes all matches'
+            expect_delete(%w(A Ba Bb C D E))
+
+            is_expected.to eq(expected_service_response(deleted: %w(A Ba Bb C D E)))
+          end
+        end
 
-      context 'with deprecated name_regex param' do
         let(:params) do
-          { 'name_regex' => '.*' }
+          { 'name_regex_delete' => '.*' }
         end
 
         it_behaves_like 'removes all matches'
+
+        context 'with deprecated name_regex param' do
+          let(:params) do
+            { 'name_regex' => '.*' }
+          end
+
+          it_behaves_like 'removes all matches'
+        end
       end
-    end
 
-    context 'with invalid regular expressions' do
-      RSpec.shared_examples 'handling an invalid regex' do
-        it 'keeps all tags' do
-          expect(Projects::ContainerRepository::DeleteTagsService)
-            .not_to receive(:new)
-          subject
+      context 'with invalid regular expressions' do
+        shared_examples 'handling an invalid regex' do
+          it 'keeps all tags' do
+            expect_no_caching
+
+            expect(Projects::ContainerRepository::DeleteTagsService)
+              .not_to receive(:new)
+
+            subject
+          end
+
+          it { is_expected.to eq(status: :error, message: 'invalid regex') }
+
+          it 'calls error tracking service' do
+            expect(Gitlab::ErrorTracking).to receive(:log_exception).and_call_original
+
+            subject
+          end
         end
 
-        it { is_expected.to eq(status: :error, message: 'invalid regex') }
+        context 'when name_regex_delete is invalid' do
+          let(:params) { { 'name_regex_delete' => '*test*' } }
+
+          it_behaves_like 'handling an invalid regex'
+        end
 
-        it 'calls error tracking service' do
-          expect(Gitlab::ErrorTracking).to receive(:log_exception).and_call_original
+        context 'when name_regex is invalid' do
+          let(:params) { { 'name_regex' => '*test*' } }
 
-          subject
+          it_behaves_like 'handling an invalid regex'
         end
-      end
 
-      context 'when name_regex_delete is invalid' do
-        let(:params) { { 'name_regex_delete' => '*test*' } }
+        context 'when name_regex_keep is invalid' do
+          let(:params) { { 'name_regex_keep' => '*test*' } }
 
-        it_behaves_like 'handling an invalid regex'
+          it_behaves_like 'handling an invalid regex'
+        end
       end
 
-      context 'when name_regex is invalid' do
-        let(:params) { { 'name_regex' => '*test*' } }
+      context 'when delete regex matching specific tags is used' do
+        let(:params) do
+          { 'name_regex_delete' => 'C|D' }
+        end
 
-        it_behaves_like 'handling an invalid regex'
-      end
+        it 'does remove C and D' do
+          expect_delete(%w(C D))
 
-      context 'when name_regex_keep is invalid' do
-        let(:params) { { 'name_regex_keep' => '*test*' } }
+          expect_no_caching
 
-        it_behaves_like 'handling an invalid regex'
-      end
-    end
+          is_expected.to eq(expected_service_response(deleted: %w(C D), before_truncate_size: 2, after_truncate_size: 2, before_delete_size: 2))
+        end
 
-    context 'when delete regex matching specific tags is used' do
-      let(:params) do
-        { 'name_regex_delete' => 'C|D' }
-      end
+        context 'with overriding allow regex' do
+          let(:params) do
+            { 'name_regex_delete' => 'C|D',
+              'name_regex_keep' => 'C' }
+          end
 
-      it 'does remove C and D' do
-        expect_delete(%w(C D))
+          it 'does not remove C' do
+            expect_delete(%w(D))
 
-        is_expected.to eq(expected_service_response(deleted: %w(C D), before_truncate_size: 2, after_truncate_size: 2, before_delete_size: 2))
-      end
+            expect_no_caching
 
-      context 'with overriding allow regex' do
-        let(:params) do
-          { 'name_regex_delete' => 'C|D',
-            'name_regex_keep' => 'C' }
+            is_expected.to eq(expected_service_response(deleted: %w(D), before_truncate_size: 1, after_truncate_size: 1, before_delete_size: 1))
+          end
         end
 
-        it 'does not remove C' do
-          expect_delete(%w(D))
+        context 'with name_regex_delete overriding deprecated name_regex' do
+          let(:params) do
+            { 'name_regex' => 'C|D',
+              'name_regex_delete' => 'D' }
+          end
+
+          it 'does not remove C' do
+            expect_delete(%w(D))
+
+            expect_no_caching
 
-          is_expected.to eq(expected_service_response(deleted: %w(D), before_truncate_size: 1, after_truncate_size: 1, before_delete_size: 1))
+            is_expected.to eq(expected_service_response(deleted: %w(D), before_truncate_size: 1, after_truncate_size: 1, before_delete_size: 1))
+          end
         end
       end
 
-      context 'with name_regex_delete overriding deprecated name_regex' do
+      context 'with allow regex value' do
         let(:params) do
-          { 'name_regex' => 'C|D',
-            'name_regex_delete' => 'D' }
+          { 'name_regex_delete' => '.*',
+            'name_regex_keep' => 'B.*' }
         end
 
-        it 'does not remove C' do
-          expect_delete(%w(D))
+        it 'does not remove B*' do
+          expect_delete(%w(A C D E))
+
+          expect_no_caching
 
-          is_expected.to eq(expected_service_response(deleted: %w(D), before_truncate_size: 1, after_truncate_size: 1, before_delete_size: 1))
+          is_expected.to eq(expected_service_response(deleted: %w(A C D E), before_truncate_size: 4, after_truncate_size: 4, before_delete_size: 4))
         end
       end
-    end
 
-    context 'with allow regex value' do
-      let(:params) do
-        { 'name_regex_delete' => '.*',
-          'name_regex_keep' => 'B.*' }
-      end
+      context 'when keeping only N tags' do
+        let(:params) do
+          { 'name_regex' => 'A|B.*|C',
+            'keep_n' => 1 }
+        end
 
-      it 'does not remove B*' do
-        expect_delete(%w(A C D E))
+        it 'sorts tags by date' do
+          expect_delete(%w(Bb Ba C))
 
-        is_expected.to eq(expected_service_response(deleted: %w(A C D E), before_truncate_size: 4, after_truncate_size: 4, before_delete_size: 4))
-      end
-    end
+          expect_no_caching
 
-    context 'when keeping only N tags' do
-      let(:params) do
-        { 'name_regex' => 'A|B.*|C',
-          'keep_n' => 1 }
+          expect(service).to receive(:order_by_date).and_call_original
+
+          is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_truncate_size: 4, after_truncate_size: 4, before_delete_size: 3))
+        end
       end
 
-      it 'sorts tags by date' do
-        expect_delete(%w(Bb Ba C))
+      context 'when not keeping N tags' do
+        let(:params) do
+          { 'name_regex' => 'A|B.*|C' }
+        end
+
+        it 'does not sort tags by date' do
+          expect_delete(%w(A Ba Bb C))
 
-        expect(service).to receive(:order_by_date).and_call_original
+          expect_no_caching
 
-        is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_truncate_size: 4, after_truncate_size: 4, before_delete_size: 3))
-      end
-    end
+          expect(service).not_to receive(:order_by_date)
 
-    context 'when not keeping N tags' do
-      let(:params) do
-        { 'name_regex' => 'A|B.*|C' }
+          is_expected.to eq(expected_service_response(deleted: %w(A Ba Bb C), before_truncate_size: 4, after_truncate_size: 4, before_delete_size: 4))
+        end
       end
 
-      it 'does not sort tags by date' do
-        expect_delete(%w(A Ba Bb C))
+      context 'when removing keeping only 3' do
+        let(:params) do
+          { 'name_regex_delete' => '.*',
+            'keep_n' => 3 }
+        end
 
-        expect(service).not_to receive(:order_by_date)
+        it 'does remove B* and C as they are the oldest' do
+          expect_delete(%w(Bb Ba C))
 
-        is_expected.to eq(expected_service_response(deleted: %w(A Ba Bb C), before_truncate_size: 4, after_truncate_size: 4, before_delete_size: 4))
-      end
-    end
+          expect_no_caching
 
-    context 'when removing keeping only 3' do
-      let(:params) do
-        { 'name_regex_delete' => '.*',
-          'keep_n' => 3 }
+          is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_delete_size: 3))
+        end
       end
 
-      it 'does remove B* and C as they are the oldest' do
-        expect_delete(%w(Bb Ba C))
+      context 'when removing older than 1 day' do
+        let(:params) do
+          { 'name_regex_delete' => '.*',
+            'older_than' => '1 day' }
+        end
+
+        it 'does remove B* and C as they are older than 1 day' do
+          expect_delete(%w(Ba Bb C))
 
-        is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_delete_size: 3))
-      end
-    end
+          expect_no_caching
 
-    context 'when removing older than 1 day' do
-      let(:params) do
-        { 'name_regex_delete' => '.*',
-          'older_than' => '1 day' }
+          is_expected.to eq(expected_service_response(deleted: %w(Ba Bb C), before_delete_size: 3))
+        end
       end
 
-      it 'does remove B* and C as they are older than 1 day' do
-        expect_delete(%w(Ba Bb C))
+      context 'when combining all parameters' do
+        let(:params) do
+          { 'name_regex_delete' => '.*',
+            'keep_n' => 1,
+            'older_than' => '1 day' }
+        end
+
+        it 'does remove B* and C' do
+          expect_delete(%w(Bb Ba C))
 
-        is_expected.to eq(expected_service_response(deleted: %w(Ba Bb C), before_delete_size: 3))
-      end
-    end
+          expect_no_caching
 
-    context 'when combining all parameters' do
-      let(:params) do
-        { 'name_regex_delete' => '.*',
-          'keep_n' => 1,
-          'older_than' => '1 day' }
+          is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_delete_size: 3))
+        end
       end
 
-      it 'does remove B* and C' do
-        expect_delete(%w(Bb Ba C))
+      context 'when running a container_expiration_policy' do
+        let(:user) { nil }
 
-        is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_delete_size: 3))
-      end
-    end
+        context 'with valid container_expiration_policy param' do
+          let(:params) do
+            { 'name_regex_delete' => '.*',
+              'keep_n' => 1,
+              'older_than' => '1 day',
+              'container_expiration_policy' => true }
+          end
 
-    context 'when running a container_expiration_policy' do
-      let(:user) { nil }
+          it 'succeeds without a user' do
+            expect_delete(%w(Bb Ba C), container_expiration_policy: true)
 
-      context 'with valid container_expiration_policy param' do
-        let(:params) do
-          { 'name_regex_delete' => '.*',
-            'keep_n' => 1,
-            'older_than' => '1 day',
-            'container_expiration_policy' => true }
+            caching_enabled ? expect_caching : expect_no_caching
+
+            is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_delete_size: 3))
+          end
         end
 
-        it 'succeeds without a user' do
-          expect_delete(%w(Bb Ba C), container_expiration_policy: true)
+        context 'without container_expiration_policy param' do
+          let(:params) do
+            { 'name_regex_delete' => '.*',
+              'keep_n' => 1,
+              'older_than' => '1 day' }
+          end
 
-          is_expected.to eq(expected_service_response(deleted: %w(Bb Ba C), before_delete_size: 3))
+          it 'fails' do
+            is_expected.to eq(status: :error, message: 'access denied')
+          end
         end
       end
 
-      context 'without container_expiration_policy param' do
+      context 'truncating the tags list' do
         let(:params) do
-          { 'name_regex_delete' => '.*',
-            'keep_n' => 1,
-            'older_than' => '1 day' }
+          {
+            'name_regex_delete' => '.*',
+            'keep_n' => 1
+          }
+        end
+
+        shared_examples 'returning the response' do |status:, original_size:, before_truncate_size:, after_truncate_size:, before_delete_size:|
+          it 'returns the response' do
+            expect_no_caching
+
+            result = subject
+
+            service_response = expected_service_response(
+              status: status,
+              original_size: original_size,
+              before_truncate_size: before_truncate_size,
+              after_truncate_size: after_truncate_size,
+              before_delete_size: before_delete_size,
+              deleted: nil
+            )
+
+            expect(result).to eq(service_response)
+          end
+        end
+
+        where(:feature_flag_enabled, :max_list_size, :delete_tags_service_status, :expected_status, :expected_truncated) do
+          false | 10 | :success | :success | false
+          false | 10 | :error   | :error   | false
+          false | 3  | :success | :success | false
+          false | 3  | :error   | :error   | false
+          false | 0  | :success | :success | false
+          false | 0  | :error   | :error   | false
+          true  | 10 | :success | :success | false
+          true  | 10 | :error   | :error   | false
+          true  | 3  | :success | :error   | true
+          true  | 3  | :error   | :error   | true
+          true  | 0  | :success | :success | false
+          true  | 0  | :error   | :error   | false
         end
 
-        it 'fails' do
-          is_expected.to eq(status: :error, message: 'access denied')
+        with_them do
+          before do
+            stub_feature_flags(container_registry_expiration_policies_throttling: feature_flag_enabled)
+            stub_application_setting(container_registry_cleanup_tags_service_max_list_size: max_list_size)
+            allow_next_instance_of(Projects::ContainerRepository::DeleteTagsService) do |service|
+              expect(service).to receive(:execute).and_return(status: delete_tags_service_status)
+            end
+          end
+
+          original_size = 7
+          keep_n = 1
+
+          it_behaves_like(
+            'returning the response',
+            status: params[:expected_status],
+            original_size: original_size,
+            before_truncate_size: original_size - keep_n,
+            after_truncate_size: params[:expected_truncated] ? params[:max_list_size] + keep_n : original_size - keep_n,
+            before_delete_size: params[:expected_truncated] ? params[:max_list_size] : original_size - keep_n - 1 # one tag is filtered out with older_than filter
+          )
         end
       end
     end
 
-    context 'truncating the tags list' do
+    context 'caching' do
       let(:params) do
         {
           'name_regex_delete' => '.*',
-          'keep_n' => 1
+          'keep_n' => 1,
+          'older_than' => '1 day',
+          'container_expiration_policy' => true
+        }
+      end
+
+      let(:tags_and_created_ats) do
+        {
+          'A' => 1.hour.ago,
+          'Ba' => 5.days.ago,
+          'Bb' => 5.days.ago,
+          'C' => 1.month.ago,
+          'D' => nil,
+          'E' => nil
         }
       end
 
-      shared_examples 'returning the response' do |status:, original_size:, before_truncate_size:, after_truncate_size:, before_delete_size:|
-        it 'returns the response' do
-          result = subject
+      let(:cacheable_tags) { tags_and_created_ats.reject { |_, value| value.nil? } }
 
-          service_response = expected_service_response(
-            status: status,
-            original_size: original_size,
-            before_truncate_size: before_truncate_size,
-            after_truncate_size: after_truncate_size,
-            before_delete_size: before_delete_size,
-            deleted: nil
-          )
+      before do
+        expect_delete(%w(Bb Ba C), container_expiration_policy: true)
+        travel_to(Time.zone.local(2021, 9, 2, 12, 0, 0))
+        # We froze time so we need to set the created_at stubs again
+        stub_digest_config('sha256:configA', 1.hour.ago)
+        stub_digest_config('sha256:configB', 5.days.ago)
+        stub_digest_config('sha256:configC', 1.month.ago)
+      end
 
-          expect(result).to eq(service_response)
-        end
+      after do
+        travel_back
       end
 
-      where(:feature_flag_enabled, :max_list_size, :delete_tags_service_status, :expected_status, :expected_truncated) do
-        false | 10 | :success | :success | false
-        false | 10 | :error   | :error   | false
-        false | 3  | :success | :success | false
-        false | 3  | :error   | :error   | false
-        false | 0  | :success | :success | false
-        false | 0  | :error   | :error   | false
-        true  | 10 | :success | :success | false
-        true  | 10 | :error   | :error   | false
-        true  | 3  | :success | :error   | true
-        true  | 3  | :error   | :error   | true
-        true  | 0  | :success | :success | false
-        true  | 0  | :error   | :error   | false
+      it 'caches the created_at values' do
+        ::Gitlab::Redis::Cache.with do |redis|
+          expect_mget(redis, tags_and_created_ats.keys)
+
+          expect_set(redis, cacheable_tags)
+        end
+
+        expect(subject).to include(cached_tags_count: 0)
       end
 
-      with_them do
+      context 'with cached values' do
         before do
-          stub_feature_flags(container_registry_expiration_policies_throttling: feature_flag_enabled)
-          stub_application_setting(container_registry_cleanup_tags_service_max_list_size: max_list_size)
-          allow_next_instance_of(Projects::ContainerRepository::DeleteTagsService) do |service|
-            expect(service).to receive(:execute).and_return(status: delete_tags_service_status)
+          ::Gitlab::Redis::Cache.with do |redis|
+            redis.set(cache_key('C'), rfc3339(1.month.ago))
+          end
+        end
+
+        it 'uses them' do
+          ::Gitlab::Redis::Cache.with do |redis|
+            expect_mget(redis, tags_and_created_ats.keys)
+
+            # because C is already in cache, it should not be cached again
+            expect_set(redis, cacheable_tags.except('C'))
+          end
+
+          # We will ping the container registry for all tags *except* for C because it's cached
+          expect(ContainerRegistry::Blob).to receive(:new).with(repository, "digest" => "sha256:configA").and_call_original
+          expect(ContainerRegistry::Blob).to receive(:new).with(repository, "digest" => "sha256:configB").twice.and_call_original
+          expect(ContainerRegistry::Blob).not_to receive(:new).with(repository, "digest" => "sha256:configC")
+          expect(ContainerRegistry::Blob).to receive(:new).with(repository, "digest" => "sha256:configD").and_call_original
+
+          expect(subject).to include(cached_tags_count: 1)
+        end
+      end
+
+      def expect_mget(redis, keys)
+        expect(redis).to receive(:mget).with(keys.map(&method(:cache_key))).and_call_original
+      end
+
+      def expect_set(redis, tags)
+        tags.each do |tag_name, created_at|
+          ex = 1.day.seconds - (Time.zone.now - created_at).seconds
+          if ex > 0
+            expect(redis).to receive(:set).with(cache_key(tag_name), rfc3339(created_at), ex: ex.to_i)
           end
         end
+      end
+
+      def cache_key(tag_name)
+        "container_repository:{#{repository.id}}:tag:#{tag_name}:created_at"
+      end
+
+      def rfc3339(date_time)
+        # DateTime rfc3339 is different ActiveSupport::TimeWithZone rfc3339
+        # The caching will use DateTime rfc3339
+        DateTime.rfc3339(date_time.rfc3339).rfc3339
+      end
+    end
+
+    context 'with container_registry_expiration_policies_caching enabled for the project' do
+      before do
+        stub_feature_flags(container_registry_expiration_policies_caching: project)
+      end
+
+      it_behaves_like 'reading and removing tags', caching_enabled: true
+    end
 
-        original_size = 7
-        keep_n = 1
+    context 'with container_registry_expiration_policies_caching disabled' do
+      before do
+        stub_feature_flags(container_registry_expiration_policies_caching: false)
+      end
+
+      it_behaves_like 'reading and removing tags', caching_enabled: false
+    end
 
-        it_behaves_like(
-          'returning the response',
-          status: params[:expected_status],
-          original_size: original_size,
-          before_truncate_size: original_size - keep_n,
-          after_truncate_size: params[:expected_truncated] ? params[:max_list_size] + keep_n : original_size - keep_n,
-          before_delete_size: params[:expected_truncated] ? params[:max_list_size] : original_size - keep_n - 1 # one tag is filtered out with older_than filter
-        )
+    context 'with container_registry_expiration_policies_caching not enabled for the project' do
+      let_it_be(:another_project) { create(:project) }
+
+      before do
+        stub_feature_flags(container_registry_expiration_policies_caching: another_project)
       end
+
+      it_behaves_like 'reading and removing tags', caching_enabled: false
     end
   end
 
@@ -368,7 +519,19 @@ RSpec.describe Projects::ContainerRepository::CleanupTagsService do
       original_size: original_size,
       before_truncate_size: before_truncate_size,
       after_truncate_size: after_truncate_size,
-      before_delete_size: before_delete_size
+      before_delete_size: before_delete_size,
+      cached_tags_count: 0
     }.compact.merge(deleted_size: deleted&.size)
   end
+
+  def expect_no_caching
+    expect(::Gitlab::Redis::Cache).not_to receive(:with)
+  end
+
+  def expect_caching
+    ::Gitlab::Redis::Cache.with do |redis|
+      expect(redis).to receive(:mget).and_call_original
+      expect(redis).to receive(:set).and_call_original
+    end
+  end
 end

spec/workers/container_expiration_policies/cleanup_container_repository_worker_spec.rb

@@ -74,6 +74,30 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
             end
           end
         end
+
+        context 'the cache hit ratio field' do
+          where(:after_truncate_size, :cached_tags_count, :ratio) do
+            nil | nil | nil
+            10  | nil | nil
+            nil | 10  | nil
+            0   | 5   | nil
+            10  | 0   | 0
+            10  | 5   | 0.5
+            3   | 10  | (10 / 3.to_f)
+          end
+
+          with_them do
+            it 'is logged properly' do
+              service_response = cleanup_service_response(status: :unfinished, repository: repository, cleanup_tags_service_before_truncate_size: after_truncate_size, cleanup_tags_service_after_truncate_size: after_truncate_size, cleanup_tags_service_cached_tags_count: cached_tags_count)
+              expect(ContainerExpirationPolicies::CleanupService)
+                .to receive(:new).with(repository).and_return(double(execute: service_response))
+              expect_log_extra_metadata(service_response: service_response, cleanup_status: :unfinished, truncated: false, cache_hit_ratio: ratio)
+              expect_log_info(project_id: project.id, container_repository_id: repository.id)
+
+              subject
+            end
+          end
+        end
       end
 
       context 'with an erroneous cleanup' do
@@ -372,7 +396,16 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
       end
     end
 
-    def cleanup_service_response(status: :finished, repository:, cleanup_tags_service_original_size: 100, cleanup_tags_service_before_truncate_size: 80, cleanup_tags_service_after_truncate_size: 80, cleanup_tags_service_before_delete_size: 50, cleanup_tags_service_deleted_size: 50)
+    def cleanup_service_response(
+      status: :finished,
+      repository:,
+      cleanup_tags_service_original_size: 100,
+      cleanup_tags_service_before_truncate_size: 80,
+      cleanup_tags_service_after_truncate_size: 80,
+      cleanup_tags_service_before_delete_size: 50,
+      cleanup_tags_service_deleted_size: 50,
+      cleanup_tags_service_cached_tags_count: 0
+    )
       ServiceResponse.success(
         message: "cleanup #{status}",
         payload: {
@@ -381,21 +414,35 @@ RSpec.describe ContainerExpirationPolicies::CleanupContainerRepositoryWorker do
           cleanup_tags_service_original_size: cleanup_tags_service_original_size,
           cleanup_tags_service_before_truncate_size: cleanup_tags_service_before_truncate_size,
           cleanup_tags_service_after_truncate_size: cleanup_tags_service_after_truncate_size,
-          cleanup_tags_service_before_delete_size: cleanup_tags_service_before_delete_size
+          cleanup_tags_service_before_delete_size: cleanup_tags_service_before_delete_size,
+          cleanup_tags_service_cached_tags_count: cleanup_tags_service_cached_tags_count
         }.compact
       )
     end
 
-    def expect_log_extra_metadata(service_response:, cleanup_status: :finished, truncated: false)
+    def expect_log_extra_metadata(service_response:, cleanup_status: :finished, truncated: false, cache_hit_ratio: 0)
       expect(worker).to receive(:log_extra_metadata_on_done).with(:container_repository_id, repository.id)
       expect(worker).to receive(:log_extra_metadata_on_done).with(:project_id, repository.project.id)
       expect(worker).to receive(:log_extra_metadata_on_done).with(:cleanup_status, cleanup_status)
 
-      %i[cleanup_tags_service_original_size cleanup_tags_service_before_truncate_size cleanup_tags_service_after_truncate_size cleanup_tags_service_before_delete_size cleanup_tags_service_deleted_size].each do |field|
+      %i[
+        cleanup_tags_service_original_size
+        cleanup_tags_service_before_truncate_size
+        cleanup_tags_service_after_truncate_size
+        cleanup_tags_service_before_delete_size
+        cleanup_tags_service_deleted_size
+        cleanup_tags_service_cached_tags_count
+      ].each do |field|
         value = service_response.payload[field]
         expect(worker).to receive(:log_extra_metadata_on_done).with(field, value) unless value.nil?
       end
       expect(worker).to receive(:log_extra_metadata_on_done).with(:cleanup_tags_service_truncated, truncated)
+
+      after_truncate_size = service_response.payload[:cleanup_tags_service_after_truncate_size]
+      if cache_hit_ratio && after_truncate_size && after_truncate_size != 0
+        expect(worker).to receive(:log_extra_metadata_on_done).with(:cleanup_tags_service_cache_hit_ratio, cache_hit_ratio)
+      end
+
       expect(worker).to receive(:log_extra_metadata_on_done).with(:running_jobs_count, 0)
 
       if service_response.error?