background: added registry event handling

From now on, registry notifications (image got pushed, image deleted)
will be handled entirely in the background. Portus itself will only
create the event on the DB, and it will expect for it to be handled
automatically by the background process.

Fixes #940
See #1353 and #1526

Signed-off-by: Miquel Sabaté Solà <msabate@suse.com>

app/controllers/api/v2/events_controller.rb

@@ -5,7 +5,7 @@ class Api::V2::EventsController < Api::BaseController
   # A new notification is coming, register it if valid.
   def create
     body = JSON.parse(request.body.read)
-    Portus::RegistryNotification.process!(body, Repository, Webhook)
+    Portus::RegistryNotification.process!(body)
     head status: :accepted
   end
 end

app/models/registry_event.rb

@@ -6,14 +6,36 @@
 #
 #  id         :integer          not null, primary key
 #  event_id   :string(255)      default("")
-#  repository :string(255)      default("")
-#  tag        :string(255)      default("")
 #  created_at :datetime         not null
 #  updated_at :datetime         not null
+#  status     :integer          default(0)
+#  data       :text(65535)
 #
 
 # RegistryEvent represents an event coming from the Registry. This model stores
 # events that are being handled by Portus or that have been handled before. This
 # way we avoid duplication of registry events
 class RegistryEvent < ActiveRecord::Base
+  HANDLERS = [Repository, Webhook].freeze
+
+  # NOTE: as a Hash because in Rails 5 we'll be able to pass a proper prefix.
+  enum status: { done: 0, progress: 1, fresh: 2 }
+
+  # Processes the notification data with the given handlers. The data is the
+  # parsed JSON body as given by the registry. A handler is a class that can
+  # call the `handle_#{event}_event` method. This method receives an `event`
+  # object, which is the event object as given by the registry.
+  def self.handle!(event)
+    RegistryEvent.where(event_id: event["id"]).update_all(status: RegistryEvent.statuses[:progress])
+
+    action = event["action"]
+    Rails.logger.info "Handling '#{event["action"]}' event:\n#{JSON.pretty_generate(event)}"
+
+    # Delegate the handling to the known handlers.
+    HANDLERS.each { |handler| handler.send("handle_#{action}_event".to_sym, event) }
+
+    # Finally mark this event as handled, so a background job does not pick it
+    # up again.
+    RegistryEvent.where(event_id: event["id"]).update_all(status: RegistryEvent.statuses[:done])
+  end
 end

bin/background.rb

@@ -12,11 +12,13 @@
 
 while ::Portus::DB.ping != :ready
   if count >= TIMEOUT
-    puts "Timeout reached, exiting with error. Check the logs..."
+    Rails.logger.tagged("Database") do
+      Rails.logger.error "Timeout reached, exiting with error. Check the logs..."
+    end
     exit 1
   end
 
-  puts "Waiting for DB to be ready"
+  Rails.logger.tagged("Database") { Rails.logger.error "Not ready yet. Waiting..." }
   sleep 5
   count += 5
 end
@@ -25,25 +27,52 @@
 # The DB is up, now let's define the different background jobs as classes.
 #
 
+require "portus/background/registry"
 require "portus/background/security_scanning"
 
-they = [::Portus::Background::SecurityScanning.new]
+they = [::Portus::Background::Registry.new, ::Portus::Background::SecurityScanning.new]
 values = they.map { |v| "'#{v}'" }.join(", ")
-Rails.logger.info "Running: #{values}"
+Rails.logger.tagged("Initialization") { Rails.logger.info "Running: #{values}" }
 
 #
-# Finally, we will loop infinitely like this:
-#   1. Each background job will execute its task.
-#   2. Then we will sleep until there's more work to be done.
+# Between each iteration of the main loop there's going to be a sleep time. This
+# sleep time is determined by the amount expected by each backend. The following
+# block defines the sleep time and the maximum value. They all have to be
+# divisible.
 #
 
-SLEEP_VALUE = 10
+SLEEP_VALUE, TOP_SLEEP_VALUE = they.minmax_by(&:sleep_value).map(&:sleep_value)
+they.each do |v|
+  value = v.sleep_value
+  next unless value % SLEEP_VALUE != 0
+
+  Rails.logger.tagged "Initialization" do
+    Rails.logger.error "Encountered '#{value}', which is not divisible by '#{SLEEP_VALUE}'"
+  end
+  exit 1
+end
+slept = 0
+
+#
+# Finally, we will loop infinitely like this:
+#   1. Each background job will execute its task if needed (given the sleep time
+#      and the `work?` method.
+#   2. Then we will go to sleep for `SLEEP_VALUE` seconds.
+#
 
-# Loop forever executing the given tasks. It will go to sleep for spans of
-# `SLEEP_VALUE` seconds, if there's nothing else to be done.
 loop do
-  they.each { |t| t.execute! if t.work? }
-  sleep SLEEP_VALUE until they.any?(&:work?)
+  they.each do |t|
+    next if slept % t.sleep_value != 0
+    t.execute! if t.work?
+  end
+
+  break if ARGV.first == "--one-shot"
+  sleep SLEEP_VALUE
+
+  # Increase the sleep value by SLEEP_VALUE. If it turns out we reached out the
+  # maximum value, reset it to zero, so the number gets too big.
+  slept += SLEEP_VALUE
+  slept = 0 if (slept % TOP_SLEEP_VALUE).zero?
 end
 
 exit 0

db/migrate/20171213093923_add_status_and_data_to_registry_event.rb

@@ -0,0 +1,6 @@
+class AddStatusAndDataToRegistryEvent < ActiveRecord::Migration
+  def change
+    add_column :registry_events, :status, :integer, default: RegistryEvent.statuses[:done]
+    add_column :registry_events, :data, :text
+  end
+end

db/migrate/20171213094038_remove_repository_and_tag_from_registry_event.rb

@@ -0,0 +1,6 @@
+class RemoveRepositoryAndTagFromRegistryEvent < ActiveRecord::Migration
+  def change
+    remove_column :registry_events, :repository, :string
+    remove_column :registry_events, :tag, :string
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 20171130095821) do
+ActiveRecord::Schema.define(version: 20171213094038) do
 
   create_table "activities", force: :cascade do |t|
     t.integer  "trackable_id",   limit: 4
@@ -90,11 +90,11 @@
   add_index "registries", ["name"], name: "index_registries_on_name", unique: true, using: :btree
 
   create_table "registry_events", force: :cascade do |t|
-    t.string   "event_id",   limit: 255, default: ""
-    t.string   "repository", limit: 255, default: ""
-    t.string   "tag",        limit: 255, default: ""
-    t.datetime "created_at",                          null: false
-    t.datetime "updated_at",                          null: false
+    t.string   "event_id",   limit: 255,   default: ""
+    t.datetime "created_at",                            null: false
+    t.datetime "updated_at",                            null: false
+    t.integer  "status",     limit: 4,     default: 0
+    t.text     "data",       limit: 65535
   end
 
   create_table "repositories", force: :cascade do |t|

lib/portus/background/registry.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Portus
+  module Background
+    # Registry represents a task that syncs pending registry events into Portus.
+    class Registry
+      # Returns how many seconds has to pass between each loop for this
+      # background service.
+      def sleep_value
+        2
+      end
+
+      # Returns always true because this does not depend on some configuration
+      # options and because the `execute!` method will deal with valid rows
+      # already.
+      def work?
+        true
+      end
+
+      def execute!
+        RegistryEvent.where(status: RegistryEvent.statuses[:fresh]).find_each do |e|
+          data = JSON.parse(e.data)
+          RegistryEvent.handle!(data)
+        end
+      end
+
+      def to_s
+        "Registry events"
+      end
+    end
+  end
+end

lib/portus/background/security_scanning.rb

@@ -7,6 +7,12 @@ module Background
     # SecurityScanning represents a task for checking vulnerabilities of tags that
     # have not been scanned yet.
     class SecurityScanning
+      # Returns how many seconds has to pass between each loop for this
+      # background service.
+      def sleep_value
+        10
+      end
+
       def work?
         ::Portus::Security.enabled? && Tag.exists?(scanned: Tag.statuses[:scan_none])
       end

lib/portus/registry_notification.rb

@@ -7,29 +7,23 @@ class RegistryNotification
     # An array with the events that a handler has to support.
     HANDLED_EVENTS = %w[push delete].freeze
 
-    # Processes the notification data with the given handlers. The data is the
-    # parsed JSON body as given by the registry. A handler is a class that can
-    # call the `handle_#{event}_event` method. This method receives an `event`
-    # object, which is the event object as given by the registry.
-    def self.process!(data, *handlers)
+    # It filters the event from the registry so the background job can actually
+    # handle this request.
+    def self.process!(data)
       data["events"].each do |event|
-        Rails.logger.debug "Incoming event:\n#{JSON.pretty_generate(event)}"
+        Rails.logger.debug "Filtering event:\n#{JSON.pretty_generate(event)}"
 
+        # Skip irrelevant or already-handled events.
         next unless should_handle?(event)
-        action = event["action"]
 
-        # Only register pushes, since they are the conflicting actions since 2.5
-        if action == "push"
-          RegistryEvent.create!(
-            event_id:   event["id"],
-            repository: event["target"]["repository"],
-            tag:        event["target"]["tag"]
-          )
-        end
-
-        # Now it's time to delegate the handling to the proper handler.
-        Rails.logger.info "Handling '#{action}' event:\n#{JSON.pretty_generate(event)}"
-        handlers.each { |handler| handler.send("handle_#{action}_event".to_sym, event) }
+        # At this point, events will be handled by
+        # ::Portus::Background::RegistryEvent. So just create the event on the
+        # DB and let the background process fetch this.
+        RegistryEvent.create!(
+          event_id: event["id"],
+          data:     event.to_json,
+          status:   RegistryEvent.statuses[:fresh]
+        )
       end
     end
 

spec/lib/portus/background/registry_spec.rb

@@ -0,0 +1,67 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+require "portus/background/registry"
+
+describe ::Portus::Background::Registry do
+  describe "#sleep_value" do
+    it "returns always 2" do
+      expect(subject.sleep_value).to eq 2
+    end
+  end
+
+  describe "#work?" do
+    it "always return true" do
+      expect(subject.work?).to be_truthy
+    end
+  end
+
+  describe "#execute!" do
+    let(:delete)    { ::Portus::Fixtures::RegistryEvent::DELETE.dup }
+    let(:version23) { ::Portus::Fixtures::RegistryEvent::VERSION23.dup }
+
+    it "calls RegistryEvent.handle! with the given events" do
+      data = '{ "key": "value" }'.to_json
+      RegistryEvent.create!(event_id: 1, data: data, status: RegistryEvent.statuses[:fresh])
+
+      count = 0
+      parsed = JSON.parse(data)
+      allow(RegistryEvent).to receive(:handle!).with(parsed) { count += 1 }
+      subject.execute!
+      expect(count).to eq 1
+    end
+
+    it "does nothing if no events happened" do
+      RegistryEvent.create!(event_id: 1, status: RegistryEvent.statuses[:done])
+
+      count = 0
+      allow(RegistryEvent).to receive(:handle!) { count += 1 }
+      subject.execute!
+      expect(count).to eq 0
+    end
+
+    it "performs well in an overall example" do
+      data = { "events" => [delete, version23] }
+
+      # All notifications are properly registered.
+      ::Portus::RegistryNotification.process!(data)
+      expect(RegistryEvent.count).to eq 2
+      expect(RegistryEvent.all.all? { |e| e.status == "fresh" }).to be_truthy
+
+      # And finally it processes the notifications.
+      expect(Webhook).to receive(:handle_delete_event).with(delete)
+      expect(Repository).to receive(:handle_delete_event).with(delete)
+      expect(Webhook).to receive(:handle_push_event).with(version23)
+      expect(Repository).to receive(:handle_push_event).with(version23)
+      subject.execute!
+
+      expect(RegistryEvent.all.all? { |e| e.status == "done" }).to be_truthy
+    end
+  end
+
+  describe "#to_s" do
+    it "works" do
+      expect(subject.to_s).to eq "Registry events"
+    end
+  end
+end

spec/lib/portus/background/security_scanning_spec.rb

@@ -17,6 +17,12 @@
     }
   end
 
+  describe "#sleep_value" do
+    it "returns always 10" do
+      expect(subject.sleep_value).to eq 10
+    end
+  end
+
   describe "#work?" do
     it "returns false if security scanning is not enabled" do
       APP_CONFIG["security"]["clair"]["server"] = ""