job + secret
secret作成
secret_name = "kubedt-pytest"
secret = V1Secret(metadata=V1ObjectMeta(
name=secret_name
), data={
"accesskey": access_key.encode('base64'),
"secretkey": secret_key.encode('base64'),
})
resp = v1.create_namespaced_secret(namespace=namespace, body=secret)
print(resp)
memory_size = '64Mi'
memory_usage = {"memory": memory_size}
access_key_env_src = V1EnvVarSource(
secret_key_ref=V1SecretKeySelector(
name=secret_name,
key="accesskey"
)
)
access_key_env = V1EnvVar(
name="ACCESS_KEY",
value_from=access_key_env_src
)
secret_key_env_src = V1EnvVarSource(
secret_key_ref=V1SecretKeySelector(
name=secret_name,
key="secretkey"
)
)
secret_key_env = V1EnvVar(
name="SECRET_KEY",
value_from=secret_key_env_src
)
args = [
"--bucket", bucket,
"--object-key", object_key,
"--parallelism", str(parallelism),
"--s3-endpoint", s3_endpoint,
"--namespace", namespace,
]
pod_template_spec = V1PodTemplateSpec(
metadata=V1ObjectMeta(name="kubedt-pytest-master", labels={"app": "kubedt-pytest-master"}),
spec=V1PodSpec(restart_policy="Never", containers=[
V1Container(args=args,
image="mumoshu/kubedt-pytest-master", image_pull_policy='IfNotPresent',
name="kubedt-pytest-master", env=[access_key_env,
secret_key_env],
# it fails like "oci runtime error: write /proc/26452/oom_score_adj: invalid argument" without resource req
resources=V1ResourceRequirements(limits=memory_usage, requests=memory_usage), )]))
job = V1Job(
api_version="batch/v1", kind='Job',
metadata=V1ObjectMeta(
name='kubedt-pytest-master',
),
spec=V1JobSpec(
active_deadline_seconds=30,
completions=1,
parallelism=1,
template=pod_template_spec
)
)
resp = batchv1.create_namespaced_job(
namespace=namespace,
body=job
)
print("Job created. status='%s'" % str(resp.status))
print(resp)
jobによって作られたpodには、controller-uidというラベルが付与される。
jobのmetadata.uidがpodのcontroller-uidラベルに対応するので、job.metadata.idをとっておけば後でそのjob管理下のpodを検索するために利用できる。
job_uid = resp.metadata.uid