stream动态upstream

1. 功能描述

stream动态upstream功能支持在stream模块内动态的添加/删除一组upstream，并且可以在添加时指定算法，以及upstream相关指令。

配置stream动态upstream模块后，可以通过动态API的方式，添加删除upstream，配合upstream api以及stream动态proxy_pass功能，可以自由的实现upstream的动态配置和自由切换。

2. 依赖模块

stream动态upstream功能依赖模块和动态共享内存功能:

njet.conf:

load_module  modules/njt_stream_dyn_upstream_module.so;

shared_slab_pool_size  100m;

njet_ctrl.conf

load_module modules/njt_stream_dyn_upstream_api_module.so;
load_module modules/njt_stream_dyn_upstream_module.so;

3. 配置说明

njet.conf

helper broker modules/njt_helper_broker_module.so conf/mqtt.conf;
helper ctrl modules/njt_helper_ctrl_module.so conf/ctrl.conf;

load_module  modules/njt_stream_dyn_upstream_module.so;
   #动态upstream模块
load_module  modules/njt_http_upstream_member_module.so;
   #upstream_api模块

user  root;
worker_processes auto;

cluster_name helper;
node_name node1;

error_log  logs/error.log info;
pid        logs/njet.pid;
events {
    worker_connections  1024;
}

shared_slab_pool_size  100m;  #动态共享内存管理指令，配置大小要大于需要配置的upstream共享内存大小

http {
    dyn_kv_conf conf/iot-work.conf;
    include       mime.types;
    default_type  application/octet-stream;

    access_log  logs/access.log;

    upstream backend1 {
  
         zone backend1 1m;
   
         server 127.0.0.1:5678;
   
      }

        server {
                listen 5555;
                server_name localhost;
  
                location / { 

                        return 200 "5555 ok";
                }

        }

}


stream {

   
        upstream backend_s1 {

        zone backend_s1 10m;

         server 192.168.40.150:5678;

        }

        server {
                listen 22222;
  
                return "22222 ok";

        }



}

njet_ctrl.conf

load_module modules/njt_http_sendmsg_module.so;
load_module modules/njt_ctrl_config_api_module.so;
load_module modules/njt_http_upstream_api_module.so;  #upstream api模块
load_module modules/njt_doc_module.so;
load_module modules/njt_stream_dyn_upstream_api_module.so;  #stream动态upstream模块
load_module modules/njt_stream_dyn_upstream_module.so;      #stream动态upstream模块
load_module modules/njt_helper_health_check_module.so;
events {
    worker_connections  1024;
}
error_log         logs/error_ctrl.log info;

http {
    dyn_sendmsg_conf  conf/iot-ctrl.conf;
    dyn_kv_conf       conf/ctrl_kv.conf;
    access_log        logs/access_ctrl.log combined;

    include           mime.types;

    server {
        listen       8081;
        keepalive_timeout 0;

        location /api {
             dyn_module_api;

        }
         location /doc {
             doc_api;
         }

  }

}


cluster_name helper;
node_name node1;

ctrl_kv.conf;

topic /ins/ups/#
topic /ins/stream_ups/#

4. 配置样例

4.1 API说明

添加一个动态upstream：

POST http://ip+port/api/v1/dyn_stream_ups

删除一个动态upstream

PUT http://ip+port/api/v1/dyn_stream_ups

特别提示：添加动态upstream时应配置持久化指令state，未配置情况下reload后会导致server列表丢失。

查询动态upstream配置：

GET http://ip+port/api/v1/config/dyn_stream_upstream

4.2 添加stream动态upstream，并使用默认轮询算法。

发送

curl -X 'POST' \
  'http:/127.0.0.1:8081/api/v1/dyn_stream_ups' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "type": "add",
  "upstream_name": "dyn_backend1",
  "upstream_body": "zone dyn_backend1 1m"
}'

返回

{
  "code": 0,
  "msg": "success."
}

4.3 添加stream动态upstream，并配置random算法。

发送

curl -X 'POST' \
  'http://127.0.0.1:8081/api/v1/dyn_stream_ups' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "type": "add",
  "upstream_name": "dyn_backend1",
  "upstream_body": "zone dyn_backend1 1m;random;"
}'

返回

{
  "code": 0,
  "msg": "success."
}

4.4 添加stream动态upstream，并配置持久化指令。

发送

curl -X 'POST' \
  'http://127.0.0.1:8081/api/v1/dyn_stream_ups' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "type": "add",
  "upstream_name": "dyn_backend1",
  "upstream_body": "zone dyn_backend1 1m;state upstream.conf;"
}'

返回

{
  "code": 0,
  "msg": "success."
}

4.5 查询stream块当前配置的所有upstream

发送

curl -X 'GET' \
  'http://127.0.0.1:8081/api/v1/config/dyn_stream_upstream' \
  -H 'accept: application/json'

返回

[
  {
    "name": "backend_s1",
    "balance": "round_robin",
    "zone": "backend1",
    "is_static": true
  }，
  {
    "name": "dyn_backend1",
    "balance": "round_robin",
    "zone": "dyn_backend1",
    "is_static": false
  }
]

4.6 删除动态添加的配置的upstream

发送

curl -X 'PUT' \
  'http://127.0.0.1:8081/api/v1/dyn_stream_ups' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "type": "del",
  "upstream_name": "dyn_backend1"
}'

返回

{
  "code": 0,
  "msg": "success."
}

4.7 在动态upstream中添加一个server

先添加一组动态upstream

curl -X 'POST' \
  'http://127.0.0.1:8081/api/v1/dyn_stream_ups' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "type": "add",
  "upstream_name": "dyn_backend1",
  "upstream_body": "zone dyn_backend1 1m;state upstream.conf;"
}'

返回值

{
  "code": 0,
  "msg": "success."
}

使用动态upstream api添加一个server

curl -X 'POST' \
  'http://127.0.0.1:8081/api/v1/upstream_api/stream/upstreams/dyn_backend1/servers/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "server": "127.0.0.1:5678",
  "weight": 4,
  "max_conns": 0,
  "max_fails": 0,
  "fail_timeout": "10s",
  "slow_start": "10s",
  "backup": false,
  "down": false
}'

返回值

{
  "id": 0,
  "server": "127.0.0.1:5678",
  "weight": 4,
  "max_conns": 0,
  "max_fails": 0,
  "fail_timeout": "10s",
  "slow_start": "10s",
  "route": "",
  "backup": false,
  "down": false
}

4.8 在动态upstream中添加一个server,并使用stream动态proxy_pass更新location的proxy_pass

stream动态proxy_pass具体配置可以参考stream动态proxy_pass章节文件同步

先添加一组动态upstream

curl -X 'POST' \
  'http://127.0.0.1:8081/api/v1/dyn_ups' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "type": "add",
  "upstream_name": "dyn_backend1",
  "upstream_body": "zone dyn_backend1 1m;state upstream.conf;"
}'

返回值

{
  "code": 0,
  "msg": "success."
}

使用动态upstream api添加一个server

curl -X 'POST' \
  'http://127.0.0.1:8081/api/v1/upstream_api/stream/upstreams/dyn_backend1/servers/' \
  -H 'accept: application/json' \
  -H 'Content-Type: application/json' \
  -d '{
  "server": "127.0.0.1:5678",
  "weight": 4,
  "max_conns": 0,
  "max_fails": 0,
  "fail_timeout": "10s",
  "slow_start": "10s",
  "backup": false,
  "down": false
}'

返回值

{
  "id": 0,
  "server": "127.0.0.1:5678",
  "weight": 4,
  "max_conns": 0,
  "max_fails": 0,
  "fail_timeout": "10s",
  "slow_start": "10s",
  "route": "",
  "backup": false,
  "down": false
}

使用动态proxy_pass接口切换upstream

curl -X 'PUT' \
  'http://127.0.0.1:8081/api/v1/config/stream_proxy_pass' \
  -H 'accept: */*' \
  -H 'Content-Type: application/json' \
  -d '{
  "servers": [
    {
      "listens": [
        "0.0.0.0:22224"
      ],
      "serverNames": [
        ""
      ],
      "proxy_pass": "dyn_backend1"
    }
  ]
}'

返回

{
  "code": 0,
  "msg": "success."
}

使用proxy_pass查询接口查看当前location的proxy_pass

发送

curl -X 'GET' \
  'http://127.0.0.1:8081/api/v1/config/stream_proxy_pass' \
  -H 'accept: application/json'

返回

{
  "servers": [
    {
      "listens": [
        "0.0.0.0:22224"
      ],
      "serverNames": [
        ""
      ],
      "proxy_pass": "dyn_backend1"
    }
  ]
}