from flask import render_template_string import requests import re class Docker(object): def __init__( self, method, prefix, host, port, api_version, card_type, tls_mode, tls_ca, tls_cert, tls_key, ): self.endpoint = None self.method = method self.prefix = prefix self.host = host self.port = port self.api_version = api_version self.card_type = card_type self.tls_mode = tls_mode self.tls_ca = tls_ca self.tls_key = tls_key self.tls_cert = tls_cert # Initialize results self.error = None self.version = "?" self.max_api_version = "?" self.name = "?" self.running = 0 self.paused = 0 self.stopped = 0 self.images = 0 self.driver = "?" self.cpu = "?" self.memory = "?" self.html_template = "" def check(self): port = "" if self.port == None else ":" + self.port if self.method.upper() == "GET": try: response = "" request = requests.get( self.prefix + self.host + port + "/v999/info", verify=self.tls_ca, cert=(self.tls_cert, self.tls_key), timeout=10, ) response = request.text if "text/plain" in request.headers["content-type"]: self.error = request.text rawdata = None elif "application/json" in request.headers["content-type"]: rawdata = request.json() else: error = request rawdata = None except Exception as e: rawdata = None self.error = f"{e}" + " " + response self.setHtml() if rawdata != None: if "message" in rawdata: regex = r"\bv?[0-9]+\.[0-9]+(?:\.[0-9]+)?\b" r = re.search(regex, rawdata["message"]) self.max_api_version = r.group(0) self.api_version = ( self.api_version if self.api_version != None else self.max_api_version ) self.endpoint = "/v" + self.api_version + "/" def getStatus(self): port = "" if self.port == None else ":" + self.port if self.method.upper() == "GET": try: rawdata = requests.get( self.prefix + self.host + port + self.endpoint + "/info", verify=self.tls_ca, cert=(self.tls_cert, self.tls_key), timeout=10, ).json() except Exception as e: rawdata = None self.error = f"{e}" self.setHtml() if rawdata != None: self.name = rawdata["Name"] self.containers = rawdata["Containers"] self.containers_running = rawdata["ContainersRunning"] self.containers_paused = rawdata["ContainersPaused"] self.containers_stopped = rawdata["ContainersStopped"] self.images = rawdata["Images"] self.warnings = rawdata["Warnings"] self.driver = rawdata["Driver"] self.cpu = rawdata["NCPU"] self.memory = self.formatSize(rawdata["MemTotal"]) if self.card_type == "Custom": self.setHtml() def formatSize(self, size): # 2**10 = 1024 power = 2 ** 10 n = 0 power_labels = {0: "", 1: "KB", 2: "MB", 3: "GB", 4: "TB"} while size > power: size /= power n += 1 return str(round(size, 1)) + " " + power_labels[n] def refresh(self): self.check() if self.error == None: self.error = "" self.getStatus() def setHtml(self): if self.error != None and self.error != "": self.html_template = """
error
Docker
Error
keyboard_arrow_down
{{ error }}
""" else: if self.tls_mode == None: img_tls = """ lock_open """ else: img_tls = """ lock """ if len(self.warnings) > 0: img_warnings = """ warning """ else: img_warnings = """ warning """ self.html_template = ( """
""" + img_tls + img_warnings + """
Docker
{{name}}
keyboard_arrow_down
Containers: {{ containers }}
Running: {{ containers_running }}
Paused: {{ containers_paused }}
Stopped: {{ containers_stopped }}
Images: {{ images }}
Driver: {{ driver }}
CPU: {{ cpu }}
Memory: {{ memory }}
""" ) def getHtml(self): return self.html_template class Platform: def docs(self): documentation = { "name": "docker", "author": "Thlb", "author_url": "https://github.com/Thlb", "version": 1.0, "description": "Display information from Docker API. Informations can be displayed on a custom card or on an app card (e.g. Portainer App)", "returns": "`value_template` as rendered string", "returns_json_keys": [ "version", "max_api_version", "name", "containers", "containers_running", "containers_paused", "containers_stopped", "images", "driver", "cpu", "memory", "warnings", "error (for debug)", ], "example": """ ```ini # Working example (using un-encrypted connection, on Portainer card) [docker-endpoint-1] platform = docker prefix = http:// host = 192.168.0.110 port = 2375 value_template = {{error}}

{{name}}
fiber_manual_record{{containers_running}}fiber_manual_record{{containers_paused}}fiber_manual_record{{containers_stopped}}

[Portainer] prefix = http:// url = 192.168.0.110:2375 icon = static/images/apps/portainer.png sidebar_icon = static/images/apps/portainer.png description = Making Docker management easy open_in = this_tab data_sources = docker-endpoint-1 # Working example (using encrypted connection, on Portainer card) ```ini [docker-endpoint-2] platform = docker prefix = https:// host = 192.168.0.110 port = 2376 tls_mode = Both tls_ca = /path/to/ca_file tls_cert = /path/to/cert_file tls_key = /path/to/key_file value_template = {{error}}

{{name}}
fiber_manual_record{{containers_running}}fiber_manual_record{{containers_paused}}fiber_manual_record{{containers_stopped}}

[Portainer] prefix = http:// url = 192.168.0.110:2375 icon = static/images/apps/portainer.png sidebar_icon = static/images/apps/portainer.png description = Making Docker management easy open_in = this_tab data_sources = docker-endpoint-2 ``` # Working example (using un-encrypted connection, on custom Docker card) ```ini [docker-endpoint-3] platform = docker prefix = http:// host = 192.168.0.110 port = 2375 card_type = Custom [Docker] type = custom data_sources = docker-endpoint-3 ``` """, "variables": [ { "variable": "[variable_name]", "description": "Name for the data source.", "default": "None, entry is required", "options": ".ini header", }, { "variable": "platform", "description": "Name of the platform.", "default": "docker", "options": "docker", }, { "variable": "prefix", "description": "The prefix for the app's url.", "default": "", "options": "web prefix, e.g. http:// or https://", }, { "variable": "host", "description": "Docker Host", "default": "", "options": "url,ip", }, { "variable": "port", "description": "Docker Port", "default": "", "options": "port", }, { "variable": "api_version", "description": "API version, by default platform will try to find latest version", "default": "", "options": "1.40", }, { "variable": "tls_mode", "description": "TLS verification mode", "default": "None", "options": "Server, Client, Both, None", }, { "variable": "tls_ca", "description": "Requiered for tls_mode=Both or tls_mode=Server", "default": "None", "options": "/path/to/ca, None", }, { "variable": "tls_cert", "description": "Requierd for tls_mode=Both or tls_mode=Client", "default": "None", "options": "/path/to/cert, None", }, { "variable": "tls_key", "description": "Requierd for tls_mode=Both or tls_mode=Client", "default": "None", "options": "/path/to/key, None", }, { "variable": "card_type", "description": "Set to Custom if you want to display informations in a custom card", "default": "App", "options": "Custom, App", }, { "variable": "value_template", "description": "Jinja template for how the returned data from API is displayed.", "default": "", "options": "jinja template", }, ], } return documentation def __init__(self, *args, **kwargs): # parse the user's options from the config entries for key, value in kwargs.items(): self.__dict__[key] = value # set defaults for omitted options if not hasattr(self, "method"): self.method = "GET" if not hasattr(self, "prefix"): self.prefix = "http://" if not hasattr(self, "host"): self.host = None if not hasattr(self, "port"): self.port = 2375 if not hasattr(self, "api_version"): self.api_version = None if not hasattr(self, "card_type"): self.card_type = "App" if not hasattr(self, "tls_ca"): self.tls_ca = None if not hasattr(self, "tls_cert"): self.tls_cert = None if not hasattr(self, "tls_key"): self.tls_key = None # Without TLS if not hasattr(self, "tls_mode"): self.tls_mode = None self.tls_ca = None self.tls_cert = None self.tls_key = None else: if self.tls_mode == "Both": if self.tls_ca == None or self.tls_cert == None or self.tls_key == None: return "tls_mode set to Both, and missing tls_ca/tls_cert/tls_key" elif self.tls_mode == "Client": self.tls_ca = False elif self.tls_mode == "Server": self.tls_cert = "" self.tls_key = "" elif self.tls_mode == "None": self.tls_ca = None self.tls_cert = None self.tls_key = None self.docker = Docker( self.method, self.prefix, self.host, self.port, self.api_version, self.card_type, self.tls_mode, self.tls_ca, self.tls_cert, self.tls_key, ) def process(self): if self.host == None: return "host missing" # TLS check if self.tls_mode == "Both": if self.tls_ca == None or self.tls_cert == None or self.tls_key == None: return "tls_mode set to Both, and missing tls_ca/tls_cert/tls_key" elif self.tls_mode == "Client": if self.tls_cert == None or self.tls_key == None: return "tls_mode set to Client, and missing tls_cert/tls_key" elif self.tls_mode == "Server": if self.tls_ca == None: return "tls_mode set to Server, and missing tls_ca" else: if self.tls_mode != None: return "Invalid tls_mode : " + self.tls_mode self.docker.refresh() if self.card_type == "Custom": return render_template_string(self.docker.getHtml(), **self.docker.__dict__) else: return render_template_string(self.value_template, **self.docker.__dict__)