Dockerfile、CronJob、Deployment

    1. FROM python:3.7.9
    2. COPY requirements.txt xjob_alert.py /
    3. RUN apt-get update && apt-get -y install cron vim
    4. RUN pip install --no-cache-dir -r requirements.txt
    5. RUN chmod 0644 xjob_alert.py
    6. RUN touch /var/log/xfep.log /var/log/xpair.log /var/log/status2.log /var/log/status109.log
    7. RUN crontab -l | { cat ; echo "*/3 * * * * /usr/local/bin/python /xjob_alert.py xpair >>/var/log/xpair.log 2>&1"; } |crontab -
    8. RUN crontab -l | { cat ; echo "*/10 * * * * /usr/local/bin/python /xjob_alert.py xfep >>/var/log/xfep.log 2>&1"; } |crontab -
    9. RUN crontab -l | { cat ; echo "*/5 * * * * /usr/local/bin/python /xjob_alert.py status2 >>/var/log/status2.log 2>&1"; } |crontab -
    10. RUN crontab -l | { cat ; echo "0 */6 * * * /usr/local/bin/python /xjob_alert.py status109 >>/var/log/status109.log 2>&1"; } |crontab -
    11. CMD cron && tail -f /var/log/xpair.log
    1. ---
    2. apiVersion: batch/v1beta1
    3. kind: CronJob
    4. metadata:
    5.   name: xpair-alert
    6. spec:
    7.   schedule: "*/5 * * * *"
    8.   successfulJobsHistoryLimit: 3
    9.   failedJobsHistoryLimit: 3
    10.   jobTemplate:
    11.     spec:
    12.       template:
    13.         spec:
    14.           containers:
    15.           - name: xjob-alert
    16.             image: 634192295060.dkr.ecr.us-east-1.amazonaws.com/development/xjob_alert:0.0.2
    17.             imagePullPolicy: IfNotPresent
    18.             command:
    19.             - /usr/bin/python
    20.             - xjob_alert.py
    21.             - xpair
    22.           restartPolicy: OnFailure
    25. ---
    26. apiVersion: batch/v1beta1
    27. kind: CronJob
    28. metadata:
    29.   name: xfep-alert
    30. spec:
    31.   schedule: "*/30 * * * *"
    32.   successfulJobsHistoryLimit: 3
    33.   failedJobsHistoryLimit: 3
    34.   jobTemplate:
    35.     spec:
    36.       template:
    37.         spec:
    38.           containers:
    39.           - name: xjob-alert
    40.             image: 634192295060.dkr.ecr.us-east-1.amazonaws.com/development/xjob_alert:0.0.2
    41.             imagePullPolicy: IfNotPresent
    42.             command:
    43.             - /usr/bin/python
    44.             - xjob_alert.py
    45.             - xfep
    46.           restartPolicy: OnFailure
    1. apiVersion: apps/v1
    2. kind: Deployment
    3. metadata:
    4.   labels:
    5.     app: xjob-alert
    6.   name: xjob-alert
    7.   namespace: production
    8. spec:
    9.   replicas: 1
    10.   revisionHistoryLimit: 10
    11.   selector:
    12.     matchLabels:
    13.       app: xjob-alert
    14.   template:
    15.     metadata:
    16.       labels:
    17.         app: xjob-alert
    18.     spec:
    19.       containers:
    20.       - name: xjob-alert
    21.         image: 634192295060.dkr.ecr.us-east-1.amazonaws.com/development/xjob_alert:0.0.3
    22.         imagePullPolicy: Always
    23.       restartPolicy: Always
    24.       imagePullSecrets:
    25.       - name: regcred
    1. #!/usr/bin/python
    2. # -*- coding: UTF-8 -*-
    3. import psycopg2
    4. import requests as requests
    5. from psycopg2 import sql
    6. import json, os, time, sys
    8. class NewDrugAlert:
    10.     def __init__(self, env):
    11.         if env == 'pro':
    12.             self.parms = {
    13.                 "database": "anakin",
    14.                 "user": "anakin",
    15.                 "password": "81b88ebb893",
    16.                 "host": "luke-pro.ct9zn8ktuvkc.us-east-1.rds.amazonaws.com",
    17.                 "port": "5432",
    18.             }
    20.         if env == 'dev' or env == '':
    21.             self.parms = {
    22.                 "database": "anakin",
    23.                 "user": "postgres",
    24.                 "password": "xtalpi123",
    25.                 "host": "luke-postgres.development.svc",
    26.                 "port": "5432",
    27.             }
    29.     def job(self,sql,job_type):
    30.         conn = psycopg2.connect(**self.parms)
    31.         cur = conn.cursor()
    32.         cur.execute(sql)
    33.         rows = cur.fetchall()
    34.         conn.close()
    35.         print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),":",job_type,"job select results:",rows)
    37.         return rows
    39.     def alert(self,msg,job_type,bussiness):
    40.         #test:fs_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/e9a0edf5-46f2-44c4-8b67-b2d12a8579e2'
    41.         #guyan:fs_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/4663ac9b-bb95-4fbc-9124-c83e347a5f89'
    42.         #xinyao:fs_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/8d738505-682a-4fc7-8cb0-21c5ea1e2d2a'
    44.         if bussiness == "guyan":
    45.             fs_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/4663ac9b-bb95-4fbc-9124-c83e347a5f89'
    46.         elif bussiness == "xinyao":
    47.             fs_url = 'https://open.feishu.cn/open-apis/bot/v2/hook/8d738505-682a-4fc7-8cb0-21c5ea1e2d2a'
    48.         headers = {'Content-Type': 'application/json;charset=utf-8'}
    49.         text = {"msg_type": "text", "content": {"msg":"","text":msg}}
    50.         res = requests.post(fs_url, json.dumps(text), headers=headers).content
    51.         print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),":",job_type,"job alert results:",res)
    53.         return res
    55.     def xpair(self):
    56.         job_type = "xpair"
    57.         sql = "select handle,\"user\",status,cost_dimension,\
    58.             date_part('minute',now()::timestamp-to_char(to_timestamp(create_at),'yyyy-MM-dd HH24:MI:SS')::timestamp) as time_diff \
    59.             from job where status in (1,2,3) \
    60.             and image LIKE '%tasksys_fep_map_py%' \
    61.             and \"group\"::text LIKE '%xpair%' \
    62.             and date_part('minute',now()::timestamp-to_char(to_timestamp(create_at),'yyyy-MM-dd HH24:MI:SS')::timestamp) >= 5 \
    63.             order by create_at desc;"
    65.         job_res=NewDrugAlert.job(self,sql,job_type)
    68.         alert_msg = ''
    69.         over_5 = "xpair任务1、2、3状态超过5分钟："
    70.         over_10 = "xpair任务1、2、3状态超过10分钟："
    71.         over_20 = "xpair任务1、2、3状态超过20分钟："
    73.         if job_res:
    74.             for m in job_res:
    75.                 if m[-1] >= 5 and m[-1] < 10:
    76.                     over_5 = over_5 + "\n" + str(m)
    77.                 elif m[-1] >= 10 and m[-1] < 20:
    78.                     over_10 = over_10 + "\n" + str(m)
    79.                 else:
    80.                     over_20 = over_20 + "\n" + str(m) 
    81.             alert_msg = alert_msg + over_20 + "\n" + over_10 + "\n" + over_5
    82.             xpair_alert = NewDrugAlert.alert(self,alert_msg,job_type,bussiness="xinyao")
    83.             print(alert_msg)
    85.     def xfep(self):
    86.         job_type = "xfep"
    87.         sql = "select handle,\"user\",status,cost_dimension,\
    88.         date_part('minute',now()::timestamp-to_char(to_timestamp(create_at),'yyyy-MM-dd HH24:MI:SS')::timestamp) as time_diff \
    89.         from job where cluster_id in ('gcloud','DefaultGPU') \
    90.         and status in (1,2,3) \
    91.         and image like '%tasksys_fep_py3%' \
    92.         and date_part('minute',now()::timestamp-to_char(to_timestamp(create_at),'yyyy-MM-dd HH24:MI:SS')::timestamp) > 30 \
    93.         order by time_diff desc;"
    95.         job_res=NewDrugAlert.job(self,sql,job_type)
    97.         alert_msg = 'xfep任务1、2、3状态持续超过30分钟：'
    99.         if job_res:
    100.             for m in job_res:
    101.                 alert_msg = alert_msg + "\n" + str(m)
    102.             xfep_alert = NewDrugAlert.alert(self,alert_msg,job_type,bussiness="xinyao")
    103.             print(alert_msg)
    105.     def status2(self):
    106.         job_type = "status 2"
    107.         sql = "select handle,\"user\",status,cluster_id,cost_dimension,job_times from  job \
    108.         where status in (2) \
    109.         and image like '%tasksys_fep_py3%' \
    110.         and cluster_id in ('gcloud','DefaultGPU') \
    111.         order by status desc, job_times asc"
    113.         job_res=NewDrugAlert.job(self,sql,job_type)
    115.         now_time = time.time()
    116.         alert_msg = 'gcloud、DefaultGPU集群2状态持续超过30分钟（请确认队列是否阻塞）：'
    117.         if job_res:
    118.             for m in job_res:
    119.                 time_diff = int((now_time - m[-1][-1][-1]) / 60)
    120.                 if time_diff > 30:
    121.                     alert_msg = alert_msg + "\n" + str(m[0]) + "," + str(m[1]) + "," + str(m[3]) + "," + str(m[4]) + "," + str(time_diff) + "min"
    122.             if alert_msg.count("xtalpi") > 0:
    123.                 status2_alert = NewDrugAlert.alert(self,alert_msg,job_type,bussiness="xinyao")
    124.                 print(alert_msg)
    126.     def status109(self):
    127.         job_type = "status 109"
    129.         sql = "select handle,\"user\",status,cluster_id,cost_dimension, \
    130.         to_char(to_timestamp(create_at),'yyyy-MM-dd HH24:MI:SS') as create_at,job_times from job \
    131.         where status in (4,109) \
    132.         and job_times::text like '%109%' \
    133.         and date_part('day',now()::timestamp-to_char(to_timestamp(create_at),'yyyy-MM-dd HH24:MI:SS')::timestamp) < 30\
    134.         order by create_at desc"
    136.         job_res=NewDrugAlert.job(self,sql,job_type)
    138.         alert_msg = "近一个月有被驱逐过记录的且目前未终止状态的handle："
    139.         if job_res:
    140.             for row in job_res:
    141.                 for m in row[-1]:
    142.                     if m[0] == 109:
    143.                         alert_msg = alert_msg + "\n" + str(row[0])\
    144.                         + "," + str(row[1]) + "," + str(row[2])  + "," + str(row[5])
    145.                         break
    146.             status109_alert = NewDrugAlert.alert(self,alert_msg,job_type,bussiness="guyan")
    147.             print(alert_msg)
    149. if __name__ == '__main__':
    150.     LUKE_ENV = 'pro'
    151.     alert = NewDrugAlert(LUKE_ENV)
    152.     if sys.argv[1] and sys.argv[1] == "xfep":
    153.         alert.xfep()
    154.     elif sys.argv[1] and sys.argv[1] == "xpair":
    155.         alert.xpair()
    156.     elif sys.argv[1] and sys.argv[1] == "status2":
    157.         alert.status2()
    158.     elif sys.argv[1] and sys.argv[1] == "status109":
    159.         alert.status109()