109 lines
4.7 KiB
Ruby
Executable File
109 lines
4.7 KiB
Ruby
Executable File
#!/usr/bin/env ruby
|
|
#
|
|
# etl [run] <pipeline.rb> [options] ('run' is optional)
|
|
#
|
|
# executable of the gem datapipeline; in this repository bin/etl loads the bundle first
|
|
#
|
|
# --from=DATE --till=DATE in DataPipeline.params, also passed on to the pipeline script
|
|
# --param key=value in DataPipeline.params[:key], repeatable
|
|
# --dry-run run everything, write nothing, print the first rows
|
|
# --limit=N only the first N source rows
|
|
# --verbose log level debug, --quiet only warnings (default info)
|
|
# --lock=FILE exclusive lock for the whole script, exit 3 when another run holds it
|
|
# --history=FILE one json line per pipeline run (DataPipeline.history)
|
|
# --notify=NAME one report for all runs of the script, notifier NAME from the connections file
|
|
# --notify-on=WHEN failure, problems (default) or always
|
|
# --name=value any other option: DataPipeline.params[:name] and passed on to the script
|
|
#
|
|
# exit code: 0 ok, 1 aborted, 2 finished with rejected rows, 3 locked by another run
|
|
#
|
|
require 'optparse'
|
|
require 'logger'
|
|
require_relative '../lib/datapipeline'
|
|
|
|
options = { settings: { log_level: Logger::INFO }, passthrough: [] }
|
|
DataPipeline.params = {}
|
|
parser = OptionParser.new do |opts|
|
|
opts.banner = "Usage: etl [run] PIPELINE.rb [options]\n\n etl examples/kunden_import.rb --dry-run\n"
|
|
opts.on('--from=DATE', String) { |v| DataPipeline.params[:from] = v; options[:passthrough] << "--from=#{v}" }
|
|
opts.on('--till=DATE', String) { |v| DataPipeline.params[:till] = v; options[:passthrough] << "--till=#{v}" }
|
|
opts.on('--param KEY=VALUE', String) do |v|
|
|
key, value = v.split('=', 2)
|
|
DataPipeline.params[key.to_sym] = value
|
|
end
|
|
opts.on('--dry-run') { options[:settings][:dry_run] = true }
|
|
opts.on('--limit=N', Integer) { |v| options[:settings][:limit] = v }
|
|
opts.on('--verbose') { options[:settings][:log_level] = Logger::DEBUG }
|
|
opts.on('--quiet') { options[:settings][:log_level] = Logger::WARN }
|
|
opts.on('--lock=FILE', String) { |v| options[:lock] = v }
|
|
opts.on('--history=FILE', String) { |v| options[:settings][:history] = v }
|
|
opts.on('--notify=NAME', String) { |v| options[:notify] = v }
|
|
opts.on('--notify-on=WHEN', String) { |v| options[:notify_on] = v }
|
|
end
|
|
|
|
# own options for the parser, everything else (also --name=value of the script) goes to the script
|
|
own = []
|
|
args = []
|
|
argv = ARGV.dup
|
|
while (arg = argv.shift)
|
|
if arg.match?(/\A--(from|till|param|dry-run|limit|verbose|quiet|lock|history|notify|notify-on)(=|\z)/)
|
|
own << arg
|
|
own << argv.shift if %w[--param --limit --from --till --lock --history --notify --notify-on].include?(arg)
|
|
else
|
|
args << arg
|
|
end
|
|
end
|
|
parser.parse(own)
|
|
args.shift if args.first == 'run'
|
|
script = args.shift
|
|
DataPipeline.params.merge!(DataPipeline.parse_params(args)) { |_key, own_value, _| own_value }
|
|
abort parser.help if script.nil?
|
|
abort "etl: #{script} not found\n\n#{parser.help}" unless File.exist?(script)
|
|
|
|
name = File.basename(script, '.rb')
|
|
DataPipeline.settings(options[:settings].merge(name:))
|
|
|
|
# one report for the whole script; created before the run, so a wrong name fails early
|
|
if options[:notify]
|
|
begin
|
|
notifier = DataPipeline::Notifier.from_connection(options[:notify])
|
|
notify_when = DataPipeline::Notifier.condition(options.fetch(:notify_on, :problems))
|
|
rescue ArgumentError, LoadError => e
|
|
abort "etl: --notify=#{options[:notify]}: #{e.message}"
|
|
end
|
|
end
|
|
summary = lambda do |error = nil|
|
|
return unless notifier && !options[:settings][:dry_run]
|
|
|
|
report = DataPipeline::Report.new(DataPipeline.runs, title: name, params: DataPipeline.params, error:)
|
|
notifier.deliver(report) if notify_when.call(report)
|
|
rescue StandardError => e
|
|
warn "#{name}: notify #{options[:notify]}: #{e.class}: #{e.message}"
|
|
end
|
|
|
|
begin
|
|
script_lock = DataPipeline::Lock.new(options[:lock]).acquire if options[:lock] # held until the process ends
|
|
rescue DataPipeline::LockError => e
|
|
warn "#{File.basename(script)}: #{e.message}"
|
|
exit 3
|
|
end
|
|
|
|
# the pipeline script sees only its own arguments
|
|
ARGV.replace(options[:passthrough] + args)
|
|
|
|
begin
|
|
load File.expand_path(script)
|
|
rescue StandardError => e
|
|
warn "#{File.basename(script)}: #{e.class}: #{e.message}"
|
|
# an aborted pipeline is already in the runs, other errors (script, lock) are reported as such
|
|
summary.call(DataPipeline.runs.any? { |stats| stats.error.equal?(e) } ? nil : e) unless e.is_a?(DataPipeline::LockError)
|
|
exit(e.is_a?(DataPipeline::LockError) ? 3 : 1)
|
|
end
|
|
|
|
summary.call
|
|
runs = DataPipeline.runs
|
|
exit 3 if runs.any? { |stats| stats.error.is_a?(DataPipeline::LockError) }
|
|
exit 1 if runs.any?(&:error)
|
|
exit 2 if runs.any? { |stats| stats.rejected.positive? }
|
|
exit 0
|