- Compass-ci - mailweb.openeuler.org

[PATCH compass-ci 1/5] add update-repo service
by Li Ping 29 Mar '21

29 Mar '21

Signed-off-by: Li Ping <1477412247(a)qq.com> --- src/lib/web_env.cr | 9 +++++++++ src/updaterepo.cr | 17 +++++++++++++++++ src/updaterepo/constants.cr | 4 ++++ src/updaterepo/updaterepo.cr | 16 ++++++++++++++++ 4 files changed, 46 insertions(+) create mode 100644 src/updaterepo.cr create mode 100644 src/updaterepo/constants.cr create mode 100644 src/updaterepo/updaterepo.cr diff --git a/src/lib/web_env.cr b/src/lib/web_env.cr index b3a90a8..d47e7e5 100644 --- a/src/lib/web_env.cr +++ b/src/lib/web_env.cr @@ -4,6 +4,7 @@ require "./sched" require "./lifecycle" require "./json_logger" +require "./updaterepo" class HTTP::Server # Instances of this class are passed to an `HTTP::Server` handler. @@ -31,5 +32,13 @@ class HTTP::Server def create_lifecycle @lifecycle = Lifecycle.new(self) end + + def repo + @repo ||= create_repo + end + + def create_repo + @repo = Repo.new(self) + end end end diff --git a/src/updaterepo.cr b/src/updaterepo.cr new file mode 100644 index 0000000..05fa4fe --- /dev/null +++ b/src/updaterepo.cr @@ -0,0 +1,17 @@ +# SPDX-License-Identifier: MulanPSL-2.0+ +# Copyright (c) 2020 Huawei Technologies Co., Ltd. All rights reserved. + +require "./updaterepo/updaterepo" +require "./updaterepo/constants" +require "./lib/json_logger" + + +module Updaterepo + log = JSONLogger.new + + begin + Kemal.run(REPO_PORT) + rescue e + log.error(e) + end +end diff --git a/src/updaterepo/constants.cr b/src/updaterepo/constants.cr new file mode 100644 index 0000000..c9daaf2 --- /dev/null +++ b/src/updaterepo/constants.cr @@ -0,0 +1,4 @@ +# SPDX-License-Identifier: MulanPSL-2.0+ +# Copyright (c) 2020 Huawei Technologies Co., Ltd. All rights reserved. + +REPO_PORT = (ENV.has_key?("REPO_PORT") ? ENV["REPO_PORT"] : 8132).to_i32 diff --git a/src/updaterepo/updaterepo.cr b/src/updaterepo/updaterepo.cr new file mode 100644 index 0000000..7ad9d20 --- /dev/null +++ b/src/updaterepo/updaterepo.cr @@ -0,0 +1,16 @@ +# SPDX-License-Identifier: MulanPSL-2.0+ +# Copyright (c) 2020 Huawei Technologies Co., Ltd. All rights reserved. + +require "kemal" + +require "../lib/web_env" +require "../lib/updaterepo" +require "../lib/json_logger" + +module Updaterepo + VERSION = "0.1.0" + post "/upload" do |env| + env.repo.upload_repo + "done" + end +end -- 2.23.0

1 0

[PATCH v2 compass-ci 3/3] sparrow/3-code/dev-env: add key EMAIL_LIMIT_COUNT
by Luan Shengde 29 Mar '21

29 Mar '21

Signed-off-by: Luan Shengde <shdluan(a)163.com> --- sparrow/3-code/dev-env | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sparrow/3-code/dev-env b/sparrow/3-code/dev-env index f96972e..3123c2a 100755 --- a/sparrow/3-code/dev-env +++ b/sparrow/3-code/dev-env @@ -12,6 +12,7 @@ server_name=$(hostname | cut -f1 -d.) : ${OS_HTTP_HOST:=$server_ip} : ${SEND_MAIL_HOST:=$server_ip} : ${SEND_MAIL_PORT:=10001} +: ${EMAIL_LIMIT_COUNT:=10} : ${SRV_HTTP_RESULT_HOST:=$server_ip} : ${SRV_HTTP_OS_HOST:=$server_ip} : ${SRV_HTTP_GIT_HOST:=$server_ip} @@ -91,6 +92,7 @@ SRV_HTTP_CCI_PORT: $SRV_HTTP_CCI_PORT TASKQUEUE_HOST: $sched_host SEND_MAIL_HOST: $SEND_MAIL_HOST SEND_MAIL_PORT: $SEND_MAIL_PORT +EMAIL_LIMIT_COUNT: ${EMAIL_LIMIT_COUNT} LOGGING_ES_HOST: $LOGGING_ES_HOST LOGGING_ES_PORT: $LOGGING_ES_PORT INITRD_HTTP_HOST: $INITRD_HTTP_HOST -- 2.23.0

1 0

[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

2024

2023

2022

2021

2020

Compass-ci