基于AI的后台进程管理

IT那活儿 2025-08-18

310

点击上方“IT那活儿”公众号--专注于企业全栈运维技术分享，不管IT什么活儿，干就完了！！！

一

场景背景

随着AI的不断发展，AI在各个领域的作用不断体现，如何使用AI对运维工作的帮助，使运维更加高效、可靠和智能，助力企业有效管理复杂IT系统。为此开发了AI-后台进程管理工具。

通过该工具可实现对整个系统中各个主机上的业务进程进行管理。

二

功能展示

AI-后台进程管理工具的进程查找与进程操作：

首先让他简单自我介绍如何使用：

以下为主机上所有进程的输出结果：

注：上图为查找进程的展示，因启动、停止、重启等操作会影响生产暂不进行展示。

三

配置智慧运维平台与AI工具步骤

3.1 首先需写一个AI能调用智慧运维平台的API

具体如下：

#!/home/bomc/miniconda2/bin/python2.7
# -*- coding:utf-8 -*-
from flask import Flask,request,jsonify
import json,requests
import time

app = Flask(__name__)


defurl_post(url,data,headers):
  r = requests.post(url,data=data,headers=headers)
if r.status_code == 200:
    return json.loads(r.text)
else:
    returnFalse

defurl_get(url,headers):
#print url
#print headers
  r = requests.get(url,headers=headers)
#print r.text.encode('gbk')
if r.status_code == 200:
    return json.loads(r.text)
else:
    returnFalse

defurl_delete(url,headers):
  r = requests.delete(url,headers=headers)
if r.status_code == 200:
    return json.loads(r.text)
else:
    returnFalse

defurl_patch(url,headers):
  r = requests.patch(url,data=data,headers=headers)
if r.status_code == 200:
    return json.loads(r.text)
else:
    returnFalse


classshsnc_zhyw_api():
def__init__(self):
    self.url = "http://xx.xx.xx.xx:xxxx"


defget_token(self):
    path = "%s/user/passport/loginCode" % self.url
    headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/xxx.xxx.xxx.xxx Safari/537.36'}
    data = {"params":{"account":"admin","password":"xxxx"}}
    app_result = url_post(path,json.dumps(data),headers)
    return app_result['data']['certification']['token']


defzhyw_rw_zx(self,data):
    token = self.get_token()
    path = "%s/atm/job/execute/orchestration" % self.url
    headers = {'snc-token': token}
    # data = {"params":{"orchestrationId":94,"orchestrationParams":[{"paramName":"hostip","paramType":"text","value":"xx.xx.xx.xx"}]}}
    app_result = url_post(path,json.dumps(data),headers)
    if  app_result['msgCode'] == 200:
      return app_result['data']['result']
    else:
      returnFalse

defzhyw_rw_stepId(self,resultid):
    token = self.get_token()
    path = "%s/atm/job/get" % self.url
    headers = {'snc-token': token}
    data = {'params':{'id':resultid}}
    app_result = url_post(path,json.dumps(data),headers)
    return app_result

defzhyw_rw_result(self,stepid):
    token = self.get_token()
    path = "%s/atm/job/findTaskScriptLog" % self.url
    headers = {'snc-token': token}
    data = {'params':{'condition':{'stepId':stepid}}}
    app_result = url_post(path,json.dumps(data),headers)
    return app_result

defzhyw_api_result(data):
  zhyw_api = shsnc_zhyw_api()
  resultid = zhyw_api.zhyw_rw_zx(data)
  timeout = 300
  interval = 5
  start_time = time.time()
if resultid:
    result_id = zhyw_api.zhyw_rw_stepId(resultid)
    id = result_id['data']['taskStepModelList'][0]['detailBeanList'][0]['stepId']
    whileTrue:
      result_data = zhyw_api.zhyw_rw_result(id)
      result_code = result_data['data']['records'][0]['resultStatus']
      print  result_data['data']['records'][0]
      print result_code
      if time.time() -start_time>timeout:
        data = "timed out"
        break
      if result_code == 300:
        try:
          data = result_data['data']['records'][0]['taskLogOutParamModelList'][0]['paramValue']
        except :
          data = result_data['data']['records'][0]['message']
        return data
        break
      elif result_code == 301:
        data = "test failed "
        break
      else:
        time.sleep(interval)
    return data






@app.route('/api/post',methods=['POST'])
defhandle_post():
    try:
        data = request.get_json()
        print data
        ifnot data:
            return jsonify({"error":"NO JSON data received"}),400
        zhyw_api = zhyw_api_result(data)
        return zhyw_api
    except Exception as e:
        return jsonify({"error":str(e)}),500

if __name__ == '__main__':
    app.run(host='0.0.0.0',port=18888,debug=True)