Class: Tengine::Job::Runtime::SshJob

Inherits:
JobBase
  • Object
show all
Includes:
Core::CollectionAccessible, StateTransition, Template::SshJob::Settings
Defined in:
lib/tengine/job/runtime/ssh_job.rb

Overview

ルートジョブネットを他のジョブネット内に展開するための特殊なテンプレート用Vertex。

Defined Under Namespace

Classes: Error, ShellClient

Instance Method Summary collapse

Instance Method Details

#ack(signal) ⇒ Object

ハンドリングするドライバ: ジョブ制御ドライバ スクリプトのプロセスのPIDを取得できたときに実行されます



389
390
391
392
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 389

def ack(signal)
  self.executing_pid = (signal.data || {})[:executing_pid]
  self.phase_key = :running
end

#activate(signal) ⇒ Object

ハンドリングするドライバ: ジョブ制御ドライバ



338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 338

def activate(signal)
  case phase_key
  when :initialized then
    # 特別ルール「starting直前stop」
    # initializedに戻されたジョブに対して、:readyになる際にtransmitで送信されたイベントを受け取って、
    # activateしようとすると状態は遷移しないが、後続のエッジを実行する。
    # (エッジを実行しようとした際、エッジがclosedならばそのジョブネットのEndに遷移する。)
    next_edges.first.transmit(signal)
  when :ready then
    self.phase_key = :starting
    self.started_at = signal.event.occurred_at

    signal.call_later do
      complete_origin_edge(signal)
      execution = signal.execution
      if execution.retry
        if execution.target_actual_ids.include?(self.id.to_s)
          execution.ack(signal)
        elsif execution.target_actuals.map{|t| t.parent.id.to_s if t.parent }.include?(self.parent.id.to_s)
          # 自身とTengine::Job::Runtime::Execution#target_actual_idsに含まれるジョブ／ジョブネットと親が同じならば、ackしない
        else
          parent.ack(signal)
        end
      else
        parent.ack(signal) # 再実行でない場合
      end
      # このコールバックはjob_control_driverでupdate_with_lockの外側から
      # 再度呼び出してもらうためにcallbackを設定しています
      signal.call_later do
        # 実際にSSHでスクリプトを実行
        execution = signal.execution
        execution.signal = signal # ackを呼び返してもらうための苦肉の策
        begin
          run(execution)
        rescue Tengine::Job::Runtime::SshJob::Error => e
          Tengine.logger.warn("error on run\nerror: #{e.inspect}\njob: #{self.inspect}\nexecution: #{execution.inspect}")
          signal.call_later do
            self.fail(signal, :message => e.message)
          end
        end
      end
    end
  when :starting then
    raise "something wrong! #{self.inspect}"
  end
end

#add_error_message(msg) ⇒ Object 



315
316
317
318
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 315

def add_error_message(msg)
  self.error_messages ||= []
  self.error_messages += [msg]
end

#build_command(execution) ⇒ Object

def ack(pid) @acked_pid = pid self.executing_pid = pid self.phase_key = :running self.previous_edges.each{|edge| edge.status_key = :transmitted} end



203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 203

def build_command(execution)
  result = []
  mm_env = build_mm_env(execution).map{|k,v| "#{k}=#{v}"}.join(" ")
  # Hadoopジョブの場合は環境変数をセットする
  if is_a?(Tengine::Job::Runtime::Jobnet) && (jobnet_type_key == :hadoop_job_run)
    mm_env << ' ' << hadoop_job_env
  end
  result << "export #{mm_env}"
  template_root = (parent ? root_or_expansion.template : nil)
  if template_root
    template_job = template_root.vertex_by_name_path(self.name_path_until_expansion)
    unless template_job
      raise "job not found #{self.name_path_until_expansion.inspect} in #{template_root.inspect}"
    end
    key = Tengine::Job::Dsl::Loader.template_block_store_key(template_job, :preparation)
    preparation_block = Tengine::Job::Dsl::Loader.template_block_store[key]
    if preparation_block
      preparation = instance_eval(&preparation_block)
      unless preparation.blank?
        result << preparation
      end
    end
  end
  unless execution.preparation_command.blank?
    result << execution.preparation_command
  end
  # cmdはユーザーが設定したスクリプトを組み立てたもので、
  # プロセスの監視／強制停止のためにtengine_job_agent/bin/tengine_job_agent_run
  # からこれらを実行させるためにはcmdを編集します。
  # tengine_job_agent_runは、標準出力に監視対象となる起動したプロセスのPIDを出力します。
  runner_path = ENV["MM_RUNNER_PATH"] || executable_command("tengine_job_agent_run")
  runner_option = ""
  # 実装するべきか要検討
  # runner_option << " --stdout" if execution.keeping_stdout
  # runner_option << " --stderr" if execution.keeping_stderr
  # script = "#{runner_path}#{runner_option} -- #{self.script}" # runnerのオプションを指定する際は -- の前に設定してください
  script = "#{runner_path}#{runner_option} #{self.script}" # runnerのオプションを指定する際は -- の前に設定してください
  result << script
  result.join(" && ")
end

#build_mm_env(execution) ⇒ Object

MMから実行されるシェルスクリプトに渡す環境変数のHashを返します。 MM_ACTUAL_JOB_ID : 実行される末端のジョブのMM上でのID MM_ACTUAL_JOB_ANCESTOR_IDS : 実行される末端のジョブの祖先のMM上でのIDをセミコロンで繋げた文字列 (テンプレートジョブ単位) MM_FULL_ACTUAL_JOB_ANCESTOR_IDS : 実行される末端のジョブの祖先のMM上でのIDをセミコロンで繋げた文字列 (expansionから展開した単位) MM_ACTUAL_JOB_NAME_PATH : 実行される末端のジョブのname_path MM_ACTUAL_JOB_SECURITY_TOKEN : 公開API呼び出しのためのセキュリティ用のワンタイムトークン MM_TEMPLATE_JOB_ID : テンプレートジョブ(=実行される末端のジョブの元となったジョブ)のID MM_TEMPLATE_JOB_ANCESTOR_IDS : テンプレートジョブの祖先のMM上でのIDをセミコロンで繋げたもの MM_SCHEDULE_ID : 実行スケジュールのID MM_SCHEDULE_ESTIMATED_TIME : 実行スケジュールの見積り時間。単位は分。 MM_SCHEDULE_ESTIMATED_END : 実行スケジュールの見積り終了時刻をYYYYMMDDHHMMSS式で。(できればISO 8601など、タイムゾーンも表現できる標準的な形式の方が良い？) MM_MASTER_SCHEDULE_ID : マスタースケジュールがあればそのID。マスタースケジュールがない場合は環境変数は指定されません。

未実装 MM_FAILED_JOB_ID : ジョブが失敗した場合にrecoverやfinally内のジョブを実行時に設定される、失敗したジョブのMM上でのID。 MM_FAILED_JOB_ANCESTOR_IDS : ジョブが失敗した場合にrecoverやfinally内のジョブを実行時に設定される、失敗したジョブの祖先のMM上でのIDをセミコロンで繋げた文字列。



268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 268

def build_mm_env(execution)
  result = {
    "MM_SERVER_NAME" => actual_server_name,  # [Tengineの仕様として追加] ジョブの実行サーバ名を設定
    "MM_ROOT_JOBNET_ID" => root.id.to_s,
    "MM_TARGET_JOBNET_ID" => (parent ? parent.id.to_s : nil),
    "MM_ACTUAL_JOB_ID" => id.to_s,
    "MM_ACTUAL_JOB_ANCESTOR_IDS" => '"%s"' % ancestors_until_expansion.map(&:id).map(&:to_s).join(';'),
    "MM_FULL_ACTUAL_JOB_ANCESTOR_IDS" => '"%s"' % ancestors.map(&:id).map(&:to_s).join(';'),
    "MM_ACTUAL_JOB_NAME_PATH" => name_path.dump,
    "MM_ACTUAL_JOB_SECURITY_TOKEN" => "", # TODO トークンの生成
    "MM_SCHEDULE_ID" => execution.id.to_s,
    "MM_SCHEDULE_ESTIMATED_TIME" => execution.estimated_time,
  }
  if estimated_end = execution.actual_estimated_end
    result["MM_SCHEDULE_ESTIMATED_END"] = estimated_end.strftime("%Y%m%d%H%M%S")
  end
  if rjt = (parent ? root.template : nil)
    t = rjt.find_descendant_by_name_path(self.name_path)
    unless t
      template_name_parts = self.name_path_until_expansion.split(Tengine::Job::Structure::NamePath::SEPARATOR).select{|s| !s.empty?}
      root_jobnet_name = template_name_parts.first
      if rjt = Tengine::Job::Template::RootJobnet.find_by_name(root_jobnet_name, :version => rjt.dsl_version)
        t = rjt.find_descendant_by_name_path(self.name_path_until_expansion)
        raise "template job #{name_path.inspect} not found in #{rjt.inspect}" unless t
      else
        raise "Tengine::Job::Template::RootJobnet not found #{self.name_path_until_expansion.inspect}"
      end
    end
    result.update({
        "MM_TEMPLATE_JOB_ID" => t.id.to_s,
        "MM_TEMPLATE_JOB_ANCESTOR_IDS" => '"%s"' % t.ancestors.map(&:id).map(&:to_s).join(';'),
    })
  end
  # if ms = execution.master_schedule
  #   result.update({
  #       "MM_MASTER_SCHEDULE_ID" => ms.id.to_s,
  #   })
  # end
  result
end

#executable_command(command) ⇒ Object 



244
245
246
247
248
249
250
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 244

def executable_command(command)
  if prefix = ENV["MM_CMD_PREFIX"]
    "#{prefix} #{command}"
  else
    command
  end
end

#execute(cmd, &block) ⇒ Object 



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 140

def execute(cmd, &block)
  raise "actual_server not found for #{self.name_path.inspect}" unless actual_server
  Tengine.logger.info("connecting to #{actual_server.hostname_or_ipv4}")
  port = actual_server.properties["ssh_port"] || 22
  keys_only = actual_credential.auth_type_cd == :ssh_public_key
  Net::SSH.start(actual_server.hostname_or_ipv4, actual_credential, :port => port, :logger => Tengine.logger, :keys_only => keys_only) do |ssh|
    # see http://net-ssh.github.com/ssh/v2/api/classes/Net/SSH/Connection/Channel.html
    c = ssh.open_channel do |ch0|
      ch0.request_pty do |channel, success|
        raise Error, "failed to request_pty" unless success

        channel.exec("#{ENV['SHELL']} -l") do |shell_ch, success|
          raise Error, "failed to \"#{ENV['SHELL']} -l\"" unless success

          shell_ch.on_extended_data do |ch, type, data|
            add_error_message(data)
            raise Error, "Failure to execute #{self.name_path} via SSH: #{data}"
          end

          client = ShellClient.new(shell_ch, cmd, block)
          shell_ch[:client] = client
          client.setup
          client.start
        end
      end
    end
    c.wait
  end
rescue Tengine::Job::Runtime::SshJob::Error
  raise
rescue Mongoid::Errors::DocumentNotFound, SocketError, Net::SSH::AuthenticationFailed => src
  error = Error.new("[#{src.class.name}] #{src.message}")
  error.set_backtrace(src.backtrace)
  raise error
rescue Exception
  # puts "[#{$!.class.name}] #{$!.message}"
  raise
end

#fail(signal, options = nil) ⇒ Object

ハンドリングするドライバ: ジョブ制御ドライバ



419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 419

def fail(signal, options = nil)
  self.phase_key = :error
  if msg = signal.event[:message]
    add_error_message(msg)
  end
  if options && (msg = options[:message])
    add_error_message(msg)
  end
  self.finished_at = signal.event.occurred_at
  event_options = {
    :exit_status => self.exit_status,
    :target_jobnet_id => parent.id,
    :target_jobnet_name_path => parent.name_path,
    :target_job_id => self.id,
    :target_job_name_path => self.name_path,
  }
  event_options.update(options) if options
  signal.fire(self, :"error.job.job.tengine", event_options)
end

#finish(signal) ⇒ Object 



396
397
398
399
400
401
402
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 396

def finish(signal)
  self.exit_status = signal.event[:exit_status]
  self.finished_at = signal.event.occurred_at
  (self.exit_status.to_s == '0') ?
  succeed(signal) :
    fail(signal)
end

#fire_stop(signal) ⇒ Object 



440
441
442
443
444
445
446
447
448
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 440

def fire_stop(signal)
  signal.fire(self, :"stop.job.job.tengine", {
      :stop_reason => signal.event[:stop_reason],
      :target_jobnet_id => parent.id,
      :target_jobnet_name_path => parent.name_path,
      :target_job_id => self.id,
      :target_job_name_path => self.name_path,
    })
end

#hadoop_job_env ⇒ Object 



309
310
311
312
313
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 309

def hadoop_job_env
  s = children.select{|c| c.is_a?(Tengine::Job::Runtime::Jobnet) && (c.jobnet_type_key == :hadoop_job)}.
    map{|c| "#{c.name}\\t#{c.id.to_s}\\n"}.join
  "MM_HADOOP_JOBS=\"#{s}\""
end

#kill(execution) ⇒ Object 



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 179

def kill(execution)
  lines = []

  if self.executing_pid.blank?
    Tengine.logger.warn("PID is blank when kill!!\n#{self.inspect}\n  " << caller.join("\n  "))
  end

  cmd = executable_command("tengine_job_agent_kill %s %d %s" % [
      self.executing_pid,
      self.actual_killing_signal_interval,
      self.actual_killing_signals.join(","),
    ])
  lines << cmd
  cmd = lines.join(' && ')
  execute(cmd)
end

#prepare_server_and_credential ⇒ Object 



24
25
26
27
28
29
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 24

def prepare_server_and_credential
  if t = template_vertex
    self.server_name = t.actual_server_name if server_name.blank?
    self.credential_name = t.actual_credential_name if credential_name.blank?
  end
end

#reset(signal, &block) ⇒ Object 



480
481
482
483
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 480

def reset(signal, &block)
  self.phase_key = :initialized
  reset_followings(signal)
end

#run(execution) ⇒ Object 



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 31

def run(execution)
  return ack(@acked_pid) if @acked_pid
  cmd = build_command(execution)
  # puts "cmd:\n" << cmd
  execute(cmd) do |ch, data|
    pids = data.strip.scan(/^\d+$/)
    case pids.length
    when 0 then
      add_error_message("expected a set of numeric charactors but not found in: " << data.inspect)
      raise Error, "Failure to execute #{self.name_path} via SSH. expected numeric charactors but got: #{data}"
    when 1 then
      pid = pids.first.strip
      Tengine.logger.info("got pid: #{pid.inspect}")
    else
      add_error_message("expected a set of numeric charactors but got #{pids.length} sets of numeric charactoers #{pids.inspect} in #{data.inspect}")
      raise Error, "Failure to execute #{self.name_path} via SSH. expected numeric charactors but got: #{data}"
    end

    if signal = execution.signal
      signal.call_later do
        signal.data = {:executing_pid => pid}

        # このブロック内の処理はupdate_with_lockによって複数回実行されることがあります。
        # 1回目と同じリロードされていないオブジェクトを2回目以降に使用すると、1回目の変更が残っているので
        # そのオブジェクトに対して処理を行うのはNGです。
        # self.ack(signal) # これはNG

        # このブロックが実行されるたびに、rootからselfと同じidのオブジェクトを新たに取得する必要があります。
        job = root.vertex(self.id)
        job.ack(signal)
      end
    end
  end
rescue Exception => e
  Tengine.logger.error("[#{e.class}] #{e.message}\n  " << e.backtrace.join("\n  "))
  raise
end

#stop(signal, &block) ⇒ Object 



451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 451

def stop(signal, &block)
  case phase_key
  when :ready then
    self.phase_key = :initialized
    self.stopped_at = signal.event.occurred_at
    self.stop_reason = signal.event[:stop_reason]
    next_edges.first.transmit(signal)
  when :starting then
    job = nil
    loop do
      # root = self.root.reload # class.find(self.root.id)
      # job = root.find_descendant(self.id)
      job = self.class.find(self.id)
      break unless job.phase_key == :starting
      yield if block_given? # テストの為にyieldしています
      sleep(0.1)
    end
    job.stop(signal, &block)
  when :running then
    self.phase_key = :dying
    self.stopped_at = signal.event.occurred_at
    self.stop_reason = signal.event[:stop_reason]
    signal.call_later do
      kill(signal.execution)
    end
  end
end

#succeed(signal) ⇒ Object

ハンドリングするドライバ: ジョブ制御ドライバ



405
406
407
408
409
410
411
412
413
414
415
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 405

def succeed(signal)
  self.phase_key = :success
  self.finished_at = signal.event.occurred_at
  signal.fire(self, :"success.job.job.tengine", {
      :exit_status => self.exit_status,
      :target_jobnet_id => parent.id,
      :target_jobnet_name_path => parent.name_path,
      :target_job_id => self.id,
      :target_job_name_path => self.name_path,
    })
end

#transmit(signal) ⇒ Object

ハンドリングするドライバ: ジョブネット制御ドライバ



324
325
326
327
328
329
330
331
332
333
 
# File 'lib/tengine/job/runtime/ssh_job.rb', line 324

def transmit(signal)
  self.phase_key = :ready
  self.started_at = signal.event.occurred_at
  signal.fire(self, :"start.job.job.tengine", {
      :target_jobnet_id => parent.id,
      :target_jobnet_name_path => parent.name_path,
      :target_job_id => self.id,
      :target_job_name_path => self.name_path,
    })
end