---
title: Charmhub | Deploy Observability Libs using Charmhub - The Open Operator Collection
description: Deploy the latest version of Observability Libs as a Kubernetes Operator
  on any cloud.
url: https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch
---

# Observability Libs

[Jon Seager](https://charmhub.io/publisher/jnsgruk "View all packages from Jon Seager")

* [Jon Seager](https://charmhub.io/publisher/jnsgruk "View all packages from Jon Seager")

Platform:

edge 454c1f8

```
juju deploy observability-libs --channel edge
```

[Learn to deploy on juju >](https://juju.is/docs/juju/manage-applications)

[Toggle side navigation](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#drawer)

## charms.observability\_libs.v0.kubernetes\_compute\_resources\_patch

* [*Docstrings*Docstrings](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch)
  [*Code*Source code](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch/source-code)
* + Download

    Fetch library

    ```
    charmcraft fetch-lib charms.observability_libs.v0.kubernetes_compute_resources_patch
    ```

    [Download kubernetes\_compute\_resources\_patch.py](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch/download)
  + *Last updated* 06 Oct 2025
  + *Revision* Library version 0.9

#### KubernetesComputeResourcesPatch Library.

This library is designed to enable developers to more simply patch the Kubernetes compute resource
limits and requests created by Juju during the deployment of a charm.

When initialised, this library binds a handler to the parent charm's `config-changed` event.
The config-changed event is used because it is guaranteed to fire on startup, on upgrade and on
pod churn. Additionally, resource limits may be set by charm config options, which would also be
caught out-of-the-box by this handler. The handler applies the patch to the app's StatefulSet.
This should ensure that the resource limits are correct throughout the charm's life. Additional
optional user-provided events for re-applying the patch are supported but discouraged.

The constructor takes a reference to the parent charm, a 'limits' and a 'requests' dictionaries
that together define the resource requirements. For information regarding the `lightkube`
`ResourceRequirements` model, please visit the `lightkube`
[docs](https://gtsystem.github.io/lightkube-models/1.23/models/core_v1/#resourcerequirements).

##### Getting Started

To get started using the library, you just need to fetch the library using `charmcraft`. **Note
that you also need to add `lightkube` and `lightkube-models` to your charm's `requirements.txt`.**

```
cd some-charm
charmcraft fetch-lib charms.observability_libs.v0.kubernetes_compute_resources_patch
cat << EOF >> requirements.txt
lightkube
lightkube-models
EOF
```

Then, to initialise the library:

```
# ...
from charms.observability_libs.v0.kubernetes_compute_resources_patch import (
    KubernetesComputeResourcesPatch,
    ResourceRequirements,
)

class SomeCharm(CharmBase):
  def __init__(self, *args):
    # ...
    self.resources_patch = KubernetesComputeResourcesPatch(
        self,
        "container-name",
        resource_reqs_func=lambda: ResourceRequirements(
            limits={"cpu": "2"}, requests={"cpu": "1"}
        ),
    )
    self.framework.observe(self.resources_patch.on.patch_failed, self._on_resource_patch_failed)

  def _on_resource_patch_failed(self, event):
    self.unit.status = BlockedStatus(event.message)
    # ...
```

Or, if, for example, the resource specs are coming from config options:

```
class SomeCharm(CharmBase):
  def __init__(self, *args):
    # ...
    self.resources_patch = KubernetesComputeResourcesPatch(
        self,
        "container-name",
        resource_reqs_func=self._resource_spec_from_config,
    )

  def _resource_spec_from_config(self) -> ResourceRequirements:
    spec = {"cpu": self.model.config.get("cpu"), "memory": self.model.config.get("memory")}
    return ResourceRequirements(limits=spec, requests=spec)
```

If you wish to pull the state of the resources patch operation and set the charm unit status based on that patch result,
you can achieve that using `get_status()` function.

```
class SomeCharm(CharmBase):
    def __init__(self, *args):
        #...
        self.framework.observe(self.on.collect_unit_status, self._on_collect_unit_status)
    #...
    def _on_collect_unit_status(self, event: CollectStatusEvent):
        event.add_status(self.resources_patch.get_status())
```

Additionally, you may wish to use mocks in your charm's unit testing to ensure that the library
does not try to make any API calls, or open any files during testing that are unlikely to be
present, and could break your tests. The easiest way to do this is during your test `setUp`:

```
# ...
from ops import ActiveStatus

@patch.multiple(
    "charm.KubernetesComputeResourcesPatch",
    _namespace="test-namespace",
    _is_patched=lambda *a, **kw: True,
    is_ready=lambda *a, **kw: True,
    get_status=lambda _: ActiveStatus(),
)
@patch("lightkube.core.client.GenericSyncClient")
def setUp(self, *unused):
    self.harness = Harness(SomeCharm)
    # ...
```

References:

* https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/
* https://gtsystem.github.io/lightkube-models/1.23/models/core\_v1/#resourcerequirements

---

Index

* [def adjust\_resource\_requirements(
  limits,
  requests,
  adhere\_to\_requests
  )](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#adjust_resource_requirements)
* [def is\_valid\_spec(
  spec,
  debug
  )](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#is_valid_spec)
* [def sanitize\_resource\_spec\_dict(
  spec
  )](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#sanitize_resource_spec_dict)
* [class K8sResourcePatchFailedEvent](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#k8sresourcepatchfailedevent)
* + [def \_\_init\_\_(
    self,
    handle,
    message)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#k8sresourcepatchfailedevent-__init__)
* + [def snapshot(
    self)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#k8sresourcepatchfailedevent-snapshot)
* + [def restore(
    self,
    snapshot)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#k8sresourcepatchfailedevent-restore)
* [class K8sResourcePatchEvents](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#k8sresourcepatchevents)
* [class ContainerNotFoundError](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#containernotfounderror)
* [class ResourcePatcher](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher)
* + [def \_\_init\_\_(
    self,
    namespace,
    statefulset\_name,
    container\_name)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-__init__)
* + [def is\_patched(
    self,
    resource\_reqs)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-is_patched)
* + [def get\_templated(
    self)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-get_templated)
* + [def get\_actual(
    self,
    pod\_name)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-get_actual)
* + [def is\_failed(
    self,
    resource\_reqs\_func)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-is_failed)
* + [def is\_in\_progress(
    self)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-is_in_progress)
* + [def is\_ready(
    self,
    pod\_name,
    resource\_reqs)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-is_ready)
* + [def apply(
    self,
    resource\_reqs,
    dry\_run)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#resourcepatcher-apply)
* [class KubernetesComputeResourcesPatch](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#kubernetescomputeresourcespatch)
* + [def \_\_init\_\_(
    self,
    charm,
    container\_name)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#kubernetescomputeresourcespatch-__init__)
* + [def is\_ready(
    self)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#kubernetescomputeresourcespatch-is_ready)
* + [def get\_status(
    self)](https://charmhub.io/observability-libs/libraries/kubernetes_compute_resources_patch#kubernetescomputeresourcespatch-get_status)

#### def adjust\_resource\_requirements(     limits,     requests,     adhere\_to\_requests: bool )

Adjust resource limits so that `limits` and `requests` are consistent with each other.

Arguments

limits

the "limits" portion of the resource spec.

requests

the "requests" portion of the resource spec.

adhere\_to\_requests

a flag indicating which portion should be adjusted when "limits" is
lower than "requests":

* if True, "limits" will be adjusted to max(limits, requests).
* if False, "requests" will be adjusted to min(limits, requests).

Returns

An adjusted (limits, requests) 2-tuple.

#### def is\_valid\_spec(     spec,     debug )

Check if the spec dict is valid.

Description

TODO: generally, the keys can be anything, not just cpu and memory. Perhaps user could pass
list of custom allowed keys in addition to the K8s ones?

#### def sanitize\_resource\_spec\_dict(spec)

Fix spec values without altering semantics.

Description

The purpose of this helper function is to correct known issues.
This function is not intended for fixing user mistakes such as incorrect keys present; that is
left for the `is_valid_spec` function.

#### class K8sResourcePatchFailedEvent

Description

Emitted when patching fails. None

Methods

K8sResourcePatchFailedEvent.
\_\_init\_\_(

*self*

,
handle

,
message
)

K8sResourcePatchFailedEvent.
snapshot(

*self*
)

Description

Save grafana source information. None

K8sResourcePatchFailedEvent.
restore(

*self*

,
snapshot
)

Description

Restore grafana source information. None

#### class K8sResourcePatchEvents

Description

Events raised by :class:`K8sResourcePatchEvents`. None

#### class ContainerNotFoundError

Description

Raised when a given container does not exist in the list of containers. None

#### class ResourcePatcher

Description

Helper class for patching a container's resource limits in a given StatefulSet. None

Methods

ResourcePatcher.
\_\_init\_\_(

*self*

,
namespace: str

,
statefulset\_name: str

,
container\_name: str
)

ResourcePatcher.
is\_patched(

*self*

,
resource\_reqs: ResourceRequirements
)

Reports if the resource patch has been applied to the StatefulSet.

Returns

bool

A boolean indicating if the service patch has been applied.

ResourcePatcher.
get\_templated(

*self*
)

Description

Returns the resource limits specified in the StatefulSet template. None

ResourcePatcher.
get\_actual(

*self*

,
pod\_name: str
)

Description

Return the resource limits that are in effect for the container in the given pod. None

ResourcePatcher.
is\_failed(

*self*

,
resource\_reqs\_func
)

Returns a tuple indicating whether a patch operation has failed along with a failure message.

Description

Implementation is based on dry running the patch operation to catch if there would be failures (e.g: Wrong spec and Auth errors).

ResourcePatcher.
is\_in\_progress(

*self*
)

Returns a boolean to indicate whether a patch operation is in progress.

Description

Implementation follows a similar approach to `kubectl rollout status statefulset` to track the progress of a rollout.
Reference: https://github.com/kubernetes/kubectl/blob/kubernetes-1.31.0/pkg/polymorphichelpers/rollout\_status.go

ResourcePatcher.
is\_ready(

*self*

,
pod\_name

,
resource\_reqs: ResourceRequirements
)

Reports if the resource patch has been applied and is in effect.

Returns

bool

A boolean indicating if the service patch has been applied and is in effect.

ResourcePatcher.
apply(

*self*

,
resource\_reqs: ResourceRequirements

,
dry\_run
)

Description

Patch the Kubernetes resources created by Juju to limit cpu or mem. None

#### class KubernetesComputeResourcesPatch

Description

A utility for patching the Kubernetes compute resources set up by Juju. None

Methods

KubernetesComputeResourcesPatch.
\_\_init\_\_(

*self*

,
charm: CharmBase

,
container\_name: str
)

Constructor for KubernetesComputeResourcesPatch.

Arguments

charm

the charm that is instantiating the library.

container\_name

the container for which to apply the resource limits.

resource\_reqs\_func

a callable returning a `ResourceRequirements`; if raises, should
only raise ValueError.

refresh\_event

an optional bound event or list of bound events which
will be observed to re-apply the patch.

Description

References:
- https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/

KubernetesComputeResourcesPatch.
is\_ready(

*self*
)

Reports if the resource patch has been applied and is in effect.

Returns

bool

A boolean indicating if the service patch has been applied and is in effect.

KubernetesComputeResourcesPatch.
get\_status(

*self*
)

Return the status of patching the resource limits in a `StatusBase` format.

Returns

StatusBase

There is a 1:1 mapping between the state of the patching operation and a `StatusBase` value that the charm can be set to.