python3 調用 ansible 模塊和 fastapi 實現項目監控接口（websocket，logs）實例

時間 2020-12-01

標籤 php html node python web shell json api websocket app 欄目 Python 简体版

原文原文鏈接

問題說明

1.pm2 經過 json 文件啓動node service時，當同時開啓多實例（instances）集羣和 --inspect 斷點 websocket 端口將沒法指定
2.並且項目可能在不一樣環境部署在不一樣的系統中php

3.開發同窗權限不足沒法查看pid從而找到websocket 端口html

{
    "apps": {
        "args": [
            "debug",
            "3121"
        ],
        "cwd": "/data/source/service-sk_platform_professional_rda/publish/",
        "error_file": "/data/pm2-log/errlogs/service-sk_platform_professional_rda-err.log",
        "exec_interpreter": "node",
        "exec_mode": "fork",
        "instances": "4",
        "log_date_format": "YYYY-MM-DD HH:mm Z",
        "max_memory_restart": "8100M",
        "max_restarts": "50",
        "merge_logs": true,
        "min_uptime": "20s",
        "name": "service-sk_platform_professional_rda",
        "node_args": "--inspect=0.0.0.0:33121 --max-old-space-size=8000",
        "out_file": "/data/pm2-log/outlogs/service-sk_platform_professional_rda-out.log",
        "pid_file": "/data/pm2-log/outlogs/service-sk_platform_professional_rda.pid",
        "script": "/data/source/service-sk_platform_professional_rda/publish/service/server.js",
        "watch": false
    }
}

python3 安裝ansible 模塊

pip3 install ansible
pip3 install ansible_runnernode

pip3 install ansible_inventorypython

pip3 install ansible_playbookweb

pip3 install fastapishell

pip3 install uvicornjson

獲取官方實例並修改

https://docs.ansible.com/ansible/latest/dev_guide/developing_api.html

核心類介紹

導入類完整路徑	功能用途
from ansible.module_utils.common.collections import ImmutableDict	用於添加選項。好比: 指定遠程用戶remote_user=None
from ansible.parsing.dataloader import DataLoader	讀取 json/ymal/ini 格式的文件的數據解析器
from ansible.vars.manager import VariableManager	管理主機和主機組的變量管理器
from ansible.inventory.manager import InventoryManager	管理資源庫的，能夠指定一個 inventory 文件等
from ansible.playbook.play import Play	用於執行 Ad-hoc 的類 ,須要傳入相應的參數
from ansible.executor.task_queue_manager import TaskQueueManager	ansible 底層用到的任務隊列管理器
ansible.plugins.callback.CallbackBase	處理任務執行後返回的狀態
from ansible import context	上下文管理器，他就是用來接收 ImmutableDict 的示例對象
import ansible.constants as C	用於獲取 ansible 產生的臨時文檔。
from ansible.executor.playbook_executor import PlaybookExecutor	執行 playbook 的核心類
from ansible.inventory.host import Group	對 *主機組* 執行操做，能夠給組添加變量等操做，擴展
from ansible.inventory.host import Host	對主機執行操做，能夠給主機添加變量等操做，擴展

[root@dev-technology-215l shell]# cat ansible2.py     
import json    
import shutil    
from ansible.module_utils.common.collections import ImmutableDict    
from ansible.parsing.dataloader import DataLoader    
from ansible.vars.manager import VariableManager    
from ansible.inventory.manager import InventoryManager    
from ansible.playbook.play import Play    
from ansible.executor.task_queue_manager import TaskQueueManager    
from ansible.plugins.callback import CallbackBase    
from ansible import context    
import ansible.constants as C

    
class ResultCallback(CallbackBase):    
    """    
    重寫callbackBase類的部分方法    
    """    
    def __init__(self, *args, **kwargs):    
        super().__init__(*args, **kwargs)    
        self.host_ok = {}    
        self.host_unreachable = {}    
        self.host_failed = {}    
        self.task_ok = {}    
    def v2_runner_on_unreachable(self, result):    
        self.host_unreachable[result._host.get_name()] = result

    def v2_runner_on_ok(self, result, **kwargs):    
        self.host_ok[result._host.get_name()] = result

    def v2_runner_on_failed(self, result, **kwargs):    
        self.host_failed[result._host.get_name()] = result

class MyAnsiable2():    
    def __init__(self,    
        connection='local',  # 鏈接方式 local 本地方式，smart ssh方式    
        remote_user=None,    # ssh 用戶    
        remote_password=None,  # ssh 用戶的密碼，應該是一個字典, key 必須是 conn_pass    
        private_key_file=None,  # 指定自定義的私鑰地址    
        sudo=None, sudo_user=None, ask_sudo_pass=None,    
        module_path=None,    # 模塊路徑，能夠指定一個自定義模塊的路徑    
        become=None,         # 是否提權    
        become_method=None,  # 提權方式 默認 sudo 能夠是 su    
        become_user=None,  # 提權後，要成爲的用戶，並不是登陸用戶    
        check=False, diff=False,    
        listhosts=None, listtasks=None,listtags=None,    
        verbosity=3,    
        syntax=None,    
        start_at_task=None,    
        inventory=None):

        # 函數文檔註釋    
        """    
        初始化函數，定義的默認的選項值，    
        在初始化的時候能夠傳參，以便覆蓋默認選項的值    
        """    
        context.CLIARGS = ImmutableDict(    
            connection=connection,    
            remote_user=remote_user,    
            private_key_file=private_key_file,    
            sudo=sudo,    
            sudo_user=sudo_user,    
            ask_sudo_pass=ask_sudo_pass,    
            module_path=module_path,    
            become=become,    
            become_method=become_method,    
            become_user=become_user,    
            verbosity=verbosity,    
            listhosts=listhosts,    
            listtasks=listtasks,    
            listtags=listtags,    
            syntax=syntax,    
            start_at_task=start_at_task,    
        )

        # 三元表達式，假如沒有傳遞 inventory, 就使用 "localhost,"    
        # 指定 inventory 文件    
        # inventory 的值能夠是一個 資產清單文件    
        # 也能夠是一個包含主機的元組，這個僅僅適用於測試    
        #  好比 ： 1.1.1.1,    # 若是隻有一個 IP 最後必須有英文的逗號    
        #  或者： 1.1.1.1, 2.2.2.2

        self.inventory = inventory if inventory else "localhost,"

        # 實例化數據解析器    
        self.loader = DataLoader()

        # 實例化 資產配置對象    
        self.inv_obj = InventoryManager(loader=self.loader, sources=self.inventory)

        # 設置密碼    
        self.passwords = remote_password

        # 實例化回調插件對象    
        self.results_callback = ResultCallback()

        # 變量管理器    
        self.variable_manager = VariableManager(self.loader, self.inv_obj)

    def run(self, hosts='localhost', gether_facts="no", module="ping", args='', task_time=0):    
        """    
        參數說明：    
        task_time -- 執行異步任務時等待的秒數，這個須要大於 0 ，等於 0 的時候不支持異步（默認值）。這個值應該等於執行任務實際耗時時間爲好    
        """    
        play_source =  dict(    
            name = "Ad-hoc",    
            hosts = hosts,    
            gather_facts = gether_facts,    
            tasks = [    
                # 這裏每一個 task 就是這個列表中的一個元素，格式是嵌套的字典    
                # 也能夠做爲參數傳遞過來，這裏就簡單化了。    
               {"action":{"module": module, "args": args}, "async": task_time, "poll": 0}])

        play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)

        tqm = None    
        try:    
            tqm = TaskQueueManager(    
                      inventory=self.inv_obj ,    
                      variable_manager=self.variable_manager,    
                      loader=self.loader,    
                      passwords=self.passwords,    
                      stdout_callback=self.results_callback)

            result = tqm.run(play)    
        finally:    
            if tqm is not None:    
                tqm.cleanup()    
            shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)

    def playbook(self,playbooks):    
        """    
        Keyword arguments:    
        playbooks --  須要是一個列表類型    
        """    
        from ansible.executor.playbook_executor import PlaybookExecutor

        playbook = PlaybookExecutor(playbooks=playbooks,    
                        inventory=self.inv_obj,    
                        variable_manager=self.variable_manager,    
                        loader=self.loader,    
                        passwords=self.passwords)

        # 使用回調函數    
        playbook._tqm._stdout_callback = self.results_callback

        result = playbook.run()

    
    def get_result(self):    
      result_raw = {'success':{},'failed':{},'unreachable':{}}

      # print(self.results_callback.host_ok)    
      for host,result in self.results_callback.host_ok.items():    
          result_raw['success'][host] = result._result    
      for host,result in self.results_callback.host_failed.items():    
          result_raw['failed'][host] = result._result    
      for host,result in self.results_callback.host_unreachable.items():    
          result_raw['unreachable'][host] = result._result

      # 最終打印結果，而且使用 JSON 繼續格式化    
      print(json.dumps(result_raw, indent=4))    
          
      return json.dumps(result_raw)

測試run ansible

[root@dev-technology-215l shell]# cat ansible_run_websocket-fastapi.py   
#!/usr/bin/python
# -*- coding:utf-8 -*-

import subprocess
import urllib.request

from fastapi import FastAPI
import platform
import socket,requests
from ansible2 import *
import ansible_runner
import os, sys, json, datetime, time
import urllib.request
from fastapi.responses import HTMLResponse

from fastapi import FastAPI

#check_sys = platform.system()
#print(check_sys) //查看系統類型


#project = 'sk_service_data_convert'
#Jenvironment = 'std'


def project_host(project,Jenvironment):
    js = urllib.request.urlopen('http://config.skong.com/index.php?apiykpjenkinsykp{tenv}ykp{tprojectname}'.format(tenv=Jenvironment, tprojectname=project))
    js2 = js.read()
    hjson = json.loads(js2.decode('utf-8'))
    #port = hjson['info']['service_port']
    #print(port)

    host = hjson['info']['hoststr']
    print(host)
    hostlist = host.split(",")
    print(hostlist)
    return hostlist



def ansible_project_pid(project,Jenvironment):
    sock5985 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock5985result = sock5985.connect_ex((project_host(project, Jenvironment)[0], 5985))

    if sock5985result != 0:
        ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

        # ansible3.run(hosts= "192.168.0.94", module="shell", args='source /etc/profile && pm2 ls|grep sk_service_design_calculate_rda')
        ansible3.run(hosts=project_host(project, Jenvironment)[0], module="shell",
                     args='source /etc/profile && pm2 ls|grep {tproject}_{tenv}|awk {tprint}'.format(tproject=project,tenv=Jenvironment,
                         tprint="'{print $12}'"))

        stdout_dict = json.loads(ansible3.get_result())
        print(stdout_dict, type(stdout_dict))
        print(stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout'])
        pid_list = stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout'].split("\n")

        print(pid_list)

        project_pid = subprocess.getoutput('ps -ef | grep {tproject}_{tenv} | grep -v grep | awk {pri}'.format(tproject=project,tenv=Jenvironment, pri="'{print $2}'"))
        # project_pid_int = int(project_pid)
    else:
        ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

        # ansible3.run(hosts= "192.168.0.94", module="shell", args='source /etc/profile && pm2 ls|grep sk_service_design_calculate_rda')
        ansible3.run(hosts=project_host(project, Jenvironment)[0], module="raw",
                     args="cmd /C 'pm2 ls|grep {tproject}_{tenv}'".format(tproject=project,tenv=Jenvironment))

        stdout_dict = json.loads(ansible3.get_result())
        # print(stdout_dict,type(stdout_dict))
        # print(stdout_dict['success']['192.168.0.42']['stdout'])
        pm2_pid_list = stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout']
        print(pm2_pid_list)
        project_pid = subprocess.getoutput('echo "{tline}"| tr -s "\n"| awk {tpri}'.format(tline=pm2_pid_list, tpri="'{print $12}'")).split("\n")


    return project_pid





def ansible_project_instance(project,Jenvironment):
    #ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

    #ansible3.run(hosts=project_host(project,Jenvironment)[0], module="shell", args='pm2 ls|grep {tproject}_{tenv}|wc -l'.format(tproject=project,tenv=Jenvironment))

    #stdout_dict = json.loads(ansible3.get_result())
    #print(stdout_dict, type(stdout_dict))
    #stdout_sum = stdout_dict['success'][project_host(project,Jenvironment)[0]]['stdout']
    js = urllib.request.urlopen('http://config.skong.com/index.php?apiykpjenkinsykp{tenv}ykp{tprojectname}'.format(tenv=Jenvironment,tprojectname=project))
    js2 = js.read()
    hjson = json.loads(js2.decode('utf-8'))
    # port = hjson['info']['service_port']
    # print(port)

    stdout_sum = hjson['info']['clusternumber']
    print(stdout_sum)

    return stdout_sum


def ansible_project_websocket_port(project,Jenvironment):
    sock5985 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock5985result = sock5985.connect_ex((project_host(project, Jenvironment)[0], 5985))

    if sock5985result != 0:
        ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

        # ansible3.run(hosts= "192.168.0.94", module="shell", args='source /etc/profile && pm2 ls|grep sk_service_design_calculate_rda')
        ansible3.run(hosts=project_host(project, Jenvironment)[0], module="shell",
                     args='source /etc/profile && pm2 ls|grep {tproject}_{tenv}|awk {tprint}'.format(tproject=project,tenv=Jenvironment,
                         tprint="'{print $12}'"))

        stdout_dict = json.loads(ansible3.get_result())
        #print(stdout_dict, type(stdout_dict))
        #print(stdout_dict['success']['192.168.0.94']['stdout'])
        pid_list = stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout'].split("\n")

        print(pid_list)

        websocket_list = []
        for i in pid_list:
            ansible3.run(hosts=project_host(project, Jenvironment)[0], module="shell",
                         args='netstat -tnlp | grep {tpid} | awk {tprint}'.format(tpid=i, tprint="'{print $4}'"))

            stdout_dict = json.loads(ansible3.get_result())
            #print(stdout_dict['success']['192.168.0.94']['stdout'])
            websocket_list.append(stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout'])

        print(websocket_list)

    else:
        ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

        ansible3.run(hosts=project_host(project, Jenvironment)[0], module="raw",
                     args="cmd /C 'pm2 ls|grep {tproject}_{tenv}'".format(tproject=project,tenv=Jenvironment))

        stdout_dict = json.loads(ansible3.get_result())
        pm2_pid_list = stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout']
        #print(pm2_pid_list)
        #pid_list = subprocess.getstatusoutput()
        pid_list = subprocess.getoutput(
            'echo "{tline}"| tr -s "\n"| awk {tpri}'.format(tline=pm2_pid_list, tpri="'{print $12}'")).split("\n")

        print(pid_list)

        # pid_list = ['14756', '13012', '10116']

        websocket_list = []
        for i in pid_list:
            ansible3.run(hosts=project_host(project, Jenvironment)[0], module="win_shell",
                         args='netstat -nao | grep 0.0.0.0 | grep {tpid}| awk {tprint}'.format(tpid=i,
                                                                                               tprint="'{print $2}'"))

            stdout_dict = json.loads(ansible3.get_result())
            #print(stdout_dict['success']['192.168.0.42']['stdout'])
            websocket_list.append(stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout'])

        print(websocket_list)

    return websocket_list




def ansible_project_logs(project,Jenvironment):
    sock5985 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock5985result = sock5985.connect_ex((project_host(project, Jenvironment)[0], 5985))

    if sock5985result != 0:
        ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

        # ansible3.run(hosts= "192.168.0.94", module="shell", args='source /etc/profile && pm2 ls|grep sk_service_design_calculate_rda')
        ansible3.run(hosts=project_host(project, Jenvironment)[0], module="shell",args="echo '############################    logs    #################################' && source /etc/profile && pm2 logs --lines 40 --nostream {tproject}_{tenv} && echo '@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@  end  @@@@@@@@@@@@@@@@@@@@@@@@'".format(tproject=project,tenv=Jenvironment))

        stdout_dict = json.loads(ansible3.get_result())
        #print(stdout_dict, type(stdout_dict))
        #print(stdout_dict['success']['192.168.0.94']['stdout'])
        project_logs = stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout']

        print(project_logs)

    else:
        ansible3 = MyAnsiable2(inventory='/data/ansible/host/hosts', connection='smart')

        ansible3.run(hosts=project_host(project, Jenvironment)[0], module="raw",
                     args="cmd /C 'pm2 logs --lines 40 --nostream {tproject}_{tenv} '".format(tproject=project,tenv=Jenvironment))

        stdout_dict = json.loads(ansible3.get_result())
        project_logs = stdout_dict['success'][project_host(project, Jenvironment)[0]]['stdout']

    return project_logs.replace('\n','<br>')






app = FastAPI()


@app.get("/")
def read_root():
    return {"messages": "hellokugou"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}



@app.get("/websocket/{Jenvironment}/{project}")
def read_project(project: str,Jenvironment: str):
    print(project, '#######################project')
    print(Jenvironment, '@@@@@@@@@@@@@@env')
    print(ansible_project_websocket_port(project,Jenvironment),"$$$ansible_project_websocket_port-LIst")

    return {"project_name": project,"env": Jenvironment , "project_host": project_host(project,Jenvironment), "project_running-sum": ansible_project_instance(project,Jenvironment),"project_websocket_port": ansible_project_websocket_port(project,Jenvironment)}


@app.get("/logs/{Jenvironment}/{project}")
def read_project_logs(project: str,Jenvironment: str):
    print(project, '#######################project')
    print(Jenvironment, '@@@@@@@@@@@@@@env')
    #print(ansible_project_websocket_port(project,Jenvironment),"$$$ansible_project_websocket_port-LIst")

    html_content = """
    <!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>config-center</title>
    </head>
    <body>
        <p>{tansiblestr}</p>
    </body>
    </html>
    """.format(tansiblestr=ansible_project_logs(project,Jenvironment))
    return HTMLResponse(content=html_content, status_code=200)




    #return ansible_project_logs(project,Jenvironment)









if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app=app,
                host="192.168.0.215",
                port=9999,
                workers=1)

運行結果：

[root@dev-technology-215l shell]# curl http://192.168.0.215:9999/websocket/fct/sk_service_design  
{"project_name":"sk_service_design","env":"fct","project_host":["192.168.0.42","192.168.0.43"],"project_running-sum":"3","project_websocket_port":["0.0.0.0:9638\r\n","0.0.0.0:9639\r\n","0.0.0.0:9640\r\n"]}[root@dev-technology-215l shell]#   
[root@dev-technology-215l shell]#

[root@dev-technology-215l shell]# python ansible_run_websocket-fastapi.py    
['192.168.0.42', '192.168.0.43']  
{  
    "success": {  
        "192.168.0.42": {  
            "start": "2020-07-07 10:05:45.305889",  
            "stdout": "0.0.0.0:9639\r\n",  
            "cmd": "netstat -nao | grep 0.0.0.0 | grep 7976| awk '{print $2}'",  
            "stderr": "",  
            "changed": true,  
            "rc": 0,  
            "delta": "0:00:00.546910",  
            "end": "2020-07-07 10:05:45.852800",  
            "stdout_lines": [  
                "0.0.0.0:9639"  
            ],  
            "stderr_lines": [],  
            "_ansible_no_log": false  
        }  
    },  
    "failed": {},  
    "unreachable": {}  
}  
192.168.0.42,192.168.0.43  
['192.168.0.42', '192.168.0.43']  
192.168.0.42,192.168.0.43  
['192.168.0.42', '192.168.0.43']  
{  
    "success": {  
        "192.168.0.42": {  
            "start": "2020-07-07 10:05:47.243515",  
            "stdout": "0.0.0.0:9640\r\n",  
            "cmd": "netstat -nao | grep 0.0.0.0 | grep 13960| awk '{print $2}'",  
            "stderr": "",  
            "changed": true,  
            "rc": 0,  
            "delta": "0:00:00.562535",  
            "end": "2020-07-07 10:05:47.806051",  
            "stdout_lines": [  
                "0.0.0.0:9640"  
            ],  
            "stderr_lines": [],  
            "_ansible_no_log": false  
        }  
    },  
    "failed": {},  
    "unreachable": {}  
}  
192.168.0.42,192.168.0.43  
['192.168.0.42', '192.168.0.43']  
['0.0.0.0:9638\r\n', '0.0.0.0:9639\r\n', '0.0.0.0:9640\r\n']  
INFO:     192.168.0.215:46513 - "GET /websocket/fct/sk_service_design HTTP/1.1" 200 OK