Portainer Edge Job - Schedule an Existing Container to Run

Question

Looking for a basic "how to schedule a container" via Portainer.. Tons of fantastic documentation, but nothing that really helps with how to perform a basic task like this.

I believe I can use Edge Jobs to schedule this, but if you have any other ideas I am open to this. Including building a container that performs my scheduling tasks if that is possible, run into problems with access to the docker Daemon :)

Would like to run my exiting container instead of deploying a new one.

Thanks in Advance!

1. Tried to run a docker container to schedule what I wanted but needed access to the docker daemon
2. Asked OpenAI
3. Tried searching for anything on the topic
4. Tried searching in the official documentation

Answer 1

Sooooo.. After a journey!! I have my solution.

I wanted more control than what Portainer was offering. I have a script that works, and due to my network implementation I have not yet got it working in a container but it will be possible.. Attached is my Python code.

Notes -

1. pip install httpie as it is a component you need. Should you miss this step, all your will get back is an error message around authentication and nothing else. So important step in addition here.. Verify working by simply open command prompt and typing 'http' and you will get some info back about Http install if correctly linked.

2.environment_id.. This one isn't easy to find. Simply keep counting up until you find yours as it seems to be based on how many attempts you made to connect to an environment.. Turns out It took me three goes

3.Provision an API key. Top right when signed into Portainer you can select "My a Account" in the drop down and simply create an Access token under "Access tokens" Give this code to "portainer_api_key" in my script under "# Portainer API details" at the top

4. Simply update # Portainer API details in the script with everything to do with your environment..

Happy coding!! Tones of other things you can do on API so might find this better than trying to work with Native Portainer management......

import requests
import subprocess
import time
import json
import schedule
import traceback

container_started = False

# Discord webhook URL
discord_webhook_url = "https://discordapp.com/api/webhooks/Something" #If you want monitoring

# Portainer API details
portainer_host = "X.X.X.X"
portainer_port = "9000" # Out of the box, but whatever the port is that you use to connect to the admin portal
portainer_api_key = "ptr_Something"

def start_container(container_id):
    print("made it to start container")
    environment_id = 3  # Replace with your actual environment ID
    start_container_cmd = f"http POST http://{portainer_host}:{portainer_port}/api/endpoints/{environment_id}/docker/containers/{container_id}/start X-API-Key:{portainer_api_key}"
    
    response = subprocess.run(start_container_cmd, shell=True, capture_output=True, text=True)
    
    if response.returncode == 0:
        print(f"Container started successfully: {container_id}")
    else:
        print(f"Failed to start container. Command return code: {response.returncode}")
        trigger_discord_alert(f"Failed to start container: {container_id}")

def get_existing_container_id(container_name):
    environment_id = 3  # Replace with your actual environment ID
    list_containers_cmd = f"http GET http://{portainer_host}:{portainer_port}/api/endpoints/{environment_id}/docker/containers/json X-API-Key:{portainer_api_key} all==true"
    response = subprocess.run(list_containers_cmd, shell=True, capture_output=True, text=True)
    
    if response.returncode == 0:
        try:
            containers = json.loads(response.stdout.strip())
            for container in containers:
                if container["Names"] and container["Names"][0] == f"/{container_name}":
                    return container["Id"]
        except json.JSONDecodeError as e:
            print(f"Error decoding JSON: {str(e)}")
    
    return None

def trigger_discord_alert(message):
    discord_data = {"content": message}
    requests.post(discord_webhook_url, json=discord_data)

def start_container_job(container_name):
    container_id = get_existing_container_id(container_name)
    print(f"Container ID for {container_name} =", container_id)
    if container_id:
        start_container(container_id)
    else:
        print(f"Container {container_name} does not exist.")
        trigger_discord_alert(f"Container {container_name} does not exist.")
        
def run_scheduler():
    global container_started
    try:
        # Run scheduler continuously
        while True:
            now = time.localtime()
            
            # First Container: Run every Wednesday at 11 am
            if now.tm_wday == 2 and now.tm_hour == 11 and now.tm_min == 0:
                if not container_started:
                    start_container("automated_Container_Name")
                    container_started = True
            
            # Second Container: Run every 3rd Sunday at 11 am
            elif now.tm_wday == 6 and now.tm_mday >= 15 and now.tm_mday <= 21 and now.tm_hour == 11 and now.tm_min == 0:
                if not container_started:
                    start_container("automated_Container_Name")
                    container_started = True
            
            # Reset flag on Wednesday
            elif now.tm_wday == 2 and now.tm_hour == 11 and now.tm_min == 0:
                container_started = False  

            time.sleep(60)  # Sleep for 1 minute before checking again

    except requests.exceptions.ConnectionError as conn_error:
        print(f"Connection error: {conn_error}")
        trigger_discord_alert("Connection error: Unable to reach Portainer host.")
        traceback.print_exc()

    except Exception as e:
        print(f"Error: {str(e)}")
        trigger_discord_alert(f"Script error: {str(e)}")
        traceback.print_exc()

if __name__ == "__main__":
    start_container("Test Container Name to verify working at start Up") # Just Testing
    run_scheduler()