说明

无法预知端口分布，所以未使用配置文件或者blackbox形式
后续可以优化添加进程名称，更具有可读性
有弊端，随机端口会有大量误告警
探测数据有1min延迟，自定义metrics新增timestamp还未想到解决方案

import os
import yaml
import socket
import prometheus_client
from prometheus_client import *
from flask import Response, Flask
from prometheus_client.core import CollectorRegistry

hostname = socket.gethostname()
ip = socket.gethostbyname(hostname)
app = Flask(__name__)

def TelPort(hostname, ip, port):
    """
        @param ip: 当前主机IP
        @param port: 目标Port
        @return: dict，端口检测结果
        """
    telnet = socket.socket()
    socket.setdefaulttimeout(0.1)
    try:
        telnet.connect((ip, int(port)))
        result_dict = {"hostname": hostname, "ip": ip, "port": str(port), "status": 1}
    except Exception as e:
        result_dict = {"hostname": hostname, "ip": ip, "port": str(port), "status": 0}
    return result_dict

def GetPortList():
    result_list = []
    """
        将 yaml 配置文件中数据格式化为字典
        """
    proPath = os.path.dirname(os.path.realpath(__file__))
    yamlPath = os.path.join(proPath, "port_config.yaml")
    f = open(yamlPath, "r")
    port_dict = yaml.load(f)
    f.close()
    for server in port_dict:
        for server_port in port_dict[server]:
            result_dict = TelPort(hostname, server, server_port.strip())
            result_list.append(result_dict)
    return result_list

@app.route("/metrics")
def PromResponse():
    check_tcp_port = GetPortList()
    # 创建metrics
    REGISTRY = CollectorRegistry(auto_describe=False)
    prometheus_metrics = Gauge("check_server_port", "result", ["hostname","ip", "port"], registry=REGISTRY)
    for data in check_tcp_port:
        hostname = data.get("hostname")
        ip = data.get("ip")
        port = data.get("port")
        status = data.get("status")
        prometheus_metrics.labels(hostname, ip, int(port)).set(status)
    return Response(prometheus_client.generate_latest(REGISTRY),mimetype="text/plain")

if __name__ == "__main__":
    app.run(host=ip, port=12345, debug=True)