- Compass-ci - mailweb.openeuler.org

[PATCH v2 compass-ci 2/3] lib/mail-post/email_limit_queue: use ENV instead of static value
by Luan Shengde 29 Mar '21

29 Mar '21

use ENV variable instead of static value. enable user-defined new value without edit the code. just change the key and restart the service. Signed-off-by: Luan Shengde <shdluan(a)163.com> --- lib/mail-post/email_limit_queue.rb | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/mail-post/email_limit_queue.rb b/lib/mail-post/email_limit_queue.rb index c7e3954..62e9634 100755 --- a/lib/mail-post/email_limit_queue.rb +++ b/lib/mail-post/email_limit_queue.rb @@ -54,9 +54,9 @@ class EmailRateLimit end def change_queue(email) - return unless (@redis.hget 'email_in_limit', email).to_i >= 10 + return unless (@redis.hget 'email_in_limit', email).to_i >= ENV['EMAIL_LIMIT_COUNT'].to_i @redis.hdel 'email_in_limit', email - @redis.hset 'email_out_limit', email, 10 + @redis.hset 'email_out_limit', email, ENV['EMAIL_LIMIT_COUNT'] end end -- 2.23.0

1 0

[PATCH v2 compass-ci 1/3] container/send-internet-mail: add key EMAIL_LIMIT_COUNT
by Luan Shengde 29 Mar '21

29 Mar '21

Signed-off-by: Luan Shengde <shdluan(a)163.com> --- container/send-internet-mail/start | 1 + 1 file changed, 1 insertion(+) diff --git a/container/send-internet-mail/start b/container/send-internet-mail/start index 136435d..57bf1f6 100755 --- a/container/send-internet-mail/start +++ b/container/send-internet-mail/start @@ -14,6 +14,7 @@ names = Set.new %w[ SEND_MAIL_PORT SENT_MAILDIR REDIS_PORT + EMAIL_LIMIT_COUNT ] defaults = relevant_defaults(names) -- 2.23.0

1 0

[PATCH compass-ci 1/2] container/send-internet-mail: add key EMAIL_LIMIT_COUNT for start
by Luan Shengde 29 Mar '21

29 Mar '21

Signed-off-by: Luan Shengde <shdluan(a)163.com> --- container/send-internet-mail/start | 1 + 1 file changed, 1 insertion(+) diff --git a/container/send-internet-mail/start b/container/send-internet-mail/start index 136435d..57bf1f6 100755 --- a/container/send-internet-mail/start +++ b/container/send-internet-mail/start @@ -14,6 +14,7 @@ names = Set.new %w[ SEND_MAIL_PORT SENT_MAILDIR REDIS_PORT + EMAIL_LIMIT_COUNT ] defaults = relevant_defaults(names) -- 2.23.0

2 2

[PATCH compass-ci 7/7] scheduler: add get_new_seqno in redis client
by Cao Xueliang 29 Mar '21

29 Mar '21

Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/scheduler/redis_client.cr | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/scheduler/redis_client.cr b/src/scheduler/redis_client.cr index 030d1b7..ed2af6e 100644 --- a/src/scheduler/redis_client.cr +++ b/src/scheduler/redis_client.cr @@ -70,4 +70,8 @@ class Redis::Client def remove_finished_job(job_id : String) @client.hdel("sched/id2job", job_id) end + + def get_new_seqno + return @client.incr("#{QUEUE_NAME_BASE}/seqno") + end end -- 2.23.0

1 0

[PATCH compass-ci 6/7] scheduler: update job parameter in etcd
by Cao Xueliang 29 Mar '21

29 Mar '21

Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/lib/sched.cr | 7 ------- src/scheduler/update_job_parameter.cr | 9 ++++++++- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/lib/sched.cr b/src/lib/sched.cr index f74df9b..6fd6657 100644 --- a/src/lib/sched.cr +++ b/src/lib/sched.cr @@ -166,13 +166,6 @@ class Sched @es.update_tbox(testbox.to_s, hash) end - def update_testbox_time(job_id) - job = @redis.get_job(job_id) - testbox = job["testbox"] - hash = {"time" => @env.get?("time").to_s} - @es.update_tbox(testbox, hash) - end - def report_ssh_port testbox = @env.params.query["tbox_name"] ssh_port = @env.params.query["ssh_port"].to_s diff --git a/src/scheduler/update_job_parameter.cr b/src/scheduler/update_job_parameter.cr index 92b905e..28ab0f2 100644 --- a/src/scheduler/update_job_parameter.cr +++ b/src/scheduler/update_job_parameter.cr @@ -27,7 +27,7 @@ class Sched job_content[parameter] = value end - @redis.update_job(job_content) + update_id2job(job_content) # json log log = job_content.dup @@ -46,4 +46,11 @@ class Sched } spawn mq_publish_confirm(JOB_MQ, mq_msg.to_json) end + + def update_testbox_time(job_id) + job = get_id2job(job_id) + testbox = job["testbox"] + hash = {"time" => @env.get?("time").to_s} + @es.update_tbox(testbox, hash) + end end -- 2.23.0

1 0

[PATCH compass-ci 5/7] scheduler: move job from in_process to extract_stats in etcd
by Cao Xueliang 29 Mar '21

29 Mar '21

Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/scheduler/close_job.cr | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/src/scheduler/close_job.cr b/src/scheduler/close_job.cr index a57b233..746f70a 100644 --- a/src/scheduler/close_job.cr +++ b/src/scheduler/close_job.cr @@ -8,7 +8,7 @@ class Sched @env.set "job_id", job_id - job = @redis.get_job(job_id) + job = get_id2job(job_id) # update job_state job_state = @env.params.query["job_state"]? @@ -21,17 +21,8 @@ class Sched raise "es set job content fail!" end - subqueue = job.subqueue - queue = (subqueue == "idle" ? job.queue : "#{job.queue}/#{subqueue}") - - response = @task_queue.hand_over_task( - "sched/#{queue}", "extract_stats", job_id - ) - if response[0] != 201 - raise "#{response}" - end - - @redis.remove_finished_job(job_id) + move_process2stats(job) + delete_id2job(job.id) job_state ||= "complete" @log.info(%({"job_id": "#{job_id}", "job_state": "#{job_state}"})) -- 2.23.0

1 0

[PATCH compass-ci 4/7] scheduler: use etcd list-watch to consume job
by Cao Xueliang 29 Mar '21

29 Mar '21

Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/scheduler/find_job_boot.cr | 113 +++++++++++++++++++++++++++------ 1 file changed, 92 insertions(+), 21 deletions(-) diff --git a/src/scheduler/find_job_boot.cr b/src/scheduler/find_job_boot.cr index 78b10ad..2596f83 100644 --- a/src/scheduler/find_job_boot.cr +++ b/src/scheduler/find_job_boot.cr @@ -69,42 +69,113 @@ class Sched default_queues = [] of String queues_str.split(',', remove_empty: true) do |item| - default_queues << item.strip + default_queues << "sched/#{item.strip}/ready" end - sub_queues = [] of String - default_queues.each do |queue| - # TODO: this could be high cost and should be improved in future : keys(pattern). - matched_queues = @redis.keys("#{QUEUE_NAME_BASE}/sched/#{queue}/*/ready") - next if matched_queues.empty? + return default_queues.uniq + end - matched_queues.each do |mq| - match_data = "#{mq}".match(%r(^#{QUEUE_NAME_BASE}/sched/(#{queue}/.+)/ready$)) - sub_queues << match_data[1] if match_data + def get_job_from_queues(queues, testbox) + job = nil + etcd_job = consume_job(queues) + job_id = etcd_job.key.split("/")[-1] + puts "#{testbox} got the job #{job_id}" + if job_id + begin + job = @es.get_job(job_id.to_s) + rescue ex + @log.warn("Invalid job (id=#{job_id}) in es. Info: #{ex}") + @log.warn(ex.inspect_with_backtrace) end end - idle_queues = [] of String - delimiter_queues = [] of String - default_queues.each do |queue| - idle_queues << "#{queue}/idle" - delimiter_queues << "#{queue}/delimiter@localhost" + if job + job.update({"testbox" => testbox}) + job.set_result_root + @log.info(%({"job_id": "#{job_id}", "result_root": "/srv#{job.result_root}", "job_state" +: "set result root"})) + set_id2job(job) end - all_queues = delimiter_queues + rand_queues(sub_queues) + idle_queues + return job + end + + def consume_job(queues) + job, revision = consume_by_list(queues) + return job if job - return all_queues.uniq + consume_by_watch(queues, revision) end - def get_job_from_queues(queues, testbox) - job = nil + def consume_by_list(queues) + jobs, revision = get_history_jobs(queues) + jobs.shuffle! + while true + return nil, revision if jobs.empty? + + job = jobs.delete_at(0) + return job, revision if ready2process(job) + end + end + def get_history_jobs(queues) + revisions = [] of Int64 + ec = EtcdClient.new + jobs = [] of Etcd::Model::Kv queues.each do |queue| - job = prepare_job("sched/#{queue}", testbox) - return job if job + job = ec.range_prefix(queue) + revisions << job.header.not_nil!.revision + jobs += job.kvs end - return job + return jobs, revisions.min + end + + def consume_by_watch(queues, revision) + ready_queues = split_ready_queues(queues) + channel = watch_queues(ready_queues, revision) + loop_handle_event(channel) + end + + def split_ready_queues(queues) + ready_queues = [] of String + queues.each do |queue| + tmp = queue.split("/ready")[0] + ready_queues << "#{tmp}/ready" + end + + ready_queues.uniq + end + + def watch_queues(queues, revision) + puts "watch #{queues}, revision #{revision}" + channel = Channel(Array(Etcd::Model::WatchEvent)).new + queues.each do |queue| + watcher = EtcdClient.new.watch_prefix(queue, start_revision: revision.to_i64, progress_notify: false ,filters: [Etcd::Watch::Filter::NODELETE]) do |events| + channel.send(events) + end + spawn { watcher.start } + Fiber.yield + end + + return channel + end + + def loop_handle_event(channel) + while true + events = channel.receive + events.each do |event| + return event.kv if ready2process(event.kv) + end + end + end + + def ready2process(job) + ec = EtcdClient.new + f_queue = job.key + t_queue = f_queue.gsub("/ready/", "/in_process/") + value = job.value + ec.move(f_queue, t_queue, value) end def get_job_boot(host, boot_type) -- 2.23.0

1 0

[PATCH compass-ci 3/7] scheduler: submit job to etcd db
by Cao Xueliang 29 Mar '21

29 Mar '21

before, we add job to taskqueue, then we save job to es, after, we save job to es at first, then we add job to etcd. Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/scheduler/submit_job.cr | 64 +++++++++++++++++++------------------ 1 file changed, 33 insertions(+), 31 deletions(-) diff --git a/src/scheduler/submit_job.cr b/src/scheduler/submit_job.cr index fe9efa1..c639ad7 100644 --- a/src/scheduler/submit_job.cr +++ b/src/scheduler/submit_job.cr @@ -58,18 +58,12 @@ class Sched next if (config["roles"].as_a.map(&.to_s) & roles).empty? queue = host.to_s - job_id = add_task("#{queue}/#{subqueue}", lab) + job_id = get_job_id(lab) # return when job_id is '0' # 2 Questions: # - how to deal with the jobs added to DB prior to this loop # - may consume job before all jobs done - return job_messages << { - "job_id" => "0", - "message" => "add task queue sched/#{queue} failed", - "job_state" => "submit", - } unless job_id - job_ids << job_id # add to job content when multi-test @@ -87,15 +81,15 @@ class Sched job["direct_macs"] = direct_macs.join(" ") job["direct_ips"] = direct_ips.join(" ") - response = add_job(job, job_id) - message = (response["error"]? ? response["error"]["root_cause"] : "") + status, msg = add_job(job, job_id) + job_id = "0" unless status job_messages << { "job_id" => job_id, - "message" => message.to_s, + "message" => msg, "job_state" => "submit", "result_root" => "/srv#{job.result_root}", } - return job_messages if response["error"]? + return job_messages unless status end cluster_id = job_ids[0] @@ -117,31 +111,31 @@ class Sched # success: [{"job_id" => job_id, "message" => "", job_state => "submit"}] # failure: [{"job_id" => "0", "message" => err_msg, job_state => "submit"}] def submit_single_job(job) - queue = "#{job.queue}/#{job.subqueue}" + job_id = get_job_id(job.lab) - job_id = add_task(queue, job.lab) + status, msg = add_job(job, job_id) + job_id = "0" unless status return [{ - "job_id" => "0", - "message" => "add task queue sched/#{queue} failed", + "job_id" => job_id, + "message" => msg, "job_state" => "submit", - }] unless job_id - - response = add_job(job, job_id) - message = (response["error"]? ? response["error"]["root_cause"] : "") - - return [{ - "job_id" => job_id, - "message" => message.to_s, - "job_state" => "submit", "result_root" => "/srv#{job.result_root}", }] end - # return job_id - def add_task(queue, lab) - task_desc = JSON.parse(%({"domain": "compass-ci", "lab": "#{lab}"})) - response = @task_queue.add_task("sched/#{queue}", task_desc) - JSON.parse(response[1].to_json)["id"].to_s if response[0] == 200 + def add_task(job, job_id) + task_desc = JSON.parse(%({"domain": "compass-ci", "id": "#{job_id}"})) + key = "sched/#{job.queue}/ready/#{job.subqueue}/#{job_id}" + response = @etcd.put(key, task_desc) + if response + return true, "" + else + return false, "add the job to queue failed: job_id #{job_id}, queue #{key}" + end + end + + def get_job_id(lab) + "#{lab}.#{(a)redis.get_new_seqno()}" end def save_secrets(job, job_id) @@ -150,11 +144,19 @@ class Sched @redis.hash_set("id2secrets", job_id, job["secrets"]?.to_json) job.delete("secrets") end - # add job content to es and return a response + def add_job(job, job_id) + # deal the job fields at first save_secrets(job, job_id) job.update_id(job_id) - @es.set_job_content(job) + + # save the job to es + response = @es.set_job_content(job) + msg = (response["error"]? ? response["error"]["root_cause"] : "") + return false, msg.to_s if response["error"]? + + # set job_id to ready queue + add_task(job, job_id) end def get_cluster_config(cluster_file, lkp_initrd_user, os_arch) -- 2.23.0

1 0

[PATCH compass-ci 2/7] extract-stats: use etcd list-watch to handle extract-stats queue
by Cao Xueliang 29 Mar '21

29 Mar '21

before, we use taskqueue to handle extract-stats queue, after, we use etcd list-watch to handle extract-stats queue. Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/extract-stats.cr | 2 +- src/extract-stats/extract_stats.cr | 67 +++++++++++++++++++------ src/extract-stats/stats_worker.cr | 78 +++++++++--------------------- 3 files changed, 75 insertions(+), 72 deletions(-) diff --git a/src/extract-stats.cr b/src/extract-stats.cr index 2f4b97d..0908a91 100644 --- a/src/extract-stats.cr +++ b/src/extract-stats.cr @@ -5,4 +5,4 @@ require "./extract-stats/extract_stats" # results data post processing -ExtractStats.in_extract_stats +ExtractStats.consume_tasks diff --git a/src/extract-stats/extract_stats.cr b/src/extract-stats/extract_stats.cr index 45310ad..4ae3e71 100644 --- a/src/extract-stats/extract_stats.cr +++ b/src/extract-stats/extract_stats.cr @@ -1,30 +1,67 @@ # SPDX-License-Identifier: MulanPSL-2.0+ # Copyright (c) 2020 Huawei Technologies Co., Ltd. All rights reserved. +require "../lib/etcd_client" require "./constants" require "./stats_worker" module ExtractStats - # Consume scheduler queue - def self.in_extract_stats - back_fill_task - STATS_WORKER_COUNT.times do - Process.fork do - consume_task + def self.consume_tasks + channel = Channel(String).new + revision = self.consume_by_list(EXTRACT_STATS_QUEUE_PATH, channel) + self.consume_by_watch(EXTRACT_STATS_QUEUE_PATH, revision, channel) + end + + def self.consume_by_list(queue, channel) + tasks, revision = self.get_history_tasks(queue) + self.handle_history_tasks(tasks, channel) + + return revision + end + + def self.consume_by_watch(queue, revision, channel) + self.watch_queue(queue, revision, channel) + self.handle_events(channel) + end + + def self.get_history_tasks(queue) + tasks = [] of Etcd::Model::Kv + range = EtcdClient.new.range_prefix(queue) + revision = range.header.not_nil!.revision + tasks += range.kvs + + return tasks, revision + end + + def self.handle_history_tasks(tasks, channel) + while true + return if tasks.empty? + + task = tasks.delete_at(0) + spawn { StatsWorker.new.handle(task.key, channel) } + + sleep 1 + end + end + + def self.watch_queue(queue, revision, channel) + watcher = EtcdClient.new.watch_prefix(queue, start_revision: revision.to_i64, filters: [Etcd::Watch::Filter::NODELETE]) do |events| + events.each do |event| + puts event + channel.send(event.kv.key) end end - # keep main-process alive - sleep + spawn { watcher.start } + Fiber.yield end - def self.consume_task - worker = StatsWorker.new - worker.consume_sched_queue(EXTRACT_STATS_QUEUE_PATH) - end + def self.handle_events(channel) + while true + key = channel.receive + spawn { StatsWorker.new.handle(key, channel) } - def self.back_fill_task - worker = StatsWorker.new - worker.back_fill_task(EXTRACT_STATS_QUEUE_PATH) + sleep 1 + end end end diff --git a/src/extract-stats/stats_worker.cr b/src/extract-stats/stats_worker.cr index 44c327d..5e0100e 100644 --- a/src/extract-stats/stats_worker.cr +++ b/src/extract-stats/stats_worker.cr @@ -1,7 +1,7 @@ # SPDX-License-Identifier: MulanPSL-2.0+ # Copyright (c) 2020 Huawei Technologies Co., Ltd. All rights reserved. -require "../lib/taskqueue_api" +require "../lib/etcd_client" require "../scheduler/elasticsearch_client" require "../scheduler/redis_client" require "../scheduler/constants" @@ -11,40 +11,25 @@ require "./constants.cr" class StatsWorker def initialize @es = Elasticsearch::Client.new - @tq = TaskQueueAPI.new + @etcd = EtcdClient.new @rc = RegressionClient.new end - def consume_sched_queue(queue_path : String) - loop do - begin - job_id = get_job_id(queue_path) - if job_id # will consume the job by post-processing - job = @es.get_job_content(job_id) - result_root = job["result_root"]? - result_post_processing(job_id, result_root.to_s, queue_path) - @tq.delete_task(queue_path + "/in_process", "#{job_id}") - end - rescue e - STDERR.puts e.message - error_message = e.message - - # incase of many error message when task-queue, ES does not work - sleep(10) - @tq = TaskQueueAPI.new if error_message && error_message.includes?("3060': Connection refused") - end - end - end - - # get job_id from task-queue - def get_job_id(queue_path : String) - response = @tq.consume_task(queue_path) - if response[0] == 200 - JSON.parse(response[1].to_json)["id"].to_s - else - # will sleep 60s if no task in task-queue - sleep(60) - nil + def handle(queue_path, channel) + begin + res = @etcd.range(queue_path) + return nil if res.count == 0 + + job_id = queue_path.split("/")[-1] + job = @es.get_job_content(job_id) + result_root = job["result_root"]? + result_post_processing(job_id, result_root.to_s, queue_path) + @etcd.delete(queue_path) + rescue e + channel.send(queue_path) + STDERR.puts e.message + # incase of many error message when task-queue, ES does not work + sleep(10) end end @@ -61,10 +46,7 @@ class StatsWorker def store_stats_es(result_root : String, job_id : String, queue_path : String) stats_path = "#{result_root}/stats.json" - unless File.exists?(stats_path) - @tq.delete_task(queue_path + "/in_process", "#{job_id}") - raise "#{stats_path} file not exists." - end + return unless File.exists?(stats_path) stats = File.open(stats_path) do |file| JSON.parse(file) @@ -81,7 +63,6 @@ class StatsWorker { :index => "jobs", :type => "_doc", :id => job_id, - :refresh => "wait_for", :body => {:doc => update_content}, } ) @@ -90,9 +71,10 @@ class StatsWorker unless new_error_ids.empty? sample_error_id = new_error_ids.sample STDOUT.puts "send a delimiter task: job_id is #{job_id}" - @tq.add_task(DELIMITER_TASK_QUEUE, JSON.parse({"error_id" => sample_error_id, - "job_id" => job_id, - "lab" => LAB}.to_json)) + queue = "#{DELIMITER_TASK_QUEUE}/#{job_id}" + value = {"error_id" => sample_error_id} + @etcd.put(queue, value) + msg = %({"job_id": "#{job_id}", "new_error_id": "#{sample_error_id}"}) system "echo '#{msg}'" end @@ -130,20 +112,4 @@ class StatsWorker end error_ids end - - def back_fill_task(queue_path) - redis_client = Redis::Client.new - # this queue may have leftover task_ids - queue_name = "queues/#{queue_path}/in_process" - begin - job_ids = redis_client.(a)client.zrange(queue_name, 0, -1) - return if job_ids.empty? - - job_ids.each do |job_id| - @tq.hand_over_task(queue_path, queue_path, job_id.to_s) - end - rescue e - STDERR.puts e.message - end - end end -- 2.23.0

1 0

[PATCH compass-ci 1/7] src/scheduler: add opt job in etcd
by Cao Xueliang 29 Mar '21

29 Mar '21

Before, we operate the job in redis. After, we operate the job in etcd. Signed-off-by: Cao Xueliang <caoxl78320(a)163.com> --- src/lib/sched.cr | 3 +++ src/scheduler/opt_job_in_etcd.cr | 36 ++++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+) create mode 100644 src/scheduler/opt_job_in_etcd.cr diff --git a/src/lib/sched.cr b/src/lib/sched.cr index b10a16d..f74df9b 100644 --- a/src/lib/sched.cr +++ b/src/lib/sched.cr @@ -7,6 +7,7 @@ require "yaml" require "./mq" require "./job" require "./web_env" +require "./etcd_client" require "./block_helper" require "./taskqueue_api" require "./remote_git_client" @@ -23,6 +24,7 @@ require "../scheduler/request_cluster_state" require "../scheduler/update_job_parameter" require "../scheduler/create_job_cpio" require "../scheduler/download_file" +require "../scheduler/opt_job_in_etcd" class Sched property es @@ -37,6 +39,7 @@ class Sched @redis = Redis::Client.instance @mq = MQClient.instance @task_queue = TaskQueueAPI.new + @etcd = EtcdClient.new @rgc = RemoteGitClient.new @env = env @log = env.log.as(JSONLogger) diff --git a/src/scheduler/opt_job_in_etcd.cr b/src/scheduler/opt_job_in_etcd.cr new file mode 100644 index 0000000..f38433a --- /dev/null +++ b/src/scheduler/opt_job_in_etcd.cr @@ -0,0 +1,36 @@ +# SPDX-License-Identifier: MulanPSL-2.0+ +# Copyright (c) 2020 Huawei Technologies Co., Ltd. All rights reserved. + +class Sched + def update_id2job(job_content) + id = job_content["id"].to_s + job = get_id2job(id) + job.update(job_content) + @etcd.update("sched/id2job/#{id}", job.dump_to_json) + end + + def set_id2job(job : Job) + @etcd.put("sched/id2job/#{job.id}", job.dump_to_json) + end + + def get_id2job(id) + response = @etcd.range("sched/id2job/#{id}") + raise "get job from etcd failed. id: #{id}" unless response.count == 1 + Job.new(JSON.parse(response.kvs[0].value.not_nil!), id) + end + + def delete_id2job(id) + @etcd.delete("sched/id2job/#{id}") + end + + def update_tbox_wtmp(testbox, wtmp_hash) + @etcd.update("sched/tbox_wtmp/#{testbox}", wtmp_hash) + end + + def move_process2stats(job : Job) + f_queue = "sched/#{job.queue}/in_process/#{job.subqueue}/#{job.id}" + t_queue = "extract_stats/#{job.id}" + value = { "id" => "#{job.id}" } + @etcd.move(f_queue, t_queue, value) + end +end -- 2.23.0

1 0