155 lines
4.6 KiB
Python
155 lines
4.6 KiB
Python
from dotenv import load_dotenv
|
|
from flask import Flask, render_template, request, session, redirect, url_for, abort
|
|
from flask_apscheduler import APScheduler
|
|
from flask_migrate import Migrate
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
import docker
|
|
import pyjokes
|
|
import time
|
|
import os
|
|
from datetime import timedelta
|
|
#https://www.youtube.com/watch?v=1nxzOrLWiic
|
|
# DEBUG
|
|
from pprint import pprint
|
|
|
|
|
|
def calculate_cpu_percent(stats, status):
|
|
if (status != "running"):
|
|
return 0
|
|
|
|
cpu_count = len(stats["cpu_stats"]["cpu_usage"]["percpu_usage"])
|
|
cpu_percent = 0.0
|
|
cpu_delta = float(stats["cpu_stats"]["cpu_usage"]["total_usage"]) - \
|
|
float(stats["precpu_stats"]["cpu_usage"]["total_usage"])
|
|
system_delta = float(stats["cpu_stats"]["system_cpu_usage"]) - \
|
|
float(stats["precpu_stats"]["system_cpu_usage"])
|
|
|
|
if system_delta > 0.0:
|
|
cpu_percent = cpu_delta / system_delta * 100.0 * cpu_count
|
|
|
|
return round(cpu_percent, 2)
|
|
|
|
|
|
def calculate_ram_percent(stats, status):
|
|
if (status != "running"):
|
|
return 0
|
|
|
|
mem_used = stats["memory_stats"]["usage"] - stats["memory_stats"]["stats"]["cache"] + \
|
|
stats["memory_stats"]["stats"]["active_file"]
|
|
limit = stats['memory_stats']['limit']
|
|
|
|
return round(mem_used / limit * 100, 2)
|
|
|
|
|
|
load_dotenv()
|
|
basedir = os.path.abspath(os.path.dirname(__file__))
|
|
app = Flask(__name__)
|
|
#SESSION
|
|
app.secret_key = "sdasdsadasdasdasdadadaasd53563da"
|
|
app.permanent_session_lifetime = timedelta(days=1)
|
|
|
|
#DATABASE
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///'+ os.path.join(basedir, 'app.db')
|
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
|
|
|
db = SQLAlchemy(app)
|
|
|
|
class users(db.Model):
|
|
_id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(255))
|
|
password = db.Column(db.String(255))
|
|
|
|
def __init__(self, name, password):
|
|
self.name = name
|
|
self.password = password
|
|
|
|
|
|
scheduler = APScheduler()
|
|
client = docker.from_env()
|
|
# migrate = Migrate(app, db)
|
|
|
|
# login_manager.init_app(app)
|
|
# https://flask-login.readthedocs.io/en/latest/
|
|
|
|
|
|
@app.route("/login", methods=["POST", "GET"])
|
|
def login():
|
|
if (request.method == "POST"):
|
|
user = request.form["email"]
|
|
|
|
#TODO: Validation
|
|
session.permanent = True
|
|
session['user'] = user
|
|
return redirect(url_for("index"))
|
|
else:
|
|
if "user" in session:
|
|
return redirect(url_for("index"))
|
|
return render_template("login.html")
|
|
|
|
|
|
@app.route("/logout", methods=['GET', 'POST'])
|
|
def logout():
|
|
session.pop("user", None)
|
|
return redirect(url_for("login"))
|
|
|
|
|
|
@app.route("/", methods=['GET'])
|
|
def index():
|
|
if "user" not in session:
|
|
return redirect(url_for("login"))
|
|
|
|
containers = client.containers.list(all=True)
|
|
punchline = pyjokes.get_joke(language="en", category="neutral")
|
|
return render_template("index.html", title="test", subtitle=punchline, containers_len=len(containers), containers=containers)
|
|
|
|
|
|
@app.route("/parts/modal/container/<container_id>", methods=['GET'])
|
|
def parts_container_modal(container_id):
|
|
if "user" not in session:
|
|
abort(401)
|
|
|
|
container = client.containers.get(container_id)
|
|
stats = container.stats(stream=False)
|
|
# com.docker.extension.detailed-description
|
|
project = ''
|
|
|
|
pprint(container.attrs)
|
|
if ('com.docker.compose.project' in container.attrs['Config']['Labels']):
|
|
project = container.attrs['Config']['Labels']['com.docker.compose.project']
|
|
|
|
return render_template("container_modal.html", name=container.name, project=project, ports=container.attrs['NetworkSettings']['Ports'], status=container.status, CPU=calculate_cpu_percent(stats, container.status), RAM=calculate_ram_percent(stats, container.status))
|
|
|
|
|
|
@app.route("/api/<container_id>/<action>", methods=['DELETE', 'POST'])
|
|
def api_stop_container(container_id, action):
|
|
if "user" not in session:
|
|
abort(401)
|
|
|
|
container = client.containers.get(container_id)
|
|
|
|
try:
|
|
match action:
|
|
case 'stop':
|
|
container.stop()
|
|
case 'restart':
|
|
container.restart()
|
|
case 'start':
|
|
container.start()
|
|
case _:
|
|
return 'False'
|
|
|
|
container = client.containers.get(container_id)
|
|
return container.status
|
|
except:
|
|
return 'False'
|
|
|
|
def scheduleTasks():
|
|
print("This test runs every 3 seconds")
|
|
|
|
if __name__ == '__main__':
|
|
# scheduler.add_job(id = 'scheduled_tasks', func=scheduleTasks, trigger="interval", seconds=60)
|
|
# scheduler.start()
|
|
app.app_context().push()
|
|
db.create_all()
|
|
app.run(host=os.getenv('IP'), port=os.getenv('port'), debug=True)
|