KubernetesStorage

class phalanx.storage.kubernetes.KubernetesStorage(context=None)

Bases: object

Storage layer for direct Kubernetes operations.

Used primarily by the installer. This uses kubectl directly rather than one of the Python Kubernetes libraries since it seemed simpler at the time.

Parameters:

context (Optional[str], default: None) – The kubectl context to specify for all kubectl commands. If this is None, then the current context will be used.

Methods Summary

annotate(resource, key, value)	Add an annotation to a namespaced resource.
create_namespace(namespace, *[, ignore_fail])	Create a Kubernetes namespace.
create_vault_secret(name, namespace, credentials)	Create a Kubernetes Secret resource for Vault credentials.
deannotate(resource, key)	Remove an annotation from a namespaced resource.
get_current_context()	Get the current context (the default Kubernetes cluster).
get_phalanx_cronjobs()	Get the names of all CronJobs in all Phalanx apps.
get_phalanx_deployments()	Get the names of all Deployments in all Phalanx apps.
get_phalanx_stateful_sets()	Get the names of all StatefulSets in all Phalanx apps.
restart(name, namespace)	Restart a kubernetes workload and wait for it to be ready again.
resume_cronjobs(cronjobs)	Resume all given cronjobs.
scale(workload, replicas)	Scale the replica count of a workload.
suspend_cronjobs(cronjobs)	Suspend all given cronjobs.
wait_for_rollout(name, namespace)	Wait for a Kubernetes rollout to complete.

Methods Documentation

annotate(resource, key, value)

Add an annotation to a namespaced resource.

Parameters:

• resource (NamespacedResource) – The Kubernetes resource to annotate.

• key (str) – The name of the annotation.

• value (str) – The value of the annotation.

Return type:

None

create_namespace(namespace, *, ignore_fail=False)

Create a Kubernetes namespace.

Parameters:

• namespace (str) – Namespace to create.

• ignore_fail (bool, default: False) – If True, ignore failures, such as when the namespace already exists.

Raises:

CommandFailedError – Raised if the namespace creation fails, and ignore_fail was not set to True.

Return type:

None

create_vault_secret(name, namespace, credentials)

Create a Kubernetes Secret resource for Vault credentials.

Parameters:

• name (str) – Name of the secret.

• namespace (str) – Namespace of the secret.

• credentials (VaultCredentials) – Vault credentials to store in the secret.

Return type:

None

deannotate(resource, key)

Remove an annotation from a namespaced resource.

Parameters:

• resource (NamespacedResource) – The Kubernetes resource from which to remove the annotation.

• key (str) – The name of the annotation.

Return type:

None

get_current_context()

Get the current context (the default Kubernetes cluster).

Returns:

Name of the current Kubernetes context.

Return type:

str

get_phalanx_cronjobs()

Get the names of all CronJobs in all Phalanx apps.

We say a CronJob is in a Phalanx app if it has an ArgoCD label.

Returns:

A list of all of the CronJobs provisioned by Phalanx apps.

Return type:

list[CronJob]

get_phalanx_deployments()

Get the names of all Deployments in all Phalanx apps.

We say a Deployment is in a Phalanx app if it has an ArgoCD label.

Returns:

A list of all of the Deployments provisioned by Phalanx apps.

Return type:

list[Deployment]

get_phalanx_stateful_sets()

Get the names of all StatefulSets in all Phalanx apps.

We say a StatefulSet is in a Phalanx app if it has an ArgoCD label.

Returns:

A list of all of the StatefulSets provisioned by Phalanx apps.

Return type:

list[StatefulSet]

restart(name, namespace)

Restart a kubernetes workload and wait for it to be ready again.

Parameters:

• name (str) – Name of the workload. This should be the type of object (usually either deployment or statefulset, followed by a slash and the name of the object.

• namespace (str) – Namespace in which the workload exists.

Return type:

None

resume_cronjobs(cronjobs)

Resume all given cronjobs.

Parameters:

cronjobs (list[CronJob]) – A list of CronJobs to resume.

Return type:

None

scale(workload, replicas)

Scale the replica count of a workload.

Parameters:

• workload (Workload) – The Kubernetes workload to scale.

• replicas (int) – The desired replica count.

Return type:

None

suspend_cronjobs(cronjobs)

Suspend all given cronjobs.

Parameters:

cronjobs (list[CronJob]) – A list of CronJobs to suspend.

Return type:

None

wait_for_rollout(name, namespace)

Wait for a Kubernetes rollout to complete.

Parameters:

• name (str) – Name of the rollout. This should be the type of object (usually either deployment or statefulset, followed by a slash and the name of the object.

• namespace (str) – Namespace in which the rollout is happening.

Return type:

None