无 IP 访问方案

静态 IP 资源有限，AI 容器太多了，分配不过来，所以只能开辟无 IP 容器，但用户仍有 ssh 直连 AI 容器调试的需求，必须要新增一层转发控制，支持无 IP 容器的连接，这包括

• HTTP 流量的请求转发
• WS 形式的 Web Terminal 流量的转发
• SSH 直连容器的需求

以 ssh 为例，在有 IP 的情况下，直接可以通过 ip 和私钥进行 ssh 直连，但现在分配不了 ip 了，所以只能用一个泛域名返回给用户，后台做一层映射转发，实现这样的需求

APISIX 网关转发

WS / HTTP 流量的转发，核心思路

1. 业务侧在更新域名 / IP 的时候，把鉴权 Token 和映射关系存到 redis，以供网关进行鉴权和代理转发
2. 用 udns 映射源域名到 apisix 网关，网关根据 envId 查 redis 找到 pod ip，再通过 apisix 插件 access 拦截将请求转发到 pod ip

流量的拦截和转发基于 APISIX（一个 Lua 网关服务），基于 OpenResty（Nginx + LuaJIT）实现，用 Etcd 存放本地数据

• APISIX AI 网关介绍 | Apache APISIX (opens new window)
• 游戏业务出海：TAPISIX 稳定运营实践 | Apache APISIX (opens new window)

apisix 插件 plugins 通过导出 _M 并提供 _M.access、_M.rewrite、_M.check_schema 等钩子函数，在请求的生命周期不同阶段被挂载调用，实现流量的前置（如鉴权）后置（如转发）处理，例如

• APISIX 在加载 / 更新插件配置时会调用 check_schema(conf)，确保用户传的 conf 合法
• access 是 APISIX 请求生命周期的"访问控制"阶段；返回数字 = HTTP 状态码并中断请求；返回 nil 表示放行

这里 ws 和 http 流量的转发就是通过插件的形式实现，通过调用 access 实现鉴权和转发

最终，这里的插件通过 Dockerfile 构建并注入现有的 apisix 服务

1. 定义 Dockerfile 将插件 copy 丢到现成的 APISIX 的 plugins 目录下
2. 配置 APISIX 的 config.yaml 插件白名单
3. 在运行阶段，如果路由命中了对应插件，APISIX 就会自动读入每个插件 _M 的函数，并按照 priority 大小顺序执行

Dockerfile 示例

FROM mirrors.tencent.com/tapisix/tapisix:3.3.1-alpha
COPY plugins/* /usr/local/tapisix/apisix/plugins/

SSH 代理直连

ssh 直连

ssh root@saulxiong-any2.dev.cloud.woa.com -p 36000

VPC 和 ENI

• VPC 即 Virtual Private Cloud 虚拟私有云/专有网络，我们所有机器的 ip 都在一个大的 VPC 下，故而所有 ip 均保持唯一性，但是唯一性不等于可达性，这个一定要注意，不是说一个 ip 地址在网络中唯一了，我就可以无脑 ping 通了，这里存在很多我还没弄清网络限制，可能是出于对 k8s 集群安全考虑的权限认证
• ENI 即 Elastic Network Interface 弹性网卡接口，网络中可以使用 ENI 为容器分配独立 ENI IP，这个 IP 一经分配，是可以在 ENI 环境中进行直连的，属于稀缺资源

通过 DNAT + IPTable 实现泛域名到 Pod 的反向代理

• UDNS（UCloud Domain Name System），域名解析服务系统，根据泛域名中某一字段，将泛域名重定向到具体域名，例如
• CLB（Classic Load Balancer），负载均衡器
• DNAT（Destination Network Address Translation），目的网络地址转换，是 Linux 系统软件的工作，将一个端口的 TCP 流量指定转发到另一 IP，通过 add 规则指定
• IP Table（IP 映射表），提供端口到 pod ip 的映射，例如 CMS_IP: 56001 映射 k8s 中的 80.133.22.231:22，一个端口映射一个 ip

codev-web-server/services/web_service.go: EnableSsh

有一个现成的分配 SSH 的接口 EnableSsh @alias=/env/enableSsh，这个接口将返回用户可以 ssh 直连的 uri，例如

response: {
  code: 200,
  data: {
    VSCode: "vscode://vscode-remote/ssh-remote+root@jaceyuan-any2.devcloud.woa.com:36000/data/workspace/FeedbackAnalyzer/?windowId=_blank", PhpStorm: "jetbr>-ains-gateway://connect#idePath=/codev/ide/PS/PhpStorm-2024.1.2&projectPath=/data/workspace/FeedbackAnalyzer/&host=jaceyuan-any2.devcloud.woa.com&user=root&port=36000&type=ssh&deploy=false",
    CodeBuddyCN: "codebuddyc://vscode-remote/codebuddy-remote-ssh+root@jaceyuan-any2.devcloud.woa.com:36000/data/workspace/FeedbackAnalyzer/?windowId=_blank",
    //...
  }
}

可以注意到这里的 url 包含了 ssh 地址、用户名、端口等等信息，我们的最终工作就是

• 针对无 IP 的容器，返回特定额外逻辑的域名 + 端口

具体而言，要在控制面实现域名到 K8s 内网 ip 的映射，有这样几个关键点

1. 找到 serverType 为 CPU 环境且禁用 eniIp 的分支
2. 采用 DNAT，映射到实体是 CVM 端口到 pod ip，这里的端口 Port 由全局的 Redis 游标维护（40000 ~ 50000）
3. CVM 的 IP 是可配置的，从七彩石里读取，这是一个 IP 列表，每次要全量下发
4. 远程命令执行，每次映射，要下发命令 bash /data/home/yingqichen/nft/nft-ctl.sh add %d %s %d 到全量的 CVM 上，添加 DNAT 规则
5. DB 读写，主要就是规则的写入和读取，还有 Redis 的存取
6. UriScheme 拼接，Scheme 写在配置文件中，在业务中通过实际的域名和 IP 替换 Scheme 中的占位符完成拼接

SSH 代理路线：用户 → 泛域名 → UDNS → CLB（Classic Load Balancer 负载均衡） → CVM（DNAT） → Pod

Apisix 流量转发

控制面缓存代理规则

EnvInstanceController 基于 Redis 缓存 proxy 信息：business/env_instance.go: CacheProxyInformation

// CacheProxyInformation 缓存proxy需要的信息（环境id对应用户名、环境id对应ip）
func (envInstance *EnvInstanceController) CacheProxyInformation(ctx tcontext.Context,
    envIns *table.DevEnvironmentInstance, envTask *table.EnvironmentTask) {
    envId := strings.ToLower(envIns.Uid)
    IpKey := fmt.Sprintf(proxy.IpKey, envId)
    UsernameKey := fmt.Sprintf(proxy.UsernameKey, envId)
    TokenKey := fmt.Sprintf(proxy.TokenKey, envId)

    if envTask.Action == EnvActionDelete {
        _, _ = config.RedisClientGlobal.Do(ctx, "del", IpKey)
        _, _ = config.RedisClientGlobal.Do(ctx, "del", UsernameKey)
        _, _ = config.RedisClientGlobal.Do(ctx, "del", TokenKey)
    } else {
        _, _ = config.RedisClientGlobal.Do(ctx, "set", IpKey, envIns.EnvironmentIP)
        _, _ = config.RedisClientGlobal.Do(ctx, "set", UsernameKey, envIns.Username)
    }
}

只要是更新实例状态，business/env_instance.go: UpdateStatus，就会根据 serverType 进行缓存，如果是 VM，一定有 IP，不需要走代理，如果是容器并且没有 IP（disableEniIp == true），则进行缓存

// UpdateStatus ...
func (envInstance *EnvInstanceController) UpdateStatus(ctx tcontext.Context, taskId, status, userName, source,
    message string, code int) error {
    //...

switch environmentInstance.ServerType {
    case vmServerType:
        // 创建cvm成功，注册默认域名和ip域名
        if environmentTask.Action == EnvActionCreate && environmentInstance.Status == runningStatus {
            _ = domainController.RegisterDefaultDomain(ctx, config.DbGlobal, &environmentInstance)
            _ = domainController.RegisterIPDomain(ctx, config.DbGlobal, &environmentInstance)
            domain, err := domainController.QueryDomainByEnvId(ctx, environmentInstance.Uid)
            if err != nil {
                return err
            }
            if domain.Id != 0 && domain.BindStatus == BindStatusWaiting {
                if _, err = domainController.BindEnvDomainInner(ctx, config.DbGlobal, &environmentInstance, domain); err != nil {
                    return err
                }
            }
        }
        ctx.Logger().Infof("envInstanceController.UpdateStatus: %s", environmentInstance.Uid)
    default:
        if environmentInstance.IsTaijiResource() {
            _ = envInstance.onTaijiEnvTaskComplete(ctx, &environmentInstance, &environmentTask)
            return nil
        }
        // 开机更新token
        if environmentTask.Action == EnvActionStart && environmentInstance.Status == runningStatus {
            _ = defaultPaaSController.UpdateEnvGitToken(ctx, &environmentInstance, false)
        }
        // 环境创建成功，绑定域名
        if environmentTask.Action == EnvActionCreate && environmentInstance.Status == runningStatus &&
            environmentInstance.ServerType != cvmContainerServerType {
            // 如果是预备环境，不注册域名
            if !environmentInstance.IsPrepareEnv() {
                _ = domainController.BindEnvDomainAfterCreate(ctx, config.DbGlobal, &environmentInstance)
            }
            // 如果是带appId创建的环境且不是预备环境，则需要判断是否要注册预缓存盘v2
            if app.IfPreheatV2Enable(ctx, config.DbSlaveGlobal, environmentInstance.AppId) &&
                !environmentInstance.IsPrepareEnv() {
                er := codeprecacheController.CreatePrecacheV2(ctx, &environmentInstance)
                if er != nil {
                    ctx.Logger().Errorf("创建预缓存盘v2失败: %+v", er)
                }
            }
        }
        // 处理预约排队相关
        DefaultQueueController.QueueTaskCallBack(ctx, &environmentTask, &environmentInstance)
        // 从any回调回来，目前默认回调信息为命令执行的结果，没有更多标志位
        if source == "any" && message != "" {
            _ = envInstance.SaveEnvInsCmdResultToRedis(ctx, environmentInstance.Uid, message)
        }
        if environmentInstance.DisableEniIp {
            envInstance.CacheProxyInformation(ctx, &environmentInstance, &environmentTask)
        }
    }
}

存储鉴权 Token：写 Redis Token 给 apisix lua 插件鉴权

business/cli/token.go: 120

// SaveTokenToRedis 持久化到redis 供tapisix查询
func (its *TokenController) SaveTokenToRedis(ctx tcontext.Context, token *table.CliToken) {
    key := cliTokenRedisKey + token.Token
    if _, err := config.RedisClientGlobal.Do(ctx, "set", key, token.Username); err != nil {
        ctx.Logger().Warn(err)
    }
}

真正的鉴权和转发由 apisix 网关执行，这里只是做控制面的存储

apisix auth 插件鉴权

codev-auth.lua 常量定义

• ngx 是 LuaJIT 启动时由 C 模块直接注册到 _G（全局环境）里的变量，里面挂着访问 Nginx 内部的所有 API，所以在任何 .lua 文件里直接写 ngx.xxx 都能用，为什么要本地 local 赋值？很简单，因为 local 查的快，全局查表慢
• core.lrucache 是 lua 的本地缓存，设置最多缓存 1000 项，每项存活 300 秒，用于缓存 Redis 中查到的 appid 信息，实现双级缓存

-- 云研发 app + token + uri 鉴权
local ngx     = ngx
local core    = require("apisix.core")
local redis_new = require("resty.redis").new
local sha256   = require("resty.sha256")
local hex      = assert(require("resty.string")).to_hex

local plugin_name = "codev-auth"
local codev_app_key = "codev_app_token"
local codev_app_uri_prefix = "codev_app_uri___"
local tonumber = tonumber
local my_cache = core.lrucache.new({
    serial_creating = true , invalid_stale = true , ttl = 300, count = 1000
})

local schema = {
    type = "object",
    properties = {
        redis_host = {
            type = "string", minLength = 2
        },
        redis_port = {
            type = "integer", minimum = 1, default = 6379,
        },
        redis_username = {
            type = "string", minLength = 1,
        },
        redis_password = {
            type = "string", minLength = 0,
        },
        redis_database = {
            type = "integer", minimum = 0, default = 0,
        },
        redis_timeout = {
            type = "integer", minimum = 1, default = 1000,
        },

    },
    required = {"redis_host"}
}

钩子函数及导出定义

local _M = {
    version = 1.0,
    priority = 2500,
    name = plugin_name,
    schema = schema,
}

function _M.check_schema(conf)
    local ok, err = core.schema.check(schema, conf)
    if not ok then
        return false, err
    end
    return true
end

function _M.access(plugin_conf, ctx)
    local conf = core.table.clone(plugin_conf)
    local err = check_codev_app_token(ctx, conf)
    if err then
        core.log.error("code auth failed: ", err)
        return 401
    end
end

return _M

鉴权的业务逻辑实现：读 redis，与请求参数比较进行鉴权

-- 连接 redis
local function redis_cli(conf)
    local red = redis_new()
    local timeout = conf.redis_timeout or 1000    -- 1sec

    red:set_timeouts(timeout, timeout, timeout)

    local ok, err = red:connect(conf.redis_host, conf.redis_port or 6379)
    if err then
        return false, err
    end
    if not ok then
        return false, "redis connect err"
    end

    local count
    count, err = red:get_reused_times()
    if 0 == count then
        if conf.redis_password and conf.redis_password ~= '' then
            local ok, err
            if conf.redis_username then
                ok, err = red:auth(conf.redis_username, conf.redis_password)
            else
                ok, err = red:auth(conf.redis_password)
            end
            if not ok then
                return nil, err
            end
        end

        -- select db
        if conf.redis_database ~= 0 then
            local ok, err = red:select(conf.redis_database)
            if not ok then
                return false, "failed to change redis db, err: " .. err
            end
        end
    elseif err then
        core.log.error(" err: ", err)
        return nil, err
    end
    return red, nil
end

sha256 加密，进行鉴权时，是对密文进行 ~= 匹配

local function gen_codev_app_enckey(token, ts, random)
    local data = token .. ts .. random
    local hash = sha256:new()
    hash:update(data)
    return string.upper(hex(hash:final()))
end

对 token 进行鉴权

local function check_codev_app_token(ctx, conf)
    -- Get Authorization header, maybe.
    local appid_header = core.request.header(ctx, "APPID")
    if not appid_header then
        return  "header APPID is empty."
    end
    local random_header = core.request.header(ctx, "RANDOM")
    if not random_header then
        return  "header RANDOM is empty."
    end
    local ts_header = core.request.header(ctx, "TIMESTP")
    if not ts_header then
        return  "header TIMESTP is empty."
    end
    local enckey_header = core.request.header(ctx, "ENCKEY")
    if not enckey_header then
        return  "header ENCKEY is empty."
    end
    local now = os.time()
    local ts = tonumber(ts_header)
    if ts + 300 < tonumber(now) then
        return "TIMESTP is expired"
    end
    
    local app_info, err = my_cache(appid_header, nil, get_appid_token_info, conf, appid_header)
    if err then
        return err
    end
    if not app_info then
        return "query app err"
    end
    local token = app_info["token"]
    local uris = app_info["uris"]
    if not token or token == ngx.null or token == "" then
        return  "APPID not exist."
    end

    local enckey = gen_codev_app_enckey(token, ts_header, random_header)
    if enckey ~= string.upper(enckey_header) then
        return "enckey is error"
    end

    local uri = ctx.var.uri
    if not uris then
        return "not exist access uri."
    end
    local allow_uri = false
    for i, uri_p in ipairs(uris) do
        local p = string.match(uri, uri_p)
        if p then
            allow_uri = true
        end
    end
    if not allow_uri then
        return "not allow access uri."
    end
    return  nil
end

staff name 请求头重写

_M.rewrite

codev-staff.lua：从用户请求中提取用户名并设置 staff_name

-- 云研发 app + token + uri 鉴权
local ngx                = ngx
local core               = require("apisix.core")
local plugin_name        = "codev-staff"
local staff_header_key   = "STAFFNAME"
local x_staff_header_key = "X-STAFFNAME"
local x_oauth_header_key = "X-Userinfo"


local schema = { 
}


local _M = {
    version = 1.0,
    priority = 600,
    name = plugin_name,
    schema = schema,
}


-- oauth认证从请求头提取用户名
local function get_staff_name(ctx)
    local user_value = core.request.header(ctx, x_oauth_header_key)
    if user_value then
        local user_info = ngx.decode_base64(user_value)
        if not user_info then
            core.log.error("Failed to decode base64 string: ", user_value)
            return 
        end
        
        local user_obj, err = core.json.decode(user_info)
        if not user_obj then
            core.log.error("Failed to decode JSON string: ", user_info, " error: ", err)
            return
        end
        local username = user_obj["user_name"]
        if  username then
            return username
        else
            core.log.warn("not found username")
        end
    end
end


function _M.rewrite(plugin_conf, ctx)
    -- staffname x-staffname 获取staffname
    local staff_name = core.request.header(ctx, staff_header_key) 
    if not staff_name then
        staff_name = core.request.header(ctx, x_staff_header_key)
    end

    -- 如果都没有从header提取
    if not staff_name then
        staff_name = get_staff_name(ctx)
    end

    -- 设置请求头
    if staff_name then
        if not core.request.header(ctx, staff_header_key) then
            core.request.set_header(ctx, staff_header_key, staff_name)
        end
        if not core.request.header(ctx, x_staff_header_key) then
            core.request.set_header(ctx, x_staff_header_key, staff_name)
        end
    end
end

return _M

proxy 反向代理

http 的代理，通过 access 转发 http 流量，和鉴权的逻辑类似，根据用户请求中的 envId 从 redis 里拿对应映射的 pod ip 地址，然后重定向把流量转发到指定 ip + port，实现代理

codev-ai-http-proxy.lua

-- anydev-ai-http-proxy.lua
-- APISIX 插件：根据 *.connect.anydev.woa.com 域名前缀从 Redis 查 Pod IP，根据 URL 端口转发

local ngx = ngx
local core = require("apisix.core")
local upstream = require("apisix.upstream")
local redis_new = require("resty.redis")
local table_insert = table.insert

local plugin_name = "anydev-ai-http-proxy"
local codev_pod_ip_prefix = "anydev-proxy:ip:"

local schema = {
    type = "object",
    properties = {
        redis_host = {
            type = "string", minLength = 2
        },
        redis_port = {
            type = "integer", minimum = 1, default = 6379,
        },
        redis_username = {
            type = "string", minLength = 1,
        },
        redis_password = {
            type = "string", minLength = 0,
        },
        redis_database = {
            type = "integer", minimum = 0, default = 0,
        },
        redis_timeout = {
            type = "integer", minimum = 1, default = 1000,
        },
        domain_suffix = {
            type = "string",
            minLength = 1,
            default = "connect.devcloud.woa.com",
            description = "Domain suffix for extracting pod id, e.g. connect-test.devcloud.woa.com or connect.devcloud.woa.com",
        },

    },
    required = {"redis_host"}
}

local _M = {
    version = 0.1,
    priority = 1000,
    name = plugin_name,
    schema = schema,
}

function _M.check_schema(conf)
    return core.schema.check(schema, conf)
end

-- 将域名后缀转为 Lua 匹配模式（转义 . 为 %.）
local function build_domain_pattern(suffix)
    local escaped = suffix:gsub("%.", "%%."):gsub("%-", "%%-")
    return "^(.+)%." .. escaped .. "$"
end

-- 从 Host 中提取 Pod ID: xxx.<domain_suffix> → xxx
local function extract_pod_id(host, domain_suffix)
    -- 去掉端口
    local h = host:match("^([^:]+)") or host
    local pattern = build_domain_pattern(domain_suffix)
    local pod_id = h:lower():match(pattern)
    return pod_id
end

-- 从 URI 中提取端口和实际路径: /8080/api/xxx → 8080, /api/xxx
local function extract_port_and_path(uri)
    local port, path = uri:match("^/(%d+)(/.*)$")
    if not port then
        port = uri:match("^/(%d+)$")
        path = "/"
    end
    return port, path
end

local function redis_cli(conf)
    local red = redis_new:new()
    local timeout = conf.redis_timeout or 1000    -- 1sec

    red:set_timeouts(timeout, timeout, timeout)

    local ok, err = red:connect(conf.redis_host, conf.redis_port or 6379)
    if err then
        return false, err
    end
    if not ok then
        return false, "redis connect err"
    end

    local count
    count, err = red:get_reused_times()
    if 0 == count then
        if conf.redis_password and conf.redis_password ~= '' then
            local ok, err
            if conf.redis_username then
                ok, err = red:auth(conf.redis_username, conf.redis_password)
            else
                ok, err = red:auth(conf.redis_password)
            end
            if not ok then
                return nil, err
            end
        end

        -- select db
        if conf.redis_database ~= 0 then
            local ok, err = red:select(conf.redis_database)
            if not ok then
                return false, "failed to change redis db, err: " .. err
            end
        end
    elseif err then
        core.log.error(" err: ", err)
        return nil, err
    end
    return red, nil
end

-- 查 Redis
local function query_redis(conf, pod_id)
    local red, err = redis_cli(conf)
    if not red then
        core.log.error("redis err: ", err)
        return nil, err
    end

    local key = codev_pod_ip_prefix .. pod_id
    local val, err = red:get(key)
    red:set_keepalive(10000, 100)

    if not val or val == ngx.null then
        return nil, "pod ip not found for key: " .. key
    end

    return val, nil
end

function _M.access(conf, ctx)
    -- 提取 Pod ID
    local host = core.request.header(ctx, "Host") or ""
    local domain_suffix = conf.domain_suffix or "connect.devcloud.woa.com"
    local pod_id = extract_pod_id(host, domain_suffix)
    core.log.info("anydev proxy, host:", host, "pod_id:", pod_id, "domain_suffix:", domain_suffix)
    if not pod_id then
        return 400, { message = "invalid domain, expected <id>." .. domain_suffix }
    end

    -- 提取端口和路径
    local port_str, real_path = extract_port_and_path(ngx.var.uri)
    if not port_str then
        return 400, { message = "missing port in URL, expected /<port>/<path>" }
    end

    local port = tonumber(port_str)
    if not port or port < 1 or port > 65535 then
        return 400, { message = "invalid port: " .. port_str }
    end

    -- 查 Redis 获取 Pod IP
    local pod_ip, err = query_redis(conf, pod_id)
    if not pod_ip then
        return 502, { message = err }
    end

    -- 动态设置 upstream
    local new_nodes = {}
    table_insert(new_nodes, { host = pod_ip, port = port, weight = 1 })
    local up_conf = {
        name = plugin_name,
        type = "roundrobin",
        nodes = new_nodes,
    }
    local matched_route = ctx.matched_route
    upstream.set(ctx, "roundrobin" .. "#route_" .. matched_route.value.id, ctx.conf_version, up_conf, matched_route)

    -- 重写 URI（去掉端口前缀）
    local args = ngx.var.args
    if args and args ~= "" then
        ngx.req.set_uri(real_path .. "?" .. args, false)
    else
        ngx.req.set_uri(real_path, false)
    end

    core.log.info("anydev proxy: ", pod_id, " → ", pod_ip, ":", port, real_path)
end

return _M

ws 的实现类似，根据域名前缀提取 pod_id，校验用户身份后转发

-- codev-ai-ws-proxy.lua
-- APISIX 插件：WebSocket 代理，根据域名前缀提取 pod_id，校验用户身份后转发

local ngx = ngx
local core = require("apisix.core")
local upstream = require("apisix.upstream")
local redis_new = require("resty.redis")
local table_insert = table.insert

local plugin_name = "codev-ai-ws-proxy"
local redis_prefix_ip = "anydev-proxy:ip:"
local redis_prefix_username = "anydev-proxy:username:"
local redis_prefix_token = "anydev-proxy:token:"

local staff_header_key = "STAFFNAME"
local x_staff_header_key = "X-STAFFNAME"
local x_oauth_header_key = "X-Userinfo"

local schema = {
    type = "object",
    properties = {
        redis_host = {
            type = "string", minLength = 2
        },
        redis_port = {
            type = "integer", minimum = 1, default = 6379,
        },
        redis_username = {
            type = "string", minLength = 1,
        },
        redis_password = {
            type = "string", minLength = 0,
        },
        redis_database = {
            type = "integer", minimum = 0, default = 0,
        },
        redis_timeout = {
            type = "integer", minimum = 1, default = 1000,
        },
        domain_suffix = {
            type = "string",
            minLength = 1,
            default = "yingqichen-ws-test.devcloud.woa.com",
            description = "Domain suffix for extracting pod id",
        },
        ws_port = {
            type = "integer", minimum = 1, default = 8900,
            description = "WebSocket port on the target pod",
        },
    },
    required = {"redis_host"}
}

local _M = {
    version = 0.1,
    priority = 1000,
    name = plugin_name,
    schema = schema,
}

function _M.check_schema(conf)
    return core.schema.check(schema, conf)
end

-- 将域名后缀转为 Lua 匹配模式（转义 . 和 -）
local function build_domain_pattern(suffix)
    local escaped = suffix:gsub("%.", "%%."):gsub("%-", "%%-")
    return "^(.+)%." .. escaped .. "$"
end

-- 从 Host 中提取 Pod ID
local function extract_pod_id(host, domain_suffix)
    local h = host:match("^([^:]+)") or host
    local pattern = build_domain_pattern(domain_suffix)
    return h:lower():match(pattern)
end

-- oauth认证从请求头提取用户名
local function get_staff_name(ctx)
    local staff_name = core.request.header(ctx, staff_header_key) 
    if not staff_name then
        staff_name = core.request.header(ctx, x_staff_header_key)
    end

    if staff_name then
        return staff_name
    end

    local user_value = core.request.header(ctx, x_oauth_header_key)
    if user_value then
        local user_info = ngx.decode_base64(user_value)
        if not user_info then
            core.log.error("Failed to decode base64 string: ", user_value)
            return 
        end
        
        local user_obj, err = core.json.decode(user_info)
        if not user_obj then
            core.log.error("Failed to decode JSON string: ", user_info, " error: ", err)
            return
        end
        local username = user_obj["user_name"]
        if  username then
            return username
        else
            core.log.warn("not found username")
        end
    end
end

-- Redis 连接
local function redis_cli(conf)
    local red = redis_new:new()
    local timeout = conf.redis_timeout or 1000

    red:set_timeouts(timeout, timeout, timeout)

    local ok, err = red:connect(conf.redis_host, conf.redis_port or 6379)
    if err then
        return false, err
    end
    if not ok then
        return false, "redis connect err"
    end

    local count
    count, err = red:get_reused_times()
    if 0 == count then
        if conf.redis_password and conf.redis_password ~= '' then
            local ok, err
            if conf.redis_username then
                ok, err = red:auth(conf.redis_username, conf.redis_password)
            else
                ok, err = red:auth(conf.redis_password)
            end
            if not ok then
                return nil, err
            end
        end

        if conf.redis_database ~= 0 then
            local ok, err = red:select(conf.redis_database)
            if not ok then
                return false, "failed to change redis db, err: " .. err
            end
        end
    elseif err then
        core.log.error("redis reused_times err: ", err)
        return nil, err
    end
    return red, nil
end

-- 查 Redis 单个 key
local function redis_get(conf, key)
    local red, err = redis_cli(conf)
    if not red then
        return nil, err
    end

    local val, err = red:get(key)
    red:set_keepalive(10000, 100)

    if not val or val == ngx.null then
        return nil, "key not found: " .. key
    end

    return val, nil
end

function _M.access(conf, ctx)
    -- 1. 从域名提取 pod_id
    local host = core.request.header(ctx, "Host") or ""
    local domain_suffix = conf.domain_suffix or "yingqichen-ws-test.devcloud.woa.com"
    local pod_id = extract_pod_id(host, domain_suffix)
    if not pod_id then
        return 400, { message = "invalid domain, expected <pod_id>." .. domain_suffix }
    end

    -- 2. 获取 staffname，没有则拒绝
    local staff_name = get_staff_name(ctx)
    if not staff_name or staff_name == "" then
        return 403, { message = "missing staffname, authentication required" }
    end

    -- 3. 从 Redis 查询 pod 对应的 username，校验是否匹配
    local pod_username, err = redis_get(conf, redis_prefix_username .. pod_id)
    if not pod_username then
        core.log.error("failed to get pod username: ", err)
        return 502, { message = "failed to get pod username: " .. (err or "") }
    end

    if pod_username ~= staff_name then
        core.log.warn("username mismatch, staff_name: ", staff_name, " pod_username: ", pod_username, " pod_id: ", pod_id)
        return 403, { message = "permission denied, you are not the owner of this pod" }
    end

    -- 4. 从 Redis 查询 token，塞到请求头
    local token, err = redis_get(conf, redis_prefix_token .. pod_id)
    if not token then
        core.log.error("failed to get pod token: ", err)
        return 502, { message = "failed to get pod token: " .. (err or "") }
    end
    core.request.set_header(ctx, "X-ANY-TOKEN", token)

    -- 5. 从 Redis 查询 pod IP，设置 upstream 转发
    local pod_ip, err = redis_get(conf, redis_prefix_ip .. pod_id)
    if not pod_ip then
        core.log.error("failed to get pod ip: ", err)
        return 502, { message = "failed to get pod ip: " .. (err or "") }
    end

    local ws_port = conf.ws_port or 8900
    local new_nodes = {}
    table_insert(new_nodes, { host = pod_ip, port = ws_port, weight = 1 })
    local up_conf = {
        name = plugin_name,
        type = "roundrobin",
        nodes = new_nodes,
    }
    local matched_route = ctx.matched_route
    upstream.set(ctx, "roundrobin" .. "#route_" .. matched_route.value.id, ctx.conf_version, up_conf, matched_route)

    -- 重写 URI 为 /
    ngx.req.set_uri("/", false)

    core.log.info("ws proxy: pod_id=", pod_id, " staff=", staff_name, " → ", pod_ip, ":", ws_port)
end

return _M

无 IP SSH 直连

基于 AIPSIX 网关的 TCP 流量转发无法通过插件的形式实现 ，只能定义规则限定指定端口到 IP 的定向传输，通过 Admin API 来一条条定义，实现一个端口对应一个 pod ip 的 tcp 流量转发，实现一个端口映射一个 IP 的需求，但似乎是性能难以保证，apisix 的负载太重，所以这里敲定的是 DNAT + IPTable 的技术方案

编码规范

你以后字段命名要注意一点，我是指沟通上，现在都是我看返回，才知道你是怎么命名的，像是这里，我截图里面命名了 sshProxyDomain，你又自己搞个 sshDomain，把 proxy 去掉了，一个字段，怎么命名，有什么改动，要告知前端

你这里后台拼好啊，还要前端来拼啊？你返回个占位符给前端

Go 的导出函数，必须要写注释，而且注释要以函数名开头

在 config.go 中新建字段并通过 yaml:"proxyCommand" 匹配赋值 cfg 中的固定参数（在生产环境中这些配置从七彩石拉下来）

sshProxyConfig:
  - id: ssh-proxy1
    domain: '%s.ssh-proxy.devcloud.woa.com'        # 访问proxy所用的泛域名, %s为环境id占位符
    instanceIds: ['ins-9bohmdl0']                  # proxy的节点
    region: 'ap-guangzhou'                         # 节点的地域
    maxPort: 50000                                 # 最大端口号
    minPort: 40000                                 # 最小端口号
    unschedulable: false                          # 是否不可调度，如果该字段为true，不往该proxy新增规则
    proxyCommand: 'bash /data/home/nft/nft-ctl.sh add %d %s %d'  # 下发新增规则的命令，占位符依次为端口号，目标ip，目标端口

拼接 urischeme 时，这里 domain 的占位符 %s 要用实际的 envId 填充，一定要在后台替换掉这个 %s，替换为真实的环境 id

// 拼接 envId + 泛域名的 domain，替换占位符
domain := fmt.Sprintf(sshConfig.Domain, environmentInstance.Uid)

sshProxyDomain 和 sshProxyPort 的优先级很高，在 eniDisable 的情况，需要替换掉 scheme 里原有的 url 和 port 进行拼接

先在 DevEnvironmentInstance 里加两个额外参数，代理 domain 和代理端口，因为拼接参数的时候只接收一个 envIns *table.DevEnvironmentInstance

models/table/devEnvironmentInstance.go

// DevEnvironmentInstance 开发环境实例表
type DevEnvironmentInstance struct {
    // gorm 字段
    // ...
    // 存储环境对应的task，逻辑字段,不映射表字段存储
    EnvironmentTask       *EnvironmentTask `gorm:"-"`
    OriginalEnvironmentIP string           `gorm:"-"` // 原始环境IP
    AutoDowngrade         bool             `gorm:"-"` // 是否自动降级，当前仅H20场景用到
    // ssh 代理相关，逻辑字段,不映射表字段存储
    SshProxyDomain string `gorm:"-"` // 对无 IP 容器进行 ssh 直连的代理泛域名
    SshProxyPort   int    `gorm:"-"` // 对无 IP 容器 ssh 代理的 dnat 中转端口
}

这里的代理域名和端口在外层处理并封装好：business/env_settings.go: GetInjectVariable

func (its *EnvSettingController) GetInjectVariable(ctx tcontext.Context,
    envIns *table.DevEnvironmentInstance) (envs map[string]string) {
    //...
    // 优先 ssh 代理泛域名
    if envIns.DisableEniIp && envIns.SshProxyDomain != "" {
        url = envIns.SshProxyDomain
    }
    //其次自定义域名
    if url == "" && domain.Id != 0 {
        url = domain.Domain
    }
    _ = json.Unmarshal([]byte(envIns.EnvAccessDomains), &domains)
    //其次默认域名
    if url == "" {
        fixUrl, ok := domains["fix"]
        if ok {
            url = fixUrl
        }
    }
    //最后ip域名
    if url == "" {
        url = domains["dynamics"]
    }
    if url == "" {
        url = envIns.EnvironmentIP
    }

    //...
    // 默认端口36000
    port := 36000
    if envIns.ServerType == cvmContainerServerType {
        containerInfo := make(map[string]string)
        if err = json.Unmarshal([]byte(envIns.ContainerInfo), &containerInfo); err == nil {
            if portStr, ok := containerInfo["port"]; ok {
                if p, err := strconv.Atoi(portStr); err == nil {
                    port = p
                }
            }
        }
    }
    // 如果有 ssh 代理端口，替换掉，优先使用代理端口
    if envIns.DisableEniIp && envIns.SshProxyPort > 0 {
        port = envIns.SshProxyPort
    }

    //...
    // 构造返回
    envs = map[string]string{
        "ANY_USERNAME":    envIns.Username,
        "ANY_ENVID":       envIns.Uid,
        "HOSTNAME":        envIns.PaasEnvironmentId,
        "hostname":        envIns.PaasEnvironmentId,
        "username":        loginUser,
        "ANY_LOGIN_USER":  loginUser,
        "host":            url,
        "ANY_HOST":        url,
        "projectPath":     projectPath,
        "ANY_PROJECTPATH": projectPath,
        "workspace":       workspace,
        "ANY_WORKSPACE":   workspace,
        // TODO 需要把urlSchema中的36000替换为占位符{{.port}}
        "port":         fmt.Sprintf("%d", port),
        "CVM_REGIONID": cvmRegion,
    }
    // ... 注入其他参数，如 workspace, ANY_WORKSPACE

    return
}

SshProxyRule 表

建表，cfg 配置中的 sshProxyConfig: id: ssh-proxy1 对应的就是集群 ID ClusterId，因为查的场景是要匹配相应 envIp 下的 active 节点，所以为这两个字段做了联合索引

• varchar10 和 varchar100 的区别：匹配时不同，数据库就是用来造的，有懂的吗

CREATE TABLE `sshProxyRule` (
  `uid` VARCHAR(45) NOT NULL COMMENT '主键',
  `environmentIp` VARCHAR(100) NOT NULL COMMENT '环境Pod IP',
  `port` INT NOT NULL COMMENT '中转机监听端口',
  `clusterId` VARCHAR(100) NOT NULL COMMENT '集群ID',
  `taskId` VARCHAR(100) NOT NULL COMMENT 'TAT任务ID',
  `status` VARCHAR(100) NOT NULL COMMENT '规则状态(active/delivering/deleted)',
  `createdAt` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  PRIMARY KEY (`uid`),
  KEY `idx_environmentIp` (`environmentIp`),
  KEY `idx_environmentIp_status` (`environmentIp`, `status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='SSH转发规则表';

写库流程，当规则下发成功后，构造完整数据结构并写入数据库

func(its *SshProxyController) SshProxy {
    //... 规则下发成功 ...
    // 写数据库，已知 environmentIp（environmentInstance.EnvironmentIP）
    // port（listen port）、clusterId（sshConfig 的 id）、taskId（invocationId）、status、createAt（now）
    // 主键 uid 通过前缀生成
    row := its.buildSshProxyRule(
        environmentInstance.EnvironmentIP, // pod ip
        listenPort,                        // cvm 上监听流量的 port
        sshConfig.Id,                      // clusterId
        invocationId,                      // tat 的任务 id
        table.SshProxyRuleStatusActive,    // tat 任务状态，下发成功直接置为 Active
    )
    if err = its.createSshProxyRule(ctx, row); err != nil {
        ctx.Logger().Errorf("写入 sshProxyRule 失败, err=%v", err)
        return "", 0, err
    }
    //...
}

sshrule 主键怎么定义？前缀 + 雪花算法生成的 uuid，调用 common 中的主键构造函数

const SshProxyRuleUidPrefix = "sshRule-"

// 构造一条 rule，通过 common GetTaskUuidByPrefix 生成 uid，time.now 获取当前时间
func (its *SshProxyController) buildSshProxyRule(environmentIP string, port int, clusterID, taskID, status string) *table.SshProxyRule {
    return &table.SshProxyRule{
        Uid:           common.GetTaskUuidByPrefix(SshProxyRuleUidPrefix),
        EnvironmentIP: environmentIP,
        Port:          port,
        ClusterId:     clusterID,
        TaskId:        taskID,
        Status:        status,
        CreatedAt:     time.Now(),
    }
}

GORM 写 SQL

// 插入一条 sshProxyRule
func (its *SshProxyController) createSshProxyRule(ctx tcontext.Context, row *table.SshProxyRule) error {
    if err := config.DbGlobal.WithContext(ctx).Create(row).Error; err != nil {
        ctx.Logger().Errorf("写入 sshProxyRule 失败, uid=%s, err=%v", row.Uid, err)
        return err
    }
    return nil
}

根据 environmentId 和 status 查规则

// 查表，根据 environmentIp 查询 sshProxyRule（原则上只有一条），未找到时返回 (nil, nil)
func (its *SshProxyController) getSshProxyRuleByEnvIP(ctx tcontext.Context, environmentIP string) (*table.SshProxyRule, error) {
    rule := &table.SshProxyRule{}
    err := config.DbGlobal.WithContext(ctx).
        Where("environmentIp = ? AND status = ?", environmentIP, table.SshProxyRuleStatusActive).
        First(rule).Error
    if err != nil {
        if err == gorm.ErrRecordNotFound {
            return nil, nil
        }
        ctx.Logger().Errorf("查询 sshProxyRule 失败, environmentIp=%s, err=%v", environmentIP, err)
        return nil, err
    }
    return rule, nil
}

Redis Port INCR

port 怎么拿？直接 Redis 字段 INCR

Redis INCR 实现每个集群 dnat 监听 port 的原子递增，虽然 incr 是原子的，但整个新增规则的流程并非原子，如果后续失败了，这个 incr 的 port 就被浪费了

上锁使用现成的 redislock，这个方法返回一个 unlock 函数，用于 defer 执行，很有意思的设计

// 加锁
func (its *SshProxyController) GetAvailablePort(ctx tcontext.Context, sshConfig *SshProxyItem) (int, error) {
    // ...
    lockKey := SshProxyPortLockPrefix + id

    // 拿带重试的分布式锁，避免并发 INCR 之间穿插初始化逻辑
    unlock := its.acquirePortLock(ctx, lockKey)
    if unlock == nil {
        return 0, fmt.Errorf("acquire ssh proxy port lock timeout, cluster=%s", id)
    }
    defer unlock()
    // ...
}

// RedisLock ...
func RedisLock(ctx tcontext.Context, client redis.Client, key string, timeout int) func() {
    ret, err := client.Do(ctx, "SET", key, 1, "NX", "EX", timeout)
    if err != nil {
        ctx.Logger().Error("lock failed:", key, err.Error())
        return nil
    }
    if ret == nil {
        // lock failed
        return nil
    }
    // lock ok
    unlock := func() {
        if _, err := client.Do(ctx, "del", key); err != nil {
            ctx.Logger().Error("unlock failed:", key, err.Error())
        }
    }
    return unlock
}

不过最后还是没有对分配 port 过程上锁，虽然会造成 port 递增过程中后续执行失败将废用 port，但容忍这种浪费，直接 incr

// Redis 操作
// 构造可用的 port，依赖 redis INCR 的原子性，SETNX 把计数器初始化为 minPort-1，INCR 拿到下一个 port
// 每个 clusterId 一个独立计数器，互不影响
func (its *SshProxyController) getAvailablePort(ctx tcontext.Context, sshConfig *config.SshProxyItem) (int, error) {

    id := sshConfig.Id
    minPort := sshConfig.MinPort
    maxPort := sshConfig.MaxPort

    counterKey := SshProxyPortKeyPrefix + id

    // 若 Key 不存在，SETNX 初始化计数器为 minPort-1，仅在 key 不存在时生效。原子操作，不需要额外锁
    if _, err := config.RedisClientGlobal.Do(ctx, "SET", counterKey, minPort-1, "NX"); err != nil {
        ctx.Logger().Errorf("初始化 redis port 计数器失败, key=%s, err=%v", counterKey, err)
        return 0, err
    }

    // 原地 INCR 拿到下一个 port
    ret, err := config.RedisClientGlobal.Do(ctx, "INCR", counterKey)
    if err != nil {
        ctx.Logger().Errorf("INCR redis port 计数器失败, key=%s, err=%v", counterKey, err)
        return 0, err
    }
    nextPort64, ok := ret.(int64)
    if !ok {
        ctx.Logger().Errorf("INCR 返回值类型异常, key=%s, ret=%T", counterKey, ret)
        return 0, fmt.Errorf("INCR 返回值类型异常")
    }
    nextPort := int(nextPort64)

    ctx.Logger().Infof("获取可用 SSH Proxy 端口, key=%s, port=%d", counterKey, nextPort)

    if nextPort > maxPort {
        ctx.Logger().Errorf("ssh proxy cluster %s 端口已耗尽, port=%d, maxPort=%d", id, nextPort, maxPort)
        return 0, fmt.Errorf("ssh proxy cluster %s 端口已耗尽 (>%d)", id, maxPort)
    }
    if nextPort < minPort {
        ctx.Logger().Errorf("ssh proxy cluster %s 端口计数器异常, port=%d, minPort=%d", id, nextPort, minPort)
        return 0, fmt.Errorf("ssh proxy cluster %s 端口计数器异常 (%d < %d)", id, nextPort, minPort)
    }
    return nextPort, nil
}

TAT 命令下发

将写 dnat 规则的脚本执行在目标 CVM 机器上：bash /data/home/yingqichen/nft/nft-ctl.sh add %d %s %d

下发不需要重试，只下发一次，然后查询命令下发状态，限定 3 秒的 timeout，每 100 ms 重试查询，只有查到所有节点任务都 success 才返回 success

tat 的命令下发接口，实际上是一个远程命令执行的后门程序 TAT...

// RunUserCommand 运行用户命令
func (its *TatController) RunUserCommand(ctx tcontext.Context, region string, params RunUserCommandParams) (invocationId string, err error){
    //...
}

下发 dnat 规则bash /data/home/yingqichen/nft/nft-ctl.sh add %d %s %d到指定 cvm，其中

• listenPort 是 redis incr 分配的可用端口，对应第一个 %d
• deliverIp 是流量转发的目标 ip，即内网的 pod ip，对应 %s
• 第二个 %d 是直连 cvm 的端口号
• sshConfig 包含了 cvm 的明细信息，如 region、instanceIds、command 等

// tat 命令下发
// DeliverSshProxyRule 通过 TAT 下发 dnat 规则到一组 proxy 节点
// proxyCommand 使用 printf 占位，依次为：listenPort(中转监听端口) / deliverIp(目标IP) / 36000(目标SSH端口)
// 如：bash /data/.../nft-ctl.sh add %d %s %d  ->  bash /data/.../nft-ctl.sh add 40123 10.x.x.x 36000
func (its *SshProxyController) deliverSshProxyRule(
    ctx tcontext.Context,
    listenPort int,
    deliverIp string,
    sshConfig *config.SshProxyItem) (string, error) {

    instanceIds := sshConfig.InstanceIds
    proxyCommand := sshConfig.ProxyCommand
    region := sshConfig.Region

    // 按 printf 占位符拼接下发命令
    cmd := fmt.Sprintf(proxyCommand, listenPort, deliverIp, sshBackendPort)
    ctx.Logger().Infof("下发 SSH Proxy 规则, region=%s, instanceIds=%v, cmd=%s", region, instanceIds, cmd)

    params := tencentcloud.RunUserCommandParams{
        CommandName: SshProxyCommandName,
        Description: "分发 SSH Proxy Dnat 到 CVM 节点",
        Content:     base64.StdEncoding.EncodeToString([]byte(cmd)),
        InstanceIds: instanceIds,
    }
    ctx.Logger().Infof("tat 下发 params: ", params)
    invocationId, err := tencentcloud.DefaultTatController.RunUserCommand(ctx, region, params)
    if err != nil {
        ctx.Logger().Errorf("tat 下发 dnat 规则失败, region=%s, err=%v", region, err)
        return "", err
    }
    return invocationId, nil
    // [DEBUG] 临时 mock，跳过真实 TAT 下发
    // ctx.Logger().Warn("[DEBUG] 跳过真实 TAT 下发, mock invocationId, cmd=", cmd)
    // return fmt.Sprintf("inv-mock-%d", listenPort), nil
}

查询命令下发状态，因为这里是全量下发，对集群中配置的所有 cvm 节点全量下发，所以查状态的时候，要对所有 instanceId 进行查询，全部成功后才 success

// CheckAllTasksSuccess 返回 invocationId 的所有节点是否成功（全部成功算成功，否则失败）
func (its *TatController) CheckAllTasksSuccess(ctx tcontext.Context, region,
    invocationId string) (string, error) {

    client, err := its.newTatClient(ctx, region)
    if err != nil {
        ctx.Logger().Errorf("new tat client error: %v", err)
        return "", err
    }
    req := tat.NewDescribeInvocationsRequest()
    req.InvocationIds = common.StringPtrs([]string{invocationId})
    rsp, err := client.DescribeInvocations(req)
    if err != nil {
        ctx.Logger().Errorf("describe invocation error: %v", err)
        return "", err
    }
    if rsp.Response.InvocationSet == nil || len(rsp.Response.InvocationSet) == 0 || len(rsp.Response.InvocationSet[0].InvocationTaskBasicInfoSet) == 0 {
        return "", fmt.Errorf("invocation 无任务记录, invocationId=%s", invocationId)
    }
    tasks := rsp.Response.InvocationSet[0].InvocationTaskBasicInfoSet
    for _, v := range tasks {
        if v == nil || v.InstanceId == nil || v.TaskStatus == nil {
            return "", fmt.Errorf("invocation 任务数据异常, invocationId=%s", invocationId)
        }
        if *v.TaskStatus != CommandStatusSuccess {
            ctx.Logger().Infof("ssh proxy 节点任务未成功, invocationId=%s, instanceId=%s, status=%s", invocationId, *v.InstanceId, *v.TaskStatus)
            return *v.TaskStatus, nil
        }
    }
    return CommandStatusSuccess, nil
}

利用一个 3s 延迟的管道实现间隔 100ms 的重试机制

func (its *SshProxyController) getDeliverSshProxyRuleResult(ctx tcontext.Context, region, invocationId string) error {
    timeoutChan := time.After(3 * time.Second)
    for {
        select {
        case <-timeoutChan:
            return goinfra.IllegalError.Newf("下发 ssh rule 命令超时")
        default:
            status, err := tencentcloud.DefaultTatController.CheckAllTasksSuccess(ctx, region, invocationId)
            if err != nil {
                return err
            }
            if status == tencentcloud.CommandStatusSuccess {
                return nil
            } else if !(status == tencentcloud.CommandStatusRunning || status == tencentcloud.CommandStatusPending ||
                status == tencentcloud.CommandStatusDelivering) {
                ctx.Logger().Errorf("tat 下发任务状态异常: %v", status)
                return goinfra.IllegalError.Newf("tat 下发任务状态异常: %v", status)
            }
            ctx.Logger().Infof("tat 下发任务状态: %v", status)
            time.Sleep(100 * time.Millisecond)
        }
    }
}