Mass Update Stacks Portainer

Mass Update Stacks Portainer.py

import httpx
from dotenv import load_dotenv
import os
import sys
import threading
import time

load_dotenv()

PORTAINER_URL=os.getenv("PORTAINER_URL")
PORTAINER_USERNAME=os.getenv("PORTAINER_USERNAME")
PORTAINER_PASSWORD=os.getenv("PORTAINER_PASSWORD")

def authenticate():
    url = f"{PORTAINER_URL}/api/auth"
    data = {
        "username": PORTAINER_USERNAME,
        "password": PORTAINER_PASSWORD
    }
    response = httpx.post(url, json=data, verify=False)
    return response.json()["jwt"]

def getStackEnv(stack: int, token: str = None):
    if not token:
        token = authenticate()
    url = f"{PORTAINER_URL}/api/stacks/{stack}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = httpx.get(url, headers=headers, verify=False)
    data = response.json()

    env = data["Env"]
    return env

def getStackCompose(stack: int, token: str = None):
    if not token:
        token = authenticate()
    url = f"{PORTAINER_URL}/api/stacks/{stack}/file"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = httpx.get(url, headers=headers, verify=False)
    data = response.json()
    
    compose = data["StackFileContent"]
    return compose

def redeployStack(endpoint: int, stack: int, token: str = None):
    start_time = time.time()
    if not token:
        token = authenticate()
    url = f"{PORTAINER_URL}/api/stacks/{stack}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    data = {
        "env": getStackEnv(stack, token),
        "prune": True,
        "pullImage": True,
        "stackFileContent": getStackCompose(stack, token)
    }
    params = {
        "endpointId": endpoint
    }
    response = httpx.put(url, headers=headers, verify=False, json=data, params=params, timeout=None)
    if response.status_code == 200:
        print(f"Redeployed stack {stack} successfully in {int(time.time() - start_time)} seconds")
    else:
        print("Failed to redeploy stack")
        print(response.text)

def getAllStacks(token: str = None):
    if not token:
        token = authenticate()
    url = f"{PORTAINER_URL}/api/stacks"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = httpx.get(url, headers=headers, verify=False)
    data = response.json()
    return data

def deployFleetUpdate(type: str):
    token = authenticate()
    all_stacks = getAllStacks(token=token)

    threads = []

    for stack in all_stacks:
        if stack["Name"].startswith(type):
            print(f"Redeploying {type} fleet to endpoint {stack['EndpointId']} at stack {stack['Id']}...")
            thread = threading.Thread(target=redeployStack, args=(stack["EndpointId"], stack["Id"], token))
            threads.append(thread)

    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
        
if __name__ == "__main__":
    if len(sys.argv) < 1:
        print("Usage: python deployUpdate.py <fleet_type>")
        sys.exit(1)
    
    fleet_type = sys.argv[1]

    print(f"Deploying {fleet_type} fleet update...")

    deployFleetUpdate(fleet_type)
    sys.exit(0)

usage.md

Context

To handle our Docker fleets, we use Portainer. It's very good and other than a few small issues we haven't had a problem. As we have scaled up to over 50 servers across the world, all running Docker for our APIs, Webdav, FTP, and file serving servers, there are a lot of containers to handle. While you could say, "oh, just switch to Kubernetes, Docker swarms, or some other solution like Terraform!". No. That's a lot of work to deal with and rewriting our infrastructure is not something I feel like doing.

Okay, well then, just use the webhook redploys! Oh, really? Take a look at Portainer's pricing. Look at it. You really think I want to pay over $200 a month just so I can have basic CI/CD? Hell no. Each of our servers are over 80 cores, and some even go higher. We have over 50 nodes in our fleet. We would have to contact sales, and have them charge us an arm and a leg for a basic ass feature.

So no, I will not be doing that. Portainer gives you their API and it's pretty simple to use, so after about a half our, I concoted this fleet redeployment, which uses Python to update the stacks of our choosing. Every update, I find myself sitting around for up to an hour reploying all of our containers, and it is a waste of time, so I built this.

To Make This Work

The most important thing is that <fleet_type> is the same name across all of your stacks across all of your endpoints. For example, if the stack name is "api" then <fleet_type> is "api". It will find all of your stacks with that name and redeploy them exactly as they are, but pulling the new images, which is all I wanted. Make sure that the stacks all are of the same name.

Then you just need the username and password of your main portainer instance (the host where the URL goes to), and you are off! Save yourself the $200+ they want for this basic feature, and be happy.