Skip to main content

​
Overview

The constants module provides commonly used timeout values and exception dictionaries for handling transient cluster errors during API operations.

​
Timeout Constants

Predefined timeout values in seconds for various operations.

​
TIMEOUT_1SEC

from ocp_resources.utils.constants import TIMEOUT_1SEC

TIMEOUT_1SEC = 1
One second timeout for quick operations. Usage: 
from ocp_resources.utils.constants import TIMEOUT_1SEC

resource.wait_for_condition(
    condition="Ready",
    status="True",
    sleep_time=TIMEOUT_1SEC
)

​
TIMEOUT_5SEC

from ocp_resources.utils.constants import TIMEOUT_5SEC

TIMEOUT_5SEC = 5
Five second timeout for short operations.

​
TIMEOUT_10SEC

from ocp_resources.utils.constants import TIMEOUT_10SEC

TIMEOUT_10SEC = 10
Ten second timeout for brief operations. This is the default timeout for cluster retry operations. Usage: 
from ocp_resources.utils.constants import TIMEOUT_10SEC
from ocp_resources import Resource

result = Resource.retry_cluster_exceptions(
    func=api_call,
    timeout=TIMEOUT_10SEC
)

​
TIMEOUT_30SEC

from ocp_resources.utils.constants import TIMEOUT_30SEC

TIMEOUT_30SEC = 30
Thirty second timeout for moderate operations.

​
TIMEOUT_1MINUTE

from ocp_resources.utils.constants import TIMEOUT_1MINUTE

TIMEOUT_1MINUTE = 60
One minute (60 seconds) timeout. This is the default timeout for most resource operations. Usage: 
from ocp_resources.utils.constants import TIMEOUT_1MINUTE

class MyResource(Resource):
    timeout_seconds = TIMEOUT_1MINUTE

​
TIMEOUT_2MINUTES

from ocp_resources.utils.constants import TIMEOUT_2MINUTES

TIMEOUT_2MINUTES = 120
Two minute (120 seconds) timeout for longer operations.

​
TIMEOUT_4MINUTES

from ocp_resources.utils.constants import TIMEOUT_4MINUTES

TIMEOUT_4MINUTES = 240
Four minute (240 seconds) timeout. This is the default timeout for resource deletion and waiting operations. Usage: 
from ocp_resources.utils.constants import TIMEOUT_4MINUTES
from ocp_resources import Pod

pod = Pod(client=client, name="my-pod", namespace="default")
pod.delete(wait=True, timeout=TIMEOUT_4MINUTES)

​
TIMEOUT_10MINUTES

from ocp_resources.utils.constants import TIMEOUT_10MINUTES

TIMEOUT_10MINUTES = 600
Ten minute (600 seconds) timeout for extended operations. Usage: 
from ocp_resources.utils.constants import TIMEOUT_10MINUTES
from ocp_resources import Deployment

deployment = Deployment(
    client=client,
    name="my-app",
    namespace="default"
)
deployment.wait_for_condition(
    condition="Available",
    status="True",
    timeout=TIMEOUT_10MINUTES
)

​
Exception Dictionaries

Dictionaries mapping exception types to lists of error message patterns. These are used with TimeoutSampler to automatically retry operations on transient cluster errors.

​
DEFAULT_CLUSTER_RETRY_EXCEPTIONS

Default exceptions that should trigger automatic retries during cluster operations. 
from ocp_resources.utils.constants import DEFAULT_CLUSTER_RETRY_EXCEPTIONS

DEFAULT_CLUSTER_RETRY_EXCEPTIONS = {
    MaxRetryError: [],
    ConnectionAbortedError: [],
    ConnectionResetError: [],
    InternalServerError: [
        "etcdserver: leader changed",
        "etcdserver: request timed out",
        "Internal error occurred: failed calling webhook",
        "rpc error:",
    ],
    ServerTimeoutError: [],
    ForbiddenError: ["context deadline exceeded"],
}

​
Exception Types

​
MaxRetryError
list[str]
Network-level retry errors. Empty list means retry on any MaxRetryError.
​
ConnectionAbortedError
list[str]
Connection aborted errors. Empty list means retry on any ConnectionAbortedError.
​
ConnectionResetError
list[str]
Connection reset errors. Empty list means retry on any ConnectionResetError.
​
InternalServerError
list[str]
Kubernetes API server internal errors. Retries only on specific error messages:
  • etcd leader changes
  • etcd timeouts
  • Webhook failures
  • RPC errors
​
ServerTimeoutError
list[str]
API server timeout errors. Empty list means retry on any ServerTimeoutError.
​
ForbiddenError
list[str]
Forbidden errors caused by timeouts. Retries only on “context deadline exceeded”.

​
Usage Example

from ocp_resources.utils.constants import DEFAULT_CLUSTER_RETRY_EXCEPTIONS
from ocp_resources import Resource
from timeout_sampler import TimeoutSampler

# Automatically retry on cluster errors
sampler = TimeoutSampler(
    wait_timeout=60,
    sleep=1,
    func=some_api_call,
    exceptions_dict=DEFAULT_CLUSTER_RETRY_EXCEPTIONS
)

for sample in sampler:
    if sample:
        break

​
PROTOCOL_ERROR_EXCEPTION_DICT

Exception dictionary for handling protocol-level errors. 
from ocp_resources.utils.constants import PROTOCOL_ERROR_EXCEPTION_DICT

PROTOCOL_ERROR_EXCEPTION_DICT = {
    ProtocolError: []
}
Usage: 
from ocp_resources.utils.constants import PROTOCOL_ERROR_EXCEPTION_DICT

resource.wait_for_status(
    status="Running",
    timeout=300,
    exceptions_dict=PROTOCOL_ERROR_EXCEPTION_DICT
)

​
NOT_FOUND_ERROR_EXCEPTION_DICT

Exception dictionary for handling resource not found errors. 
from ocp_resources.utils.constants import NOT_FOUND_ERROR_EXCEPTION_DICT

NOT_FOUND_ERROR_EXCEPTION_DICT = {
    NotFoundError: []
}
Usage: 
from ocp_resources.utils.constants import NOT_FOUND_ERROR_EXCEPTION_DICT
from timeout_sampler import TimeoutSampler

# Wait for resource to exist, ignoring NotFoundError
sampler = TimeoutSampler(
    wait_timeout=60,
    sleep=1,
    func=lambda: resource.exists,
    exceptions_dict=NOT_FOUND_ERROR_EXCEPTION_DICT
)

for sample in sampler:
    if sample:
        break

​
Complete Example

Here’s a comprehensive example using multiple constants: 
from ocp_resources import Pod
from ocp_resources.resource import get_client
from ocp_resources.utils.constants import (
    DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    NOT_FOUND_ERROR_EXCEPTION_DICT,
    PROTOCOL_ERROR_EXCEPTION_DICT,
    TIMEOUT_1MINUTE,
    TIMEOUT_4MINUTES,
    TIMEOUT_10MINUTES,
)
from timeout_sampler import TimeoutSampler

# Create client
client = get_client()

# Create pod with custom timeout
pod = Pod(
    client=client,
    name="my-pod",
    namespace="default"
)
pod.timeout_seconds = TIMEOUT_1MINUTE

# Create with retry logic
pod.create(wait=True)

# Wait for pod with combined exception handling
exceptions_dict = {
    **PROTOCOL_ERROR_EXCEPTION_DICT,
    **NOT_FOUND_ERROR_EXCEPTION_DICT,
    **DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
}

sampler = TimeoutSampler(
    wait_timeout=TIMEOUT_10MINUTES,
    sleep=5,
    func=lambda: pod.instance.status.phase == "Running",
    exceptions_dict=exceptions_dict
)

for sample in sampler:
    if sample:
        print("Pod is running!")
        break

# Delete with timeout
pod.delete(wait=True, timeout=TIMEOUT_4MINUTES)

​
Exception Dictionary Merging

You can combine multiple exception dictionaries for comprehensive error handling: 
from ocp_resources.utils.constants import (
    DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
    NOT_FOUND_ERROR_EXCEPTION_DICT,
    PROTOCOL_ERROR_EXCEPTION_DICT,
)

# Merge dictionaries for comprehensive retry logic
all_exceptions = {
    **PROTOCOL_ERROR_EXCEPTION_DICT,
    **NOT_FOUND_ERROR_EXCEPTION_DICT,
    **DEFAULT_CLUSTER_RETRY_EXCEPTIONS,
}

resource.wait(
    timeout=300,
    sleep=1,
    exceptions_dict=all_exceptions
)

​
Custom Exception Patterns

You can extend exception dictionaries with your own patterns: 
from ocp_resources.utils.constants import DEFAULT_CLUSTER_RETRY_EXCEPTIONS
from kubernetes.dynamic.exceptions import InternalServerError

# Add custom error patterns
custom_exceptions = DEFAULT_CLUSTER_RETRY_EXCEPTIONS.copy()
custom_exceptions[InternalServerError].append("custom error pattern")

resource.create(
    wait=True,
    exceptions_dict=custom_exceptions
)

Build docs developers (and LLMs) love

Get started for freeTalk to us