Kubernetes Pythonクライアントの使い方

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