Practical infra coding

Flag Kubernetes workloads with missing resource limits

mediumCSV, JSON and YAML Must-do

Problem statement

A cluster policy says every container must set both resources.limits.cpu and resources.limits.memory. Write a CI check that reads manifests and lists every container that breaks the rule, then exits non-zero if it found any.

The manifests would normally be YAML. The Python standard library has no YAML parser, so the input here is the JSON that kubectl get deploy,sts,ds,cronjob,svc -o json produces: a List with items. The logic is identical once the document is loaded.

Where the pod spec lives depends on the kind:

  • Pod: spec
  • Deployment, StatefulSet, DaemonSet, ReplicaSet, Job: spec.template.spec
  • CronJob: spec.jobTemplate.spec.template.spec

Check initContainers too. Ignore kinds that do not run pods, such as Service. A missing namespace means default.

manifests.json

JSON
{
"apiVersion": "v1",
"kind": "List",
"items": [
{"kind": "Deployment", "metadata": {"name": "web", "namespace": "shop"},
"spec": {"template": {"spec": {
"initContainers": [{"name": "migrate", "image": "shop/migrate:3"}],
"containers": [{"name": "nginx", "image": "nginx:1.27",
"resources": {"limits": {"cpu": "500m", "memory": "256Mi"}}}]}}}},
{"kind": "Deployment", "metadata": {"name": "api", "namespace": "payments"},
"spec": {"template": {"spec": {"containers": [
{"name": "api", "image": "pay/api:2.1", "resources": {"limits": {"memory": "512Mi"}}},
{"name": "envoy", "image": "envoyproxy/envoy:v1.31",
"resources": {"limits": {"cpu": "200m", "memory": "128Mi"}}}]}}}},
{"kind": "Service", "metadata": {"name": "api", "namespace": "payments"},
"spec": {"ports": [{"port": 80}]}},
{"kind": "CronJob", "metadata": {"name": "report", "namespace": "ops"},
"spec": {"jobTemplate": {"spec": {"template": {"spec": {"containers": [
{"name": "report", "image": "ops/report:1"}]}}}}}},
{"kind": "StatefulSet", "metadata": {"name": "redis", "namespace": "shop"},
"spec": {"template": {"spec": {"containers": [
{"name": "redis", "image": "redis:7",
"resources": {"requests": {"cpu": "100m", "memory": "128Mi"}}}]}}}},
{"kind": "DaemonSet", "metadata": {"name": "node-exporter", "namespace": "monitoring"},
"spec": {"template": {"spec": {"containers": [
{"name": "exporter", "image": "prom/node-exporter:v1.8",
"resources": {"limits": {"cpu": "100m", "memory": "64Mi"}}}]}}}}
]
}

Examples

Example 1

Input: python solution.py manifests.json; echo $?

Output: Deployment shop/web initContainer=migrate missing: cpu, memory Deployment payments/api container=api missing: cpu CronJob ops/report container=report missing: cpu, memory StatefulSet shop/redis container=redis missing: cpu, memory 4 container(s) without full limits in 5 workload(s)

Explanation: The exit code is 1. redis sets requests but not limits, which does not satisfy the policy. The Service is not a workload, so 5 of the 6 items are checked.

Hints

Approach

Optimal

Separate where the containers are from what to check on each one.

  1. Load the document and accept three shapes: a List with items, a bare array, or a single object.
  2. pod_spec maps each kind to its pod spec path and returns None for non-workloads. Every step uses (x.get(k) or {}), so a half-written manifest yields no containers rather than an AttributeError.
  3. For every container in initContainers and containers, read limits and list the required keys that are absent or empty.
  4. Print the findings and a summary, and return 1 if anything was found.

c is the total number of containers and f the number of findings. Invalid JSON is reported with its line number and a non-zero exit, not a traceback.

In a real repository you would load YAML with yaml.safe_load_all(f) from PyYAML, which yields one dict per --- document, and pass those dicts to the same find_missing_limits.

ComplexityTime O(c)Space O(f)
Python
import json
import sys
REQUIRED = ("cpu", "memory")
def pod_spec(obj):
"""Return the pod spec inside a workload, or None for non-workloads."""
kind = obj.get("kind")
spec = obj.get("spec") or {}
if kind == "Pod":
return spec
if kind in ("Deployment", "StatefulSet", "DaemonSet", "ReplicaSet", "Job"):
return (spec.get("template") or {}).get("spec")
if kind == "CronJob":
job = (spec.get("jobTemplate") or {}).get("spec") or {}
return (job.get("template") or {}).get("spec")
return None
def find_missing_limits(items):
findings = []
workloads = 0
for i, obj in enumerate(items):
if not isinstance(obj, dict):
print(f"item {i}: not an object, skipped", file=sys.stderr)
continue
spec = pod_spec(obj)
if spec is None:
continue
workloads += 1
meta = obj.get("metadata") or {}
ref = f"{obj['kind']} {meta.get('namespace', 'default')}/{meta.get('name', '?')}"
for field in ("initContainers", "containers"):
for c in spec.get(field) or []:
# "or {}" covers both a missing key and an explicit null
limits = (c.get("resources") or {}).get("limits") or {}
missing = [r for r in REQUIRED if not limits.get(r)]
if missing:
label = "initContainer" if field == "initContainers" else "container"
findings.append(f"{ref} {label}={c.get('name', '?')} missing: {', '.join(missing)}")
return findings, workloads
def main(path):
with open(path, encoding="utf-8") as f:
try:
doc = json.load(f)
except json.JSONDecodeError as e:
sys.exit(f"{path}: invalid JSON at line {e.lineno}: {e.msg}")
# accept a kubectl List, a bare array, or a single object
items = doc.get("items", [doc]) if isinstance(doc, dict) else doc
findings, workloads = find_missing_limits(items)
for line in findings:
print(line)
print(f"{len(findings)} container(s) without full limits in {workloads} workload(s)")
return 1 if findings else 0 # non-zero exit fails a CI step
if __name__ == "__main__":
sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "manifests.json"))

Follow-up questions

  • Also flag memory limits above 8Gi. How do you parse 512Mi, 1G and 1.5Gi into bytes?
  • Allow exceptions through an annotation such as policy/skip-limits: "true" on the workload.
  • Read a directory of YAML files with several --- documents each. What changes, and what if you may not install PyYAML?

Frequently asked questions

The rule here is about limits, which cap what a container can use. A missing memory limit lets one leaking pod push the node into memory pressure and get its neighbours evicted. Requests matter for scheduling, and many policies require both. The check is one line to extend: add a second list of required request keys.

Platform and DevOps teams enforce policies like this in CI before tools such as OPA Gatekeeper or Kyverno are in place, and often alongside them. The task tests careful traversal of nested, optional structures, which is most of the work with Kubernetes objects, and exit-code discipline for CI.

Render them first with helm template, then check the output. Checking the templates themselves does not work, because the limits often come from values files and conditionals. The same applies to Kustomize (kustomize build).