X workflow4 robusta actions

xWorkflows/xWorkflow4/daemonSetCpu.py

@@ -0,0 +1,84 @@
+import logging
+from robusta.api import (
+    Finding,
+    FindingSource,
+    FindingType,
+    MarkdownBlock,
+    DaemonSetEvent,
+    action,
+    ActionParams,
+)
+from kubernetes import client, config
+
+class resizeParams(ActionParams):
+    updateCpuRequest: str
+
+@action
+def daemonSetCpu(event: DaemonSetEvent, params: resizeParams):
+    """
+    Change the CPU requests of a DaemonSet
+    """
+    if not event.get_daemonset():
+        # Log an error message if the DaemonSet is not found
+        logging.error("Failed to get the DaemonSet for CPU adjustment")
+        return
+
+    daemonSet = event.get_daemonset()
+    daemonSetName = daemonSet.metadata.name
+    daemonSetNamespace = daemonSet.metadata.namespace
+
+    # Load the kube config
+    config.load_incluster_config()
+
+    # Create a Kubernetes API client
+    apps_v1 = client.AppsV1Api()
+
+    try:
+        # Read the existing DaemonSet specification
+        daemonSetSpec = apps_v1.read_namespaced_daemon_set(name=daemonSetName, namespace=daemonSetNamespace)
+
+        # Update the CPU requests in the DaemonSet spec
+        for container in daemonSetSpec.spec.template.spec.containers:
+            container.resources.requests['cpu'] = params.updateCpuRequest
+
+        # Update the DaemonSet with the modified spec
+        apps_v1.patch_namespaced_daemon_set(
+            name=daemonSetName,
+            namespace=daemonSetNamespace,
+            body={"spec": daemonSetSpec.spec}
+        )
+
+        logging.info(f"DaemonSet {daemonSetName} in namespace {daemonSetNamespace} has been updated with new CPU requests.")
+
+        functionName = "Change CPU requests of DaemonSet"
+        finding = Finding(
+            title="DaemonSet CPU upgradation",
+            source=FindingSource.MANUAL,
+            aggregation_key=functionName,
+            finding_type=FindingType.REPORT,
+            failure=False,
+        )
+        print(params)
+        finding.add_enrichment(
+            [
+                MarkdownBlock(f"{daemonSetName} CPU is Upgraded."),
+            ]
+        )
+    except client.exceptions.ApiException as e:
+        logging.error(f"Exception when modifying DaemonSet: {e}")
+        functionName = "Change CPU requests of DaemonSet Failed"
+        finding = Finding(
+            title="DaemonSet CPU upgradation Failed",
+            source=FindingSource.MANUAL,
+            aggregation_key=functionName,
+            finding_type=FindingType.REPORT,
+            failure=False,
+        )
+        print(params)
+        finding.add_enrichment(
+            [
+                MarkdownBlock(f"{daemonSetName} CPU Upgradation Failed."),
+            ]
+        )
+
+    event.add_finding(finding)

xWorkflows/xWorkflow4/daemonSetMemory.py

@@ -0,0 +1,85 @@
+import logging
+
+from robusta.api import (
+    Finding,
+    FindingSource,
+    FindingType,
+    MarkdownBlock,
+    DaemonSetEvent,
+    action,
+    ActionParams,
+)
+from kubernetes import client, config
+
+class resizeParams(ActionParams):
+    updateMemoryRequest: str
+
+@action
+def daemonSetMemory(event: DaemonSetEvent, params: resizeParams):
+    """
+    Change the Memory requests of a DaemonSet
+    """
+    if not event.get_daemonset():
+        # Log an error message if the DaemonSet is not found
+        logging.error("Failed to get the DaemonSet for Memory adjustment")
+        return
+
+    daemonSet = event.get_daemonset()
+    daemonSetName = daemonSet.metadata.name
+    daemonSetNamespace = daemonSet.metadata.namespace
+
+    # Load the kube config
+    config.load_incluster_config()
+
+    # Create a Kubernetes API client
+    apps_v1 = client.AppsV1Api()
+
+    try:
+        # Read the existing DaemonSet specification
+        daemonSetSpec = apps_v1.read_namespaced_daemon_set(name=daemonSetName, namespace=daemonSetNamespace)
+
+        # Update the Memory requests in the DaemonSet spec
+        for container in daemonSetSpec.spec.template.spec.containers:
+            container.resources.requests['memory'] = params.updateMemoryRequest
+
+        # Update the DaemonSet with the modified spec
+        apps_v1.patch_namespaced_daemon_set(
+            name=daemonSetName,
+            namespace=daemonSetNamespace,
+            body={"spec": daemonSetSpec.spec}
+        )
+
+        logging.info(f"DaemonSet {daemonSetName} in namespace {daemonSetNamespace} has been updated with new Memory requests.")
+
+        functionName = "Change Memory requests of DaemonSet"
+        finding = Finding(
+            title="DaemonSet Memory upgradation",
+            source=FindingSource.MANUAL,
+            aggregation_key=functionName,
+            finding_type=FindingType.REPORT,
+            failure=False,
+        )
+        print(params)
+        finding.add_enrichment(
+            [
+                MarkdownBlock(f"{daemonSetName} Memory is Upgraded."),
+            ]
+        )
+    except client.exceptions.ApiException as e:
+        logging.error(f"Exception when modifying DaemonSet: {e}")
+        functionName = "Change Memory requests of DaemonSet Failed"
+        finding = Finding(
+            title="DaemonSet Memory upgradation Failed",
+            source=FindingSource.MANUAL,
+            aggregation_key=functionName,
+            finding_type=FindingType.REPORT,
+            failure=True,
+        )
+        print(params)
+        finding.add_enrichment(
+            [
+                MarkdownBlock(f"{daemonSetName} Memory Upgradation Failed."),
+            ]
+        )
+
+    event.add_finding(finding)

xWorkflows/xWorkflow4/deploymentCpu.py

@@ -0,0 +1,70 @@
+import logging
+import time
+from robusta.api import (
+    Finding,
+    FindingSource,
+    FindingType,
+    MarkdownBlock,
+    DeploymentEvent,
+    action,
+    ActionParams,
+)
+from kubernetes import client, config
+
+class resizeParams(ActionParams):
+    updateCpuRequest: str
+
+@action
+def deploymentCpu(event: DeploymentEvent, params: resizeParams):
+    """
+    Change the CPU requests of a Deployment
+    """
+    if not event.get_deployment():
+        # Log an error message if the deployment is not found
+        logging.error("Failed to get the deployment for CPU adjustment")
+        return
+
+    deployment = event.get_deployment()
+    deploymentName = deployment.metadata.name
+    deploymentNamespace = deployment.metadata.namespace
+
+    # Load the kube config
+    config.load_incluster_config()
+
+    # Create a Kubernetes API client
+    apps_v1 = client.AppsV1Api()
+
+    try:
+        # Read the existing Deployment specification
+        deploymentSpec = apps_v1.read_namespaced_deployment(name=deploymentName, namespace=deploymentNamespace)
+
+        # Update the CPU requests in the Deployment spec
+        for container in deploymentSpec.spec.template.spec.containers:
+            container.resources.requests['cpu'] = params.updateCpuRequest
+
+        # Update the Deployment with the modified spec
+        apps_v1.patch_namespaced_deployment(
+            name=deploymentName,
+            namespace=deploymentNamespace,
+            body={"spec": deploymentSpec.spec}
+        )
+
+        logging.info(f"Deployment {deploymentName} in namespace {deploymentNamespace} has been updated with new CPU requests.")
+    except client.exceptions.ApiException as e:
+        logging.error(f"Exception when modifying deployment: {e}")
+
+    functionName = "Change CPU requests of Deployment"
+    finding = Finding(
+        title="Deployment CPU upgradation",
+        source=FindingSource.MANUAL,
+        aggregation_key=functionName,
+        finding_type=FindingType.REPORT,
+        failure=False,
+    )
+    print(params)
+    finding.add_enrichment(
+        [
+            MarkdownBlock(f"{deploymentName} CPU is Upgraded."),
+        ]
+    )
+    event.add_finding(finding)

xWorkflows/xWorkflow4/deploymentMemory.py

@@ -0,0 +1,70 @@
+import logging
+import time
+from robusta.api import (
+    Finding,
+    FindingSource,
+    FindingType,
+    MarkdownBlock,
+    DeploymentEvent,
+    action,
+    ActionParams,
+)
+from kubernetes import client, config
+
+class resizeParams(ActionParams):
+    updateMemoryRequest: str
+
+@action
+def deploymentMemory(event: DeploymentEvent, params: resizeParams):
+    """
+    Change the Memory requests of a Deployment
+    """
+    if not event.get_deployment():
+        # Log an error message if the deployment is not found
+        logging.error("Failed to get the deployment for Memory adjustment")
+        return
+
+    deployment = event.get_deployment()
+    deploymentName = deployment.metadata.name
+    deploymentNamespace = deployment.metadata.namespace
+
+    # Load the kube config
+    config.load_incluster_config()
+
+    # Create a Kubernetes API client
+    apps_v1 = client.AppsV1Api()
+
+    try:
+        # Read the existing Deployment specification
+        deploymentSpec = apps_v1.read_namespaced_deployment(name=deploymentName, namespace=deploymentNamespace)
+
+        # Update the Memory requests in the Deployment spec
+        for container in deploymentSpec.spec.template.spec.containers:
+            container.resources.requests['memory'] = params.updateMemoryRequest
+
+        # Update the Deployment with the modified spec
+        apps_v1.patch_namespaced_deployment(
+            name=deploymentName,
+            namespace=deploymentNamespace,
+            body={"spec": deploymentSpec.spec}
+        )
+
+        logging.info(f"Deployment {deploymentName} in namespace {deploymentNamespace} has been updated with new Memory requests.")
+    except client.exceptions.ApiException as e:
+        logging.error(f"Exception when modifying deployment: {e}")
+
+    functionName = "Change Memory requests of Deployment"
+    finding = Finding(
+        title="Deployment Memory upgradation",
+        source=FindingSource.MANUAL,
+        aggregation_key=functionName,
+        finding_type=FindingType.REPORT,
+        failure=False,
+    )
+    print(params)
+    finding.add_enrichment(
+        [
+            MarkdownBlock(f"{deploymentName} Memory is Upgraded."),
+        ]
+    )
+    event.add_finding(finding)