Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 38 additions & 1 deletion src/robusta/patch/patch.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import logging
from dataclasses import InitVar, is_dataclass
from dataclasses import InitVar, dataclass, field, is_dataclass
from inspect import getmodule, signature
from typing import Dict, List, Optional, Union, get_type_hints

Expand Down Expand Up @@ -44,9 +44,46 @@ def create_monkey_patches():
logging.info("Creating kubernetes ContainerImage monkey patch")
EventsV1Event.event_time = EventsV1Event.event_time.setter(event_time)
patch_on_pod_conditions()
patch_lifecycle_handler_sleep()
monkey_patches_applied = True


@dataclass
class SleepAction(HikaruBase):
"""LifecycleHandler.sleep action, GA in Kubernetes 1.34.

Hikaru's bundled v1 model (rel_1_26) predates this field, so we add it here
to ensure preStop/postStart sleep handlers are parsed and diffed correctly.
"""

seconds: int = 0


def patch_lifecycle_handler_sleep():
from hikaru.model.rel_1_26 import v1 as v1_pkg
from hikaru.model.rel_1_26.v1 import LifecycleHandler

if "sleep" in LifecycleHandler.__dataclass_fields__:
return

LifecycleHandler.__annotations__["sleep"] = Optional[SleepAction]
LifecycleHandler.sleep = None
sleep_field = field(default=None)
sleep_field.name = "sleep"
sleep_field.type = Optional[SleepAction]
# Borrow _field_type from an existing field so dataclasses.fields() picks
# this one up, without importing the private dataclasses._FIELD sentinel.
existing_field = next(iter(LifecycleHandler.__dataclass_fields__.values()))
sleep_field._field_type = existing_field._field_type
LifecycleHandler.__dataclass_fields__["sleep"] = sleep_field

v1_pkg.SleepAction = SleepAction
v1_pkg.v1.SleepAction = SleepAction

LifecycleHandler.cached_args = None
LifecycleHandler.cached_hints = None


def patch_on_pod_conditions():
# This fixes https://github.com/kubernetes-client/python/issues/2056 before the
# k8s people take care of it (it's urgent for us).
Expand Down
Loading