Nginx动态修改Upstream模块(ngx_http_dyups_module)源码分析

1. 模块概述

ngx_http_dyups_module是Nginx的一个第三方模块，允许动态修改upstream配置而无需重新加载Nginx。主要功能包括：

• 动态添加、删除和修改upstream配置
• 多进程间配置同步
• 引用计数管理
• 健康检查集成

2. 核心数据结构

2.1 主要配置结构

typedef struct {
    ngx_array_t dy_upstreams;  // 存储动态upstream配置的数组
    ngx_str_t shm_name;       // 共享内存名称
    ngx_uint_t shm_size;      // 共享内存大小
    ngx_msec_t read_msg_timeout; // 读取消息超时时间
} ngx_http_dyups_main_conf_t;

2.2 Upstream配置结构

typedef struct {
    ngx_http_upstream_srv_conf_t *upstream; // 指向原始upstream配置
    ngx_uint_t dynamic;        // 是否为动态upstream
    ngx_uint_t deleted;        // 删除标记
    ngx_uint_t idx;            // 在数组中的索引
    ngx_http_conf_ctx_t *ctx;  // 配置上下文
    ngx_pool_t *pool;          // 内存池
    ngx_atomic_t *ref;         // 引用计数
} ngx_http_dyups_srv_conf_t;

2.3 共享内存结构

typedef struct {
    ngx_queue_t msg_queue;     // 消息队列
    ngx_uint_t version;        // 版本号
    ngx_dyups_status_t *status; // 进程状态数组
} ngx_dyups_shctx_t;

3. 模块初始化流程

3.1 配置创建阶段

static void * ngx_http_dyups_create_main_conf(ngx_conf_t *cf) {
    // 初始化dy_upstreams数组，容量1024
    if (ngx_array_init(&dmcf->dy_upstreams, cf->pool, 1024, 
        sizeof(ngx_http_dyups_srv_conf_t)) != NGX_OK) {
        return NULL;
    }
}

3.2 共享内存初始化

static char *ngx_http_dyups_init_shm(ngx_conf_t *cf, void *conf) {
    // 添加共享内存区域
    shm_zone = ngx_shared_memory_add(cf, &dmcf->shm_name, 
                dmcf->shm_size, &ngx_http_dyups_module);
    shm_zone->data = cf->pool;
    shm_zone->init = ngx_http_dyups_init_shm_zone;
}

3.3 进程初始化

static ngx_int_t ngx_http_dyups_init_process(ngx_cycle_t *cycle) {
    // 设置定时器读取消息
    timer = &ngx_dyups_global_ctx.msg_timer;
    timer->handler = ngx_http_dyups_read_msg;
    ngx_add_timer(timer, dmcf->read_msg_timeout);
    
    // 初始化共享内存状态
    if (sh->status == NULL) {
        sh->status = ngx_slab_alloc_locked(shpool, 
            sizeof(ngx_dyups_status_t) * ccf->worker_processes);
    }
    
    // 检查版本号判断是否需要恢复
    if (sh->version != 0) {
        // 恢复流程...
    }
}

4. 核心功能实现

4.1 消息同步机制

static void ngx_http_dyups_read_msg_locked(ngx_event_t *ev) {
    // 更新进程状态
    for (i = 0; i < ccf->worker_processes; i++) {
        status = &sh->status[i];
        if (status->pid == 0 || status->pid == ngx_pid) {
            status->pid = ngx_pid;
            status->time = now;
            break;
        }
    }
    
    // 处理消息队列
    for (q = ngx_queue_last(&sh->msg_queue); 
         q != ngx_queue_sentinel(&sh->msg_queue); 
         q = ngx_queue_prev(q)) {
        
        // 已同步的消息移除
        if (msg->count == ccf->worker_processes) {
            ngx_queue_remove(q);
            ngx_dyups_destroy_msg(shpool, msg);
            continue;
        }
        
        // 检查是否已处理
        for (i = 0; i < msg->count; i++) {
            if (msg->pid[i] == ngx_pid) {
                found = 1;
                break;
            }
        }
        
        // 未处理的消息执行同步
        if (!found) {
            msg->pid[i] = ngx_pid;
            msg->count++;
            rc = ngx_dyups_sync_cmd(pool, &name, &content, msg->flag);
        }
    }
}

4.2 Upstream更新流程

static ngx_int_t ngx_dyups_do_update(ngx_str_t *name, ngx_buf_t *buf, ngx_str_t *rv) {
    // 查找现有upstream
    duscf = ngx_dyups_find_upstream(name, &idx);
    
    if (idx == -1) {
        // 创建新的upstream配置
        duscf = ngx_array_push(&dumcf->dy_upstreams);
        uscfp = ngx_array_push(&umcf->upstreams);
        idx = umcf->upstreams.nelts - 1;
    }
    
    // 初始化upstream
    rc = ngx_dyups_init_upstream(duscf, name, idx);
    
    // 添加server
    rc = ngx_dyups_add_server(duscf, buf);
    
    // 发送同步消息
    if (ngx_http_dyups_send_msg(name, buf, NGX_DYUPS_ADD)) {
        ngx_str_set(rv, "alert: update success but not sync to other process");
        return NGX_HTTP_INTERNAL_SERVER_ERROR;
    }
    
    return NGX_OK;
}

4.3 Upstream删除流程

static ngx_int_t ngx_dyups_do_delete(ngx_str_t *name, ngx_str_t *rv) {
    // 查找upstream
    duscf = ngx_dyups_find_upstream(name, &dumy);
    
    if (duscf == NULL || duscf->deleted) {
        return NGX_HTTP_NOT_FOUND;
    }
    
    // 标记删除
    ngx_dyups_mark_upstream_delete(duscf);
    
    // 发送删除消息
    if (ngx_http_dyups_send_msg(name, NULL, NGX_DYUPS_DELETE)) {
        ngx_str_set(rv, "alert: delete success but not sync to other process");
        return NGX_HTTP_INTERNAL_SERVER_ERROR;
    }
    
    return NGX_OK;
}

4.4 引用计数管理

static ngx_int_t ngx_http_dyups_init_peer(ngx_http_request_t *r, 
    ngx_http_upstream_srv_conf_t *us) {
    
    // 初始化上下文
    ctx = ngx_pcalloc(r->pool, sizeof(ngx_http_dyups_ctx_t));
    ctx->scf = dscf;
    ctx->data = r->upstream->peer.data;
    ctx->get = r->upstream->peer.get;
    ctx->free = r->upstream->peer.free;
    
    // 设置peer方法
    r->upstream->peer.data = ctx;
    r->upstream->peer.get = ngx_http_dyups_get_peer;
    r->upstream->peer.free = ngx_http_dyups_free_peer;
    
    // 增加引用计数
    dscf->ref++;
    
    // 注册清理函数
    cln = ngx_pool_cleanup_add(r->pool, 0);
    cln->handler = ngx_http_dyups_clean_request;
    cln->data = &dscf->ref;
}

5. 关键设计点

5.1 多进程同步机制

1. 共享内存：使用共享内存存储配置变更消息
2. 定时器：每个worker进程定期检查消息队列
3. 版本控制：通过version字段检测进程异常重启
4. 状态跟踪：记录每个进程的处理状态

5.2 安全删除策略

1. 标记删除：先标记为NGX_DYUPS_DELETING状态
2. 引用检查：等待引用计数降为0才真正删除
3. 健康检查集成：关闭健康检查peer

5.3 动态配置解析

1. 模拟配置解析：重用Nginx原生配置解析逻辑
2. 沙箱测试：先测试配置有效性再应用
3. 内存管理：为每个动态upstream创建独立内存池

6. 性能优化

1. 批量操作：支持一次更新多个server
2. 延迟删除：引用计数机制避免立即删除正在使用的upstream
3. 最小化锁：只在必要时获取共享内存锁

7. 使用建议

1. 共享内存大小：根据upstream数量调整shm_size
2. 超时设置：read_msg_timeout影响同步延迟
3. 监控：关注version字段检测同步问题
4. 异常处理：处理同步失败时的配置不一致问题

8. 扩展点

1. 健康检查：集成第三方健康检查模块
2. 持久化：添加配置持久化支持
3. 权限控制：增加API访问控制
4. 监控接口：暴露运行时状态信息

通过深入分析ngx_http_dyups_module的实现，我们可以学习到Nginx模块开发的高级技巧，包括共享内存管理、多进程同步、动态配置处理等核心机制。