开源运维平台：面向中小型企业设计的无 Agent的自动化运维平台，整合了主机管理、主机批量执行、主机在线终端、等一系列功能

# Copyright: (c) OpenSpug Organization. https://github.com/openspug/spug # Copyright: (c) <spug.dev@gmail.com> # Released under the AGPL-3.0 License. from django.views.generic import View from django.db.models import F from django.conf import settings from django.http.response import HttpResponseBadRequest from django_redis import get_redis_connection from libs import json_response, JsonParser, Argument, human_datetime, human_time, auth from apps.deploy.models import DeployRequest from apps.app.models import Deploy, DeployExtend2 from apps.repository.models import Repository from apps.deploy.utils import dispatch, Helper from apps.host.models import Host from collections import defaultdict from threading import Thread from datetime import datetime import subprocess import json import os class RequestView(View): @auth('deploy.request.view') def get(self, request): data, query, counter = [], {}, {} if not request.user.is_supper: perms = request.user.deploy_perms query['deploy__app_id__in'] = perms['apps'] query['deploy__env_id__in'] = perms['envs'] for item in DeployRequest.objects.filter(**query).annotate( env_id=F('deploy__env_id'), env_name=F('deploy__env__name'), app_id=F('deploy__app_id'), app_name=F('deploy__app__name'), app_host_ids=F('deploy__host_ids'), app_extend=F('deploy__extend'), rep_extra=F('repository__extra'), do_by_user=F('do_by__nickname'), approve_by_user=F('approve_by__nickname'), created_by_user=F('created_by__nickname')): tmp = item.to_dict() tmp['env_id'] = item.env_id tmp['env_name'] = item.env_name tmp['app_id'] = item.app_id tmp['app_name'] = item.app_name tmp['app_extend'] = item.app_extend tmp['host_ids'] = json.loads(item.host_ids) tmp['fail_host_ids'] = json.loads(item.fail_host_ids) tmp['extra'] = json.loads(item.extra) if item.extra else None tmp['rep_extra'] = json.loads(item.rep_extra) if item.rep_extra else None tmp['app_host_ids'] = json.loads(item.app_host_ids) tmp['status_alias'] = item.get_status_display() tmp['created_by_user'] = item.created_by_user tmp['approve_by_user'] = item.approve_by_user tmp['do_by_user'] = item.do_by_user if item.app_extend == '1': tmp['visible_rollback'] = item.deploy_id not in counter counter[item.deploy_id] = True data.append(tmp) return json_response(data) @auth('deploy.request.del') def delete(self, request): form, error = JsonParser( Argument('id', type=int, required=False), Argument('mode', filter=lambda x: x in ('count', 'expire', 'deploy'), required=False, help='参数错误'), Argument('value', required=False), ).parse(request.GET) if error is None: if form.id: deploy = DeployRequest.objects.filter(pk=form.id).first() if not deploy or deploy.status not in ('0', '1', '-1'): return json_response(error='未找到指定发布申请或当前状态不允许删除') deploy.delete() return json_response() count = 0 if form.mode == 'count': if not str(form.value).isdigit() or int(form.value) < 1: return json_response(error='请输入正确的保留数量') counter, form.value = defaultdict(int), int(form.value) for item in DeployRequest.objects.all(): counter[item.deploy_id] += 1 if counter[item.deploy_id] > form.value: count += 1 item.delete() elif form.mode == 'expire': for item in DeployRequest.objects.filter(created_at__lt=form.value): count += 1 item.delete() elif form.mode == 'deploy': app_id, env_id = str(form.value).split(',') for item in DeployRequest.objects.filter(deploy__app_id=app_id, deploy__env_id=env_id): count += 1 item.delete() return json_response(count) return json_response(error=error) class RequestDetailView(View): @auth('deploy.request.view') def get(self, request, r_id): req = DeployRequest.objects.filter(pk=r_id).first() if not req: return json_response(error='未找到指定发布申请') hosts = Host.objects.filter(id__in=json.loads(req.host_ids)) outputs = {x.id: {'id': x.id, 'title': x.name, 'data': f'{human_time()} 读取数据... '} for x in hosts} response = {'outputs': outputs, 'status': req.status} if req.is_quick_deploy: outputs['local'] = {'id': 'local', 'data': ''} if req.deploy.extend == '2': outputs['local'] = {'id': 'local', 'data': f'{human_time()} 读取数据... '} response['s_actions'] = json.loads(req.deploy.extend_obj.server_actions) response['h_actions'] = json.loads(req.deploy.extend_obj.host_actions) if not response['h_actions']: response['outputs'] = {'local': outputs['local']} rds, key, counter = get_redis_connection(), f'{settings.REQUEST_KEY}:{r_id}', 0 data = rds.lrange(key, counter, counter + 9) while data: for item in data: counter += 1 item = json.loads(item.decode()) if item['key'] in outputs: if 'data' in item: outputs[item['key']]['data'] += item['data'] if 'step' in item: outputs[item['key']]['step'] = item['step'] if 'status' in item: outputs[item['key']]['status'] = item['status'] data = rds.lrange(key, counter, counter + 9) response['index'] = counter if counter == 0: for item in outputs: outputs[item]['data'] += '\r\n\r\n未读取到数据，Spug 仅保存最近2周的日志信息。' if req.is_quick_deploy: if outputs['local']['data']: outputs['local']['data'] = f'{human_time()} 读取数据... ' + outputs['local']['data'] else: outputs['local'].update(step=100, data=f'{human_time()} 已构建完成忽略执行。') return json_response(response) @auth('deploy.request.do') def post(self, request, r_id): form, _ = JsonParser(Argument('mode', default='all')).parse(request.body) query = {'pk': r_id} if not request.user.is_supper: perms = request.user.deploy_perms query['deploy__app_id__in'] = perms['apps'] query['deploy__env_id__in'] = perms['envs'] req = DeployRequest.objects.filter(**query).first() if not req: return json_response(error='未找到指定发布申请') if req.status not in ('1', '-3'): return json_response(error='该申请单当前状态还不能执行发布') host_ids = req.fail_host_ids if form.mode == 'fail' else req.host_ids hosts = Host.objects.filter(id__in=json.loads(host_ids)) message = f'{human_time()} 等待调度... ' outputs = {x.id: {'id': x.id, 'title': x.name, 'step': 0, 'data': message} for x in hosts} req.status = '2' req.do_at = human_datetime() req.do_by = request.user req.save() Thread(target=dispatch, args=(req, form.mode == 'fail')).start() if req.is_quick_deploy: if req.repository_id: