Cache created_at values for tags that are kept

This is to avoid network requests to the container registry api. This
cache will only work if the `older_than` parameter is set.
Tags that are destroyed are guaranted to have their `created_at` freshly
fetched from the container registry API.

By reducing the amount of pings to the container registry API, the
cleanup tags service is more efficient and takes less time to go through
a set of tags.

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

@@ -22,6 +22,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
@@ -148,13 +149,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?