Class: Ductwork::Configuration

Inherits:
Object
  • Object
show all
Defined in:
lib/ductwork/configuration.rb

Overview

rubocop:todo Metrics/ClassLength

Defined Under Namespace

Classes: InvalidRoleError

Constant Summary collapse

DEFAULT_ENV = 
:default
DEFAULT_FILE_PATH = 
"config/ductwork.yml"
DEFAULT_FORKING =

fork pipeline advancer and job workers

"default"
DEFAULT_JOB_WORKER_COUNT =

threads

5
DEFAULT_JOB_WORKER_MAX_CRASH =

attempts

10
DEFAULT_JOB_WORKER_MAX_RETRY =

attempts

3
DEFAULT_JOB_WORKER_POLLING_TIMEOUT =

second

1
DEFAULT_JOB_WORKER_SHUTDOWN_TIMEOUT =

seconds

20
DEFAULT_LOGGER_LEVEL = 
::Logger::INFO
DEFAULT_LOGGER_SOURCE =

Logger instance writing to STDOUT

"default"
DEFAULT_PIPELINE_ADVANCER_MAX_CRASH =

attempts

10
DEFAULT_PIPELINE_ADVANCER_MAX_RETRY =

attempts

10
DEFAULT_PIPELINE_ADVANCER_THREAD_COUNT =

threads

3
DEFAULT_PIPELINE_POLLING_TIMEOUT =

second

1
DEFAULT_PIPELINE_SHUTDOWN_TIMEOUT =

seconds

20
DEFAULT_ROLE =

supervisor, pipeline advancer, and job workers

"all"
DEFAULT_STEPS_MAX_DEPTH =

unlimited count

-1 # unlimited count
DEFAULT_SUPERVISOR_POLLING_TIMEOUT =

second

1
DEFAULT_SUPERVISOR_REAPER_TIMEOUT =

seconds

300
DEFAULT_SUPERVISOR_SHUTDOWN_TIMEOUT =

seconds

30
DEFAULT_LOGGER = 
::Logger.new($stdout)
PIPELINES_DIRECTORIES = 
%w[app/pipelines app/workflows].freeze
PIPELINES_WILDCARD = 
"*"
VALID_ROLES = 
%w[all advancer worker].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(path: DEFAULT_FILE_PATH, role: nil) ⇒ Configuration

Returns a new instance of Configuration.



43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/ductwork/configuration.rb', line 43

def initialize(path: DEFAULT_FILE_PATH, role: nil)
  full_path = Pathname.new(path)
  data = ActiveSupport::ConfigurationFile.parse(full_path).deep_symbolize_keys
  env = defined?(Rails) ? Rails.env.to_sym : DEFAULT_ENV
  @config = if role.present?
              data[env].merge(role:)
            else
              data[env]
            end
rescue Errno::ENOENT
  @config = { role: }.compact
end

Instance Attribute Details

#job_worker_count(pipeline) ⇒ Object 



83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/ductwork/configuration.rb', line 83

def job_worker_count(pipeline)
  return @job_worker_count if instance_variable_defined?(:@job_worker_count)

  raw_count = config.dig(:job_worker, :count) || DEFAULT_JOB_WORKER_COUNT

  if raw_count.is_a?(Hash)
    raw_count[pipeline.to_sym]
  else
    raw_count
  end
end

#job_worker_max_crash(pipeline: nil, step: nil) ⇒ Object

rubocop:disable Metrics



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/ductwork/configuration.rb', line 95

def job_worker_max_crash(pipeline: nil, step: nil) # rubocop:disable Metrics
  return @job_worker_max_crash if instance_variable_defined?(:@job_worker_max_crash)

  pipeline ||= :default
  step ||= :default
  base_config = config.dig(:job_worker, :max_crash)

  if base_config.is_a?(Hash) && base_config[pipeline.to_sym].is_a?(Hash)
    pipeline_config = config.dig(:job_worker, :max_crash, pipeline.to_sym)

    pipeline_config[step.to_sym] || pipeline_config[:default] || DEFAULT_JOB_WORKER_MAX_CRASH
  elsif base_config.is_a?(Hash)
    base_config[pipeline.to_sym] || base_config[:default] || DEFAULT_JOB_WORKER_MAX_CRASH
  else
    base_config || DEFAULT_JOB_WORKER_MAX_CRASH
  end
end

#job_worker_max_retry(pipeline: nil, step: nil) ⇒ Object

rubocop:disable Metrics



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/ductwork/configuration.rb', line 113

def job_worker_max_retry(pipeline: nil, step: nil) # rubocop:disable Metrics
  return @job_worker_max_retry if instance_variable_defined?(:@job_worker_max_retry)

  pipeline ||= :default
  step ||= :default
  base_config = config.dig(:job_worker, :max_retry)

  if base_config.is_a?(Hash) && base_config[pipeline.to_sym].is_a?(Hash)
    pipeline_config = config.dig(:job_worker, :max_retry, pipeline.to_sym)

    pipeline_config[step.to_sym] || pipeline_config[:default] || DEFAULT_JOB_WORKER_MAX_RETRY
  elsif base_config.is_a?(Hash)
    base_config[pipeline.to_sym] || base_config[:default] || DEFAULT_JOB_WORKER_MAX_RETRY
  else
    base_config || DEFAULT_JOB_WORKER_MAX_RETRY
  end
end

#job_worker_polling_timeout(pipeline = nil) ⇒ Object 



131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/ductwork/configuration.rb', line 131

def job_worker_polling_timeout(pipeline = nil)
  pipeline ||= :default
  default = DEFAULT_JOB_WORKER_POLLING_TIMEOUT
  base_config = config.dig(:job_worker, :polling_timeout)

  if instance_variable_defined?(:@job_worker_polling_timeout)
    @job_worker_polling_timeout
  elsif base_config.is_a?(Hash)
    base_config[pipeline.to_sym] || base_config[:default] || default
  else
    base_config || default
  end
end

#job_worker_shutdown_timeout ⇒ Object 



145
146
147
 
# File 'lib/ductwork/configuration.rb', line 145

def job_worker_shutdown_timeout
  @job_worker_shutdown_timeout ||= fetch_job_worker_shutdown_timeout
end

#logger_level ⇒ Object 



149
150
151
 
# File 'lib/ductwork/configuration.rb', line 149

def logger_level
  @logger_level ||= fetch_logger_level
end

#pipeline_advancer_count(pipeline = nil) ⇒ Object 



157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/ductwork/configuration.rb', line 157

def pipeline_advancer_count(pipeline = nil)
  pipeline ||= :default
  default = DEFAULT_PIPELINE_ADVANCER_THREAD_COUNT
  base_config = config.dig(:pipeline_advancer, :count)

  if instance_variable_defined?(:@pipeline_advancer_count)
    @pipeline_advancer_count
  elsif base_config.is_a?(Hash)
    base_config[pipeline.to_sym] || base_config[:default] || default
  else
    base_config || default
  end
end

#pipeline_advancer_max_crash ⇒ Object 



171
172
173
 
# File 'lib/ductwork/configuration.rb', line 171

def pipeline_advancer_max_crash
  @pipeline_advancer_max_crash ||= fetch_pipeline_advancer_max_crash
end

#pipeline_advancer_max_retry ⇒ Object 



175
176
177
 
# File 'lib/ductwork/configuration.rb', line 175

def pipeline_advancer_max_retry
  @pipeline_advancer_max_retry ||= fetch_pipeline_advancer_max_retry
end

#pipeline_polling_timeout(pipeline = nil) ⇒ Object 



179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/ductwork/configuration.rb', line 179

def pipeline_polling_timeout(pipeline = nil)
  pipeline ||= :default
  default = DEFAULT_PIPELINE_POLLING_TIMEOUT
  base_config = config.dig(:pipeline_advancer, :polling_timeout)

  if instance_variable_defined?(:@pipeline_polling_timeout)
    @pipeline_polling_timeout
  elsif base_config.is_a?(Hash)
    base_config[pipeline.to_sym] || base_config[:default] || default
  else
    base_config || default
  end
end

#pipeline_shutdown_timeout ⇒ Object 



193
194
195
 
# File 'lib/ductwork/configuration.rb', line 193

def pipeline_shutdown_timeout
  @pipeline_shutdown_timeout ||= fetch_pipeline_shutdown_timeout
end

#steps_max_depth(pipeline: nil, step: nil) ⇒ Object

rubocop:disable Metrics



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
 
# File 'lib/ductwork/configuration.rb', line 197

def steps_max_depth(pipeline: nil, step: nil) # rubocop:disable Metrics
  return @steps_max_depth if instance_variable_defined?(:@steps_max_depth)

  pipeline ||= :default
  step ||= :default
  base_config = config.dig(:pipeline_advancer, :steps, :max_depth)

  if base_config.is_a?(Hash) && base_config[pipeline.to_sym].is_a?(Hash)
    pipeline_config = config.dig(:pipeline_advancer, :steps, :max_depth, pipeline.to_sym)

    pipeline_config[step.to_sym] ||
      pipeline_config[:default] ||
      DEFAULT_STEPS_MAX_DEPTH
  elsif base_config.is_a?(Hash)
    base_config[pipeline.to_sym] ||
      base_config[:default] ||
      DEFAULT_STEPS_MAX_DEPTH
  else
    base_config || DEFAULT_STEPS_MAX_DEPTH
  end
end

#supervisor_polling_timeout ⇒ Object 



219
220
221
 
# File 'lib/ductwork/configuration.rb', line 219

def supervisor_polling_timeout
  @supervisor_polling_timeout ||= fetch_supervisor_polling_timeout
end

#supervisor_reaper_timeout ⇒ Object 



223
224
225
 
# File 'lib/ductwork/configuration.rb', line 223

def supervisor_reaper_timeout
  @supervisor_reaper_timeout ||= fetch_supervisor_reaper_timeout
end

#supervisor_shutdown_timeout ⇒ Object 



227
228
229
 
# File 'lib/ductwork/configuration.rb', line 227

def supervisor_shutdown_timeout
  @supervisor_shutdown_timeout ||= fetch_supervisor_shutdown_timeout
end

Instance Method Details

#database ⇒ Object 



79
80
81
 
# File 'lib/ductwork/configuration.rb', line 79

def database
  config[:database]
end

#forking ⇒ Object 



62
63
64
 
# File 'lib/ductwork/configuration.rb', line 62

def forking
  config[:forking] || DEFAULT_FORKING
end

#logger_source ⇒ Object 



153
154
155
 
# File 'lib/ductwork/configuration.rb', line 153

def logger_source
  @logger_source ||= fetch_logger_source
end

#pipelines ⇒ Object 



66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/ductwork/configuration.rb', line 66

def pipelines
  raw_pipelines = config[:pipelines] || []

  if raw_pipelines == PIPELINES_WILDCARD
    PIPELINES_DIRECTORIES
      .flat_map { |dir| Dir.glob("**/*.rb", base: dir) }
      .map { |path| path.delete_suffix(".rb").camelize }
      .uniq
  else
    raw_pipelines.map(&:strip)
  end
end

#role ⇒ Object 



56
57
58
59
60
 
# File 'lib/ductwork/configuration.rb', line 56

def role
  r = config[:role] || DEFAULT_ROLE
  validate_role!(r)
  r
end