Add required logic for replication to start working

Author: ereslibre

app/controllers/api/v1/objects_controller.rb

@@ -3,14 +3,25 @@ class Api::V1::ObjectsController < ApplicationController
   before_action :has_permissions?
 
   def create
-    object = params[:object][:payload].read
-    object_name = Digest::SHA256.hexdigest object
-    File.open("#{@volume}/#{object_name}", "w") do |f|
-      f.write object
-    end
-    Brain.request path: "/cluster-api/v1/upload-tokens/#{params[:upload_token]}",
-                  method: :delete,
-                  query: { client_ip: request.remote_ip },
+    object_contents = params[:object][:payload].read
+    digest = Cell.digest_contents object_contents
+    object_name = digest[:sha256sum]
+    object_path = File.join @volume, object_name
+    File.open(object_path, "w") { |f| f.write object }
+    Brain.request path: "/cluster-api/v1/objects",
+                  method: :post,
+                  payload: {
+                    client_ip: request.remote_ip,
+                    upload_token: params[:upload_token],
+                    object: {
+                      uuid: SecureRandom.uuid,
+                      name: params[:object][:payload].original_filename,
+                      size: File.size(object_path),
+                      md5sum: digest[:md5sum],
+                      sha1sum: digest[:sha1sum],
+                      sha256sum: digest[:sha256sum]
+                    }
+                  },
                   access_token: @jwt
     ok
   end

app/controllers/cluster_api/v1/objects/download_controller.rb

@@ -1,6 +1,38 @@
 class ClusterApi::V1::Objects::DownloadController < ClusterApplicationController
 
+  before_action :has_permissions?
+  before_action :object_matches?
+  before_action :source_allowed?
+  before_action :target_ip_address_allowed?
+
   def show
+    send_file File.join(@source_cell_volume, @object_sha256sum)
+  end
+
+  private
+
+  def has_permissions?
+    if !Brain.ok?(path: "/cluster-api/v1/sync-tokens/#{params[:sync_token]}") do |json_response|
+         @object_uuid = json_response[:object][:uuid]
+         @object_sha256sum = json_response[:object][:sha256sum]
+         @source_cell_uuid = json_response[:source_cell][:uuid]
+         @source_cell_volume = json_response[:source_cell][:volume]
+         @target_cell_ip_address = json_response[:target_cell][:ip_address]
+       end
+      forbidden
+    end
+  end
+
+  def object_matches?
+    forbidden if @object_uuid != params[:uuid]
+  end
+
+  def source_allowed?
+    forbidden if @source_cell_uuid != Cell.machine_uuid
+  end
+
+  def target_ip_address_allowed?
+    forbidden if @target_cell_ip_address != request.remote_ip
   end
 
 end

app/controllers/cluster_api/v1/objects_controller.rb

@@ -0,0 +1,34 @@
+class ClusterApi::V1::ObjectsController < ClusterApplicationController
+
+  before_action :has_permissions?
+  before_action :target_allowed?
+
+  def create
+    object_contents, _ = Cell.request cell_ip: @source_cell_ip_address,
+                                      path: "/cluster-api/v1/objects/#{@object_uuid}/download",
+                                      query: { sync_token: params[:sync_token] }
+    object_path = File.join @target_cell_volume, @object_sha256sum
+    File.open(object_path, "w") { |f| f.write object_contents.body }
+    # FIXME: notify brain
+    ok
+  end
+
+  private
+
+  def has_permissions?
+    if !Brain.ok?(path: "/cluster-api/v1/sync-tokens/#{params[:sync_token]}") do |json_response|
+         @object_uuid = json_response[:object][:uuid]
+         @object_sha256sum = json_response[:object][:sha256sum]
+         @target_cell_uuid = json_response[:target_cell][:uuid]
+         @target_cell_volume = json_response[:target_cell][:volume]
+         @source_cell_ip_address = json_response[:source_cell][:ip_address]
+       end
+      forbidden
+    end
+  end
+
+  def target_allowed?
+    forbidden if @target_cell_uuid != Cell.machine_uuid
+  end
+
+end

config/initializers/register_cell.rb

@@ -4,6 +4,7 @@
                 cell: {
                   uuid: Cell.machine_uuid,
                   fqdn: Cell.fqdn,
-                  volumes: Cell.storage_volumes
+                  volumes: Cell.storage_volumes,
+                  public_ip_address: ENV["HOST_IP_ADDRESS"]
                 }
               })

config/routes.rb

@@ -1,3 +1,5 @@
+require "sidekiq/web"
+
 Rails.application.routes.draw do
   namespace :api do
     namespace :v1 do
@@ -10,11 +12,12 @@
   end
   namespace :cluster_api, path: "cluster-api" do
     namespace :v1 do
-      resources :objects, only: [], param: :uuid do
+      resources :objects, only: :create, param: :uuid do
         member do
           resource :download, only: :show, controller: "objects/download"
         end
       end
     end
   end
+  mount Sidekiq::Web => "/sidekiq" unless Rails.env.production?
 end

lib/cell.rb

@@ -1,14 +1,24 @@
 class Cell
 
   require "socket"
+  require "fileutils"
   require "securerandom"
   require "sys/filesystem"
 
+  def self.digest_content(contents)
+    {
+      md5sum: Digest::MD5.hexdigest(contents),
+      sha1sum: Digest::SHA1.hexdigest(contents),
+      sha256sum: Digest::SHA256.hexdigest(contents)
+    }
+  end
+
   def self.machine_uuid
-    File.read("/etc/brain/machine-uuid").strip
+    File.read("/etc/cell/machine-uuid").strip
   rescue Errno::ENOENT
+    FileUtils.mkdir_p "/etc/cell"
     SecureRandom.uuid.tap do |machine_uuid|
-      File.open("/etc/brain/machine-uuid", "w") { |f| f.write machine_uuid }
+      File.open("/etc/cell/machine-uuid", "w") { |f| f.write machine_uuid }
     end
   end
 
@@ -45,4 +55,36 @@ def self.storage_mountpoints
     ]
   end
 
+  def self.request(cell_ip:, path:, method: :get, payload: nil, query: nil)
+    uri = URI("http://#{cell_ip}:#{ENV["CELL_SERVICE_PORT"]}#{path}")
+    uri.query = URI.encode_www_form(query) if query
+    http = Net::HTTP.new uri.host, uri.port
+    req = case method
+          when :head
+            Net::HTTP::Head.new uri.request_uri
+          when :get
+            Net::HTTP::Get.new uri.request_uri
+          when :post
+            Net::HTTP::Post.new uri.request_uri, { "Content-Type" => "application/json" }
+          when :put
+            Net::HTTP::Put.new uri.request_uri, { "Content-Type" => "application/json" }
+          when :patch
+            Net::HTTP::Patch.new uri.request_uri, { "Content-Type" => "application/json" }
+          when :delete
+            Net::HTTP::Delete.new uri.request_uri
+          else
+            raise "unknown method"
+          end
+    if !%i(head get delete).include?(method) && payload
+      req.body = payload.to_json
+    end
+    response = http.request req
+    json_response = JSON.parse response.body, symbolize_names: true rescue nil
+    if block_given?
+      yield response, json_response
+    else
+      return response, json_response
+    end
+  end
+
 end