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

`add_service_load_balancer_ip`(service, ip)	Assign an IP address associated with a LoadBalancerService.
`annotate`(resource, key, value, *[, overwrite])	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_namespaced_resource`(kind, name, namespace)	Get a resource by namespace and name.
`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_load_balancer_services`()	Get all Services of type LoadBalancer for all Phalanx apps.
`get_phalanx_stateful_sets`()	Get the names of all StatefulSets in all Phalanx apps.
`get_pvs`()	Get all of the PersistentVolumes in the cluster.
`get_sasquatch_kafka_cluster_id`()	Get the Strimzi cluster id from a Kafka node PVC.
`get_service`(name, namespace)	Get a Service by namespace and name.
`pause_sasquatch_kafka_reconciliation`()	Pause Strimzi reconciliation of the Sasquatch Kafka cluster.
`remove_service_load_balancer_ip`(service)	Remove the IP address associated with a LoadBalancerService.
`restart`(name, namespace)	Restart a kubernetes workload and wait for it to be ready again.
`resume_cronjobs`(cronjobs)	Resume all given cronjobs.
`resume_sasquatch_kafka_reconciliation`()	Resume Strimzi reconciliation of the Sasquatch Kafka cluster.
`retain_pvs`()	Set the persistentVolumeReclaimPolicy to Retain for all PVs.
`scale`(workload, replicas)	Scale the replica count of a workload.
`set_sasquatch_cluster_id`(cluster_id)	Configure a Strimzi Kafka cluster with an explicitly specified ID.
`set_service_to_cluster_ip`(service)	Set the type of a Service to ClusterIP.
`set_service_to_load_balancer`(service)	Set the type of a Service to LoadBalancer.
`suspend_cronjobs`(cronjobs)	Suspend all given cronjobs.
`update_service_external_traffic_policy`(...)	Update the Service spec.externalTrafficPolicy value.
`version`()	Get the version of the kubectl client and kubernetes server.
`wait_for_resource_no_finalizers`(resource, ...)	Wait for a resource to NOT have the given finalizers.
`wait_for_rollout`(name, namespace)	Wait for a Kubernetes rollout to complete.
`wait_for_service_ingress`(service)	Wait for a LoadBalancer Service to have an associated ingress.

Methods Documentation

add_service_load_balancer_ip(service, ip)#

Assign an IP address associated with a LoadBalancerService.

Parameters:

service (Service) – The LoadBalancer Service to assign the IP to.
ip (IPv4Address) – The IP address to set the spec.loadBalancerIP field to.

Return type:

None

annotate(resource, key, value, *, overwrite=False)#

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.
overwrite (bool, default: False) – Whether or not to overwrite the annotation if it already exists.

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_namespaced_resource(kind, name, namespace)#

Get a resource by namespace and name.

Returns:

A namespaced resource.

Return type:

NamespacedResource

Parameters:

kind (str)
name (str)
namespace (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_load_balancer_services()#

Get all Services of type LoadBalancer for all Phalanx apps.

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

Returns:: A list of all of the LoadBalancer Services provisioned by Phalanx apps.
Return type:: list[Service]

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]

get_pvs()#

Get all of the PersistentVolumes in the cluster.

Returns:: A list of all of the PersistentVolumes in the cluster.
Return type:: list[PersistentVolume]

get_sasquatch_kafka_cluster_id()#

Get the Strimzi cluster id from a Kafka node PVC.

When recovering a Strimzi Kafka cluster from backed-up PersistentVolumes, the cluster id generated from applying a Strimzi Kafka resource will not match the cluster id in the data in the recovered volumes. We need to get the cluster id from the data on the volume before we let strimzi reconcile the Kafka resource, or the Kafka pods will not be able to start.

Command is from the Strimzi recovery docs here: https://strimzi.io/docs/operators/latest/full/deploying#proc-cluster-recovery-volume-str

This method assumes a lot about what various resources are called.

Return type:: str

get_service(name, namespace)#

Get a Service by namespace and name.

Returns:

A Service.

Return type:

Service

Parameters:

name (str)
namespace (str)

pause_sasquatch_kafka_reconciliation()#

Pause Strimzi reconciliation of the Sasquatch Kafka cluster.

During some recovery operations, we want to modify resources that are managed by the Strimzi operator. The Strimzi operator will automatically revert any changes we make, so we have to pause Strimzi reconciliation if we want our changes to persist.

https://strimzi.io/docs/operators/latest/full/deploying#proc-pausing-reconciliation-str

Return type:: None

remove_service_load_balancer_ip(service)#

Remove the IP address associated with a LoadBalancerService.

If the services is associated with a cloud load balancer with a static IP, and you want the IP to be released in the cloud, you also need to modify the type of the load balancer to be a ClusterIP.

Parameters:: service (Service) – The LoadBalancer Service to remove the IP from.
Return type:: None

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

resume_sasquatch_kafka_reconciliation()#

Resume Strimzi reconciliation of the Sasquatch Kafka cluster.

https://strimzi.io/docs/operators/latest/full/deploying#proc-pausing-reconciliation-str

Return type:: None

retain_pvs()#

Set the persistentVolumeReclaimPolicy to Retain for all PVs.

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

set_sasquatch_cluster_id(cluster_id)#

Configure a Strimzi Kafka cluster with an explicitly specified ID.

When we’re recovering a Strimzi Kafka cluster from backed-up persistent storage, We need to manually change the cluster ID to match the original cluster ID.

For more info on Strimzi Kafka cluster recovery, see: https://strimzi.io/docs/operators/latest/deploying#assembly-cluster-recovery-volume-str

Parameters:: cluster_id (str)
Return type:: None

set_service_to_cluster_ip(service)#

Set the type of a Service to ClusterIP.

Parameters:: service (Service) – The service to change.
Return type:: None

set_service_to_load_balancer(service)#

Set the type of a Service to LoadBalancer.

Parameters:: service (Service) – The service to change.
Return type:: None

suspend_cronjobs(cronjobs)#

Suspend all given cronjobs.

Parameters:: cronjobs (list[CronJob]) – A list of CronJobs to suspend.
Return type:: None

update_service_external_traffic_policy(service, policy)#

Update the Service spec.externalTrafficPolicy value.

When we convert a LoadBalancer service to a ClusterIP service, then back to a LoadBalancer service, spec.externalTrafficPolicy always gets set to “Cluster”, even if it was set to “Local” originally.

Parameters:

service (Service) – The LoadBalancer Service to assign the IP to.
policy (ServiceExternalTrafficPolicy) – The value to set the spec.externalTrafficPolicy field to.

Return type:

None

version()#

Get the version of the kubectl client and kubernetes server.

Useful for checking that kubectl can connect to a cluster using a given context.

Return type:: None

wait_for_resource_no_finalizers(resource, finalizers)#

Wait for a resource to NOT have the given finalizers.

This is useful for waiting for some controller to do some reconcilliation triggered by changing a resource.

Parameters:

resource (NamespacedResource) – The service to wait for.
finalizers (list[str])

Raises:

ResourceNoFinalizersTimeoutError – The associated finalizers are not removed after waiting.

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

wait_for_service_ingress(service)#

Wait for a LoadBalancer Service to have an associated ingress.

Parameters:: service (Service) – The service to wait for.
Return type:: None