k8s_functions.functions.k8s_subject(obj=None, resource_path=None, prefix='k8s:')

Returns a k8s subject instance providing a kubernetes resource.

rulesdata = [
    {
        rulename: "...",
        subscibre_to: "...",
        ruledata: {
            filters: [
                Filter(
                    K8sObjectsQuery(
                        returns=lambda obj: (
                            k8s_subject(obj).ext_name == "my-pod"
                        )
                    )
                )
            ]
            processing: [
                ...
            ]
        }
    }
]

Parameters

• obj – K8s object from which the subject will be created

• resource_path – If is None is equal to object selfLink. BE CAREFUL from Kubernetes 1.20 selfLink is deprecated![default None]

• prefix – [default k8s:]

k8s_functions.functions.k8s_object(subject, renew=False)

Returns the k8s resource providing a subject instance.

rulesdata = [
    {
        rulename: "...",
        subscibre_to: "...",
        ruledata: {
            filters: [
                Filter(
                    lambda subject: "my-label" in k8s_object(subject).get("metadata", {}).get("labels", {})
                )
            ]
            processing: [
                ...
            ]
        }
    }
]

Parameters

• subject – Subject from which will be get the K8s object

• renew – If True reset subject property to the corresponding resource[default False]

k8s_functions.functions.k8s_event_create(api, producer, action, message, reason, type, reporting_component=None, reporting_instance=None, involved_object=None, namespace=None, source_component=None, first_timestamp=None, last_timestamp=None)

Crate a Kubernetes reosurce of type v1/Event.

class k8s_functions.functions.K8sObjectsQuery

Search for a list of K8s resources using the given filters.

rulesdata = [
    {
        rulename: "...",
        subscibre_to: "...",
        ruledata: {
            filters: [
                Filter(
                    K8sObjectsQuery(
                        kind="Pod",
                        returns=lambda payload: lambda obj: (
                            payload.set_default("pods", obj)
                        )
                    )
                )
            ]
            processing: [
                ...
            ]
        }
    }
]

execute(apiversion=None, kind=None, foreach=None, returns=None, **filters)

Parameters

• apiversion – resources api version, if None get this value from subject extended properties[default None]

• kind – resources kind version, if None get this value from subject extended properties[default None]

• foreach – an optional callable that will be execute for each query element[default None]

• returns – an optional callable that will received as parameters the whole query elements[default None]

• **filters – query kwargs

class k8s_functions.functions.K8sObjectUpdate

Update a K8s object.

rulesdata = [
    {
        rulename: "...",
        subscibre_to: "...",
        ruledata: {
            filters: [
                ...
            ]
            processing: [
                K8sObjectUpdate(
                    lambda obj: (
                        obj["metadata"]["labels"].update({
                            "updated": "my-pod"
                        })
                    ),
                    name="my-pod",
                    apiversion="v1",
                    kind="Pod",
                    namespace="default"
                ),
            ]
        }
    }
]

execute(patch, name=None, apiversion=None, kind=None, subresource=None, is_strategic=True, **filters)

Parameters

• patch – could be a callable that receive a dict or a dict itself

• name – resource name, if None get this value from subject extended properties[default None]

• apiversion – resource api version, if None get this value from subject extended properties[default None]

• kind – resource kind version, if None get this value from subject extended properties[default None]

• subresource – subresource parameter of pykube object update[default None]

• is_strategic – is_strategic parameter of pykube object update[default True]

• **filters – query kwargs

class k8s_functions.functions.K8sObjectCreate

Create a K8s object.

rulesdata = [
    {
        rulename: "...",
        subscibre_to: "...",
        ruledata: {
            filters: [
                ...
            ]
            processing: [
                K8sObjectCreate({
                    "apiVersion": "v1",
                    "kind": "Pod",
                    "metadata": {
                        "name": "my-pod",
                        "labels": {
                            "app": "pytest-temp"
                        }
                    },
                    "spec": {
                        "containers": [{
                            "name": "hello",
                            "image": "karthequian/helloworld"
                        }]
                    }
                })
            ]
        }
    }
]

execute(obj)

Parameters

obj – dict representing the structure of the object you want to create.

class k8s_functions.functions.K8sObjectDelete

Delete a K8s object.

rulesdata = [
    {
        rulename: "...",
        subscibre_to: "...",
        ruledata: {
            filters: [
                ...
            ]
            processing: [
                K8sObjectCreate(
                    name="my-pod",
                    apiversion="v1",
                    kind="Pod",
                    namespace="default"
                )
            ]
        }
    }
]

execute(name=None, apiversion=None, kind=None, **filters)

Parameters

• name – resource name, if None get this value from subject extended properties[default None]

• apiversion – resource api version, if None get this value from subject extended properties[default None]

• kind – resource kind version, if None get this value from subject extended properties[default None]

• **filters – query kwargs