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

本文涉及的产品
全局流量管理 GTM,标准版 1个月
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
云解析 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了。

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



相关文章
|
1月前
|
Python
【python从入门到精通】-- 第五战:函数大总结
【python从入门到精通】-- 第五战:函数大总结
66 0
|
29天前
|
Python
Python之函数详解
【10月更文挑战第12天】
Python之函数详解
|
30天前
|
存储 数据安全/隐私保护 索引
|
20天前
|
测试技术 数据安全/隐私保护 Python
探索Python中的装饰器:简化和增强你的函数
【10月更文挑战第24天】在Python编程的海洋中,装饰器是那把可以令你的代码更简洁、更强大的魔法棒。它们不仅能够扩展函数的功能,还能保持代码的整洁性。本文将带你深入了解装饰器的概念、实现方式以及如何通过它们来提升你的代码质量。让我们一起揭开装饰器的神秘面纱,学习如何用它们来打造更加优雅和高效的代码。
|
22天前
|
Python
Python编程中正则表达式的使用
【10月更文挑战第22天】正则表达式,一种强大的文本处理工具,在Python编程中有着广泛的应用。本文将介绍如何使用Python中的re库来使用正则表达式,包括如何创建、匹配、查找和替换字符串等。通过学习本文,你将能够掌握Python中正则表达式的基本使用方法。
|
22天前
|
弹性计算 安全 数据处理
Python高手秘籍:列表推导式与Lambda函数的高效应用
列表推导式和Lambda函数是Python中强大的工具。列表推导式允许在一行代码中生成新列表,而Lambda函数则是用于简单操作的匿名函数。通过示例展示了如何使用这些工具进行数据处理和功能实现,包括生成偶数平方、展平二维列表、按长度排序单词等。这些工具在Python编程中具有高度的灵活性和实用性。
|
25天前
|
Python
python的时间操作time-函数介绍
【10月更文挑战第19天】 python模块time的函数使用介绍和使用。
27 4
|
26天前
|
存储 Python
[oeasy]python038_ range函数_大小写字母的起止范围_start_stop
本文介绍了Python中`range`函数的使用方法及其在生成大小写字母序号范围时的应用。通过示例展示了如何利用`range`和`for`循环输出指定范围内的数字,重点讲解了小写和大写字母对应的ASCII码值范围,并解释了`range`函数的参数(start, stop)以及为何不包括stop值的原因。最后,文章留下了关于为何`range`不包含stop值的问题,留待下一次讨论。
19 1
|
1月前
|
索引 Python
Python中的其他内置函数有哪些
【10月更文挑战第12天】Python中的其他内置函数有哪些
15 1
|
26天前
|
安全 数据处理 数据安全/隐私保护
python中mod函数怎么用
通过这些实例,我们不仅掌握了Python中 `%`运算符的基础用法,还领略了它在解决实际问题中的灵活性和实用性。在诸如云计算服务提供商的技术栈中,类似的数学运算逻辑常被应用于数据处理、安全加密等关键领域,凸显了基础运算符在复杂系统中的不可或缺性。
18 0