summaryrefslogtreecommitdiff
path: root/lib/gitlab/ci/trace.rb
blob: 84eb860a168f9c00da22ba216f07aad297cebb1f (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
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
# frozen_string_literal: true

module Gitlab
  module Ci
    class Trace
      include ::Gitlab::ExclusiveLeaseHelpers
      include Checksummable

      LOCK_TTL = 10.minutes
      LOCK_RETRIES = 2
      LOCK_SLEEP = 0.001.seconds
      WATCH_FLAG_TTL = 10.seconds

      UPDATE_FREQUENCY_DEFAULT = 60.seconds
      UPDATE_FREQUENCY_WHEN_BEING_WATCHED = 3.seconds

      LOAD_BALANCING_STICKING_NAMESPACE = 'ci/build/trace'

      ArchiveError = Class.new(StandardError)
      AlreadyArchivedError = Class.new(StandardError)
      LockedError = Class.new(StandardError)

      attr_reader :job

      delegate :old_trace, to: :job

      def initialize(job)
        @job = job
      end

      def html(last_lines: nil)
        read do |stream|
          stream.html(last_lines: last_lines)
        end
      end

      def raw(last_lines: nil)
        read do |stream|
          stream.raw(last_lines: last_lines)
        end
      end

      def extract_coverage(regex)
        read do |stream|
          stream.extract_coverage(regex)
        end
      end

      def extract_sections
        read do |stream|
          stream.extract_sections
        end
      end

      def set(data)
        write('w+b') do |stream|
          data = job.hide_secrets(data)
          stream.set(data)
        end
      end

      def append(data, offset)
        write('a+b') do |stream|
          current_length = stream.size
          break current_length unless current_length == offset

          data = job.hide_secrets(data)
          stream.append(data, offset)
          stream.size
        end
      end

      def exist?
        archived_trace_exist? || live_trace_exist?
      end

      def archived_trace_exist?
        trace_artifact&.exists?
      end

      def live_trace_exist?
        job.trace_chunks.any? || current_path.present? || old_trace.present?
      end

      def read(&block)
        read_stream(&block)
      rescue Errno::ENOENT, ChunkedIO::FailedToGetChunkError
        job.reset
        read_stream(&block)
      end

      def write(mode, &blk)
        in_write_lock do
          unsafe_write!(mode, &blk)
        end
      end

      def erase_trace_chunks!
        job.trace_chunks.fast_destroy_all # Destroy chunks of a live trace
      end

      def erase!
        ##
        # Erase the archived trace
        trace_artifact&.destroy!

        ##
        # Erase the live trace
        erase_trace_chunks!
        FileUtils.rm_f(current_path) if current_path # Remove a trace file of a live trace
        job.erase_old_trace! if job.has_old_trace? # Remove a trace in database of a live trace
      ensure
        @current_path = nil
      end

      def archive!
        in_write_lock do
          unsafe_archive!
        end
      end

      def update_interval
        if being_watched?
          UPDATE_FREQUENCY_WHEN_BEING_WATCHED
        else
          UPDATE_FREQUENCY_DEFAULT
        end
      end

      def being_watched!
        Gitlab::Redis::SharedState.with do |redis|
          redis.set(being_watched_cache_key, true, ex: WATCH_FLAG_TTL)
        end
      end

      def being_watched?
        Gitlab::Redis::SharedState.with do |redis|
          redis.exists(being_watched_cache_key)
        end
      end

      def lock(&block)
        in_write_lock(&block)
      rescue FailedToObtainLockError
        raise LockedError, "build trace `#{job.id}` is locked"
      end

      private

      def read_stream
        stream = Gitlab::Ci::Trace::Stream.new do
          if trace_artifact
            trace_artifact.open
          elsif job.trace_chunks.any?
            Gitlab::Ci::Trace::ChunkedIO.new(job)
          elsif current_path
            File.open(current_path, "rb")
          elsif old_trace
            StringIO.new(old_trace)
          end
        end

        yield stream
      ensure
        stream&.close
      end

      def unsafe_write!(mode, &blk)
        stream = Gitlab::Ci::Trace::Stream.new do
          if trace_artifact
            raise AlreadyArchivedError, 'Could not write to the archived trace'
          elsif current_path
            File.open(current_path, mode)
          elsif Feature.enabled?(:ci_enable_live_trace, job.project)
            Gitlab::Ci::Trace::ChunkedIO.new(job)
          else
            File.open(ensure_path, mode)
          end
        end

        yield(stream).tap do
          job.touch if job.needs_touch?
        end
      ensure
        stream&.close
      end

      def unsafe_archive!
        raise ArchiveError, 'Job is not finished yet' unless job.complete?

        if trace_artifact
          unsafe_trace_cleanup! if Feature.enabled?(:erase_traces_from_already_archived_jobs_when_archiving_again, job.project, default_enabled: :yaml)

          raise AlreadyArchivedError, 'Could not archive again'
        end

        if job.trace_chunks.any?
          Gitlab::Ci::Trace::ChunkedIO.new(job) do |stream|
            archive_stream!(stream)
            destroy_stream(job) { stream.destroy! }
          end
        elsif current_path
          File.open(current_path) do |stream|
            archive_stream!(stream)
            FileUtils.rm(current_path)
          end
        elsif old_trace
          StringIO.new(old_trace, 'rb').tap do |stream|
            archive_stream!(stream)
            job.erase_old_trace!
          end
        end
      end

      def unsafe_trace_cleanup!
        return unless trace_artifact

        if trace_artifact.archived_trace_exists?
          # An archive already exists, so make sure to remove the trace chunks
          erase_trace_chunks!
        else
          # An archive already exists, but its associated file does not, so remove it
          trace_artifact.destroy!
        end
      end

      def in_write_lock(&blk)
        lock_key = "trace:write:lock:#{job.id}"
        in_lock(lock_key, ttl: LOCK_TTL, retries: LOCK_RETRIES, sleep_sec: LOCK_SLEEP, &blk)
      end

      def archive_stream!(stream)
        clone_file!(stream, JobArtifactUploader.workhorse_upload_path) do |clone_path|
          create_build_trace!(job, clone_path)
        end
      end

      def clone_file!(src_stream, temp_dir)
        FileUtils.mkdir_p(temp_dir)
        Dir.mktmpdir("tmp-trace-#{job.id}", temp_dir) do |dir_path|
          temp_path = File.join(dir_path, "job.log")
          FileUtils.touch(temp_path)
          size = IO.copy_stream(src_stream, temp_path)
          raise ArchiveError, 'Failed to copy stream' unless size == src_stream.size

          yield(temp_path)
        end
      end

      def create_build_trace!(job, path)
        File.open(path) do |stream|
          # TODO: Set `file_format: :raw` after we've cleaned up legacy traces migration
          # https://gitlab.com/gitlab-org/gitlab-foss/merge_requests/20307
          job.create_job_artifacts_trace!(
            project: job.project,
            file_type: :trace,
            file: stream,
            file_sha256: self.class.hexdigest(path))
        end
      end

      def ensure_path
        return current_path if current_path

        ensure_directory
        default_path
      end

      def ensure_directory
        unless Dir.exist?(default_directory)
          FileUtils.mkdir_p(default_directory)
        end
      end

      def current_path
        @current_path ||= paths.find do |trace_path|
          File.exist?(trace_path)
        end
      end

      def paths
        [default_path]
      end

      def default_directory
        File.join(
          Settings.gitlab_ci.builds_path,
          job.created_at.utc.strftime("%Y_%m"),
          job.project_id.to_s
        )
      end

      def default_path
        File.join(default_directory, "#{job.id}.log")
      end

      def trace_artifact
        read_trace_artifact(job) { job.job_artifacts_trace }
      end

      def destroy_stream(build)
        if consistent_archived_trace?(build)
          ::Gitlab::Database::LoadBalancing::Sticking
            .stick(LOAD_BALANCING_STICKING_NAMESPACE, build.id)
        end

        yield
      end

      def read_trace_artifact(build)
        if consistent_archived_trace?(build)
          ::Gitlab::Database::LoadBalancing::Sticking
            .unstick_or_continue_sticking(LOAD_BALANCING_STICKING_NAMESPACE, build.id)
        end

        yield
      end

      def consistent_archived_trace?(build)
        ::Feature.enabled?(:gitlab_ci_archived_trace_consistent_reads, build.project, default_enabled: false)
      end

      def being_watched_cache_key
        "gitlab:ci:trace:#{job.id}:watched"
      end
    end
  end
end