apps/workbench/app/models/arvados_api_client.rb

   1 require 'httpclient'
   2 require 'thread'
   3
   4 class ArvadosApiClient
   5   class ApiError < StandardError
   6     attr_reader :api_response, :api_response_s, :api_status, :request_url
   7
   8     def initialize(request_url, errmsg)
   9       @request_url = request_url
  10       @api_response ||= {}
  11       errors = @api_response[:errors]
  12       if not errors.is_a?(Array)
  13         @api_response[:errors] = [errors || errmsg]
  14       end
  15       super(errmsg)
  16     end
  17   end
  18
  19   class NoApiResponseException < ApiError
  20     def initialize(request_url, exception)
  21       @api_response_s = exception.to_s
  22       super(request_url,
  23             "#{exception.class.to_s} error connecting to API server")
  24     end
  25   end
  26
  27   class InvalidApiResponseException < ApiError
  28     def initialize(request_url, api_response)
  29       @api_status = api_response.status_code
  30       @api_response_s = api_response.content
  31       super(request_url, "Unparseable response from API server")
  32     end
  33   end
  34
  35   class ApiErrorResponseException < ApiError
  36     def initialize(request_url, api_response)
  37       @api_status = api_response.status_code
  38       @api_response_s = api_response.content
  39       @api_response = Oj.load(@api_response_s, :symbol_keys => true)
  40       errors = @api_response[:errors]
  41       if errors.respond_to?(:join)
  42         errors = errors.join("\n\n")
  43       else
  44         errors = errors.to_s
  45       end
  46       super(request_url, "#{errors} [API: #{@api_status}]")
  47     end
  48   end
  49
  50   class AccessForbiddenException < ApiErrorResponseException; end
  51   class NotFoundException < ApiErrorResponseException; end
  52   class NotLoggedInException < ApiErrorResponseException; end
  53
  54   ERROR_CODE_CLASSES = {
  55     401 => NotLoggedInException,
  56     403 => AccessForbiddenException,
  57     404 => NotFoundException,
  58   }
  59
  60   @@profiling_enabled = Rails.configuration.profiling_enabled
  61   @@discovery = nil
  62
  63   # An API client object suitable for handling API requests on behalf
  64   # of the current thread.
  65   def self.new_or_current
  66     # If this thread doesn't have an API client yet, *or* this model
  67     # has been reloaded since the existing client was created, create
  68     # a new client. Otherwise, keep using the latest client created in
  69     # the current thread.
  70     unless Thread.current[:arvados_api_client].andand.class == self
  71       Thread.current[:arvados_api_client] = new
  72     end
  73     Thread.current[:arvados_api_client]
  74   end
  75
  76   def initialize *args
  77     @api_client = nil
  78     @client_mtx = Mutex.new
  79   end
  80
  81   def api(resources_kind, action, data=nil, tokens={})
  82
  83     profile_checkpoint
  84
  85     if not @api_client
  86       @client_mtx.synchronize do
  87         @api_client = HTTPClient.new
  88         @api_client.ssl_config.timeout = Rails.configuration.api_client_connect_timeout
  89         @api_client.connect_timeout = Rails.configuration.api_client_connect_timeout
  90         @api_client.receive_timeout = Rails.configuration.api_client_receive_timeout
  91         if Rails.configuration.arvados_insecure_https
  92           @api_client.ssl_config.verify_mode = OpenSSL::SSL::VERIFY_NONE
  93         else
  94           # Use system CA certificates
  95           ["/etc/ssl/certs/ca-certificates.crt",
  96            "/etc/pki/tls/certs/ca-bundle.crt"]
  97             .select { |ca_path| File.readable?(ca_path) }
  98             .each { |ca_path| @api_client.ssl_config.add_trust_ca(ca_path) }
  99         end
 100         if Rails.configuration.api_response_compression
 101           @api_client.transparent_gzip_decompression = true
 102         end
 103       end
 104     end
 105
 106     resources_kind = class_kind(resources_kind).pluralize if resources_kind.is_a? Class
 107     url = "#{self.arvados_v1_base}/#{resources_kind}#{action}"
 108
 109     # Clean up /arvados/v1/../../discovery/v1 to /discovery/v1
 110     url.sub! '/arvados/v1/../../', '/'
 111
 112     query = {
 113       'api_token' => (tokens[:arvados_api_token] ||
 114                       Thread.current[:arvados_api_token] ||
 115                       ''),
 116       'reader_tokens' => ((tokens[:reader_tokens] ||
 117                            Thread.current[:reader_tokens] ||
 118                            []) +
 119                           [Rails.configuration.anonymous_user_token]).to_json,
 120       'current_request_id' => (Thread.current[:current_request_id] || ''),
 121     }
 122     if !data.nil?
 123       data.each do |k,v|
 124         if v.is_a? String or v.nil?
 125           query[k] = v
 126         elsif v == true
 127           query[k] = 1
 128         elsif v == false
 129           query[k] = 0
 130         else
 131           query[k] = Oj.dump(v, mode: :compat)
 132         end
 133       end
 134     else
 135       query["_method"] = "GET"
 136     end
 137
 138     if @@profiling_enabled
 139       query["_profile"] = "true"
 140     end
 141
 142     header = {"Accept" => "application/json"}
 143
 144     profile_checkpoint { "Prepare request #{query["_method"] or "POST"} #{url} #{query[:uuid]} #{query.inspect[0,256]}" }
 145     msg = @client_mtx.synchronize do
 146       begin
 147         @api_client.post(url, query, header: header)
 148       rescue => exception
 149         raise NoApiResponseException.new(url, exception)
 150       end
 151     end
 152     profile_checkpoint 'API transaction'
 153     if @@profiling_enabled
 154       if msg.headers['X-Runtime']
 155         Rails.logger.info "API server: #{msg.headers['X-Runtime']} runtime reported"
 156       end
 157       Rails.logger.info "Content-Encoding #{msg.headers['Content-Encoding'].inspect}, Content-Length #{msg.headers['Content-Length'].inspect}, actual content size #{msg.content.size}"
 158     end
 159
 160     begin
 161       resp = Oj.load(msg.content, :symbol_keys => true)
 162     rescue Oj::ParseError
 163       resp = nil
 164     end
 165
 166     if not resp.is_a? Hash
 167       raise InvalidApiResponseException.new(url, msg)
 168     elsif msg.status_code != 200
 169       error_class = ERROR_CODE_CLASSES.fetch(msg.status_code,
 170                                              ApiErrorResponseException)
 171       raise error_class.new(url, msg)
 172     end
 173
 174     if resp[:_profile]
 175       Rails.logger.info "API client: " \
 176       "#{resp.delete(:_profile)[:request_time]} request_time"
 177     end
 178     profile_checkpoint 'Parse response'
 179     resp
 180   end
 181
 182   def self.patch_paging_vars(ary, items_available, offset, limit, links=nil)
 183     if items_available
 184       (class << ary; self; end).class_eval { attr_accessor :items_available }
 185       ary.items_available = items_available
 186     end
 187     if offset
 188       (class << ary; self; end).class_eval { attr_accessor :offset }
 189       ary.offset = offset
 190     end
 191     if limit
 192       (class << ary; self; end).class_eval { attr_accessor :limit }
 193       ary.limit = limit
 194     end
 195     if links
 196       (class << ary; self; end).class_eval { attr_accessor :links }
 197       ary.links = links
 198     end
 199     ary
 200   end
 201
 202   def unpack_api_response(j, kind=nil)
 203     if j.is_a? Hash and j[:items].is_a? Array and j[:kind].match(/(_list|List)$/)
 204       ary = j[:items].collect { |x| unpack_api_response x, x[:kind] }
 205       links = ArvadosResourceList.new Link
 206       links.results = (j[:links] || []).collect do |x|
 207         unpack_api_response x, x[:kind]
 208       end
 209       self.class.patch_paging_vars(ary, j[:items_available], j[:offset], j[:limit], links)
 210     elsif j.is_a? Hash and (kind || j[:kind])
 211       oclass = self.kind_class(kind || j[:kind])
 212       if oclass
 213         j.keys.each do |k|
 214           childkind = j["#{k.to_s}_kind".to_sym]
 215           if childkind
 216             j[k] = self.unpack_api_response(j[k], childkind)
 217           end
 218         end
 219         oclass.new.private_reload(j)
 220       else
 221         j
 222       end
 223     else
 224       j
 225     end
 226   end
 227
 228   def arvados_login_url(params={})
 229     if Rails.configuration.respond_to? :arvados_login_base
 230       uri = Rails.configuration.arvados_login_base
 231     else
 232       uri = self.arvados_v1_base.sub(%r{/arvados/v\d+.*}, '/login')
 233     end
 234     if params.size > 0
 235       uri += '?' << params.collect { |k,v|
 236         CGI.escape(k.to_s) + '=' + CGI.escape(v.to_s)
 237       }.join('&')
 238     end
 239     uri
 240   end
 241
 242   def arvados_logout_url(params={})
 243     arvados_login_url(params).sub('/login','/logout')
 244   end
 245
 246   def arvados_v1_base
 247     Rails.configuration.arvados_v1_base
 248   end
 249
 250   def discovery
 251     @@discovery ||= api '../../discovery/v1/apis/arvados/v1/rest', ''
 252   end
 253
 254   def kind_class(kind)
 255     kind.match(/^arvados\#(.+?)(_list|List)?$/)[1].pluralize.classify.constantize rescue nil
 256   end
 257
 258   def class_kind(resource_class)
 259     resource_class.to_s.underscore
 260   end
 261
 262   def self.class_kind(resource_class)
 263     resource_class.to_s.underscore
 264   end
 265
 266   protected
 267   def profile_checkpoint label=nil
 268     return if !@@profiling_enabled
 269     label = yield if block_given?
 270     t = Time.now
 271     if label and @profile_t0
 272       Rails.logger.info "API client: #{t - @profile_t0} #{label}"
 273     end
 274     @profile_t0 = t
 275   end
 276 end