Action: Check the status of K8s CronJob pods

Kubernetes/legos/k8s_check_cronjob_pod_status/README.md

@@ -0,0 +1,24 @@
+[<img align="left" src="https://unskript.com/assets/favicon.png" width="100" height="100" style="padding-right: 5px">]
+(https://unskript.com/assets/favicon.png)
+<h1>Checks the status of CronJob pods</h1>
+
+## Description
+This action checks the status of CronJob pods
+
+## Lego Details
+	k8s_check_cronjob_pod_status(handle, namespace: str="")
+		handle: Object of type unSkript K8S Connector.
+		namespace: Namespace where the CronJob is deployed.
+
+
+## Lego Input
+This Lego takes inputs handle namespace(Optional)
+
+## Lego Output
+Here is a sample output.
+<img src="./1.png">
+<img src="./2.png">
+
+## See it in Action
+
+You can see this Lego in action following this link [unSkript Live](https://us.app.unskript.io)

Kubernetes/legos/k8s_check_cronjob_pod_status/k8s_check_cronjob_pod_status.json

@@ -0,0 +1,15 @@
+{
+  "action_title": "Check the status of K8s CronJob pods",
+  "action_description": "This action checks the status of CronJob pods",
+  "action_type": "LEGO_TYPE_K8S",
+  "action_entry_function": "k8s_check_cronjob_pod_status",
+  "action_needs_credential": true,
+  "action_output_type": "ACTION_OUTPUT_TYPE_LIST",
+  "action_is_check": true,
+  "action_next_hop": [
+    ""
+  ],
+  "action_next_hop_parameter_mapping": {},
+  "action_supports_iteration": true,
+  "action_supports_poll": true
+}

Kubernetes/legos/k8s_check_cronjob_pod_status/k8s_check_cronjob_pod_status.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+#
+# Copyright (c) 2023 unSkript.com
+# All rights reserved.
+#
+from datetime import datetime, timezone
+from kubernetes import client
+from typing import Tuple, Optional
+from pydantic import BaseModel, Field
+from croniter import croniter
+
+
+class InputSchema(BaseModel):
+    namespace: Optional[str] = Field(..., description='k8s Namespace', title='Namespace')
+
+
+def k8s_check_cronjob_pod_status_printer(output):
+    status, issues = output
+    if status:
+        print("CronJobs are running as expected.")
+    else:
+        for issue in issues:
+            print(f"CronJob '{issue['cronjob_name']}' Alert: {issue['message']}")
+
+
+def k8s_check_cronjob_pod_status(handle, namespace: str='') -> Tuple:
+    """
+    Checks the status of the CronJob pods.
+
+    :type handle: object
+    :param handle: The Kubernetes client handle.
+
+    :type name: str
+    :param namespace: Namespace where the CronJob is deployed.
+
+    :return: A tuple where the first item has the status if the check and second has a list of failed objects.
+    """
+    # Initialize the K8s API clients
+    batch_v1 = client.BatchV1Api(api_client=handle)
+    batch_v1beta1 = client.BatchV1beta1Api(api_client=handle)
+    core_v1 = client.CoreV1Api(api_client=handle)
+
+    issues = []
+
+    # Get namespaces to check
+    if namespace:
+        namespaces = [namespace]
+    else:
+        ns_obj = core_v1.list_namespace()
+        namespaces = [ns.metadata.name for ns in ns_obj.items]
+
+    for ns in namespaces:
+        # Fetch all CronJobs in the namespace
+        cronjobs = batch_v1beta1.list_namespaced_cron_job(ns).items
+
+        for cronjob in cronjobs:
+            schedule = cronjob.spec.schedule
+
+            # Calculate the next expected run
+            now = datetime.now(timezone.utc)
+            iter = croniter(schedule, now)
+            next_run = iter.get_next(datetime)
+            time_to_next_run = next_run - now
+
+            # Fetch the most recent Job associated with the CronJob
+            jobs = batch_v1.list_namespaced_job(ns)  # Fetch all jobs, and then filter by prefix.
+
+            associated_jobs = [job for job in jobs.items if job.metadata.name.startswith(cronjob.metadata.name)]
+            if not associated_jobs:
+                issues.append({"cronjob_name": cronjob.metadata.name, "namespace": ns, "message": "CronJob has no associated Jobs yet."})
+                continue
+
+            latest_job = sorted(associated_jobs, key=lambda x: x.status.start_time, reverse=True)[0]
+
+            # Check job's pods for any issues
+            pods = core_v1.list_namespaced_pod(ns, label_selector=f"job-name={latest_job.metadata.name}")
+
+            for pod in pods.items:
+                if pod.status.phase == 'Pending' and now - pod.status.start_time > time_to_next_run:
+                    issues.append({"cronjob_name": cronjob.metadata.name, "namespace": ns, "message": "CronJob's Pod is stuck in 'Pending' state."})
+                elif pod.status.phase not in ['Running', 'Succeeded']:
+                    issues.append({"cronjob_name": cronjob.metadata.name, "namespace": ns, "message": f"CronJob's Pod is in unexpected state: {pod.status.phase}."})
+
+    if not issues:
+        return (True, None)
+    else:
+        return (False, issues)
+
+
+
+