Skip to content

GKE Extensions

Google Kubernetes Engine specific resources.

Models

Gateway

GatewaySpec

Bases: BaseModel

GKE Gateway specification.

Example

GatewaySpec( ... name="main-gateway", ... namespace="gateway", ... gateway_class_name="gke-l7-global-external-managed", ... listeners=[{"name": "https", "port": 443, "protocol": "HTTPS"}], ... )

Source code in src/k8smith/gke/models.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class GatewaySpec(BaseModel):
    """GKE Gateway specification.

    Example:
        >>> GatewaySpec(
        ...     name="main-gateway",
        ...     namespace="gateway",
        ...     gateway_class_name="gke-l7-global-external-managed",
        ...     listeners=[{"name": "https", "port": 443, "protocol": "HTTPS"}],
        ... )
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    namespace: str
    gateway_class_name: str = Field(alias="gatewayClassName")
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    listeners: list[dict] = Field(default_factory=list)
    addresses: list[dict] = Field(default_factory=list)

HTTPRoute

HTTPRouteSpec

Bases: BaseModel

GKE HTTPRoute specification.

Example

HTTPRouteSpec( ... name="app-route", ... namespace="production", ... parent_refs=[{"name": "main-gateway", "namespace": "gateway"}], ... hostnames=["app.example.com"], ... rules=[{"matches": [{"path": {"value": "/"}}], "backendRefs": [...]}], ... )

Source code in src/k8smith/gke/models.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class HTTPRouteSpec(BaseModel):
    """GKE HTTPRoute specification.

    Example:
        >>> HTTPRouteSpec(
        ...     name="app-route",
        ...     namespace="production",
        ...     parent_refs=[{"name": "main-gateway", "namespace": "gateway"}],
        ...     hostnames=["app.example.com"],
        ...     rules=[{"matches": [{"path": {"value": "/"}}], "backendRefs": [...]}],
        ... )
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    namespace: str
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    parent_refs: list[dict] = Field(alias="parentRefs")
    hostnames: list[str] = Field(default_factory=list)
    rules: list[dict] = Field(default_factory=list)

HealthCheckPolicy

HealthCheckPolicySpec

Bases: BaseModel

GKE HealthCheckPolicy specification.

Example

HealthCheckPolicySpec( ... name="app-healthcheck", ... namespace="production", ... target_ref={"group": "", "kind": "Service", "name": "app"}, ... config={"type": "HTTP", "httpHealthCheck": {"requestPath": "/health"}}, ... )

Source code in src/k8smith/gke/models.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class HealthCheckPolicySpec(BaseModel):
    """GKE HealthCheckPolicy specification.

    Example:
        >>> HealthCheckPolicySpec(
        ...     name="app-healthcheck",
        ...     namespace="production",
        ...     target_ref={"group": "", "kind": "Service", "name": "app"},
        ...     config={"type": "HTTP", "httpHealthCheck": {"requestPath": "/health"}},
        ... )
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    namespace: str
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    target_ref: dict = Field(alias="targetRef")
    config: dict = Field(default_factory=dict)

GCPBackendPolicy

GCPBackendPolicySpec

Bases: BaseModel

GKE GCPBackendPolicy specification.

Example

GCPBackendPolicySpec( ... name="app-backend-policy", ... namespace="production", ... target_ref={"group": "", "kind": "Service", "name": "app"}, ... config={"timeoutSec": 30, "connectionDraining": {"drainingTimeoutSec": 60}}, ... )

Source code in src/k8smith/gke/models.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class GCPBackendPolicySpec(BaseModel):
    """GKE GCPBackendPolicy specification.

    Example:
        >>> GCPBackendPolicySpec(
        ...     name="app-backend-policy",
        ...     namespace="production",
        ...     target_ref={"group": "", "kind": "Service", "name": "app"},
        ...     config={"timeoutSec": 30, "connectionDraining": {"drainingTimeoutSec": 60}},
        ... )
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    namespace: str
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    target_ref: dict = Field(alias="targetRef")
    config: dict = Field(default_factory=dict)

PodMonitoring

PodMonitoringSpec

Bases: BaseModel

GKE PodMonitoring specification (Cloud Monitoring).

Example

PodMonitoringSpec( ... name="app-monitoring", ... namespace="production", ... selector={"matchLabels": {"app": "web"}}, ... endpoints=[{"port": "metrics", "interval": "30s"}], ... )

Source code in src/k8smith/gke/models.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class PodMonitoringSpec(BaseModel):
    """GKE PodMonitoring specification (Cloud Monitoring).

    Example:
        >>> PodMonitoringSpec(
        ...     name="app-monitoring",
        ...     namespace="production",
        ...     selector={"matchLabels": {"app": "web"}},
        ...     endpoints=[{"port": "metrics", "interval": "30s"}],
        ... )
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    namespace: str
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    selector: dict = Field(default_factory=dict)
    endpoints: list[dict] = Field(default_factory=list)
    target_labels: dict | None = Field(default=None, alias="targetLabels")

ClusterPodMonitoring

ClusterPodMonitoringSpec

Bases: BaseModel

GKE ClusterPodMonitoring specification.

Example

ClusterPodMonitoringSpec( ... name="cluster-monitoring", ... selector={"matchLabels": {"monitoring": "enabled"}}, ... endpoints=[{"port": "metrics", "interval": "60s"}], ... )

Source code in src/k8smith/gke/models.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class ClusterPodMonitoringSpec(BaseModel):
    """GKE ClusterPodMonitoring specification.

    Example:
        >>> ClusterPodMonitoringSpec(
        ...     name="cluster-monitoring",
        ...     selector={"matchLabels": {"monitoring": "enabled"}},
        ...     endpoints=[{"port": "metrics", "interval": "60s"}],
        ... )
    """

    model_config = ConfigDict(populate_by_name=True)

    name: str
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    selector: dict = Field(default_factory=dict)
    endpoints: list[dict] = Field(default_factory=list)
    target_labels: dict | None = Field(default=None, alias="targetLabels")

Builders

build_gateway(spec)

Build a GKE Gateway resource.

Parameters:

Name Type Description Default
spec GatewaySpec

Gateway specification

 required

Returns:

Type Description
dict

Gateway resource as a dict
Source code in src/k8smith/gke/gateway.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def build_gateway(spec: GatewaySpec) -> dict:
    """Build a GKE Gateway resource.

    Args:
        spec: Gateway specification

    Returns:
        Gateway resource as a dict
    """
    gateway: dict = {
        "apiVersion": "gateway.networking.k8s.io/v1",
        "kind": "Gateway",
        "metadata": {
            "name": spec.name,
            "namespace": spec.namespace,
        },
        "spec": {
            "gatewayClassName": spec.gateway_class_name,
            "listeners": spec.listeners,
        },
    }

    # Add optional metadata fields
    if spec.labels:
        gateway["metadata"]["labels"] = spec.labels
    if spec.annotations:
        gateway["metadata"]["annotations"] = spec.annotations

    # Add optional spec fields
    if spec.addresses:
        gateway["spec"]["addresses"] = spec.addresses

    return gateway

build_httproute(spec)

Build a GKE HTTPRoute resource.

Parameters:

Name Type Description Default
spec HTTPRouteSpec

HTTPRoute specification

 required

Returns:

Type Description
dict

HTTPRoute resource as a dict
Source code in src/k8smith/gke/httproute.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def build_httproute(spec: HTTPRouteSpec) -> dict:
    """Build a GKE HTTPRoute resource.

    Args:
        spec: HTTPRoute specification

    Returns:
        HTTPRoute resource as a dict
    """
    httproute: dict = {
        "apiVersion": "gateway.networking.k8s.io/v1",
        "kind": "HTTPRoute",
        "metadata": {
            "name": spec.name,
            "namespace": spec.namespace,
        },
        "spec": {
            "parentRefs": spec.parent_refs,
        },
    }

    # Add optional metadata fields
    if spec.labels:
        httproute["metadata"]["labels"] = spec.labels
    if spec.annotations:
        httproute["metadata"]["annotations"] = spec.annotations

    # Add optional spec fields
    if spec.hostnames:
        httproute["spec"]["hostnames"] = spec.hostnames
    if spec.rules:
        httproute["spec"]["rules"] = spec.rules

    return httproute

build_healthcheckpolicy(spec)

Build a GKE HealthCheckPolicy resource.

Parameters:

Name Type Description Default
spec HealthCheckPolicySpec

HealthCheckPolicy specification

 required

Returns:

Type Description
dict

HealthCheckPolicy resource as a dict
Source code in src/k8smith/gke/healthcheck.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def build_healthcheckpolicy(spec: HealthCheckPolicySpec) -> dict:
    """Build a GKE HealthCheckPolicy resource.

    Args:
        spec: HealthCheckPolicy specification

    Returns:
        HealthCheckPolicy resource as a dict
    """
    healthcheck: dict = {
        "apiVersion": "networking.gke.io/v1",
        "kind": "HealthCheckPolicy",
        "metadata": {
            "name": spec.name,
            "namespace": spec.namespace,
        },
        "spec": {
            "targetRef": spec.target_ref,
            "default": spec.config,
        },
    }

    # Add optional metadata fields
    if spec.labels:
        healthcheck["metadata"]["labels"] = spec.labels
    if spec.annotations:
        healthcheck["metadata"]["annotations"] = spec.annotations

    return healthcheck

build_gcp_backend_policy(spec)

Build a GKE GCPBackendPolicy resource.

Parameters:

Name Type Description Default
spec GCPBackendPolicySpec

GCPBackendPolicy specification

 required

Returns:

Type Description
dict

GCPBackendPolicy resource as a dict
Source code in src/k8smith/gke/backendpolicy.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def build_gcp_backend_policy(spec: GCPBackendPolicySpec) -> dict:
    """Build a GKE GCPBackendPolicy resource.

    Args:
        spec: GCPBackendPolicy specification

    Returns:
        GCPBackendPolicy resource as a dict
    """
    policy: dict = {
        "apiVersion": "networking.gke.io/v1",
        "kind": "GCPBackendPolicy",
        "metadata": {
            "name": spec.name,
            "namespace": spec.namespace,
        },
        "spec": {
            "targetRef": spec.target_ref,
            "default": spec.config,
        },
    }

    # Add optional metadata fields
    if spec.labels:
        policy["metadata"]["labels"] = spec.labels
    if spec.annotations:
        policy["metadata"]["annotations"] = spec.annotations

    return policy

build_pod_monitoring(spec)

Build a GKE PodMonitoring resource.

Parameters:

Name Type Description Default
spec PodMonitoringSpec

PodMonitoring specification

 required

Returns:

Type Description
dict

PodMonitoring resource as a dict
Source code in src/k8smith/gke/monitoring.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def build_pod_monitoring(spec: PodMonitoringSpec) -> dict:
    """Build a GKE PodMonitoring resource.

    Args:
        spec: PodMonitoring specification

    Returns:
        PodMonitoring resource as a dict
    """
    monitoring: dict = {
        "apiVersion": "monitoring.googleapis.com/v1",
        "kind": "PodMonitoring",
        "metadata": {
            "name": spec.name,
            "namespace": spec.namespace,
        },
        "spec": {
            "selector": spec.selector,
            "endpoints": spec.endpoints,
        },
    }

    # Add optional metadata fields
    if spec.labels:
        monitoring["metadata"]["labels"] = spec.labels
    if spec.annotations:
        monitoring["metadata"]["annotations"] = spec.annotations

    # Add optional spec fields
    if spec.target_labels:
        monitoring["spec"]["targetLabels"] = spec.target_labels

    return monitoring

build_cluster_pod_monitoring(spec)

Build a GKE ClusterPodMonitoring resource.

Parameters:

Name Type Description Default
spec ClusterPodMonitoringSpec

ClusterPodMonitoring specification

 required

Returns:

Type Description
dict

ClusterPodMonitoring resource as a dict
Source code in src/k8smith/gke/monitoring.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def build_cluster_pod_monitoring(spec: ClusterPodMonitoringSpec) -> dict:
    """Build a GKE ClusterPodMonitoring resource.

    Args:
        spec: ClusterPodMonitoring specification

    Returns:
        ClusterPodMonitoring resource as a dict
    """
    monitoring: dict = {
        "apiVersion": "monitoring.googleapis.com/v1",
        "kind": "ClusterPodMonitoring",
        "metadata": {
            "name": spec.name,
        },
        "spec": {
            "selector": spec.selector,
            "endpoints": spec.endpoints,
        },
    }

    # Add optional metadata fields
    if spec.labels:
        monitoring["metadata"]["labels"] = spec.labels
    if spec.annotations:
        monitoring["metadata"]["annotations"] = spec.annotations

    # Add optional spec fields
    if spec.target_labels:
        monitoring["spec"]["targetLabels"] = spec.target_labels

    return monitoring