feat!: stream image build, pull and push progress

lib/docker_engine_ruby.rb

@@ -44,6 +44,7 @@
 require_relative "docker_engine_ruby/internal/type/hash_of"
 require_relative "docker_engine_ruby/internal/type/base_model"
 require_relative "docker_engine_ruby/internal/type/base_page"
+require_relative "docker_engine_ruby/internal/jsonl_stream"
 require_relative "docker_engine_ruby/internal/type/request_parameters"
 require_relative "docker_engine_ruby/internal"
 require_relative "docker_engine_ruby/request_options"
@@ -227,6 +228,13 @@
 require_relative "docker_engine_ruby/models/volume_prune_response"
 require_relative "docker_engine_ruby/models/volume_update_params"
 require_relative "docker_engine_ruby/models/wait_response"
+require_relative "docker_engine_ruby/models/error_detail"
+require_relative "docker_engine_ruby/models/progress_detail"
+require_relative "docker_engine_ruby/models/image_id"
+require_relative "docker_engine_ruby/models/build_info"
+require_relative "docker_engine_ruby/models/create_image_info"
+require_relative "docker_engine_ruby/models/push_image_info"
+require_relative "docker_engine_ruby/helpers/buildkit_status"
 require_relative "docker_engine_ruby/models"
 require_relative "docker_engine_ruby/resources/auth"
 require_relative "docker_engine_ruby/resources/configs"

lib/docker_engine_ruby/errors.rb

@@ -33,6 +33,20 @@ def initialize(on:, method:, target:, value:, cause: nil)
       end
     end
 
+    # Docker reports a failed build, pull or push inside a 200 response, as the last
+    # message of the progress stream.
+    class StreamError < DockerEngineRuby::Errors::Error
+      # @return [DockerEngineRuby::Models::ErrorDetail, nil]
+      attr_reader :detail
+
+      # @param message [String]
+      # @param detail [DockerEngineRuby::Models::ErrorDetail, nil]
+      def initialize(message, detail: nil)
+        @detail = detail
+        super(message)
+      end
+    end
+
     class APIError < DockerEngineRuby::Errors::Error
       # @return [URI::Generic]
       attr_accessor :url

lib/docker_engine_ruby/helpers/buildkit_status.rb

@@ -0,0 +1,136 @@
+# frozen_string_literal: true
+
+require "base64"
+
+module DockerEngineRuby
+  module Helpers
+    # Progress of a BuildKit build (`version: :"2"`). The daemon sends it as `BuildInfo` messages
+    # with `id: "moby.buildkit.trace"` and a base64 protobuf `StatusResponse` in `aux`
+    # (moby/buildkit `api/services/control/control.proto`). Only the fields a progress log
+    # needs are decoded; the rest are skipped.
+    module BuildkitStatus
+      TRACE_ID = "moby.buildkit.trace"
+
+      Vertex = Struct.new(:digest, :name, :cached, :started, :completed, :error, keyword_init: true)
+      Log = Struct.new(:vertex, :stream, :msg, keyword_init: true)
+      Status = Struct.new(:vertexes, :logs, keyword_init: true)
+
+      class << self
+        # @param info [DockerEngineRuby::Models::BuildInfo]
+        # @return [Boolean]
+        def trace?(info) = info.id == TRACE_ID && info.aux.is_a?(String)
+
+        # @param aux [String] base64-encoded `StatusResponse`
+        # @return [DockerEngineRuby::Helpers::BuildkitStatus::Status]
+        def decode(aux)
+          status = Status.new(vertexes: [], logs: [])
+          each_field(Base64.decode64(aux)) do |number, value|
+            case number
+            in 1 then status.vertexes << decode_vertex(value)
+            in 3 then status.logs << decode_log(value)
+            else nil
+            end
+          end
+          status
+        end
+
+        private
+
+        def decode_vertex(bytes)
+          vertex = Vertex.new(cached: false, started: false, completed: false)
+          each_field(bytes) do |number, value|
+            case number
+            in 1 then vertex.digest = utf8(value)
+            in 3 then vertex.name = utf8(value)
+            in 4 then vertex.cached = value != 0
+            in 5 then vertex.started = true
+            in 6 then vertex.completed = true
+            in 7 then vertex.error = utf8(value)
+            else nil
+            end
+          end
+          vertex
+        end
+
+        def decode_log(bytes)
+          log = Log.new(stream: 0, msg: +"")
+          each_field(bytes) do |number, value|
+            case number
+            in 1 then log.vertex = utf8(value)
+            in 3 then log.stream = value
+            in 4 then log.msg = utf8(value)
+            else nil
+            end
+          end
+          log
+        end
+
+        def utf8(bytes) = bytes.dup.force_encoding(Encoding::UTF_8).scrub
+
+        # Protobuf wire format: yields [field number, varint Integer | length-delimited String].
+        def each_field(bytes)
+          reader = StringIO.new(bytes.b)
+          until reader.eof?
+            key = read_varint(reader)
+            number = key >> 3
+            case key & 7
+            in 0 then yield(number, read_varint(reader))
+            in 1 then reader.read(8)
+            in 2 then yield(number, reader.read(read_varint(reader)).to_s)
+            in 5 then reader.read(4)
+            end
+          end
+        end
+
+        def read_varint(reader)
+          result = 0
+          shift = 0
+          loop do
+            byte = reader.readbyte
+            result |= (byte & 0x7f) << shift
+            return result if byte < 0x80
+
+            shift += 7
+          end
+        end
+      end
+
+      # Renders statuses as `docker build --progress=plain` does: vertexes are numbered in the
+      # order they appear, and each state change or log chunk becomes a line.
+      class PlainPrinter
+        # @param out [#call] receives each rendered line
+        def initialize(&out)
+          @out = out
+          @numbers = {}
+          @printed = {}
+        end
+
+        # @param status [DockerEngineRuby::Helpers::BuildkitStatus::Status]
+        def print(status)
+          status.vertexes.each { print_vertex(_1) }
+          status.logs.each do |log|
+            prefix = "##{number(log.vertex)} "
+            log.msg.each_line { @out.call(prefix + _1.chomp) }
+          end
+        end
+
+        private
+
+        def number(digest) = @numbers[digest] ||= @numbers.size + 1
+
+        def print_vertex(vertex)
+          n = number(vertex.digest)
+          state = if vertex.error then "ERROR: #{vertex.error}"
+          elsif vertex.cached then "CACHED"
+          elsif vertex.completed then "DONE"
+          elsif vertex.started then vertex.name
+          end
+          return if state.nil? || @printed[[n, state]]
+
+          @printed[[n, state]] = true
+          @out.call(state == vertex.name ? "##{n} #{vertex.name}" : "##{n} #{state}")
+        end
+      end
+    end
+  end
+end

lib/docker_engine_ruby/internal/jsonl_stream.rb

@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+module DockerEngineRuby
+  module Internal
+    # Progress streams of Docker (`/build`, `/images/create`, `/images/{name}/push`) are one JSON
+    # object per line, sent as `application/json`. Each line is coerced to `model` as it arrives.
+    #
+    # @generic Elem
+    class JSONLStream
+      include Enumerable
+
+      # @param model [DockerEngineRuby::Internal::Type::Converter, Class]
+      # @param stream [Enumerable<String>] raw response lines
+      def initialize(model:, stream:, **)
+        @model = model
+        @lines = stream
+      end
+
+      # @yieldparam message [generic<Elem>]
+      def each
+        return enum_for(__method__) unless block_given?
+
+        @lines.each do |line|
+          next if line.strip.empty?
+
+          yield(DockerEngineRuby::Internal::Type::Converter.coerce(
+            @model,
+            JSON.parse(line, symbolize_names: true)
+          ))
+        end
+      end
+    end
+  end
+end

lib/docker_engine_ruby/internal/transport/base_client.rb

@@ -537,7 +537,14 @@ def request(req)
             return nil
           end
 
-          decoded = DockerEngineRuby::Internal::Util.decode_content(headers, stream: stream)
+          # A stream class takes the raw lines: Docker sends its progress streams as `application/json`,
+          # which `decode_content` would join and parse as a single document.
+          decoded =
+            if req.key?(:stream)
+              DockerEngineRuby::Internal::Util.decode_lines(stream)
+            else
+              DockerEngineRuby::Internal::Util.decode_content(headers, stream: stream)
+            end
           case req
           in {stream: Class => st}
             st.new(

lib/docker_engine_ruby/internal/type/union.rb

@@ -150,6 +150,8 @@ def coerce(value, state:)
             target = variant_fn.call
             exact = state[:exactness] = {yes: 0, no: 0, maybe: 0}
             state[:branched] += 1
+            # A failed attempt must not leave its error behind for the variant that fits.
+            state[:error] = nil
 
             coerced = DockerEngineRuby::Internal::Type::Converter.coerce(target, value, state: state)
             yes, no, maybe = exact.values
@@ -158,7 +160,7 @@ def coerce(value, state:)
               state[:exactness] = exactness
               return coerced
             elsif maybe.positive?
-              alternatives << [[-yes, -maybe, no], exact, coerced]
+              alternatives << [[-yes, -maybe, no], exact, coerced, state[:error]]
             end
           end
 
@@ -167,8 +169,9 @@ def coerce(value, state:)
             exactness[:no] += 1
             state[:error] = ArgumentError.new("no matching variant for #{value.inspect}")
             value
-          in [[_, exact, coerced], *]
+          in [[_, exact, coerced, error], *]
             exact.each { exactness[_1] += _2 }
+            state[:error] = error
             coerced
           end
             .tap { state[:exactness] = exactness }

lib/docker_engine_ruby/models.rb

@@ -331,4 +331,16 @@ module DockerEngineRuby
   VolumeUpdateParams = DockerEngineRuby::Models::VolumeUpdateParams
 
   WaitResponse = DockerEngineRuby::Models::WaitResponse
+
+  BuildInfo = DockerEngineRuby::Models::BuildInfo
+
+  CreateImageInfo = DockerEngineRuby::Models::CreateImageInfo
+
+  ErrorDetail = DockerEngineRuby::Models::ErrorDetail
+
+  ImageID = DockerEngineRuby::Models::ImageID
+
+  ProgressDetail = DockerEngineRuby::Models::ProgressDetail
+
+  PushImageInfo = DockerEngineRuby::Models::PushImageInfo
 end

lib/docker_engine_ruby/models/build_info.rb

@@ -0,0 +1,81 @@
+# frozen_string_literal: true
+
+module DockerEngineRuby
+  module Models
+    # Сообщение из потока `/build`.
+    class BuildInfo < DockerEngineRuby::Internal::Type::BaseModel
+      # @!attribute id
+      #
+      #   @return [String, nil]
+      optional :id, String
+
+      # @!attribute stream
+      #   Строка вывода сборки.
+      #
+      #   @return [String, nil]
+      optional :stream, String
+
+      # @!attribute error
+      #   Устаревшее поле: то же, что `error_detail.message`.
+      #
+      #   @return [String, nil]
+      optional :error, String
+
+      # @!attribute error_detail
+      #
+      #   @return [DockerEngineRuby::Models::ErrorDetail, nil]
+      optional :error_detail, -> { DockerEngineRuby::ErrorDetail }, api_name: :errorDetail
+
+      # @!attribute status
+      #
+      #   @return [String, nil]
+      optional :status, String
+
+      # @!attribute progress
+      #   Устаревшее поле: прогресс строкой.
+      #
+      #   @return [String, nil]
+      optional :progress, String
+
+      # @!attribute progress_detail
+      #
+      #   @return [DockerEngineRuby::Models::ProgressDetail, nil]
+      optional :progress_detail, -> { DockerEngineRuby::ProgressDetail }, api_name: :progressDetail
+
+      # @!attribute aux
+      #   Классический сборщик кладёт сюда ID образа, BuildKit — base64 protobuf `StatusResponse` при `id` = `moby.buildkit.trace`.
+      #
+      #   @return [DockerEngineRuby::Models::ImageID, String, nil]
+      optional :aux, union: -> { DockerEngineRuby::BuildInfo::Aux }
+
+      # @!method initialize(id: nil, stream: nil, error: nil, error_detail: nil, status: nil, progress: nil, progress_detail: nil, aux: nil)
+      #   @param id [String]
+      #
+      #   @param stream [String] Строка вывода сборки.
+      #
+      #   @param error [String] Устаревшее поле: то же, что `error_detail.message`.
+      #
+      #   @param error_detail [DockerEngineRuby::Models::ErrorDetail]
+      #
+      #   @param status [String]
+      #
+      #   @param progress [String] Устаревшее поле: прогресс строкой.
+      #
+      #   @param progress_detail [DockerEngineRuby::Models::ProgressDetail]
+      #
+      #   @param aux [DockerEngineRuby::Models::ImageID, String] Классический сборщик кладёт сюда ID образа, BuildKit — base64 protobuf `StatusResponse` при `id` = `moby.buildkit.trace`.
+
+      # @see DockerEngineRuby::Models::BuildInfo#aux
+      module Aux
+        extend DockerEngineRuby::Internal::Type::Union
+
+        variant -> { DockerEngineRuby::ImageID }
+
+        variant String
+
+        # @!method self.variants
+        #   @return [Array(DockerEngineRuby::Models::ImageID, String)]
+      end
+    end
+  end
+end