REFACTOR: Simplify converter steps in migration tooling (#29779)

* Remove unused `report_progress_in_percent` option from step
* Remove `use_custom_progress_increment` option from the step because we can figure it out by looking at the progress
* Introduce `StepTracker` to for logging warnings and errors and tracking step progress
* Make it easier to log warnings and errors in all methods of `Step` without the need to pass around a `stats` object
This commit is contained in:
Gerhard Schlager 2024-11-19 23:54:37 +01:00 committed by GitHub
parent a48af2f120
commit 5ac69076c1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 153 additions and 161 deletions

View File

@ -9,7 +9,6 @@ en:
estimated: "ETA: %{duration}"
elapsed: "Time: %{duration}"
processed:
percentage: "Processed: %{percentage}"
progress: "Processed: %{current}"
progress_with_max: "Processed: %{current} / %{max}"

View File

@ -4,14 +4,8 @@ require "ruby-progressbar"
module Migrations
class ExtendedProgressBar
def initialize(
max_progress: nil,
report_progress_in_percent: false,
use_custom_progress_increment: false
)
def initialize(max_progress: nil)
@max_progress = max_progress
@report_progress_in_percent = report_progress_in_percent
@use_custom_progress_increment = use_custom_progress_increment
@warning_count = 0
@error_count = 0
@ -31,16 +25,16 @@ module Migrations
nil
end
def update(stats)
def update(progress, warning_count, error_count)
extra_information_changed = false
if stats.warning_count > 0
@warning_count += stats.warning_count
if warning_count > 0
@warning_count += warning_count
extra_information_changed = true
end
if stats.error_count > 0
@error_count += stats.error_count
if error_count > 0
@error_count += error_count
extra_information_changed = true
end
@ -59,10 +53,10 @@ module Migrations
@progressbar.format = "#{@base_format}#{@extra_information}"
end
if @use_custom_progress_increment
@progressbar.progress += stats.progress
else
if progress == 1
@progressbar.increment
else
@progressbar.progress += progress
end
end
@ -70,9 +64,7 @@ module Migrations
def setup_progressbar
format =
if @report_progress_in_percent
I18n.t("progressbar.processed.percentage", percentage: "%J%")
elsif @max_progress
if @max_progress
I18n.t("progressbar.processed.progress_with_max", current: "%c", max: "%C")
else
I18n.t("progressbar.processed.progress", current: "%c")

View File

@ -73,7 +73,7 @@ module Migrations::Converters::Base
default_args = { settings: settings }
args = default_args.merge(step_args(step_class))
step_class.new(args)
step_class.new(StepTracker.new, args)
end
end
end

View File

@ -4,7 +4,7 @@ module Migrations::Converters::Base
class ParallelJob
def initialize(step)
@step = step
@stats = ProgressStats.new
@tracker = step.tracker
@offline_connection = ::Migrations::Database::OfflineConnection.new
@ -14,16 +14,16 @@ module Migrations::Converters::Base
end
def run(item)
@stats.reset!
@tracker.reset_stats!
@offline_connection.clear!
begin
@step.process_item(item, @stats)
@step.process_item(item)
rescue StandardError => e
@stats.log_error("Failed to process item", exception: e, details: item)
@tracker.log_error("Failed to process item", exception: e, details: item)
end
[@offline_connection.parametrized_insert_statements, @stats]
[@offline_connection.parametrized_insert_statements, @tracker.stats]
end
def cleanup

View File

@ -10,7 +10,7 @@ module Migrations::Converters::Base
raise NotImplementedError
end
def process_item(item, stats)
def process_item(item)
raise NotImplementedError
end
@ -22,22 +22,6 @@ module Migrations::Converters::Base
def run_in_parallel?
@run_in_parallel == true
end
def report_progress_in_percent(value)
@report_progress_in_percent = !!value
end
def report_progress_in_percent?
@report_progress_in_percent == true
end
def use_custom_progress_increment(value)
@use_custom_progress_increment = !!value
end
def use_custom_progress_increment?
@use_custom_progress_increment == true
end
end
end
end

View File

@ -39,7 +39,7 @@ module Migrations::Converters::Base
with_progressbar do |progressbar|
@step.items.each do |item|
stats = job.run(item)
progressbar.update(stats)
progressbar.update(stats.progress, stats.warning_count, stats.error_count)
end
end
end
@ -76,11 +76,7 @@ module Migrations::Converters::Base
def with_progressbar
::Migrations::ExtendedProgressBar
.new(
max_progress: @max_progress,
report_progress_in_percent: @step.class.report_progress_in_percent?,
use_custom_progress_increment: @step.class.use_custom_progress_increment?,
)
.new(max_progress: @max_progress)
.run { |progressbar| yield progressbar }
end
@ -94,7 +90,7 @@ module Migrations::Converters::Base
::Migrations::Database::IntermediateDB.insert(sql, *parameters)
end
progressbar.update(stats)
progressbar.update(stats.progress, stats.warning_count, stats.error_count)
end
end
end

View File

@ -4,19 +4,19 @@ module Migrations::Converters::Base
class SerialJob
def initialize(step)
@step = step
@stats = ProgressStats.new
@tracker = step.tracker
end
def run(item)
@stats.reset!
@tracker.reset_stats!
begin
@step.process_item(item, @stats)
@step.process_item(item)
rescue StandardError => e
@stats.log_error("Failed to process item", exception: e, details: item)
@tracker.log_error("Failed to process item", exception: e, details: item)
end
@stats
@tracker.stats
end
def cleanup

View File

@ -5,9 +5,18 @@ module Migrations::Converters::Base
IntermediateDB = ::Migrations::Database::IntermediateDB
attr_accessor :settings
attr_reader :tracker
def initialize(args = {})
args.each { |arg, value| instance_variable_set("@#{arg}", value) if respond_to?(arg, true) }
# inside of Step it might make more sense to access it as `step` instead of `tracker`
alias step tracker
def initialize(tracker, args = {})
@tracker = tracker
args.each do |arg, value|
setter = "#{arg}=".to_sym
public_send(setter, value) if respond_to?(setter, true)
end
end
def execute

View File

@ -0,0 +1,5 @@
# frozen_string_literal: true
module Migrations::Converters::Base
StepStats = Struct.new(:progress, :warning_count, :error_count)
end

View File

@ -1,17 +1,22 @@
# frozen_string_literal: true
module Migrations::Converters::Base
class ProgressStats
attr_accessor :progress, :warning_count, :error_count
class StepTracker
attr_reader :stats
def initialize
reset!
@stats = StepStats.new
reset_stats!
end
def reset!
@progress = 1
@warning_count = 0
@error_count = 0
def reset_stats!
@stats.progress = 1
@stats.warning_count = 0
@stats.error_count = 0
end
def progress=(value)
@stats.progress = value
end
def log_info(message, details: nil)
@ -19,20 +24,15 @@ module Migrations::Converters::Base
end
def log_warning(message, exception: nil, details: nil)
@warning_count += 1
@stats.warning_count += 1
log(::Migrations::Database::IntermediateDB::LogEntry::WARNING, message, exception:, details:)
end
def log_error(message, exception: nil, details: nil)
@error_count += 1
@stats.error_count += 1
log(::Migrations::Database::IntermediateDB::LogEntry::ERROR, message, exception:, details:)
end
def ==(other)
other.is_a?(ProgressStats) && progress == other.progress &&
warning_count == other.warning_count && error_count == other.error_count
end
private
def log(type, message, exception: nil, details: nil)

View File

@ -4,14 +4,7 @@ require "oj"
module Migrations::Converters::Base
class Worker
OJ_SETTINGS = {
mode: :custom,
create_id: "^o",
create_additions: true,
cache_keys: true,
class_cache: true,
symbol_keys: true,
}
OJ_SETTINGS = { mode: :object, class_cache: true, symbol_keys: true }
def initialize(index, input_queue, output_queue, job)
@index = index

View File

@ -8,11 +8,11 @@ module Migrations::Converters::Example
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
end
def process_item(item, stats)
def process_item(item)
sleep(0.5)
stats.warning_count += 1 if item.in?([3, 7, 9])
stats.error_count += 1 if item.in?([6, 10])
step.log_warning("Test", details: item) if item.in?([3, 7, 9])
step.log_error("Test", details: item) if item.in?([6, 10])
IntermediateDB::LogEntry.create!(type: "info", message: "Step2 - #{item}")
end

View File

@ -12,9 +12,11 @@ module Migrations::Converters::Example
(1..1000).map { |i| { counter: i } }
end
def process_item(item, stats)
def process_item(item)
sleep(0.5)
step.log_warning("Test", details: item) if item[:counter] > 10 && item[:counter] < 20
IntermediateDB::LogEntry.create!(type: "info", message: "Step3 - #{item[:counter]}")
end
end

View File

@ -5,14 +5,17 @@ RSpec.describe ::Migrations::Converters::Base::ParallelJob do
let(:step) { instance_double(::Migrations::Converters::Base::ProgressStep) }
let(:item) { { key: "value" } }
let(:stats) { instance_double(::Migrations::Converters::Base::ProgressStats) }
let(:tracker) { instance_double(::Migrations::Converters::Base::StepTracker) }
let(:stats) { ::Migrations::Converters::Base::StepStats.new }
let(:intermediate_db) { class_double(::Migrations::Database::IntermediateDB).as_stubbed_const }
before do
allow(::Migrations::Converters::Base::ProgressStats).to receive(:new).and_return(stats)
allow(step).to receive(:tracker).and_return(tracker)
allow(tracker).to receive(:reset_stats!)
allow(tracker).to receive(:log_error)
allow(tracker).to receive(:stats).and_return(stats)
allow(stats).to receive(:reset!)
allow(stats).to receive(:log_error)
allow(intermediate_db).to receive(:setup)
allow(intermediate_db).to receive(:close)
end
@ -52,7 +55,7 @@ RSpec.describe ::Migrations::Converters::Base::ParallelJob do
it "resets stats and clears the offline connection" do
job.run(item)
expect(stats).to have_received(:reset!)
expect(tracker).to have_received(:reset_stats!)
expect(offline_connection).to have_received(:clear!)
end
@ -61,7 +64,7 @@ RSpec.describe ::Migrations::Converters::Base::ParallelJob do
job.run(item)
expect(stats).to have_received(:log_error).with(
expect(tracker).to have_received(:log_error).with(
"Failed to process item",
exception: an_instance_of(StandardError),
details: item,

View File

@ -5,20 +5,26 @@ RSpec.describe ::Migrations::Converters::Base::SerialJob do
let(:step) { instance_double(::Migrations::Converters::Base::ProgressStep) }
let(:item) { "Item" }
let(:stats) do
instance_double(::Migrations::Converters::Base::ProgressStats, reset!: nil, log_error: nil)
end
let(:tracker) { instance_double(::Migrations::Converters::Base::StepTracker) }
let(:stats) { ::Migrations::Converters::Base::StepStats.new }
before { allow(::Migrations::Converters::Base::ProgressStats).to receive(:new).and_return(stats) }
before do
allow(step).to receive(:tracker).and_return(tracker)
allow(tracker).to receive(:reset_stats!)
allow(tracker).to receive(:log_error)
allow(tracker).to receive(:stats).and_return(stats)
end
describe "#run" do
it "resets stats and processes item" do
allow(step).to receive(:process_item).and_return(stats)
job.run(item)
result = job.run(item)
expect(result).to eq(stats)
expect(stats).to have_received(:reset!)
expect(step).to have_received(:process_item).with(item, stats)
expect(tracker).to have_received(:reset_stats!)
expect(step).to have_received(:process_item).with(item)
end
it "logs error if processing item raises an exception" do
@ -26,7 +32,7 @@ RSpec.describe ::Migrations::Converters::Base::SerialJob do
job.run(item)
expect(stats).to have_received(:log_error).with(
expect(tracker).to have_received(:log_error).with(
"Failed to process item",
exception: an_instance_of(StandardError),
details: item,

View File

@ -1,6 +1,8 @@
# frozen_string_literal: true
RSpec.describe ::Migrations::Converters::Base::Step do
let(:tracker) { instance_double(::Migrations::Converters::Base::StepTracker) }
before do
Object.const_set(
"TemporaryModule",
@ -32,13 +34,13 @@ RSpec.describe ::Migrations::Converters::Base::Step do
describe "#initialize" do
it "works when no arguments are supplied" do
step = nil
expect { step = TemporaryModule::Users.new }.not_to raise_error
expect { step = TemporaryModule::Users.new(tracker) }.not_to raise_error
expect(step.settings).to be_nil
end
it "initializes the `settings` attribute if given" do
settings = { a: 1, b: 2 }
step = TemporaryModule::Users.new(settings:)
step = TemporaryModule::Users.new(tracker, settings:)
expect(step.settings).to eq(settings)
end
@ -49,7 +51,7 @@ RSpec.describe ::Migrations::Converters::Base::Step do
foo = "a string"
bar = false
step = TemporaryModule::Users.new(settings:, foo:, bar:, non_existent: 123)
step = TemporaryModule::Users.new(tracker, settings:, foo:, bar:, non_existent: 123)
expect(step.settings).to eq(settings)
expect(step.foo).to eq(foo)
expect(step.bar).to eq(bar)

View File

@ -1,59 +1,79 @@
# frozen_string_literal: true
RSpec.describe ::Migrations::Converters::Base::ProgressStats do
subject(:stats) { described_class.new }
RSpec.describe ::Migrations::Converters::Base::StepTracker do
subject(:tracker) { described_class.new }
before { allow(::Migrations::Database::IntermediateDB::LogEntry).to receive(:create!) }
describe "#initialize" do
it "starts at the correct values" do
stats = tracker.stats
expect(stats.progress).to eq(1)
expect(stats.warning_count).to eq(0)
expect(stats.error_count).to eq(0)
end
end
describe "attribute accessors" do
it "allows reading and writing for :progress" do
stats.progress = 10
expect(stats.progress).to eq(10)
end
it "allows reading and writing for :warning_count" do
stats.warning_count = 5
expect(stats.warning_count).to eq(5)
end
it "allows reading and writing for :error_count" do
stats.error_count = 3
expect(stats.error_count).to eq(3)
describe "#progress=" do
it "allows setting progress" do
tracker.progress = 10
expect(tracker.stats.progress).to eq(10)
end
end
describe "#reset!" do
before do
stats.progress = 5
stats.warning_count = 2
stats.error_count = 3
stats.reset!
end
describe "#stats" do
it "returns correct stats" do
expect(tracker.stats).to eq(
::Migrations::Converters::Base::StepStats.new(
progress: 1,
warning_count: 0,
error_count: 0,
),
)
it "resets progress to 1" do
expect(stats.progress).to eq(1)
end
tracker.progress = 5
2.times { tracker.log_warning("Foo") }
3.times { tracker.log_error("Foo") }
it "resets warning_count to 0" do
expect(stats.warning_count).to eq(0)
expect(tracker.stats).to eq(
::Migrations::Converters::Base::StepStats.new(
progress: 5,
warning_count: 2,
error_count: 3,
),
)
end
end
it "resets error_count to 0" do
expect(stats.error_count).to eq(0)
describe "#reset_stats!" do
it "correctly resets stats" do
tracker.progress = 5
2.times { tracker.log_warning("Foo") }
3.times { tracker.log_error("Foo") }
expect(tracker.stats).to eq(
::Migrations::Converters::Base::StepStats.new(
progress: 5,
warning_count: 2,
error_count: 3,
),
)
tracker.reset_stats!
expect(tracker.stats).to eq(
::Migrations::Converters::Base::StepStats.new(
progress: 1,
warning_count: 0,
error_count: 0,
),
)
end
end
describe "#log_info" do
before { allow(::Migrations::Database::IntermediateDB::LogEntry).to receive(:create!) }
it "logs an info message" do
stats.log_info("Info message")
tracker.log_info("Info message")
expect(::Migrations::Database::IntermediateDB::LogEntry).to have_received(:create!).with(
type: ::Migrations::Database::IntermediateDB::LogEntry::INFO,
@ -64,7 +84,7 @@ RSpec.describe ::Migrations::Converters::Base::ProgressStats do
end
it "logs an info message with details" do
stats.log_info("Info message", details: { key: "value" })
tracker.log_info("Info message", details: { key: "value" })
expect(::Migrations::Database::IntermediateDB::LogEntry).to have_received(:create!).with(
type: ::Migrations::Database::IntermediateDB::LogEntry::INFO,
@ -78,10 +98,10 @@ RSpec.describe ::Migrations::Converters::Base::ProgressStats do
end
describe "#log_warning" do
before { allow(::Migrations::Database::IntermediateDB::LogEntry).to receive(:create!) }
it "logs a warning message and increments warning_count" do
expect { stats.log_warning("Warning message") }.to change { stats.warning_count }.by(1)
expect { tracker.log_warning("Warning message") }.to change {
tracker.stats.warning_count
}.by(1)
expect(::Migrations::Database::IntermediateDB::LogEntry).to have_received(:create!).with(
type: ::Migrations::Database::IntermediateDB::LogEntry::WARNING,
@ -95,8 +115,8 @@ RSpec.describe ::Migrations::Converters::Base::ProgressStats do
exception = StandardError.new("Warning exception")
expect {
stats.log_warning("Warning message", exception: exception, details: { key: "value" })
}.to change { stats.warning_count }.by(1)
tracker.log_warning("Warning message", exception: exception, details: { key: "value" })
}.to change { tracker.stats.warning_count }.by(1)
expect(::Migrations::Database::IntermediateDB::LogEntry).to have_received(:create!).with(
type: ::Migrations::Database::IntermediateDB::LogEntry::WARNING,
@ -110,10 +130,8 @@ RSpec.describe ::Migrations::Converters::Base::ProgressStats do
end
describe "#log_error" do
before { allow(::Migrations::Database::IntermediateDB::LogEntry).to receive(:create!) }
it "logs an error message and increments error_count" do
expect { stats.log_error("Error message") }.to change { stats.error_count }.by(1)
expect { tracker.log_error("Error message") }.to change { tracker.stats.error_count }.by(1)
expect(::Migrations::Database::IntermediateDB::LogEntry).to have_received(:create!).with(
type: ::Migrations::Database::IntermediateDB::LogEntry::ERROR,
@ -127,8 +145,8 @@ RSpec.describe ::Migrations::Converters::Base::ProgressStats do
exception = StandardError.new("Error exception")
expect {
stats.log_error("Error message", exception: exception, details: { key: "value" })
}.to change { stats.error_count }.by(1)
tracker.log_error("Error message", exception: exception, details: { key: "value" })
}.to change { tracker.stats.error_count }.by(1)
expect(::Migrations::Database::IntermediateDB::LogEntry).to have_received(:create!).with(
type: ::Migrations::Database::IntermediateDB::LogEntry::ERROR,
@ -140,17 +158,4 @@ RSpec.describe ::Migrations::Converters::Base::ProgressStats do
)
end
end
describe "#==" do
let(:other_stats) { described_class.new }
it "returns true for objects with the same values" do
expect(stats).to eq(other_stats)
end
it "returns false for objects with different values" do
other_stats.progress = 2
expect(stats).not_to eq(other_stats)
end
end
end

View File

@ -49,11 +49,7 @@ RSpec.describe ::Migrations::Converters::Base::Worker do
end
def create_progress_stats(progress: 1, warning_count: 0, error_count: 0)
stats = ::Migrations::Converters::Base::ProgressStats.new
stats.progress = progress
stats.warning_count = warning_count
stats.error_count = error_count
stats
::Migrations::Converters::Base::StepStats.new(progress:, warning_count:, error_count:)
end
it "writes objects to the `output_queue`" do