Refactor to JSON messages and support version selection

bin/client.rb

@@ -46,7 +46,16 @@ class PipelineClient
   end
 
   def analyze(track_slug, exercise_slug, solution_slug, iteration_folder)
-    send_msg("analyze_#{track_slug}|#{exercise_slug}|#{solution_slug}|#{iteration_folder}", 10000)
+    params = {
+      action: "analyze_iteration",
+      track_slug: track_slug,
+      container_version: "v0.0.5",
+      exercise_slug: exercise_slug,
+      solution_slug: solution_slug,
+      iteration_folder: iteration_folder
+    }
+    msg = params.to_json
+    send_msg(msg, 10000)
   end
 
 end

bin/worker

@@ -6,6 +6,6 @@ require "pipeline"
 
 env_base = ARGV[0]
 
-Pipeline.load_config(File.expand_path('../../config/pipeline.yml', __FILE__))
+# Pipeline.load_config(File.expand_path('../../config/pipeline.yml', __FILE__))
-server = Pipeline::Rpc::Server.new(env_base)
+server = Pipeline::Rpc::Worker.new(env_base)
 server.listen

lib/pipeline.rb

@@ -76,7 +76,7 @@ module Pipeline
 end
 
 require "pipeline/rpc/router"
-require "pipeline/rpc/server"
+require "pipeline/rpc/worker"
 require "pipeline/analyzer_repo"
 require "pipeline/container_repo"
 require "pipeline/validation/check_invokable"

lib/pipeline/rpc/router.rb

@@ -104,7 +104,7 @@ module Pipeline::Rpc
       puts ">>>> #{msg}"
       if (msg[2] == "describe_analysers")
         analyzer_spec = {
-          "ruby" => "master"
+          "ruby" => [ "v0.0.3", "v0.0.5" ]
         }
         reply = [msg.first, "", analyzer_spec.to_json]
         front_end_socket.send_strings(reply)

lib/pipeline/rpc/worker.rb

@@ -1,4 +1,4 @@
-class Pipeline::Rpc::Server
+class Pipeline::Rpc::Worker
 
   attr_reader :context, :incoming, :outgoing, :environment
 
@@ -22,12 +22,15 @@ class Pipeline::Rpc::Server
 
     environment.prepare
 
-    analyzer_spec.each do |language_slug, version|
+    analyzer_spec.each do |language_slug, versions|
-      if environment.released?(language_slug)
+      puts "Preparing #{language_slug} #{versions}"
+      versions.each do |version|
+        if environment.released?(language_slug, version)
           puts "Already installed #{language_slug}"
         else
           puts "Installed #{language_slug}"
-        environment.release_analyzer(language_slug)
+          environment.release_analyzer(language_slug, version)
+        end
       end
     end
   end
@@ -41,12 +44,12 @@ class Pipeline::Rpc::Server
       incoming.recv_strings(msg)
       puts "Received request. Data: #{msg.inspect}"
       return_address = msg[0].unpack('c*')
-      puts return_address
+      raw_request = msg[2]
-      request = msg[2]
+      request = JSON.parse(raw_request)
-      if request.start_with? "analyze_"
+      puts request
-        _, arg = request.split("_", 2)
+      action = request["action"]
-        track, exercise_slug, solution_slug, location = arg.split("|")
+      if action == "analyze_iteration"
-        result = analyze(track, exercise_slug, solution_slug, location)
+        result = analyze(request)
         result["return_address"] = return_address
         result['msg_type'] = 'response'
         outgoing.send_string(result.to_json)
@@ -56,8 +59,20 @@ class Pipeline::Rpc::Server
     end
   end
 
-  def analyze(language_slug, exercise_slug, solution_slug, location)
+  def analyze(request)
-    analysis_run = environment.new_analysis(language_slug, exercise_slug, solution_slug)
+    language_slug = request["track_slug"]
+    exercise_slug = request["exercise_slug"]
+    solution_slug = request["solution_slug"]
+    location = request["iteration_folder"]
+    container_version = request["container_version"]
+
+    unless environment.released?(language_slug, container_version)
+      return {
+        error: "Container #{language_slug}:#{container_version} isn't available"
+      }
+    end
+
+    analysis_run = environment.new_invocation(language_slug, container_version, exercise_slug, solution_slug)
     analysis_run.prepare_iteration do |iteration_folder|
       location_uri = URI(location)
       bucket = location_uri.host

lib/pipeline/runtime/runtime_environment.rb

@@ -11,16 +11,16 @@ module Pipeline::Runtime
       FileUtils.mkdir_p env_base
     end
 
-    def released?(track_slug)
+    def released?(track_slug, version)
-      track_dir = "#{env_base}/#{track_slug}"
+      track_dir = "#{env_base}/#{track_slug}/#{version}"
       current_dir = "#{track_dir}/current"
       File.exist? current_dir
     end
 
-    def release_analyzer(track_slug)
+    def release_analyzer(track_slug, version)
       registry_endpoint = Pipeline.config["registry_endpoint"]
 
-      track_dir = "#{env_base}/#{track_slug}"
+      track_dir = "#{env_base}/#{track_slug}/#{version}"
       release_dir = "#{track_dir}/releases/#{Time.now.to_i}_release"
       current_dir = "#{track_dir}/current"
       FileUtils.mkdir_p release_dir
@@ -57,16 +57,18 @@ module Pipeline::Runtime
       FileUtils.symlink(release_dir, current_dir, force: true)
     end
 
-    def create_analyzer_workdir(track_slug)
+    # def create_analyzer_workdir(track_slug, version, version)
-      track_dir = "#{env_base}/#{track_slug}"
+    #   container_slug = "#{track_slug}/#{version}"
-      current_dir = "#{track_dir}/current"
+    #   track_dir = "#{env_base}/#{container_slug}"
-      iterations = "#{track_dir}/iterations"
+    #   current_dir = "#{track_dir}/current"
-      FileUtils.mkdir_p iterations
+    #   iterations = "#{track_dir}/iterations"
-    end
+    #   FileUtils.mkdir_p iterations
+    # end
 
-    def new_analysis(track_slug, exercise_slug, solution_slug)
+    def new_invocation(track_slug, version, exercise_slug, solution_slug)
-      puts "AnalysisRun: #{track_slug} #{exercise_slug} #{solution_slug}"
+      container_slug = "#{track_slug}/#{version}"
-      track_dir = "#{env_base}/#{track_slug}"
+      puts "AnalysisRun: #{container_slug} #{exercise_slug} #{solution_slug}"
+      track_dir = "#{env_base}/#{container_slug}"
       AnalysisRun.new(track_dir, exercise_slug, solution_slug)
     end