使用镜像搭建ntp服务器，并在客户端调用进行定时时间同步。

一、 ntp服务器搭建

查询到有现成的ntp服务器docker镜像， 并配有完整使用方法，参考链接如下：

• https://hub.docker.com/r/cturra/ntp

docker run启动容器，国内时间源可以通过环境变量指定：

docker run --name=ntp --restart=always --detach --publish=123:123/udp --env=NTP_SERVERS="ntp.ntsc.ac.cn,ntp1.aliyun.com" cturra/ntp

• 可通过docker exec ntp chronyc tracking查看ntp服务器状态：

Reference ID    : 78197314 (120.25.115.20)
Stratum         : 3
Ref time (UTC)  : Thu Dec 08 08:07:08 2022
System time     : 0.000000000 seconds fast of NTP time
Last offset     : -0.107805088 seconds
RMS offset      : 0.107805088 seconds
Frequency       : 0.000 ppm slow
Residual freq   : -188.944 ppm
Skew            : 1000000.000 ppm
Root delay      : 0.045836508 seconds
Root dispersion : 7.620638371 seconds
Update interval : 2.1 seconds
Leap status     : Normal

• 通过docker exec ntp chronyc sources查看ntp源状态

MS Name/IP address         Stratum Poll Reach LastRx Last sample
===============================================================================
^? 114.118.7.163                 2   6     1    53   -105ms[ -105ms] +/-  103ms
^* 120.25.115.20                 2   6    33   116   -111ms[ -111ms] +/-   24ms

二、 客户端使用ntp服务器进行时间同步

• 服务端安装ntpdate: sudo apt install ntpdate
• 测试是否能根据ntp服务器更新,例如ntpdate -d <ntp server>
• 时间同步：ntpdate {ntp_server}

三、 Python配置定时ntp同步任务

使用python-crontab库控制linux crontab定时任务，实现ntp同步任务配置

• 安装python-crontab： pip install python-crontab
• 示例代码如下：

import datetime
import os
import re

from crontab import CronTab


NTP_JOB_NAME = "ntpdate"

# crontab任务处理类
class LinuxCrontab:
    def __init__(self, user=True):
        self.cron = CronTab(user=user)

    def set_interval_job(self, command, comment="", cron_str=None, minute=None):
        job = self.cron.new(command=command, comment=comment)
        if cron_str:
            job.setall(cron_str)
        elif minute:
            job.minute.every(minute)
        self.cron.write()

    def find_job_by_comment(self, comment):
        comment_abs_compile = re.compile(f"^{comment}$")
        jobs = list(self.cron.find_comment(comment_abs_compile))
        return jobs[0] if jobs else None

    def remove_job_by_comment(self, comment):
        job = self.find_job_by_comment(comment)
        if job:
            result = self.cron.remove(job)
            self.cron.write()
            return result
        return 0

# 设置ntp同步任务，或者直接设置系统时间
def set_server_time(time_str: str = None, ntp_config: dict = None):
    """linux服务器设置时间
        time_str： 若参数存在，则指定时间字符串，强制指定时间为系统时间
        ntp_config： 若参数存在，则解析指定ntp服务器定时同步时间
    """
    cron_handler = LinuxCrontab()

    if ntp_config:
        ntp_server = ntp_config.get('host')
        interval_minute = ntp_config.get("interval_minute")
        ntpdate_cmd = f"/usr/sbin/ntpdate {ntp_server} >> /var/log/ti/system_ntp.log 2>&1"
        cron_handler.remove_job_by_comment(NTP_JOB_NAME)
        cron_handler.set_interval_job(command=ntpdate_cmd, comment=NTP_JOB_NAME, minute=interval_minute)
        print("设置ntp定时任务")
    else:
        # cmd = f'date -s "{real_time}";hwclock -w'   # 设置时间并写入bios
        cmd = f'date -s "{time_str}"'   # 设置时间
        result = os.popen(cmd)
        # 删除定时任务
        cron_handler.remove_job_by_comment(NTP_JOB_NAME)
        print(result)
    return True

# 获取ntp同步任务配置或者当前系统时间
ef get_server_time_conf():
    cron_handler = LinuxCrontab()
    job = cron_handler.find_job_by_comment(comment=NTP_JOB_NAME)
    if job:
        command = job.command
        minute_cron = str(job.minute)  # "*/2"
        if minute_cron == "*":
            minute = 1
        else:
            minute = int(minute_cron.split("/")[1])
        return {
            "user_ntp": True,
            "ntp_config": {
                "host": command.split(" ")[1],
                "port": 123,
                "interval_minute": minute
            },
            "local_time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }
    else:
        return {
            "user_ntp": False,
            "ntp_config": {},
            "local_time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        }

# 测试ntp服务器连接
def test_ntp_server(ntp_server):
    ntpdate_test_cmd = f"/usr/sbin/ntpdate -d {ntp_server}"
    exe_result = os.popen(ntpdate_test_cmd).read()
    ok, exe_info = False, "\n".join([line.strip() for line in exe_result.splitlines()])
    if re.search(r"adjust time server .*? offset", exe_result):
        ok = True
    return ok, exe_info


if __name__ == '__main__':
    ntp_config = {"host": "ntp.ntsc.ac.cn", "interval_minute": 3}
    set_server_time(ntp_config=ntp_config)
    result = get_server_time_conf()
    print(result)