function set_upstream_by_reids(ctx, conf, redis_key)

    local red, err2 = redis.new(conf)
    if not red then
        return false, err2
    end

    local cluster, get_err = red:get(redis_key)
    if get_err then
        return false, get_err

    elseif not cluster or cluster == nil or (type(cluster) == "userdata" and tostring(cluster) == "userdata: NULL") then
        core.response.set_header("x-cluster", conf.default_upstream_id)
        ctx.upstream_id = conf.default_upstream_id

        return true
    else
        core.response.set_header("x-cluster", cluster)
        ctx.upstream_id = cluster
        return true
    end
end