Building a Slack Bot with Python and Flask for Kubernetes Management

Introduction

Slack bots are becoming an integral part of modern DevOps due to their ability to automate tasks and streamline operations. In this comprehensive guide, we’ll walk you through creating a Slack bot that interacts with a Kubernetes cluster. We’ll use Python and Flask for the backend service.

Slack bots have become a staple in modern DevOps and SRE tooling. They’re not just gimmicks; they simplify life, especially in a distributed, fast-paced working environment. Today, I’ll take you through creating a Slack bot that integrates with Kubernetes (k8s) to run commands like get pods or describe nodes via Slack messages.

Here’s a step-by-step guide to create such a bot using Python, Flask, and the Slack API. The bot will listen to mentions, present a dropdown menu for selecting commands, and call a Python-based backend service to execute Kubernetes operations using kubectl.

Below App is a Baseline where you can start from and build on it.

Prerequisites

1. Python (>=3.6)
2. Flask
3. Slack workspace and a Slack bot
4. Kubernetes cluster
5. SlackEventAdapter, WebClient (from the Slack SDK)
6. ngrok installed

Setting Up Your Slack Workspace

Step 1: Create a Slack Account and Workspace

If you don’t already have a Slack account, head over to Slack’s website to create one. Once you have an account, create a new workspace or use an existing one.

Step 2: Create a Slack App

Navigate to the Slack API and click on “Create New App.”

Name your app and associate it with your workspace.

Navigate to “OAuth & Permissions” and add the chat:write, commands, and users:read scopes under “Bot Token Scopes.”

Click “Install App to Workspace” and authorize the app.

Environment Variables

Make sure to set the following environment variables:

• SLACK_SIGNING_SECRET
• SLACK_BOT_TOKEN
• VERIFICATION_TOKEN

You can obtain these tokens from your Slack bot’s dashboard.

You can find the Bot user Token in OAuth & Permissions page

Backend Service with Python and Flask

Setting Up Your Python Environment

Create a virtual environment and install the necessary packages:

python -m virtualenv env
source env/bin/activate
pip install Flask slack_sdk slackeventsapi aiohttp

Code Walkthrough

Importing Libraries

from flask import Flask, Response, request, jsonify
from slackeventsapi import SlackEventAdapter
import os
import subprocess
import json
from threading import Thread
from slack import WebClient

• Flask: For creating the web service.
• SlackEventAdapter: To handle events from Slack.
• os, subprocess, json: For environment variables, shell commands, and JSON handling.
• Thread: For asynchronous tasks.
• WebClient: Slack’s Python client for API calls.

Initializing Variables and Configurations

Creating an Instance

app = Flask(__name__)

Initializes an instance of the Flask class.

After this line, the app variable becomes a Flask application object, which can now be used to define routes, error handlers, and other elements of the web application.

Variables

SLACK_SIGNING_SECRET = os.environ['SLACK_SIGNING_SECRET']
slack_token = os.environ['SLACK_BOT_TOKEN']
VERIFICATION_TOKEN = os.environ['VERIFICATION_TOKEN']

Creates an instance of the Slack WebClient class, initialized with a Slack bot token.

slack_client = WebClient(slack_token)

Initializes a new instance of the SlackEventAdapter class. This is part of Slack's Python SDK and is used to handle event payloads sent by Slack to your application.

slack_events_adapter = SlackEventAdapter(
    SLACK_SIGNING_SECRET, "/slack/events", app
)

Available Kubernetes Commands

Here we specify available kubectl commands and sub-commands.

available_commands = ["get", "describe", "logs"]
available_sub_commands = {
    "get": ["pods", "nodes", "services"],
    "describe": ["pods", "nodes", "services"],
    "logs": ["pods"]
}

Utility Functions

We need some utility functions to get available namespaces and pods and to run kubectl commands.

• get_available_namespaces(): Fetches available Kubernetes namespaces.
• get_available_pods(namespace): Fetches available pods in a given namespace.
• run_kubectl_command(channel_id, command): Executes a kubectl command and posts the output to Slack.

selected_actions = {}

def get_available_namespaces():
    try:
        command = ["kubectl", "get", "namespaces", "-o", "jsonpath='{.items[*].metadata.name}'"]
        result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True)
        namespaces = result.stdout.strip("'").split()
        return namespaces
    except subprocess.CalledProcessError as e:
        print("Error running kubectl command:", e)
        return []

def get_available_pods(namespace):
    try:
        command = ["kubectl", "get", "pods", "-n", namespace, "-o", "jsonpath='{.items[*].metadata.name}'"]
        result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=True)
        pods = result.stdout.strip("'").split()
        return pods
    except subprocess.CalledProcessError as e:
        print("Error running kubectl command:", e)
        return []

def run_kubectl_command(channel_id, command):
    try:
        print(f"Running command: {command}")  
        output = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, text=True)
        slack_client.chat_postMessage(channel=channel_id, text=f"```\n{output}\n```")
    except subprocess.CalledProcessError as e:
        slack_client.chat_postMessage(channel=channel_id, text=f"Error executing command:\n```\n{e.output}\n```")

Event Listener for Bot Mentions

We will listen for events where the bot is mentioned and then show a dropdown for available commands.

@slack_events_adapter.on("app_mention")
def handle_mention(event_data):
    def send_kubectl_options(value):
        event_data = value
        message = event_data["event"]
        if message.get("subtype") is None:
            channel_id = message["channel"]
            user_id = message["user"]

            response_message = {
                "blocks": [
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": f"Hello <@{user_id}>! Please select a kubectl command:"
                        }
                    },
                    {
                        "type": "actions",
                        "elements": [
                            {
                                "type": "static_select",
                                "placeholder": {
                                    "type": "plain_text",
                                    "text": "Select a command"
                                },
                                "options": [
                                    {
                                        "text": {
                                            "type": "plain_text",
                                            "text": command
                                        },
                                        "value": command
                                    }
                                    for command in available_commands
                                ],
                                "action_id": "kubectl_command_select"
                            }
                        ]
                    }
                ]
            }

            slack_client.chat_postMessage(channel=channel_id, blocks=response_message["blocks"])

    thread = Thread(target=send_kubectl_options, kwargs={"value": event_data})
    thread.start()
    return Response(status=200)

This function is triggered when the bot is mentioned in Slack. It sends a message with a dropdown menu to select a kubectl command.

Handling Interactions

Here we will handle different interactions after the user selects options from the dropdowns.

@app.route("/interactions", methods=["POST"])
def handle_interactions():
    payload = json.loads(request.form.get("payload"))
    channel_id = payload["channel"]["id"]
    user_id = payload["user"]["id"]
    action_id = payload["actions"][0]["action_id"]

    if action_id == "kubectl_command_select":
        selected_command = payload["actions"][0]["selected_option"]["value"]
        selected_actions[channel_id] = {"command": selected_command}
        sub_command_menu = {
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": "Please select a sub-command:"
                    }
                },
                {
                    "type": "actions",
                    "elements": [
                        {
                            "type": "static_select",
                            "placeholder": {
                                "type": "plain_text",
                                "text": "Select a sub-command"
                            },
                            "options": [
                                {
                                    "text": {
                                        "type": "plain_text",
                                        "text": sub_command
                                    },
                                    "value": sub_command
                                }
                                for sub_command in available_sub_commands.get(selected_command, [])
                            ],
                            "action_id": "kubectl_sub_command_select"
                        }
                    ]
                }
            ]
        }
        slack_client.chat_postMessage(channel=channel_id, blocks=sub_command_menu["blocks"])

    elif action_id == "kubectl_sub_command_select":
        selected_sub_command = payload["actions"][0]["selected_option"]["value"]
        if channel_id in selected_actions:
            selected_actions[channel_id]["sub_command"] = selected_sub_command
        available_namespaces = get_available_namespaces()
        namespaces_menu = {
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": "Please select a namespace:"
                    }
                },
                {
                    "type": "actions",
                    "elements": [
                        {
                            "type": "static_select",
                            "placeholder": {
                                "type": "plain_text",
                                "text": "Select a namespace"
                            },
                            "options": [
                                {
                                    "text": {
                                        "type": "plain_text",
                                        "text": namespace
                                    },
                                    "value": namespace
                                }
                                for namespace in available_namespaces
                            ],
                            "action_id": "kubectl_namespace_select"
                        }
                    ]
                }
            ]
        }
        slack_client.chat_postMessage(channel=channel_id, blocks=namespaces_menu["blocks"])

    elif action_id == "kubectl_namespace_select":
        selected_namespace = payload["actions"][0]["selected_option"]["value"]
        if channel_id in selected_actions:
            selected_actions[channel_id]["namespace"] = selected_namespace

        selected_command = selected_actions.get(channel_id, {}).get("command", "get")
        selected_sub_command = selected_actions.get(channel_id, {}).get("sub_command", "nodes")

        if selected_sub_command == "pods" and selected_command in ["describe", "logs"]:
            available_pods = get_available_pods(selected_namespace)
            pods_menu = {
                "blocks": [
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": "Please select a pod:"
                        }
                    },
                    {
                        "type": "actions",
                        "elements": [
                            {
                                "type": "static_select",
                                "placeholder": {
                                    "type": "plain_text",
                                    "text": "Select a pod"
                                },
                                "options": [
                                    {
                                        "text": {
                                            "type": "plain_text",
                                            "text": pod
                                        },
                                        "value": pod
                                    }
                                    for pod in available_pods
                                ],
                                "action_id": "kubectl_pod_select"
                            }
                        ]
                    }
                ]
            }
            slack_client.chat_postMessage(channel=channel_id, blocks=pods_menu["blocks"])
        else:
            command = f"kubectl {selected_command} {selected_sub_command} -n {selected_namespace}"
            run_kubectl_command(channel_id, command)

    elif action_id == "kubectl_pod_select":
        selected_pod = payload["actions"][0]["selected_option"]["value"]
        selected_namespace = selected_actions.get(channel_id, {}).get("namespace", "")
        selected_command = selected_actions.get(channel_id, {}).get("command", "describe")

        if selected_namespace:  # Ensure namespace is not empty
            if selected_command in ["logs", "describe"]:
                command = f"kubectl {selected_command} {selected_pod} -n {selected_namespace}"  # Removed the word "pod"
            else:
                command = f"kubectl {selected_command} pod {selected_pod} -n {selected_namespace}"
            run_kubectl_command(channel_id, command)
        else:
            slack_client.chat_postMessage(channel=channel_id, text="Namespace not selected. Please start over.")

    return Response(status=200)

This function handles the user’s selections and executes the corresponding kubectl commands.

LAST Piece

serves as the entry point for running the Flask application when the script is executed directly.

if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=3000)

Running the App

Finally, run your Flask app:

FLASK_APP=app.py FLASK_ENV=development flask run --debug --host=0.0.0.0 --port=3000

you should get a running app:

* Debug mode: on
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
 * Running on all addresses (0.0.0.0)
 * Running on <http://127.0.0.1:3000>
 * Running on <http://192.168.1.3:3000>
Press CTRL+C to quit
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 144-626-857

Exposing the Local Server Using ngrok

Normally, your local Flask application is only accessible on your local network. However, Slack needs a public URL to send events to your application. ngrok can expose your local server to the internet.

1. Installation: If you haven’t already, download and install ngrok from ngrok's website.
2. Run ngrok: Open a new terminal window and run:

ngrok http 3000

This exposes port 3000 to the internet, which is the port our Flask app is running on.

1. Public URL: ngrok will provide you with a public URL (e.g., https://abcd1234.ngrok.io). Use this URL to configure Slack's Event Subscription settings.
2. Event Subscriptions: In your Slack App’s dashboard, navigate to the “Event Subscriptions” page and set the “Request URL” to your ngrok URL appended with /slack/events (e.g., https://abcd1234.ngrok.io/slack/events).
3. Also check Interactivity & Shortcuts as below

Restart ngrok: If you stop your ngrok session, you'll get a new URL the next time you start it. You will need to update the Slack Event Subscriptions page with the new URL.

By following these steps, you should now have a working Slack bot capable of running kubectl commands via a Flask backend, exposed securely to the internet through ngrok.

Testing Your Slack Bot

1. Mention your Slack bot in a message.
2. You should see a dropdown with Kubernetes commands.
3. After making a selection, you’ll get another dropdown, possibly asking for more details like namespaces or pods.
4. The bot will then execute the Kubernetes command and return the output.

Screens

kubectl get pods -n argocd

kubectl logs argocd-application-controller-0 -n argocd

While using the slack bot and communicate with your Python Service you can see the logs in the Terminal

Running command: kubectl get pods -n argocd
127.0.0.1 - - [01/Sep/2023 02:40:26] "POST /interactions HTTP/1.1" 200 -
127.0.0.1 - - [01/Sep/2023 02:40:28] "POST /slack/events HTTP/1.1" 200 -

Running command: kubectl logs argocd-application-controller-0 -n argocd
127.0.0.1 - - [01/Sep/2023 02:45:46] "POST /interactions HTTP/1.1" 200 -
127.0.0.1 - - [01/Sep/2023 02:45:47] "POST /slack/events HTTP/1.1" 200 -

I am printing the structured command in the stdout to be used in troubleshooting

Github Repository

You can find the Full Python and Flask Code Here .

Features and Types of Slack Bot

• Interactive Dropdowns: For selecting commands and namespaces.
• Real-time Feedback: Provides real-time output from Kubernetes.
• Asynchronous Execution: Uses Python threading for non-blocking operations.

Benefits for Tech Teams

1. Automation: Automates the process of querying Kubernetes clusters.
2. Quick Access: Provides quick access to cluster information directly from Slack.
3. Collaboration: Team members can see the output and understand the cluster state without leaving Slack.

Conclusion

Building a Slack bot with Python and Flask offers a robust way to integrate Kubernetes management directly into your Slack workspace. It not only simplifies tasks but also enhances team collaboration. With this guide, you should have all the information you need to build your own.