Live trace PoC

lib/gitlab/ci/trace.rb

@@ -61,6 +61,8 @@ module Gitlab
         stream = Gitlab::Ci::Trace::Stream.new do
           if trace_artifact
             trace_artifact.open
+          elsif LiveIO.exists?(job.id)
+            LiveIO.new(job.id)
           elsif current_path
             File.open(current_path, "rb")
           elsif old_trace
@@ -75,7 +77,7 @@ module Gitlab
 
       def write
         stream = Gitlab::Ci::Trace::Stream.new do
-          File.open(ensure_path, "a+b")
+          LiveIO.new(job.id)
         end
 
         yield(stream).tap do
@@ -142,19 +144,6 @@ module Gitlab
         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)

lib/gitlab/ci/trace/chunked_io.rb

+##
+# This class is compatible with IO class (https://ruby-doc.org/core-2.3.1/IO.html)
+# source: https://gitlab.com/snippets/1685610
+module Gitlab
+  module Ci
+    class Trace
+      class ChunkedIO
+        attr_reader :size
+        attr_reader :tell
+        attr_reader :chunk, :chunk_range
+
+        alias_method :pos, :tell
+
+        def initialize(size)
+          @size = size
+          @tell = 0
+        end
+
+        def close
+          # no-op
+        end
+
+        def binmode
+          # no-op
+        end
+
+        def binmode?
+          true
+        end
+
+        def path
+          nil
+        end
+
+        def seek(pos, where = IO::SEEK_SET)
+          new_pos =
+            case where
+            when IO::SEEK_END
+              size + pos
+            when IO::SEEK_SET
+              pos
+            when IO::SEEK_CUR
+              tell + pos
+            else
+              -1
+            end
+
+          raise 'new position is outside of file' if new_pos < 0 || new_pos > size
+
+          @tell = new_pos
+        end
+
+        def eof?
+          tell == size
+        end
+
+        def each_line
+          until eof?
+            line = readline
+            break if line.nil?
+
+            yield(line)
+          end
+        end
+
+        def read(length = nil)
+          out = ""
+
+          until eof? || (length && out.length >= length)
+            data = get_chunk
+            break if data.empty?
+
+            out << data
+            @tell += data.bytesize
+          end
+
+          out = out[0, length] if length && out.length > length
+
+          out
+        end
+
+        def readline
+          out = ""
+
+          until eof?
+            data = get_chunk
+            new_line = data.index("\n")
+
+            if !new_line.nil?
+              out << data[0..new_line]
+              @tell += new_line + 1
+              break
+            else
+              out << data
+              @tell += data.bytesize
+            end
+          end
+
+          out
+        end
+
+        def write(data)
+          raise NotImplementedError
+        end
+
+        def truncate(offset)
+          raise NotImplementedError
+        end
+
+        def flush
+          raise NotImplementedError
+        end
+
+        def present?
+          true
+        end
+
+        private
+
+        ##
+        # To be overridden by superclasses
+        #
+        def get_chunk
+          raise NotImplementedError
+        end
+
+        def in_range?
+          @chunk_range&.include?(tell)
+        end
+
+        def chunk_offset
+          tell % BUFFER_SIZE
+        end
+
+        def chunk_start
+          (tell / BUFFER_SIZE) * BUFFER_SIZE
+        end
+
+        def chunk_end
+          [chunk_start + BUFFER_SIZE, size].min
+        end
+      end
+    end
+  end
+end

lib/gitlab/ci/trace/http_io.rb

-##
-# This class is compatible with IO class (https://ruby-doc.org/core-2.3.1/IO.html)
-# source: https://gitlab.com/snippets/1685610
 module Gitlab
   module Ci
     class Trace
-      class HttpIO
-        BUFFER_SIZE = 128.kilobytes
-
-        InvalidURLError = Class.new(StandardError)
+      class HttpIO < ChunkedIO
         FailedToGetChunkError = Class.new(StandardError)
+        InvalidURLError = Class.new(StandardError)
 
-        attr_reader :uri, :size
-        attr_reader :tell
-        attr_reader :chunk, :chunk_range
+        BUFFER_SIZE = 128.kilobytes
 
-        alias_method :pos, :tell
+        attr_reader :uri
 
         def initialize(url, size)
           raise InvalidURLError unless ::Gitlab::UrlSanitizer.valid?(url)
 
           @uri = URI(url)
-          @size = size
-          @tell = 0
-        end
-
-        def close
-          # no-op
-        end
-
-        def binmode
-          # no-op
-        end
-
-        def binmode?
-          true
-        end
 
-        def path
-          nil
+          super
         end
 
         def url
           @uri.to_s
         end
 
-        def seek(pos, where = IO::SEEK_SET)
-          new_pos =
-            case where
-            when IO::SEEK_END
-              size + pos
-            when IO::SEEK_SET
-              pos
-            when IO::SEEK_CUR
-              tell + pos
-            else
-              -1
-            end
-
-          raise 'new position is outside of file' if new_pos < 0 || new_pos > size
-
-          @tell = new_pos
-        end
-
-        def eof?
-          tell == size
-        end
-
-        def each_line
-          until eof?
-            line = readline
-            break if line.nil?
-
-            yield(line)
-          end
-        end
-
-        def read(length = nil)
-          out = ""
-
-          until eof? || (length && out.length >= length)
-            data = get_chunk
-            break if data.empty?
-
-            out << data
-            @tell += data.bytesize
-          end
-
-          out = out[0, length] if length && out.length > length
-
-          out
-        end
-
-        def readline
-          out = ""
-
-          until eof?
-            data = get_chunk
-            new_line = data.index("\n")
-
-            if !new_line.nil?
-              out << data[0..new_line]
-              @tell += new_line + 1
-              break
-            else
-              out << data
-              @tell += data.bytesize
-            end
-          end
-
-          out
-        end
-
         def write(data)
           raise NotImplementedError
         end
@@ -123,19 +33,10 @@ module Gitlab
           raise NotImplementedError
         end
 
-        def present?
-          true
-        end
-
         private
 
         ##
-        # The below methods are not implemented in IO class
-        #
-        def in_range?
-          @chunk_range&.include?(tell)
-        end
-
+        # Override
         def get_chunk
           unless in_range?
             response = Net::HTTP.start(uri.hostname, uri.port, use_ssl: uri.scheme == 'https') do |http|
@@ -169,18 +70,6 @@ module Gitlab
             request.set_range(chunk_start, BUFFER_SIZE)
           end
         end
-
-        def chunk_offset
-          tell % BUFFER_SIZE
-        end
-
-        def chunk_start
-          (tell / BUFFER_SIZE) * BUFFER_SIZE
-        end
-
-        def chunk_end
-          [chunk_start + BUFFER_SIZE, size].min
-        end
       end
     end
   end

lib/gitlab/ci/trace/live_io.rb

+module Gitlab
+  module Ci
+    class Trace
+      class LiveIO < ChunkedIO
+        BUFFER_SIZE = 32.kilobytes
+
+        class << self
+          def exists?(job_id)
+            exists_in_redis? || exists_in_database?
+          end
+
+          def exists_in_redis?(job_id)
+            Gitlab::Redis::Cache.with do |redis|
+              redis.exists(buffer_key(job_id))
+            end
+          end
+
+          def exists_in_database?(job_id)
+            Ci::JobTraceChunk.exists?(job_id: job_id)
+          end
+
+          def buffer_key(job_id)
+            "ci:live_trace_buffer:#{job_id}"
+          end
+        end
+
+        attr_reader :job_id
+
+        def initialize(job_id)
+          @job_id = job_id
+
+          super
+        end
+
+        def write(data)
+          # TODO: 
+        end
+
+        def truncate(offset)
+          # TODO: 
+        end
+
+        def flush
+          # TODO: 
+        end
+
+        private
+
+        ##
+        # Override
+        def get_chunk
+          # TODO: 
+        end
+      end
+    end
+  end
+end