VPN automation and load balancing using the hosting API
In this post I will tell you how to manage such a service – and automate some points.
Content:
-
We count clients
-
We count the average number of clients and notify the admin
-
Create a VDS clone
TimeWeb API or how to automate most systems
To automate the process, we need that when the high average online is reached within 3-4 hours, the system independently rents a new VDS, making a clone from the previous one.
First we need to count how many people are online now
We import the subprocess module, which allows you to execute commands in the terminal from a Python script.
2. Execute the command wg show by using subprocess.run()capturing the output of the command with a parameter capture_output=True and converting it to a string using the parameter text=True.
3. Check the return code of the command. If it is zero, then the command was successfully executed and we can process the output. If the return code is non-zero, then an error has occurred and we print an error message.
4. Break the output into lines using the method split('\n').
5. Count the number of lines in the output (number of clients), subtracting 1 for the header line.
6. Output the result as a string using the function print().
import subprocess
result = subprocess.run(['wg', 'show'], capture_output=True, text=True)
if result.returncode == 0:
output = result.stdout.strip().split('\n')
clients = len(output) - 1 # subtract 1 for the header line
print(f"Number of connected clients: {clients}")
else:
print("Error:", result.stderr)
Next, we need to track in order to understand the average number of clients for 12 hours and then decide whether to rent a vds clone or not.
1. We define variables avg_clients, count And max_count. avg_clients will be used to store the average number of clients, count – to store the number of checks, and max_count is the maximum number of checks we want to perform.
2. We start an endless loop while count < max_countin which we will check the number of connected clients and increase count.
3. We execute the command wg showas in the previous script, and process the output.
4. We add the number of customers to avg_clients and increase count by 1.
5. We display the number of connected clients and wait 10 minutes using the function time.sleep(600).
6. After doing all the checks, we calculate the average number of customers by dividing avg_clients on count.
7. If the average number of clients is greater than or equal to 75, we print “Server is overloaded”, otherwise we print “Server is running fine”.
Script code that displays the number of connected Wireguard clients and checks it for 12 hours:
import subprocess
import time
avg_clients = 0
count = 0
max_count = 72 # 12 hours with a check every 10 minutes
while count < max_count:
result = subprocess.run(['wg', 'show'], capture_output=True, text=True)
if result.returncode == 0:
output = result.stdout.strip().split('\n')
clients = len(output) - 1 # subtract 1 for the header line
avg_clients += clients
count += 1
print(f"Number of connected clients: {clients}")
time.sleep(600) # wait for 10 minutes
else:
print("Error:", result.stderr)
avg_clients //= count
if avg_clients >= 75:
print("Server is overloaded")
else:
print("Server is running fine")
Script code that prints the number of connected Wireguard clients in 12 hours and checks if the average is above 75:
python
import subprocess
import time
clients = []
for i in range(0, 144): # 12 hours * 12 updates per hour = 144 updates
result = subprocess.run(['wg', 'show'], capture_output=True, text=True)
if result.returncode == 0:
output = result.stdout.strip().split('\n')
clients.append(len(output) - 1) # subtract 1 for the header line
else:
print("Error:", result.stderr)
clients.append(0) # add 0 to the list to maintain the length
time.sleep(300) # wait 5 minutes between updates
average_clients = sum(clients) / len(clients)
if average_clients > 75:
print("Server is overloaded")
else:
print(f"Number of connected clients: {clients[-1]}")
```1. Create an empty `clients` list, to which we will add the number of connected clients on each pass of the loop.
2. Use a `for` loop to perform 144 updates (12 hours * 12 updates per hour).
3. We execute the `wg show` command and process the output in the same way as the previous script.
4. Add the number of connected clients to the `clients` list.
5. If an error occurs, add 0 to the `clients` list to keep the length of the list.
6. Use the `time.sleep()` function to wait 5 minutes between updates.
7. We calculate the average value of the number of clients in the list `clients`.
8. If the average value is greater than 75, display the message “Server is overloaded”.
9. Otherwise print the last value of the `clients` list as a string using the `print()` function.
All the same, I lied a little, not everything is automated, an alert comes, it’s more convenient for me in Telegram that the server is full, and the Administrator decides whether we have enough money for a new server or temporarily close the distribution of keys to WG
To send an alert to telegram, we will use the library python-telegram-bot. Before you can start working with this library, you need to create a bot and get a token. To do this, you need to contact BotFather in Telegram and follow the instructions.
After receiving the token, you can write a script for sending a message to telegram:
import telegram
bot_token = 'YOUR_BOT_TOKEN_HERE'
chat_id = 'YOUR_CHAT_ID_HERE'
bot = telegram.Bot(token=bot_token)
message = "Server is overloaded"
bot.send_message(chat_id=chat_id, text=message)
Now you can combine this code with the code of the previous script to send messages to telegram if the server is overloaded:
import subprocess
import time
import telegram
bot_token = 'YOUR_BOT_TOKEN_HERE'
chat_id = 'YOUR_CHAT_ID_HERE'
clients = []
for i in range(0, 144): # 12 hours * 12 updates per hour = 144 updates
result = subprocess.run(['wg', 'show'], capture_output=True, text=True)
if result.returncode == 0:
output = result.stdout.strip().split('\n')
clients.append(len(output) - 1) # subtract 1 for the header line
else:
print("Error:", result.stderr)
clients.append(0) # add 0 to the list to maintain the length
time.sleep(300) # wait 5 minutes between updates
average_clients = sum(clients) / len(clients)
if average_clients > 75:
message = "Server is overloaded"
bot = telegram.Bot(token=bot_token)
bot.send_message(chat_id=chat_id, text=message)
else:
print(f"Number of connected clients: {clients[-1]}")
The code for sending a message to telegram has been added to the condition block, which checks if the average value is greater than 75. If it is, then the message “Server is overloaded” is sent to telegram. If not, then the number of connected clients is displayed, as before.
“Server is overloaded” – we need to clone the VPS and raise it
Connect to timeweb api (api key in your account)
Cloning a VPS on Timeweb involves several steps:
1. Get a list of available VPS.
2. Select the VPS you want to clone.
3. Create a copy of the VPS.
4. Wait until the copy is created.
5. Get VPS copy data.
6. Save the data to the Wireguard configuration file.
Below I have written a Python code that performs these steps:
import requests
import json
# Задайте свой ключ API
api_key = "YOUR_API_KEY_HERE"
# 1. Получите список доступных VPS.
url = "https://api.timeweb.com/v1/vps"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
response = requests.get(url, headers=headers)
vps_list = response.json()["data"]
print("Available VPS:", vps_list)
# 2. Выберите VPS, которую нужно клонировать.
vps_to_clone = None
for v in vps_list:
if v["hostname"] == "VPS_TO_CLONE":
vps_to_clone = v
break
if vps_to_clone is None:
print("VPS to clone not found.")
exit(1)
print("VPS to clone:", vps_to_clone)
# 3. Создайте копию VPS.
url = "https://api.timeweb.com/v1/vps/clone"
data = {
"vps_id": vps_to_clone["id"]
}
response = requests.post(url, headers=headers, data=json.dumps(data))
clone_id = response.json()["data"]["id"]
print("Clone created:", clone_id)
# 4. Дождитесь, пока копия будет создана.
url = f"https://api.timeweb.com/v1/vps/clone/{clone_id}"
while True:
response = requests.get(url, headers=headers)
status = response.json()["data"]["status"]
if status == "completed":
print("Clone processed.")
break
elif status == "failed":
print("Clone failed.")
exit(1)
print("Clone status:", status)
time.sleep(10)
# 5. Получите данные копии VPS.
url = f"https://api.timeweb.com/v1/vps/{clone_id}"
response = requests.get(url, headers=headers)
clone_vps = response.json()["data"]
print("Clone details:", clone_vps)
# 6. Сохраните данные в файл конфигурации Wireguard.
with open("wg0.conf", "a") as f:
f.write(f"[Peer]\n")
f.write(f"PublicKey = {clone_vps['wireguard_public_key']}\n")
f.write(f"AllowedIPs = {clone_vps['wireguard_allowed_ips']}\n")
f.write(f"Endpoint = {clone_vps['wireguard_endpoint']}\n")
What did we get
The script receives data on the number of clients on the server, understands that the average number exceeds the limit, notifies the administrator, the administrator sends a request to clone the server, a clone of the server is created, it remains only to configure the bot that issues keys – issuing to a new server.
You also need to understand that some servers can be empty, this has already happened during the year, we noticed that for some reason people take new keys, clients remove wireguard or simply do not pour vpn, in this case, you need to fill up such servers, but so far there are no servers a lot we just reconfigure the bot to issue keys to a specific server for pre-population.
Outcome: this is a very artisanal method of administration, we do not receive profit from this service, but in the future we plan to create a mini admin panel where you can track the parameters of the load on the CPU, network.
This article is not an advertisement, we just made a good project and are ready to share what we have done, you can use our experience to organize your vpn service. Work for the soul!
We are always happy, you can test our project in the telegram channel https://t.me/blacktemple_space
