Airflow标记任务成功/失败后Golang进程无法优雅终止的问题

核心疑问解答 #

当你在Airflow中点击“标记成功/标记失败”终止运行中的任务时：

• Airflow会先向任务进程发送SIGTERM信号，等待一段默认时长（5秒）后，如果进程仍未退出，就会发送SIGKILL强制终止进程。
• 如果进程被SIGKILL终止，Golang程序会被立即强制停止，剩余代码不会执行，优雅退出逻辑（比如你的closeConnections）也无法触发。
• 如果进程在收到SIGTERM后及时响应，剩余代码可以执行，但前提是程序能捕获到SIGTERM并在Airflow的等待超时内完成退出流程。

你的现有代码逻辑未执行，核心原因有两个：

1. SIGKILL无法被捕获、阻塞或忽略，你在signal.Notify中添加syscall.SIGKILL完全无效；
2. Airflow的默认等待超时（5秒）可能过短，你的closeConnections还没执行完，进程就被强制杀掉了。

解决方案 #

1. 修复Golang程序的信号处理逻辑 #

移除无效的SIGKILL监听，改用上下文（context）管理主逻辑与连接，确保收到终止信号后能快速响应并关闭资源：

package main

import (
    "context"
    "os"
    "os/signal"
    "syscall"
)

func main() {
    // 创建可取消的上下文，用于传递终止信号给主逻辑
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 启动信号监听goroutine
    go func() {
        sigChan := make(chan os.Signal, 1)
        // 只监听可捕获的终止信号
        signal.Notify(sigChan, syscall.SIGINT, syscall.SIGHUP, syscall.SIGQUIT, syscall.SIGTERM)
        <-sigChan

        // 触发上下文取消，通知主逻辑停止
        cancel()
        // 执行连接关闭逻辑
        closeConnections()
        // 主动退出程序
        os.Exit(0)
    }()

    // 主业务逻辑，通过ctx监听终止信号
    runMainTask(ctx)
}

func runMainTask(ctx context.Context) {
    // 示例主逻辑：定期检查上下文是否被取消
    for {
        select {
        case <-ctx.Done():
            // 收到终止信号，立即退出主逻辑
            return
        default:
            // 执行你的核心业务操作（比如数据处理、调用服务等）
            // 注意：避免长时间阻塞的操作，确保能及时响应ctx.Done()
        }
    }
}

func closeConnections() {
    // 实现实际的连接关闭逻辑：比如数据库连接、HTTP客户端连接等
    // 确保所有打开的资源都被正确释放
}

2. 调整Airflow配置，延长优雅退出等待时间 #

修改Airflow的airflow.cfg配置文件，增加任务终止的等待超时，给Golang程序足够时间完成资源清理：

# 调整为合适的时长，比如30秒
killed_task_cleanup_time = 30

如果使用CeleryExecutor，还需确保Worker能正确传递信号，可检查worker_send_task_events = True配置是否开启。

3. 降级方案：外部信号触发退出（针对无法捕获SIGTERM的场景） #

如果因为Airflow执行器限制，程序始终无法收到SIGTERM，可以在Golang程序中定期检查外部标志（比如本地文件、Redis键值），当标志存在时主动关闭连接并退出：

func checkTerminateSignal() bool {
    // 示例：检查是否存在终止标志文件
    _, err := os.Stat("/tmp/terminate_task")
    return err == nil
}

// 在runMainTask的循环中加入检查
func runMainTask(ctx context.Context) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
            if checkTerminateSignal() {
                closeConnections()
                os.Exit(0)
            }
            // 执行业务操作
        }
    }
}

此时你需要在Airflow标记任务时，额外执行创建标志文件的操作（比如通过BashOperator提前创建）。

内容的提问来源于stack exchange，提问作者user22168259