怎么用python+Element實現(xiàn)主機Host操作
這篇文章主要講解了“怎么用python+Element實現(xiàn)主機Host操作”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“怎么用python+Element實現(xiàn)主機Host操作”吧！
創(chuàng)新互聯(lián)建站長期為1000多家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊從業(yè)經(jīng)驗10年，關(guān)注不同地域、不同群體，并針對不同對象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為廣元企業(yè)提供專業(yè)的成都做網(wǎng)站、網(wǎng)站設(shè)計，廣元網(wǎng)站改版等技術(shù)服務(wù)。擁有十多年豐富建站經(jīng)驗和眾多成功案例,為您定制開發(fā)。
1.前端HTML和CSS
<%inherit file="/base.html"/>
<div id="app">
<div class="col-md-10 col-md-offset-1" >
<div class="container-fluid mb0 " >
<div class="row">
<div class="col-md-4">
<form class="form-horizontal">
<div class="form-group clearfix ">
<label class="col-sm-4 control-label bk-lh40 pt0">選擇業(yè)務(wù)：</label>
<div class="col-sm-8">
<el-select v-model="searchBiz" placeholder="請選擇業(yè)務(wù)名稱" @change="changeBiz">
<el-option v-for="item in bkBizData" :label="item.bk_biz_name" :value="item.bk_biz_id"></el-option>
</el-select>
</div>
</div>
</form>
</div>
<div class="col-md-6">
<form class="form-horizontal">
<div class="form-group clearfix ">
<label class="col-sm-4 control-label bk-lh40 pt0">請輸入主機IP：</label>
<div class="col-sm-8">
<el-input v-model="searchHostIp" placeholder="請輸入主機IP" ></el-input>
</div>
</div>
</form>
</div>
<div class="col-md-2">
<form class="form-horizontal">
<div class="form-group clearfix ">
<div class="col-sm-8">
<el-button type="primary" @click="getSearch">搜索</el-button>
</div>
</div>
</form>
</div>
</div>
</div>
<el-table :data="hostData" border >
<el-table-column prop="ip" label="內(nèi)網(wǎng)IP"></el-table-column>
<el-table-column prop="bk_os_name" label="系統(tǒng)名"></el-table-column>
<el-table-column prop="host_name" label="主機名"></el-table-column>
<el-table-column prop="cloud_name" label="云區(qū)域"></el-table-column>
<el-table-column prop="mem_use" label="Mem(%)"></el-table-column>
<el-table-column prop="disk_use" label="Disk(%)"></el-table-column>
<el-table-column prop="cpu_use" label="CPU(%)"></el-table-column>
<el-table-column label="操作" width="300%">
<template slot-scope="scope">
<el-button :type='scope.row.is_monitored ? "primary":"success"' size="small" @click="editHost(scope.row)">{{ scope.row.is_monitored ? "移除監(jiān)控":"加入監(jiān)控" }}</el-button>
<el-button type="warning" size="small" @click="getPerformData(scope.row)">查看性能</el-button>
<el-button type="danger" size="small" @click="getStatus(scope.row)">查看狀態(tài)</el-button>
</template>
</el-table-column>
</el-table>
<!-- 設(shè)置面板End -->
</div>
</div>2.前端JS
<script type="text/javascript">
new Vue({
el: '#app',
data: {
bkBizData: [],
hostData: [],
searchBiz:'',
searchHostIp:''
},
mounted() {
// 頁面加載就獲取所有模板
this.init();
this.getSearch();
},
methods: {
init() {
axios.get(site_url + "get_biz_list/").then(res => {
if (res.data.result){
this.bkBizData = res.data.data;
}else{
this.$message.error('獲取業(yè)務(wù)失敗');
}
},'json');
this.getSearch();
},
getSearch() {
axios.get(site_url + "host_view/?search_biz_id=" + this.searchBiz + "&query_str=" + this.searchHostIp).then(res => {
if (res.data.result){
this.hostData = res.data.data;
}else{
this.$message.error('獲取模板失敗');
}
},'json');
},
changeBiz(){
this.getSearch();
},
editHost(row) {
const tmpText = row.is_monitored ? "主機移除監(jiān)控隊列, 是否繼續(xù)?" : "主機加入監(jiān)控隊列, 是否繼續(xù)?";
this.$confirm(tmpText, '提示', {
confirmButtonText: '確定',
cancelButtonText: '取消',
type: 'warning'
}).then(() => {
axios.post(site_url + "host_view/", {"host_id": row.host_id,"is_monitored": row.is_monitored}).then(res => {
if (res.data.result) {
if(row.is_monitored){
this.$message.success('主機移除監(jiān)控隊列成功');
} else {
this.$message.warning('主機加入監(jiān)控隊列成功');
}
this.getSearch();
} else {
this.$message.error('更新主機監(jiān)控狀態(tài)失敗');
}
}, 'json');
}).catch(() => {
this.$message({type: 'info', message: '已取消刪除'});
});
},
getPerformData(row) {
this.$confirm("是否查詢系統(tǒng)資源使用情況？", '提示', {
confirmButtonText: '確定',
cancelButtonText: '取消',
type: 'warning'
}).then(() => {
const params = {
"host_id": row.host_id,
"bk_biz_id": row.bk_biz_id,
"ip": row.ip,
"cloud_id": row.cloud_id,
};
axios.post(site_url + "get_perform_data/", params).then(res => {
if (res.data.result) {
row.mem_use = res.data.data.mem;
row.cpu_use = res.data.data.cpu;
row.disk_use = res.data.data.disk;
} else {
this.$message.error('查詢主機系統(tǒng)資源使用情況失敗');
}
}, 'json');
}).catch(() => {
this.$message({type: 'info', message: '已取消查詢'});
});
},
getStatus(row) {
location.href = site_url + 'status/'
}
}
})
</script>3.Django代碼
urls.py文件內(nèi)容
from django.conf.urls import patterns from home_application.temp import views as temp_view from home_application.job import views as job_view from home_application.host import views as host_view from home_application.exam import views as exam_view urlpatterns = patterns( 'home_application.views', (r'^$', job_view.job), (r'^dev-guide/$', 'dev_guide'), (r'^contactus/$', 'contactus'), (r'^api/test/$', 'test'), (r'^temp/$', 'temp'), (r'^host/$', host_view.host), (r'^status/$', host_view.status), (r'^host_view/$', host_view.HostView.as_view()), (r'^get_all_hosts/$', host_view.get_all_hosts), (r'^get_perform_data/$', host_view.get_perform_data), ... )
host\views.py文件內(nèi)容
import json
from django.views.generic import View
from django.views.decorators.csrf import csrf_exempt
from django.utils.decorators import method_decorator
from django.http import JsonResponse
from common.mymako import render_mako_context
from home_application.models import Host, LoadData
from home_application.utils.job_api import FastJobApi
perform_script = """
#!/bin/bash
MEMORY=$(free -m | awk 'NR==2{printf "%.2f%%",$3*100/$2}')
DISK=$(df -h| awk '$NF=="/"{printf "%s",$5}')
CPU=$(top -bn1 | grep load | awk '{printf "%.2f%%",$(NF-2)}')
DATE=$(date "+%Y-%m-%d %H:%M:%S")
echo -e "$DATE|$MEMORY|$DISK|$CPU"
"""
def host(request):
return render_mako_context(request, '/home_application/host.html')
def status(request):
return render_mako_context(request, "/home_application/status.html")
class CsrfExemptView(View):
@method_decorator(csrf_exempt)
def dispatch(self, request, *args, **kwargs):
return super(CsrfExemptView, self).dispatch(request, *args, **kwargs)
def get_all_hosts(request):
from home_application.utils.cc_by_request import cc_search_host
res_data = cc_search_host().get("info")
for host in res_data:
bk_host_innerip = host.get("host", {}).get("bk_host_innerip")
bk_host_id = host.get("host", {}).get("bk_host_id")
bk_host_name = host.get("host", {}).get("bk_host_name")
bk_os_name = host.get("host", {}).get("bk_os_name")
bk_biz_id = host.get("biz", [])[0].get("bk_biz_id")
bk_biz_name = host.get("biz", [])[0].get("bk_biz_name")
cloud_name = host.get("host", {}).get("bk_cloud_id")[0].get("bk_inst_name")
cloud_id = host.get("host", {}).get("bk_cloud_id")[0].get("bk_inst_id")
host_obj = {
"ip": bk_host_innerip,
"bk_biz_id": bk_biz_id,
"os_name": bk_os_name,
"host_name": bk_host_name,
"bk_biz_name": bk_biz_name,
"cloud_id": cloud_id,
"cloud_name": cloud_name,
}
Host.objects.update_or_create(host_id=bk_host_id, defaults=host_obj)
return JsonResponse({"result": res_data})
@csrf_exempt
def get_perform_data(request):
data = json.loads(request.body)
host_id = data.get("host_id")
try:
obj = LoadData.objects.filter(host_id=host_id).last()
if obj:
res_data = {
"cpu": obj.cpu,
"mem": obj.mem,
"disk": obj.disk,
}
return JsonResponse({"result": True, "data": res_data})
bk_biz_id = int(data.get("bk_biz_id"))
ip = data.get("ip")
cloud_id = data.get("cloud_id")
# res_data = execute_script_and_return(request,ip_list, bk_biz_id, perform_script)
obj = FastJobApi(bk_biz_id, cloud_id, ip, perform_script)
job_res = obj.execute_script_and_return()
if job_res:
log_content = job_res[0].get("log_content")
res_data = {
"result": True,
"cpu": log_content.split("|")[3],
"mem": log_content.split("|")[1],
"disk": log_content.split("|")[2],
}
return JsonResponse({"result": True, "data": res_data})
return JsonResponse({"result": False})
except Exception:
return JsonResponse({"result": False})
class HostView(CsrfExemptView):
def get(self, request, *args, **kwargs):
try:
host_query = Host.objects.all()
except Exception:
return JsonResponse({"result": False})
search_biz_id = request.GET.get("search_biz_id")
query_str = request.GET.get("query_str")
if search_biz_id:
host_query = host_query.filter(bk_biz_id=search_biz_id)
if query_str:
host_query = host_query.filter(ip__in=query_str.split(","))
host_query = host_query[:30] if host_query.count() > 10 else host_query
res_data = [i.to_dict() for i in host_query]
return JsonResponse({"result": True, "data": res_data})
def post(self, request, *args, **kwargs):
try:
data = json.loads(request.body)
host_id = data.get("host_id")
is_monitored = data.get("is_monitored")
if is_monitored:
Host.objects.filter(host_id=host_id).update(is_monitored=False)
else:
Host.objects.filter(host_id=host_id).update(is_monitored=True)
return JsonResponse({"result": True})
except Exception:
return JsonResponse({"result": False})models.py文件內(nèi)容
from django.db import models
from home_application.utils.parse_time import parse_datetime_to_timestr
class Host(models.Model):
host_id = models.IntegerField(u"主機ID", primary_key=True, unique=True)
ip = models.CharField(u"IP地址", max_length=32, blank=True, null=True)
bk_biz_id = models.CharField(u"業(yè)務(wù)ID", max_length=16, blank=True, null=True)
bk_biz_name = models.CharField(u"業(yè)務(wù)名稱", max_length=512, blank=True, null=True)
os_name = models.CharField(u"系統(tǒng)名", max_length=128, blank=True, null=True)
host_name = models.CharField(u"主機名", max_length=128, blank=True, null=True)
cloud_id = models.IntegerField(u"云區(qū)域ID", blank=True, null=True)
cloud_name = models.CharField(u"云區(qū)域名稱", max_length=32, blank=True, null=True)
is_monitored = models.BooleanField(u"是否已監(jiān)控", default=False)
def to_dict(self):
return {
"host_id": self.host_id,
"ip": self.ip,
"pk": self.pk,
"bk_biz_id": self.bk_biz_id,
"bk_biz_name": self.bk_biz_name,
"os_name": self.os_name,
"host_name": self.host_name,
"cloud_name": self.cloud_name,
"cloud_id": self.cloud_id,
"is_monitored": self.is_monitored,
"mem_use": "-",
"cpu_use": "-",
"disk_use": "-",
}
def get_load_data(self):
load_query = LoadData.objects.filter(host_id=self.pk).order_by("create_time")
return [i.to_dict() for i in load_query]
class LoadData(models.Model):
host_id = models.IntegerField(u"主機ID", default=0)
cpu = models.IntegerField(u"CPU使用率", default=0)
mem = models.IntegerField(u"內(nèi)存使用率", default=0)
disk = models.IntegerField(u"硬盤使用率", default=0)
create_time = models.DateTimeField(u"創(chuàng)建時間", auto_now_add=True)
def to_dict(self):
return {
"host_id": self.host_id,
"cpu": self.cpu,
"mem": self.mem,
"disk": self.disk,
"create_time": parse_datetime_to_timestr(self.create_time)
}FastJobApi部分代碼
import base64
import time
from conf.default import APP_ID, APP_TOKEN
from blueking.component.shortcuts import get_client_by_user, get_client_by_request
count = 0
class FastJobApi(object):
def __init__(self, bk_biz_id, bk_cloud_id, ip_list, script_content):
self.client = get_client_by_user('admin')
self.username = "admin"
self.biz_id = bk_biz_id
self.script_content = script_content
self.ip_list = [{"bk_cloud_id": bk_cloud_id, "ip": i} for i in ip_list.split(",")]
def _fast_execute_script(self, execute_account="root", param_content='', script_timeout=1000):
"""
快速執(zhí)行腳本
:param execute_account: 執(zhí)行腳本的賬戶名
:param param_content: 執(zhí)行腳本的參數(shù)
:param script_timeout: 腳本執(zhí)行超時時間
:return: job_instance_id
"""
kwargs = {
"bk_app_code": APP_ID,
"bk_app_secret": APP_TOKEN,
"bk_biz_id": self.biz_id,
"bk_username": self.username,
"script_content": base64.b64encode(self.script_content),
"ip_list": self.ip_list,
"script_type": 1,
"account": execute_account,
"script_param": base64.b64encode(param_content),
"script_timeout": script_timeout
}
result = self.client.job.fast_execute_script(kwargs)
if result["result"]:
return result.get("data").get("job_instance_id")
return False
def _get_job_instance_status(self, task_id):
"""
獲取腳本執(zhí)行狀態(tài)
:param task_id: 執(zhí)行腳本的 job_instance_id
:return:
"""
global count
count += 1
# 查詢執(zhí)行狀態(tài)
resp = self.client.job.get_job_instance_status(
bk_username=self.username,
bk_biz_id=self.biz_id,
job_instance_id=task_id
)
if resp.get('data').get('is_finished'):
count = 0
return True
elif not resp.get('data').get('is_finished') and count <= 5:
time.sleep(2)
return self._get_job_instance_status(task_id)
else:
count = 0
return False
def _get_job_instance_log(self, task_id):
"""
查詢作業(yè)日志
:param task_id: 執(zhí)行腳本的 job_instance_id
:return:
"""
if self._get_job_instance_status(task_id):
resp = self.client.job.get_job_instance_log(
job_instance_id=task_id,
bk_biz_id=self.biz_id,
bk_username='admin'
)
return resp['data'][0]['step_results'][0]['ip_logs']
def execute_script_and_return(self):
"""
執(zhí)行腳本并獲取腳本執(zhí)行結(jié)果
:return: 腳本執(zhí)行結(jié)果
"""
job_instance_id = self._fast_execute_script()
if job_instance_id:
ip_logs = self._get_job_instance_log(job_instance_id)
return ip_logs實現(xiàn)效果
感謝各位的閱讀，以上就是“怎么用python+Element實現(xiàn)主機Host操作”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對怎么用python+Element實現(xiàn)主機Host操作這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！
分享名稱：怎么用python+Element實現(xiàn)主機Host操作
URL分享：http://bm7419.com/article36/goegpg.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站策劃、外貿(mào)網(wǎng)站建設(shè)、搜索引擎優(yōu)化、虛擬主機、網(wǎng)站設(shè)計、Google
聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)
- 哈爾濱app軟件開發(fā)之音樂歌曲 2023-03-27
- 一家做網(wǎng)站建設(shè)和小程序軟件開發(fā)的網(wǎng)絡(luò)公司！ 2017-03-12
- 武漢app軟件開發(fā)流程 2023-02-12
- 行業(yè)軟件開發(fā)遇到問題該如何解決 2022-07-18
- 成都物流軟件開發(fā)該注意哪些問題？ 2022-06-08
- 保姆APP軟件開發(fā)，尋找保姆方便快捷 2022-06-28
- 電商APP軟件開發(fā)應(yīng)如何選擇軟件開發(fā)公司？ 2020-11-13
- 做好項目管理，才是軟件開發(fā)管理的關(guān)鍵 2016-08-30
- 軟件開發(fā)可用性原則 2014-07-24
- 美容院加盟運營系統(tǒng)軟件開發(fā) 2021-05-14
- 成都軟件開發(fā)公司哪家好?如何選擇？ 2020-08-21
- 創(chuàng)新互聯(lián)：企業(yè)如何選擇專業(yè)APP軟件開發(fā)公司 2022-11-06