X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/55888e63181a0847d3e00344fa9c7c5e747082ae..72aa70eec7693bfb5d46a4bdac3619b3c6b1f79c:/services/api/lib/eventbus.rb diff --git a/services/api/lib/eventbus.rb b/services/api/lib/eventbus.rb index b7579d307d..1754fc0ae9 100644 --- a/services/api/lib/eventbus.rb +++ b/services/api/lib/eventbus.rb @@ -49,54 +49,99 @@ class EventBus end # Push out any pending events to the connection +ws+ - # +id+ the id of the most recent row in the log table, may be nil - def push_events ws, id = nil - begin - # Must have at least one filter set up to receive events - if ws.filters.length > 0 - # Start with log rows readable by user, sorted in ascending order - logs = Log.readable_by(ws.user).order("id asc") - - if ws.last_log_id - # Client is only interested in log rows that are newer than the - # last log row seen by the client. - logs = logs.where("logs.id > ?", ws.last_log_id) - elsif id - # No last log id, so only look at the most recently changed row - logs = logs.where("logs.id = ?", id.to_i) - else - return - end + # +notify_id+ the id of the most recent row in the log table, may be nil + # + # This accepts a websocket and a notify_id (this is the row id from Postgres + # LISTEN/NOTIFY, it may be nil if called from somewhere else) + # + # It queries the database for log rows that are either + # a) greater than ws.last_log_id, which is the last log id which was a candidate to be sent out + # b) if ws.last_log_id is nil, then it queries rows starting with notify_id + # + # Regular Arvados permissions are applied using readable_by() and filters using record_filters() + # To avoid clogging up the database, queries are limited to batches of 100. It will schedule a new + # push_events call if there are more log rows to send. + def push_events ws, notify_id + begin + if !notify_id.nil? and !ws.last_log_id.nil? and notify_id <= ws.last_log_id + # This notify is for a row we've handled already. + return + end + + # Must have at least one filter set up to receive events + if ws.filters.length > 0 + # Start with log rows readable by user, sorted in ascending order + logs = Log.readable_by(ws.user).order("id asc") + + cond_id = nil + cond_out = [] + param_out = [] - # Now process filters provided by client - cond_out = [] - param_out = [] - ws.filters.each do |filter| - ft = record_filters filter.filters, Log - cond_out += ft[:cond_out] + if !ws.last_log_id.nil? + # Client is only interested in log rows that are newer than the + # last log row seen by the client. + cond_id = "logs.id > ?" + param_out << ws.last_log_id + elsif !notify_id.nil? + # No last log id, so look at rows starting with notify id + cond_id = "logs.id >= ?" + param_out << notify_id + else + # No log id to start from, nothing to do, return + return + end + + # Now build filters provided by client + ws.filters.each do |filter| + ft = record_filters filter.filters, Log + if ft[:cond_out].any? + # Join the clauses within a single subscription filter with AND + # so it is consistent with regular queries + cond_out << "(#{ft[:cond_out].join ') AND ('})" param_out += ft[:param_out] end + end - # Add filters to query - if cond_out.any? - logs = logs.where('(' + cond_out.join(') OR (') + ')', *param_out) - end + # Add filters to query + if cond_out.any? + # Join subscriptions with OR + logs = logs.where(cond_id + " AND ((#{cond_out.join ') OR ('}))", *param_out) + else + logs = logs.where(cond_id, *param_out) + end - # Finally execute query and actually send the matching log rows - logs.each do |l| - ws.send(l.as_api_response.to_json) - ws.last_log_id = l.id + # Execute query and actually send the matching log rows + count = 0 + limit = 100 + + logs.limit(limit).each do |l| + ws.send(l.as_api_response.to_json) + ws.last_log_id = l.id + count += 1 + end + + if count == limit + # Number of rows returned was capped by limit(), we need to schedule + # another query to get more logs (will start from last_log_id + # reported by current query) + EventMachine::schedule do + push_events ws, nil end - elsif id - # No filters set up, so just record the sequence number - ws.last_log_id = id.to_i + elsif !notify_id.nil? and (ws.last_log_id.nil? or notify_id > ws.last_log_id) + # Number of rows returned was less than cap, but the notify id is + # higher than the last id visible to the client, so update last_log_id + ws.last_log_id = notify_id end - rescue Exception => e - puts "Error publishing event: #{$!}" - puts "Backtrace:\n\t#{e.backtrace.join("\n\t")}" - ws.send ({status: 500, message: 'error'}.to_json) - ws.close + elsif !notify_id.nil? + # No filters set up, so just record the sequence number + ws.last_log_id = notify_id end + rescue => e + Rails.logger.warn "Error publishing event: #{$!}" + Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}" + ws.send ({status: 500, message: 'error'}.to_json) + ws.close + end end # Handle inbound subscribe or unsubscribe message. @@ -121,7 +166,7 @@ class EventBus ws.send ({status: 200, message: 'subscribe ok', filter: p}.to_json) # Send any pending events - push_events ws + push_events ws, nil else ws.send ({status: 403, message: "maximum of #{MAX_FILTERS} filters allowed per connection"}.to_json) end @@ -142,9 +187,9 @@ class EventBus end rescue Oj::Error => e ws.send ({status: 400, message: "malformed request"}.to_json) - rescue Exception => e - puts "Error handling message: #{$!}" - puts "Backtrace:\n\t#{e.backtrace.join("\n\t")}" + rescue => e + Rails.logger.warn "Error handling message: #{$!}" + Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}" ws.send ({status: 500, message: 'error'}.to_json) ws.close end @@ -204,9 +249,12 @@ class EventBus # channel and calls #push_events to actually dispatch the # events to the client. conn.wait_for_notify do |channel, pid, payload| - @channel.push payload + @channel.push payload.to_i end end + rescue NoMemoryError + EventMachine::stop_event_loop + abort "Out of memory" ensure # Don't want the connection to still be listening once we return # it to the pool - could result in weird behavior for the next