蓝鲸平台Celery后台任务进度获取代码

墨蓝 2022-04-22 11:18 298阅读 0赞

**celery\_tasks.py**

# -*- coding: utf-8 -*-
    """
    Tencent is pleased to support the open source community by making 蓝鲸智云(BlueKing) available.
    Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
    Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
    You may obtain a copy of the License at http://opensource.org/licenses/MIT
    Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
    an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and limitations under the License.
    
    celery 任务示例
    
    本地启动celery命令: python  manage.py  celery  worker  --settings=settings
    周期性任务还需要启动celery调度命令：python  manage.py  celerybeat --settings=settings
    """
    import datetime
    import json
    import traceback
    import urllib2
    import base64
    from celery.task import task
    import time
    from celery.schedules import crontab
    from celery.task import periodic_task
    from common.log import logger
    
    from home_application.dce import DCE_MAIN_IP, ADMIN_NAME, ADMIN_PWD, check_deployment_podcnt
    
    
    @task(bind=True)
    def update_deployment(self, app_name, deployment_name, replicas, labels, image, container_port, cpu, memory,
                          health_check_url, namespace='default'):
        # 缩replicas为replicas-1，空出一个ip
        post_url = 'http://' + DCE_MAIN_IP + '/apis/apps/v1/namespaces/' + namespace + '/deployments/' + deployment_name
        data = {"metadata": {"annotations": {"kubernetes.io/change-cause": "update replica"}},
                "spec": {"replicas": replicas - 1}}
        req = urllib2.Request(post_url, json.dumps(data))
        req.add_header('Content-Type', 'application/strategic-merge-patch+json')
        req.add_header("Authorization", "Basic " + base64.b64encode(ADMIN_NAME + ':' + ADMIN_PWD))
        req.get_method = lambda: 'PATCH'
        print(data)
        try:
            response = urllib2.urlopen(req)
            if response.code == 200:
                # result=json.loads(response.readline())
                print('replicas->replicas-1')
        except:
            print('error para:url:' + post_url)
            print('error data:' + json.dumps(data))
            traceback.print_exc()
            return 'RequestError'
    
        # 等待缩replicas完毕
        i = 0
        while check_deployment_podcnt(deployment_name, namespace) != replicas - 1 and i < 100:
            i += 5
            self.update_state(state='PROGRESS', meta={'i': i})
            time.sleep(5)
            print('sleep 5')
    
        labels = {} if not labels else labels  # 防止为None
        post_url = 'http://' + DCE_MAIN_IP + '/apis/apps/v1beta1/namespaces/' + namespace + '/deployments/' + deployment_name
        data = {
            "apiVersion": "apps/v1beta1",
            "kind": "Deployment",
            "metadata": {
                "name": deployment_name,
                "labels": {
                    "dce.daocloud.io/app": app_name,
                    "dce.daocloud.io/component": deployment_name
                }
            },
            "spec": {
                "terminationGracePeriodSeconds": 66,
                "selector": {
                    "matchLabels": {
                        "dce.daocloud.io/component": deployment_name}},
                "revisionHistoryLimit": 10,
                "template": {
                    "spec": {
                        "volumes": [
                            {"name": deployment_name + "-logs",
                             "hostPath":
                                 {"path": "/logs/" + deployment_name,
                                  "type": ""}}],
                        "containers": [
                            {
                                "image": image,
                                "name": deployment_name,
                                "livenessProbe":  # kubernetes认为该pod是存活的,不存活则需要重启
                                    {"httpGet":
                                         {"path": health_check_url,
                                          "port": container_port,
                                          "scheme": "HTTP"},
                                     "initialDelaySeconds": 60,
                                     # equals to the maximum startup time of the application + couple of seconds
                                     "timeoutSeconds": 5,
                                     "successThreshold": 1,
                                     "failureThreshold": 5},
                                "readinessProbe":  # kubernetes认为该pod是启动成功的
                                    {"httpGet":
                                         {"path": health_check_url,
                                          "port": container_port,
                                          "scheme": "HTTP", },
                                     "initialDelaySeconds": 60,  # equals to minimum startup time of the application
                                     "timeoutSeconds": 5,
                                     "successThreshold": 1,
                                     "failureThreshold": 5},
                                "resources": {"limits":
                                                  {"cpu": cpu,
                                                   "memory": memory}},
                                "ports": [
                                    {
                                        "containerPort": container_port
                                    }
                                ],
                                "volumeMounts": [
                                    {"name": deployment_name + "-logs",
                                     "mountPath": "/home/tomcat/apache-tomcat-9.0.8/logs"}]
                            }
                        ],
                        "affinity": {
                            "nodeAffinity": {
                                "preferredDuringSchedulingIgnoredDuringExecution": [
                                    {"weight": 1,
                                     "preference":
                                         {"matchExpressions": [
                                             {"key": "as.stat",
                                              "operator": "In",
                                              "values":
                                                  ["active"]}
                                         ]}
                                     }]
                            },
                            "podAntiAffinity": {
                                "requiredDuringSchedulingIgnoredDuringExecution": [
                                    {"labelSelector":
                                        {"matchExpressions": [
                                            {"key": "name", "operator": "In", "values": [deployment_name]}
                                        ]}
                                        ,
                                        "topologyKey": "kubernetes.io/hostname"}
                                ]}}
                    },
                    "metadata": {
                        "name": deployment_name,
                        "labels": {
                            "dce.daocloud.io/app": app_name,
                            "dce.daocloud.io/component": deployment_name
                        }
                    }
                },
                "replicas": replicas - 1,
                "strategy":
                    {"rollingUpdate":  # 由于replicas为3,则整个升级,pod个数在2-3个之间
                         {"maxSurge": 0,  # 滚动升级时会先启动0个pod
                          "maxUnavailable": 1}}  # 滚动升级时允许的最大Unavailable的pod个数
            }
        }
        data['metadata']['labels'].update(labels)
        data['spec']['template']["metadata"]["labels"].update(labels)
        req = urllib2.Request(post_url, json.dumps(data))
        req.add_header('Content-Type', 'application/json')
        req.add_header("Authorization", "Basic " + base64.b64encode(ADMIN_NAME + ':' + ADMIN_PWD))
        req.get_method = lambda: 'PUT'
        print(data)
        try:
            response = urllib2.urlopen(req)
            if response.code == 200:
                # result=json.loads(response.readline())
                print('Created')
            else:
                return 'RequestError'
        except:
            print('error para:url:' + post_url)
            print('error data:' + json.dumps(data))
            traceback.print_exc()
            return 'RequestError'
    
        # 等待update完毕
        while check_deployment_podcnt(deployment_name, namespace) != replicas - 1 and i < 100:
            i += 5
            self.update_state(state='PROGRESS', meta={'i': i})
            time.sleep(5)
            print('sleep 5')
        # 缩replicas为replicas-1，空出一个ip
        post_url = 'http://' + DCE_MAIN_IP + '/apis/apps/v1/namespaces/' + namespace + '/deployments/' + deployment_name
        data = {"metadata": {"annotations": {"kubernetes.io/change-cause": "update replica"}},
                "spec": {"replicas": replicas}}
        req = urllib2.Request(post_url, json.dumps(data))
        req.add_header('Content-Type', 'application/strategic-merge-patch+json')
        req.add_header("Authorization", "Basic " + base64.b64encode(ADMIN_NAME + ':' + ADMIN_PWD))
        req.get_method = lambda: 'PATCH'
    
        print(data)
        try:
            response = urllib2.urlopen(req)
            if response.code == 200:
                # result=json.loads(response.readline())
                print('replicas-1->replicas')
                return 'Created'
        except:
            print('error para:url:' + post_url)
            print('error data:' + json.dumps(data))
            traceback.print_exc()
            return 'RequestError'

**[views.py][]**

def deploy_image_dce(proj_code, category, name, module_id, ver, urlpath, instance_num, node_port, cpu, memory):
    	async_task_id = '0'
     	if result == 'RequestError':
            return '{"code":"ERROR","info":"read_namespaced_deployment:RequestError"}'
        elif result == 'Find':
            # ---------------------------------------------Celery后台任务---------------------------------------------------
            async_result = update_deployment.delay(proj_code, deployment_name, replicas, labels, image, container_port, cpu,
                                                   memory, urlpath, namespace)
            async_task_id = async_result.id
            # ---------------------------------------------Celery后台任务---------------------------------------------------
        DCE_MAIN_IP = '1.1.1.1'
        return '{"code":"OK","async_task_id":"' + str(async_task_id) + '","info":"http://' + DCE_MAIN_IP + ':' + str(
            node_port) + (urlpath if urlpath else '/') + '"}'
    
    def task_status(request):
        """
        后台任务状态查询
        :param request:
        :return:
        """
        the_task_id = request.GET.get('the_task_id')
        the_task = update_deployment.AsyncResult(the_task_id)
        print_util('状态==>', "任务：{0} 当前的 state 为：{1}".format(the_task_id, the_task.state))
        if the_task.state == 'PROGRESS':
            resp = {'state': 'progress', 'progress': the_task.info.get('i', 0)}
        elif the_task.state == 'SUCCESS':
            resp = {'state': "success", 'progress': 100}
        elif the_task.state == 'PENDING':  # 任务处于排队之中
            resp = {'state': 'waiting', 'progress': 0}
        else:
            resp = {'state': the_task.state, 'progress': the_task.info.get('i', 0)}
        return JsonResponse(resp)

deploy.html

[views.py]: http://views.py