Module: ActiveJobKubernetes

Defined in:
lib/active_job_kubernetes.rb,
lib/active_job_kubernetes/railtie.rb,
lib/active_job_kubernetes/version.rb

Defined Under Namespace

Classes: Railtie

Constant Summary collapse

SERIALIZED_JOB = 
'SERIALIZED_JOB'
SUSPEND = 
'active-job-kubernetes/suspend'
SUSPEND_UNTIL = 
'active-job-kubernetes/suspend-until'
VERSION = 
'0.2.0'

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.kubeclient ⇒ Object 



23
24
25
26
27
28
29
30
31
32
 
# File 'lib/active_job_kubernetes.rb', line 23

def kubeclient
  @kubeclient ||= lambda do |scope|
    Kubeclient::Client.new(
      "https://kubernetes.default.svc#{scope}",
      'v1',
      auth_options: { bearer_token_file: "#{SERVICE_ACCOUNT}/token" },
      ssl_options: { ca_file: "#{SERVICE_ACCOUNT}/ca.crt" }
    )
  end
end

.namespace ⇒ Object 



34
35
36
 
# File 'lib/active_job_kubernetes.rb', line 34

def namespace
  @namespace ||= File.read("#{SERVICE_ACCOUNT}/namespace")
end

Class Method Details

.create_job(job, timestamp = nil) ⇒ Object 



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
 
# File 'lib/active_job_kubernetes.rb', line 38

def create_job(job, timestamp = nil)
  serialized_job = JSON.dump(job.serialize)
  kube_job = Kubeclient::Resource.new(job.manifest)
  kube_job.metadata.namespace ||= namespace

  kube_job.spec.template.spec.containers.each do |container|
    container.env ||= []
    container.env.push({
      'name' => SERIALIZED_JOB,
      'value' => serialized_job
    })
  end

  if timestamp
    kube_job.spec.suspend = true

    kube_job.metadata.labels ||= {}
    kube_job.metadata.labels[SUSPEND] = 'true'

    kube_job.metadata.annotations ||= {}
    kube_job.metadata.annotations[SUSPEND_UNTIL] = Time.at(timestamp).utc.iso8601
  end

  job.provider_job_id = kubeclient_for(job).create_job(kube_job).metadata.name
end

.unsuspend_jobs ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/active_job_kubernetes.rb', line 64

def unsuspend_jobs
  now = Time.now
  client = kubeclient.call(BATCH_API)
  continue = nil

  loop do
    kube_jobs = client.get_jobs(
      namespace: namespace,
      label_selector: "#{SUSPEND}=true",
      limit: CHUNK_SIZE,
      continue: continue
    )

    kube_jobs.each do |kube_job|
      suspend_until = kube_job.metadata.annotations&.[](SUSPEND_UNTIL)
      next unless suspend_until && Time.iso8601(suspend_until) <= now

      client.patch_job(
        kube_job.metadata.name,
        { metadata: { labels: { SUSPEND => 'false' } }, spec: { suspend: false } },
        namespace
      )
    end

    continue = kube_jobs.continue
    break if kube_jobs.last?
  end
end