blob: 3027d46b8b1d03af8299808730c60cbb9df62ad1 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
# frozen_string_literal: true
module ContainerExpirationPolicies
class CleanupContainerRepositoryWorker
include ApplicationWorker
sidekiq_options retry: 3
include LimitedCapacity::Worker
include Gitlab::Utils::StrongMemoize
queue_namespace :container_repository
feature_category :container_registry
tags :exclude_from_kubernetes
urgency :low
worker_resource_boundary :unknown
idempotent!
LOG_ON_DONE_FIELDS = %i[
cleanup_status
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
].freeze
def perform_work
return unless throttling_enabled?
return unless container_repository
log_extra_metadata_on_done(:container_repository_id, container_repository.id)
log_extra_metadata_on_done(:project_id, project.id)
unless allowed_to_run?
container_repository.cleanup_unscheduled!
log_extra_metadata_on_done(:cleanup_status, :skipped)
return
end
result = ContainerExpirationPolicies::CleanupService.new(container_repository)
.execute
log_on_done(result)
end
def max_running_jobs
return 0 unless throttling_enabled?
::Gitlab::CurrentSettings.container_registry_expiration_policies_worker_capacity
end
def remaining_work_count
total_count = cleanup_scheduled_count + cleanup_unfinished_count
log_info(
cleanup_scheduled_count: cleanup_scheduled_count,
cleanup_unfinished_count: cleanup_unfinished_count,
cleanup_total_count: total_count
)
total_count
end
private
def container_repository
strong_memoize(:container_repository) do
ContainerRepository.transaction do
# We need a lock to prevent two workers from picking up the same row
container_repository = next_container_repository
container_repository&.tap(&:cleanup_ongoing!)
end
end
end
def next_container_repository
# rubocop: disable CodeReuse/ActiveRecord
next_one_requiring = ContainerRepository.requiring_cleanup
.order(:expiration_policy_cleanup_status, :expiration_policy_started_at)
.limit(1)
.lock('FOR UPDATE SKIP LOCKED')
.first
return next_one_requiring if next_one_requiring
ContainerRepository.with_unfinished_cleanup
.order(:expiration_policy_started_at)
.limit(1)
.lock('FOR UPDATE SKIP LOCKED')
.first
# rubocop: enable CodeReuse/ActiveRecord
end
def cleanup_scheduled_count
strong_memoize(:cleanup_scheduled_count) do
limit = max_running_jobs + 1
ContainerExpirationPolicy.with_container_repositories
.runnable_schedules
.limit(limit)
.count
end
end
def cleanup_unfinished_count
strong_memoize(:cleanup_unfinished_count) do
limit = max_running_jobs + 1
ContainerRepository.with_unfinished_cleanup
.limit(limit)
.count
end
end
def allowed_to_run?
return false unless policy&.enabled && policy&.next_run_at
now = Time.zone.now
policy.next_run_at < now || (now + max_cleanup_execution_time.seconds < policy.next_run_at)
end
def throttling_enabled?
Feature.enabled?(:container_registry_expiration_policies_throttling)
end
def max_cleanup_execution_time
::Gitlab::CurrentSettings.container_registry_delete_tags_service_timeout
end
def log_info(extra_structure)
logger.info(structured_payload(extra_structure))
end
def log_on_done(result)
if result.error?
log_extra_metadata_on_done(:cleanup_status, :error)
log_extra_metadata_on_done(:cleanup_error_message, result.message)
end
LOG_ON_DONE_FIELDS.each do |field|
value = result.payload[field]
next if value.nil?
log_extra_metadata_on_done(field, value)
end
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
project.container_expiration_policy
end
def project
container_repository.project
end
end
end
|
