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

本文涉及的产品
云解析 DNS,旗舰版 1个月
云解析DNS,个人版 1个月
全局流量管理 GTM,标准版 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了。

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



相关文章
|
6天前
|
Python
高阶函数如`map`, `filter`, `reduce`和`functools.partial`在Python中用于函数操作
【6月更文挑战第20天】高阶函数如`map`, `filter`, `reduce`和`functools.partial`在Python中用于函数操作。装饰器如`@timer`接收或返回函数,用于扩展功能,如记录执行时间。`timer`装饰器通过包裹函数并计算执行间隙展示时间消耗,如`my_function(2)`执行耗时2秒。
15 3
|
3天前
|
测试技术 开发者 Python
Python中的装饰器:提升函数的灵活性和可重用性
在Python编程中,装饰器是一种强大的工具,它可以在不修改函数本身的情况下,动态地扩展函数的功能。本文将介绍装饰器的工作原理及其在实际开发中的应用,帮助读者更好地理解和利用这一特性。
|
2天前
|
存储 Python
在Python中,匿名函数(lambda表达式)是一种简洁的创建小型、一次性使用的函数的方式。
【6月更文挑战第24天】Python的匿名函数,即lambda表达式,用于创建一次性的小型函数,常作为高阶函数如`map()`, `filter()`, `reduce()`的参数。lambda表达式以`lambda`开头,后跟参数列表,冒号分隔参数和单行表达式体。例如,`lambda x, y: x + y`定义了一个求和函数。在调用时,它们与普通函数相同。例如,`map(lambda x: x ** 2, [1, 2, 3, 4, 5])`会返回一个列表,其中包含原列表元素的平方。
13 4
|
3天前
|
JSON 数据格式 索引
Python内置函数如`print()`输出信息,`len()`计算长度
【6月更文挑战第23天】Python内置函数如`print()`输出信息,`len()`计算长度,`type()`识别类型,`range()`生成序列,`sum()`求和,`min()`和`max()`找极值,`abs()`取绝对值,`round()`四舍五入,`sorted()`排序,`zip()`和`enumerate()`组合及遍历,`map()`和`filter()`应用函数。标准库如`os`用于操作系统交互,`sys`处理解释器信息,`math`提供数学运算,`re`支持正则表达式,`json`处理JSON数据。学习这些能提升编程效率。
17 5
|
2天前
|
Python
在Python中,高阶函数是指那些可以接受一个或多个函数作为参数,并返回一个新的函数的函数。
【6月更文挑战第24天】Python的高阶函数简化代码,增强可读性。示例:`map()`检查用户名合法性,如`["Alice", "Bob123", "Charlie!", "David7890"]`;`reduce()`与`lambda`结合计算阶乘,如1到10的阶乘为3628800;`filter()`找出1到100中能被3整除的数,如[3, 6, 9, ..., 99]。
12 3
|
1天前
|
jenkins 持续交付 API
使用Python操作Jenkins的过程详解
Python作为一种简洁、灵活且功能丰富的编程语言,可以与各种API轻松集成,Jenkins的API也不例外。借助于Python中的python-jenkins模块,我们可以轻松地编写脚本来连接到Jenkins服务器,并执行各种操作,如创建、删除、构建Jobs等。这种自动化的方式不仅提高了效率,还使得CI/CD流程更加灵活和可控。
|
3天前
|
分布式计算 大数据 调度
MaxCompute产品使用问题之为什么用python写的udf函数跑起来比本地还要慢
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。
|
6天前
|
数据安全/隐私保护 Python
Python装饰器是高阶函数,用于在不修改代码的情况下扩展或修改函数行为。它们提供可重用性、模块化和无侵入性的功能增强。
【6月更文挑战第20天】Python装饰器是高阶函数,用于在不修改代码的情况下扩展或修改函数行为。它们提供可重用性、模块化和无侵入性的功能增强。例如,`@simple_decorator` 包装`my_function`,在调用前后添加额外操作。装饰器还能接受参数,如`@logged("INFO", "msg")`,允许动态定制功能。
16 6
|
3天前
|
SQL 分布式计算 大数据
MaxCompute产品使用问题之建了一个python 的 UDF脚本,生成函数引用总是说类不存在,是什么导致的
MaxCompute作为一款全面的大数据处理平台,广泛应用于各类大数据分析、数据挖掘、BI及机器学习场景。掌握其核心功能、熟练操作流程、遵循最佳实践,可以帮助用户高效、安全地管理和利用海量数据。以下是一个关于MaxCompute产品使用的合集,涵盖了其核心功能、应用场景、操作流程以及最佳实践等内容。