class Dynflow::PersistenceAdapters::Sequel
Constants
- MAX_RETRIES
- META_DATA
- RETRY_DELAY
- SERIALIZABLE_COLUMNS
- TABLES
Attributes
db[R]
Public Class Methods
migrations_path()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 311 def self.migrations_path File.expand_path('../sequel_migrations', __FILE__) end
new(config)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 50 def initialize(config) migrate = true config = config.dup @additional_responsibilities = { coordinator: true, connector: true } if config.is_a?(Hash) @additional_responsibilities.merge!(config.delete(:additional_responsibilities)) if config.key?(:additional_responsibilities) migrate = config.fetch(:migrate, true) end @db = initialize_db config migrate_db if migrate end
Public Instance Methods
abort_if_pending_migrations!()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 288 def abort_if_pending_migrations! ::Sequel::Migrator.check_current(db, self.class.migrations_path, table: 'dynflow_schema_info') end
connector_feature!()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 207 def connector_feature! unless @additional_responsibilities[:connector] raise "The sequel persistence adapter connector feature used but not enabled in additional_features" end end
coordinator_feature!()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 244 def coordinator_feature! unless @additional_responsibilities[:coordinator] raise "The sequel persistence adapter coordinator feature used but not enabled in additional_features" end end
delete_coordinator_record(class_name, record_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 260 def delete_coordinator_record(class_name, record_id) coordinator_feature! table(:coordinator_record).where(class: class_name, id: record_id).delete end
delete_delayed_plans(filters, batch_size = 1000)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 124 def delete_delayed_plans(filters, batch_size = 1000) count = 0 filter(:delayed, table(:delayed), filters).each_slice(batch_size) do |plans| uuids = plans.map { |p| p.fetch(:execution_plan_uuid) } @db.transaction do count += table(:delayed).where(execution_plan_uuid: uuids).delete end end count end
delete_execution_plans(filters, batch_size = 1000, backup_dir = nil)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 91 def delete_execution_plans(filters, batch_size = 1000, backup_dir = nil) count = 0 filter(:execution_plan, table(:execution_plan), filters).each_slice(batch_size) do |plans| uuids = plans.map { |p| p.fetch(:uuid) } @db.transaction do table(:delayed).where(execution_plan_uuid: uuids).delete steps = table(:step).where(execution_plan_uuid: uuids) backup_to_csv(:step, steps, backup_dir, 'steps.csv') if backup_dir steps.delete table(:output_chunk).where(execution_plan_uuid: uuids).delete actions = table(:action).where(execution_plan_uuid: uuids) backup_to_csv(:action, actions, backup_dir, 'actions.csv') if backup_dir actions.delete execution_plans = table(:execution_plan).where(uuid: uuids) backup_to_csv(:execution_plan, execution_plans, backup_dir, 'execution_plans.csv') if backup_dir count += execution_plans.delete end end return count end
delete_output_chunks(execution_plan_id, action_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 203 def delete_output_chunks(execution_plan_id, action_id) filter(:output_chunk, table(:output_chunk), { execution_plan_uuid: execution_plan_id, action_id: action_id }).delete end
filtering_by()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 27 def filtering_by META_DATA.fetch :execution_plan end
find_coordinator_records(options)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 265 def find_coordinator_records(options) coordinator_feature! options = options.dup filters = (options[:filters] || {}).dup exclude_owner_id = filters.delete(:exclude_owner_id) data_set = filter(:coordinator_record, table(:coordinator_record), filters) if exclude_owner_id data_set = data_set.exclude(:owner_id => exclude_owner_id) end data_set.all.map { |record| load_data(record) } end
find_execution_plan_counts(options = {})
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 77 def find_execution_plan_counts(options = {}) filter(:execution_plan, table(:execution_plan), options[:filters]).count end
find_execution_plan_statuses(options)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 81 def find_execution_plan_statuses(options) plans = filter(:execution_plan, table(:execution_plan), options[:filters]) .select(:uuid, :state, :result) plans.each_with_object({}) do |current, acc| uuid = current.delete(:uuid) acc[uuid] = current end end
find_execution_plans(options = {})
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 66 def find_execution_plans(options = {}) table_name = :execution_plan options[:order_by] ||= :started_at data_set = filter(table_name, order(table_name, paginate(table(table_name), options), options), options[:filters]) data_set.all.map { |record| execution_plan_column_map(load_data(record, table_name)) } end
find_old_execution_plans(age)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 135 def find_old_execution_plans(age) table_name = :execution_plan table(table_name) .where(::Sequel.lit('ended_at <= ? AND state = ?', age, 'stopped')) .all.map { |plan| execution_plan_column_map(load_data plan, table_name) } end
find_past_delayed_plans(time)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 142 def find_past_delayed_plans(time) table_name = :delayed table(table_name) .where(::Sequel.lit('start_at <= ? OR (start_before IS NOT NULL AND start_before <= ?)', time, time)) .where(:frozen => false) .order_by(:start_at) .all .map { |plan| load_data(plan, table_name) } end
insert_coordinator_record(value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 250 def insert_coordinator_record(value) coordinator_feature! save :coordinator_record, {}, value end
load_action(execution_plan_id, action_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 175 def load_action(execution_plan_id, action_id) load :action, execution_plan_uuid: execution_plan_id, id: action_id end
load_actions(execution_plan_id, action_ids)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 179 def load_actions(execution_plan_id, action_ids) load_records :action, { execution_plan_uuid: execution_plan_id, id: action_ids } end
load_actions_attributes(execution_plan_id, attributes)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 183 def load_actions_attributes(execution_plan_id, attributes) load_records :action, { execution_plan_uuid: execution_plan_id }, attributes end
load_delayed_plan(execution_plan_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 152 def load_delayed_plan(execution_plan_id) load :delayed, execution_plan_uuid: execution_plan_id rescue KeyError return nil end
load_execution_plan(execution_plan_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 116 def load_execution_plan(execution_plan_id) execution_plan_column_map(load :execution_plan, uuid: execution_plan_id) end
load_output_chunks(execution_plan_id, action_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 199 def load_output_chunks(execution_plan_id, action_id) load_records :output_chunk, { execution_plan_uuid: execution_plan_id, action_id: action_id }, [:timestamp, :kind, :chunk] end
load_step(execution_plan_id, step_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 162 def load_step(execution_plan_id, step_id) load :step, execution_plan_uuid: execution_plan_id, id: step_id end
load_steps(execution_plan_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 166 def load_steps(execution_plan_id) load_records :step, execution_plan_uuid: execution_plan_id end
migrate_db()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 284 def migrate_db ::Sequel::Migrator.run(db, self.class.migrations_path, table: 'dynflow_schema_info') end
ordering_by()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 31 def ordering_by META_DATA.fetch :execution_plan end
pagination?()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 23 def pagination? true end
prune_envelopes(receiver_ids)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 234 def prune_envelopes(receiver_ids) connector_feature! table(:envelope).where(receiver_id: receiver_ids).delete end
prune_undeliverable_envelopes()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 239 def prune_undeliverable_envelopes connector_feature! table(:envelope).where(receiver_id: table(:coordinator_record).select(:id)).invert.delete end
pull_envelopes(receiver_id)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 218 def pull_envelopes(receiver_id) connector_feature! db.transaction do data_set = table(:envelope).where(receiver_id: receiver_id).all envelopes = data_set.map { |record| load_data(record) } table(:envelope).where(id: data_set.map { |d| d[:id] }).delete return envelopes end end
push_envelope(envelope)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 229 def push_envelope(envelope) connector_feature! table(:envelope).insert(prepare_record(:envelope, envelope)) end
save_action(execution_plan_id, action_id, value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 187 def save_action(execution_plan_id, action_id, value) save :action, { execution_plan_uuid: execution_plan_id, id: action_id }, value, with_data: false end
save_delayed_plan(execution_plan_id, value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 158 def save_delayed_plan(execution_plan_id, value) save :delayed, { execution_plan_uuid: execution_plan_id }, value, with_data: false end
save_envelope(data)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 213 def save_envelope(data) connector_feature! save :envelope, {}, data end
save_execution_plan(execution_plan_id, value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 120 def save_execution_plan(execution_plan_id, value) save :execution_plan, { uuid: execution_plan_id }, value, with_data: false end
save_output_chunks(execution_plan_id, action_id, chunks)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 191 def save_output_chunks(execution_plan_id, action_id, chunks) chunks.each do |chunk| chunk[:execution_plan_uuid] = execution_plan_id chunk[:action_id] = action_id save :output_chunk, {}, chunk, with_data: false end end
save_step(execution_plan_id, step_id, value, update_conditions = {})
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 170 def save_step(execution_plan_id, step_id, value, update_conditions = {}) save :step, { execution_plan_uuid: execution_plan_id, id: step_id }, value, with_data: false, update_conditions: update_conditions end
to_hash()
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 277 def to_hash { execution_plans: table(:execution_plan).all.to_a, steps: table(:step).all.to_a, actions: table(:action).all.to_a, envelopes: table(:envelope).all.to_a } end
transaction(&block)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 62 def transaction(&block) db.transaction(&block) end
update_coordinator_record(class_name, record_id, value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 255 def update_coordinator_record(class_name, record_id, value) coordinator_feature! save :coordinator_record, { class: class_name, :id => record_id }, value end
Private Instance Methods
backup_to_csv(table_name, dataset, backup_dir, file_name)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 423 def backup_to_csv(table_name, dataset, backup_dir, file_name) ensure_backup_dir(backup_dir) csv_file = File.join(backup_dir, file_name) appending = File.exist?(csv_file) columns = dataset.columns File.open(csv_file, 'a') do |csv| csv << columns.to_csv unless appending dataset.each do |row| values = columns.map do |col| value = row[col] value = value.unpack('H*').first if value && SERIALIZABLE_COLUMNS.fetch(table_name, []).include?(col.to_s) value end csv << values.to_csv end end dataset end
delete(what, condition)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 442 def delete(what, condition) table(what).where(Utils.symbolize_keys(condition)).delete end
dump_data(value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 452 def dump_data(value) return if value.nil? packed = MessagePack.pack(Type!(value, Hash, Array, Integer, String)) ::Sequel.blob(packed) end
ensure_backup_dir(backup_dir)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 419 def ensure_backup_dir(backup_dir) FileUtils.mkdir_p(backup_dir) unless File.directory?(backup_dir) end
execution_plan_column_map(plan)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 531 def execution_plan_column_map(plan) plan[:id] = plan[:uuid] unless plan[:uuid].nil? plan end
extract_metadata(what, value)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 446 def extract_metadata(what, value) meta_keys = META_DATA.fetch(what) - SERIALIZABLE_COLUMNS.fetch(what, []) value = Utils.indifferent_hash(value) meta_keys.inject({}) { |h, k| h.update k.to_sym => value[k] } end
filter(what, data_set, filters)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 480 def filter(what, data_set, filters) Type! filters, NilClass, Hash return data_set if filters.nil? filters = filters.each.with_object({}) { |(k, v), hash| hash[k.to_s] = v } unknown = filters.keys - META_DATA.fetch(what) if what == :execution_plan unknown -= %w[uuid caller_execution_plan_id caller_action_id delayed] if filters.key?('caller_action_id') && !filters.key?('caller_execution_plan_id') raise ArgumentError, "caller_action_id given but caller_execution_plan_id missing" end if filters.key?('caller_execution_plan_id') data_set = data_set.join_table(:inner, TABLES[:action], :execution_plan_uuid => :uuid) .select_all(TABLES[:execution_plan]).distinct end if filters.key?('delayed') filters.delete('delayed') data_set = data_set.join_table(:inner, TABLES[:delayed], :execution_plan_uuid => :uuid) .select_all(TABLES[:execution_plan]).distinct end end unless unknown.empty? raise ArgumentError, "unkown columns: #{unknown.inspect}" end data_set.where Utils.symbolize_keys(filters) end
initialize_db(db_path)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 306 def initialize_db(db_path) logger = Logger.new($stderr) if ENV['DYNFLOW_SQL_LOG'] ::Sequel.connect db_path, logger: logger end
load_data(record, what = nil)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 406 def load_data(record, what = nil) hash = if record[:data].nil? SERIALIZABLE_COLUMNS.fetch(what, []).each do |key| key = key.to_sym record[key] = MessagePack.unpack(record[key].to_s) unless record[key].nil? end record else MessagePack.unpack(record[:data].to_s) end Utils.indifferent_hash(hash) end
load_record(what, condition)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 366 def load_record(what, condition) table = table(what) if (record = with_retry { table.first(Utils.symbolize_keys(condition)) }) load_data(record, what) else raise KeyError, "searching: #{what} by: #{condition.inspect}" end end
Also aliased as: load
load_records(what, condition, keys = nil)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 385 def load_records(what, condition, keys = nil) table = table(what) records = with_retry do filtered = table.filter(Utils.symbolize_keys(condition)) # Filter out requested columns which the table doesn't have, load data just in case unless keys.nil? columns = table.columns & keys columns |= [:data] if table.columns.include?(:data) filtered = filtered.select(*columns) end filtered.all end records = records.map { |record| load_data(record, what) } return records if keys.nil? records.map do |record| keys.reduce({}) do |acc, key| acc.merge(key => record[key]) end end end
order(what, data_set, options)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 470 def order(what, data_set, options) order_by = (options[:order_by]).to_s return data_set if order_by.empty? unless META_DATA.fetch(what).include? order_by raise ArgumentError, "unknown column #{order_by.inspect}" end order_by = order_by.to_sym data_set.order_by options[:desc] ? ::Sequel.desc(order_by) : order_by end
paginate(data_set, options)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 458 def paginate(data_set, options) page = Integer(options[:page]) if options[:page] per_page = Integer(options[:per_page]) if options[:per_page] if page raise ArgumentError, "page specified without per_page attribute" unless per_page data_set.limit per_page, per_page * page else data_set end end
prepare_record(table_name, value, base = {}, with_data = true)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 315 def prepare_record(table_name, value, base = {}, with_data = true) record = base.dup has_data_column = table(table_name).columns.include?(:data) if with_data && has_data_column record[:data] = dump_data(value) else if has_data_column record[:data] = nil else record.delete(:data) end record.merge! serialize_columns(table_name, value) end record.merge! extract_metadata(table_name, value) record.each { |k, v| record[k] = v.to_s if v.is_a? Symbol } record end
prune_unchanged(what, object, record)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 375 def prune_unchanged(what, object, record) record = record.dup table(what).columns.each do |column| record.delete(column) if object[column] == record[column] end record end
save(what, condition, value, with_data: true, update_conditions: {})
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 345 def save(what, condition, value, with_data: true, update_conditions: {}) table = table(what) existing_record = with_retry { table.first condition } unless condition.empty? if value record = prepare_record(what, value, (existing_record || condition), with_data) if existing_record record = prune_unchanged(what, existing_record, record) return value if record.empty? condition = update_conditions.merge(condition) return with_retry { table.where(condition).update(record) } else with_retry { table.insert record } end else existing_record and with_retry { table.where(condition).delete } end value end
serialize_columns(table_name, record)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 335 def serialize_columns(table_name, record) record.reduce({}) do |acc, (key, value)| if SERIALIZABLE_COLUMNS.fetch(table_name, []).include?(key.to_s) acc.merge(key.to_sym => dump_data(value)) else acc end end end
table(which)
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 302 def table(which) db[TABLES.fetch(which)] end
with_retry() { || ... }
click to toggle source
# File lib/dynflow/persistence_adapters/sequel.rb, line 511 def with_retry attempts = 0 begin yield rescue ::Sequel::DatabaseConnectionError, ::Sequel::DatabaseDisconnectError => e attempts += 1 log(:error, e) if attempts > MAX_RETRIES log(:error, "The number of MAX_RETRIES exceeded") raise Errors::FatalPersistenceError.delegate(e) else log(:error, "Persistence retry no. #{attempts}") sleep RETRY_DELAY retry end rescue Exception => e raise Errors::PersistenceError.delegate(e) end end