做一个可通过jenkins定时任务Cron表达式设置的python定时函数

本文涉及的产品
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
全局流量管理 GTM,标准版 1个月
云解析 DNS,旗舰版 1个月
简介: 用python代码,来解析jenkins定时任务表达式,并获取最近的执行任务时间戳

image.png

jenkins定时任务Cron表达式简介

jenkins是目前市场上最主流最好用的持续集成平台,几乎每个科技公司都在使用:

image.png

jenkins定时任务一直以来非常受欢迎,通过简单的命令,既可以实现几乎无所不能的定时需求。

命令举例如下:

image.png

背景

笔者之前所在的企业里,经常会有很多内部自动化平台或者自动跑批等平台开发出来,之前这些平台的定时任务都是交给公司内唯一的一套jenkins来执行触发。


但可能是因为要支持的任务太多,jnekins表现越来越卡了,甚至一度宕机。

那天,运维同事终于忍不了了,叫我们自己去搭建jenkins,公司内的唯一一套jenkins不给我们随便玩了...


于是,各组纷纷搭建自己的jenkins服务,但因为对运维方面的知识并不专业,加上也没有专用的好服务器来运行jenksin,导致服务异常不稳。后来我们总结了原因,领导说,一个平台所需要的支撑服务越多,链路越长,就越不稳定。我们之前成功跨越了组件化阶段,实现了业务/技术中台,虽然省心。但现在来看,反而可能会牵一发动全身,非常被动也不灵活。


于是他们决定要把部分不稳定的支持服务回退到组件化阶段!这其中,就包括了jenkins服务。

用领导的意思说就是,我们很多内部自动化平台,其实仅仅就是用一下jenkins的定时任务罢了,没必要单独部署一整套jenkins还要占个服务器。既然公司共用的jenkins不给我们玩了,那我们自己做一套jenkins的定时任务工具就好,然后最好做成可pip install 直接下载的第三方库,以后用的时候直接导入就用,轻便快捷,还稳定。


乍一听,不就是一个时间定时函数么?能有啥难得,于是大家纷纷表示可以搞。但下一个瞬间,大家突然发现好像不对....

jenkins的定时表达式,读起来非常简单,但如果解析并实现这个底层貌似有些难搞呀?


于是在领导问谁来负责的时候,不出意外的,大家朝着我伸出了他们那罪恶的食指....


我的思考

要解析这个表达式,并成功按照指定时间触发。我觉得最好的办法,就是根据当前时间,计算出下一次要触发的时间,然后一个轮询等到时间就执行就好。

结果里面居然引发出来好多难点。比如如何计算出下一次触发的时间,如何要节省的去轮询?多少秒一次?允许误差是几秒?是否需要做个高幂等性锁来防止触发两次?等等问题,让我感到压力。


不过好在,把难题拆分后,还是可以坚持搞下去的。否则我真的当场GG了。


正文

为了贯彻我们领导的尽量精简链路思想,所以我选择用python,并且只用一个函数来实现。函数接收jenkins的定时任务表达式字符串为唯一入参。返回为距离现在最近一次要触发的时间戳。因为时间戳在我们具体开发中才是易于使用的而非给人看的日期格式。

简单易用!

def get_next_time(s='* * * * *'):
    if s == '* * * * *':
        return time.time() + 60

    s = s.split(' ')
    for i in range(len(s)):
        fg = ''
        if '/' in s[i]:
            fg = int(s[i].split('/')[1])
            s[i] = s[i].split('/')[0]
        if '-' in s[i]:
            s[i] = list(range(int(s[i].split('-')[0]),int(s[i].split('-')[1])+1))
        elif ',' in s[i]:
            s[i] = [int(j) for j in s[i].split(',')]
        elif '*' in s[i]:
            s[i] = [list(range(0,60)),list(range(0,24)),list(range(1,32)),list(range(1,13)),list(range(1,8))][i]
        else:
            s[i] = [int(s[i])]
        if fg:
            s[i] = list(range(s[i][0],s[i][-1]+1,fg))

    tmp = list(itertools.product(s[0],s[1],s[2],s[3]) )
    all = []
    for i in tmp:
        t = '%s-%s-%s %s:%s:00'%(time.localtime(time.time())[0],i[3],i[2],i[1],i[0])
        try:
            s_t = time.strptime(t, "%Y-%m-%d %H:%M:%S")
        except:
            continue
        mkt = int(time.mktime(s_t))
        if mkt > time.time():
            rq = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(mkt))
            dw = datetime.strptime(rq.split(' ')[0],'%Y-%m-%d').isoweekday()
            if dw in s[-1]:
                all.append(mkt)
    all.sort()
    if all:
        return all[0]
    else:
        return None

简单来讲解一下(虽然这是第一版,并不完美,也没有完全解析的了jenkins表达式。但是已经足够我们组内使用了!)


思路:

先把表达式拆分成每一段:分、时、日、月、星期

然后把每一段符合的时间数字,做成一个列表。最终整个表达式字符串就变成了一个二维大列表。

然后这里每个列表的不同组合,就是所有符合的时间节点了。

我们再从者所有时间节点中,找到距离现在最近的下一个执行时间节点即可。

但是要进行时间排序的话,干脆就全变成时间戳,然后通过sort方法自动排序快。然后截取现在时间戳之后的片段,再拿到最小的那个,就是最终结果了!就是下一个要执行触发的时间节点。


首先,就是把这个表达式按照空格进行了拆解,每一部分再分析其所属类别:/ - , * 其他等等。

针对每个部分,都安排了算法进行解析。(注意,尽量避免出现60分或24时,可用0代表,也可以自行修改设计。我们组有自己业务上的考虑才故意这样写...)

对每个部分的可能值都放进了一个列表。

所以中间就变成了一个二维列表。

举例:希望每天的6:30 启动触发任务

image.png


经过前半部分的分段解析后,二维列表如下:

image.png

这就是当前,可以触发的所有时间日期节点。

然后我们再在里面找出距离现在时间最近的一个未来时间戳。

这个未来时间都是放在all变量中,all变量经过.sort()后会自动排序,从小到大。所以用时间戳而不是具体日期格式来计算出最近节点是可靠并简单的。

所以我们返回这个all[0]为最终输出。

结果如下:

image.png

那么这个时间到底对不对呢?

看一下我限制写这篇文章的时间:是6.5号深夜了

image.png

然后我们把这个时间戳翻译成日期:(见证奇迹的时刻来了!!!)

image.png


计算出的最近触发时间节点是:6.6号的早上6:30

完全正确!非常nice!


好,到此算是初步完成了领导的要求了。

虽然领导表示,你这个算法貌似不够完善,也不够健壮,更不符合pep8风格标准,甚至可能还存在着边界值bug。

但是整个的这个思路设计,非常好啊!非常值得组内推广!

于是,上传成了pypi的一个工具:python_jenkins_monitor

以后组员下载方便了,直接命令行:pip3 install python_jenkins_monitor 即可。

image.png

image.png

虽然后续该代码被领导给优化过了(加了好多标点符号和注释既视感),但是狗领导不给更新到pypi了。

大家如果想用的,自行优化下,或者留言讨论下吧~



相关文章
|
4月前
|
调度 索引 Python
python3创建定时任务
python3创建定时任务
44 2
|
1月前
|
Python
Python编程中正则表达式的使用
【10月更文挑战第22天】正则表达式,一种强大的文本处理工具,在Python编程中有着广泛的应用。本文将介绍如何使用Python中的re库来使用正则表达式,包括如何创建、匹配、查找和替换字符串等。通过学习本文,你将能够掌握Python中正则表达式的基本使用方法。
|
3月前
|
Python
Python中正则表达式(re模块)用法详解
Python中正则表达式(re模块)用法详解
54 2
|
3月前
|
数据采集 NoSQL 调度
flask celery python 每月定时任务
flask celery python 每月定时任务
|
3月前
|
API Python
使用Python requests库下载文件并设置超时重试机制
使用Python的 `requests`库下载文件时,设置超时参数和实现超时重试机制是确保下载稳定性的有效方法。通过这种方式,可以在面对网络波动或服务器响应延迟的情况下,提高下载任务的成功率。
187 1
|
2月前
|
程序员 Python
Python中Lambda表达式的优缺点及使用场景
Python中Lambda表达式的优缺点及使用场景
42 0
|
2月前
|
存储 算法 API
Python学习五:函数、参数(必选、可选、可变)、变量、lambda表达式、内置函数总结、案例
这篇文章是关于Python函数、参数、变量、lambda表达式、内置函数的详细总结,包含了基础知识点和相关作业练习。
33 0
|
2月前
|
Python 容器
AutoDL Python实现 自动续签 防止实例过期释放 小脚本 定时任务 apscheduler requests
AutoDL Python实现 自动续签 防止实例过期释放 小脚本 定时任务 apscheduler requests
40 0
|
3月前
|
Python
python之定时任务schedule
python之定时任务schedule
|
4月前
|
运维 Java API
探索Java中的Lambda表达式自动化运维的魔法:如何利用Python脚本提升效率
【8月更文挑战第29天】Lambda表达式是Java 8中引入的一个新特性,它允许我们将功能作为方法参数,或者代码作为数据来处理。在这篇文章中,我们将深入探讨Java中的Lambda表达式,包括它的语法、使用场景以及如何在实际编程中应用它。我们将通过一些简单的示例来演示Lambda表达式的强大功能和灵活性,让你更好地理解和掌握这一新特性。
下一篇
DataWorks