Shulou(Shulou.com)06/02 Report--

This article mainly explains "how to use python+Element to achieve host Host operation", the content of the article is simple and clear, easy to learn and understand, the following please follow the editor's ideas slowly in depth, together to study and learn "how to use python+Element to achieve host Host operation" bar!

1. Front-end HTML and CSS select business: Please enter the host IP: Search Cable {{scope.row.is_monitored? "remove Monitoring": "join Monitoring"}} View performance View status 2. Front-end JS new Vue ({el:'# app', data: {bkBizData: [], hostData: [], searchBiz:'', searchHostIp:''}, mounted () {/ / page load to get all templates 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 ('failed to acquire business');}}, '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 ('failed to get template');}}, 'json');}, changeBiz () {this.getSearch () }, editHost (row) {const tmpText = row.is_monitored? "the host removes the monitoring queue, do you want to continue?": "the host joins the monitoring queue, do you want to continue?" This.$confirm (tmpText, 'prompt', {confirmButtonText:'OK', cancelButtonText: 'cancel', 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 ('host removed monitoring queue successfully') } else {this.$message.warning ('host joined monitoring queue successfully');} this.getSearch () } else {this.$message.error ('failed to update host monitoring status');}}, 'json');}) .catch () = > {this.$message ({type:' info', message: 'deleted'}) });}, getPerformData (row) {this.$confirm ("do you want to query system resource usage?" , 'prompt', {confirmButtonText:'OK', cancelButtonText: 'cancel', 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 ('failed to query host system resource usage');}}, 'json') }) .catch (() = > {this.$message ({type: 'info', message:' query cancelled'});});}, getStatus (row) {location.href = site_url + 'status/'}}) 3.Django code

Urls.py file content

From django.conf.urls import patternsfrom home_application.temp import views as temp_viewfrom home_application.job import views as job_viewfrom home_application.host import views as host_viewfrom home_application.exam import views as exam_viewurlpatterns = 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) .)

Contents of host\ views.py file

Import jsonfrom django.views.generic import Viewfrom django.views.decorators.csrf import csrf_exemptfrom django.utils.decorators import method_decoratorfrom django.http import JsonResponsefrom common.mymako import render_mako_contextfrom home_application.models import Host, LoadDatafrom home_application.utils.job_api import FastJobApiperform_script = "" #! / bin/bashMEMORY=$ (free-m | awk 'NR==2 {printf "% .2f%", $3x100) 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 | $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_exemptdef 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 file content

From django.db import modelsfrom home_application.utils.parse_time import parse_datetime_to_timestrclass Host (models.Model): host_id = models.IntegerField (u "host ID", primary_key=True, unique=True) ip = models.CharField (u "IP address", max_length=32, blank=True, null=True) bk_biz_id = models.CharField (u "business ID", max_length=16, blank=True, null=True) bk_biz_name = models.CharField (u "business name", max_length=512, blank=True Null=True) os_name = models.CharField (u "system name", max_length=128, blank=True, null=True) host_name = models.CharField (u "hostname", max_length=128, blank=True, null=True) cloud_id = models.IntegerField (u "cloud region ID", blank=True, null=True) cloud_name = models.CharField (u "cloud region name", max_length=32, blank=True, null=True) is_monitored = models.BooleanField (u "whether monitored" Default=False) def to_dict (self): return {"host_id": self.host_id, "ip": self.ip, "Competition": 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 "host ID", default=0) cpu = models.IntegerField (u "CPU usage", default=0) mem = models.IntegerField (u "memory usage" Default=0) disk = models.IntegerField (u "hard disk usage", default=0) create_time = models.DateTimeField (u "creation time", 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 partial code

Import base64import timefrom conf.default import APP_ID, APP_TOKENfrom blueking.component.shortcuts import get_client_by_user, get_client_by_requestcount = 0class 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): "" Quick execute script: param execute_account: execute script account name: param param_content: execute script parameters: param script_timeout: script execution timeout: 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): "" get script execution status: param task_id: job_instance_id: return: "global count count + = 1 # query execution status 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

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.