feat(consul): filter nodes in upstream with metadata

apisix/discovery/consul/init.lua

@@ -21,6 +21,8 @@ local core_sleep         = require("apisix.core.utils").sleep
 local resty_consul       = require('resty.consul')
 local http               = require('resty.http')
 local util               = require("apisix.cli.util")
+local discovery_utils    = require("apisix.utils.discovery")
+local cjson              = require('cjson')
 local ipairs             = ipairs
 local error              = error
 local ngx                = ngx
@@ -41,6 +43,7 @@ local pcall              = pcall
 local null               = ngx.null
 local type               = type
 local next               = next
+local cjson_null         = cjson.null
 
 local all_services = core.table.new(0, 5)
 local default_service
@@ -78,15 +81,19 @@ function _M.all_nodes()
     return all_services
 end
 
-
-function _M.nodes(service_name)
+function _M.nodes(service_name, discovery_args)
     if not all_services then
         log.error("all_services is nil, failed to fetch nodes for : ", service_name)
         return
     end
 
     local resp_list = all_services[service_name]
 
+    local metadata_match = discovery_args and discovery_args.metadata_match
+    if metadata_match then
+        resp_list = discovery_utils.nodes_metadata_match(resp_list, metadata_match)
+    end
+
     if not resp_list then
         log.error("fetch nodes failed by ", service_name, ", return default service")
         return default_service and {default_service}
@@ -98,7 +105,6 @@ function _M.nodes(service_name)
     return resp_list
 end
 
-
 local function update_all_services(consul_server_url, up_services)
     -- clean old unused data
     local old_services = consul_services[consul_server_url] or {}
@@ -511,11 +517,14 @@ function _M.connect(premature, consul_server, retry_delay)
                 local nodes = up_services[service_name]
                 local nodes_uniq = {}
                 for _, node in ipairs(result.body) do
-                    if not node.Service then
+                    local service = node.Service
+                    if not service then
                         goto CONTINUE
                     end
 
-                    local svc_address, svc_port = node.Service.Address, node.Service.Port
+                    local svc_address = service.Address
+                    local svc_port = service.Port
+                    local metadata = service.Meta
                     -- Handle nil or 0 port case - default to 80 for HTTP services
                     if not svc_port or svc_port == 0 then
                         svc_port = 80
@@ -527,12 +536,23 @@ function _M.connect(premature, consul_server, retry_delay)
                     end
                     -- not store duplicate service IDs.
                     local service_id = svc_address .. ":" .. svc_port
+                    -- ensure that metadata is an accessible table,
+                    -- avoid `null` returned by cjson
+                    if metadata == cjson_null then
+                        metadata = nil
+                    elseif type(metadata) ~= "table" then
+                        log.error("service ", service_id,
+                                " has invalid metadata, use nil as default: ",
+                                json_delay_encode(metadata))
+                        metadata = nil
+                    end
                     if not nodes_uniq[service_id] then
                         -- add node to nodes table
                         core.table.insert(nodes, {
                             host = svc_address,
                             port = tonumber(svc_port),
                             weight = default_weight,
+                            metadata = metadata
                         })
                         nodes_uniq[service_id] = true
                     end

apisix/schema_def.lua

@@ -488,6 +488,18 @@ local upstream_schema = {
                     description = "group name",
                     type = "string",
                 },
+                metadata_match = {
+                    description = "metadata for filtering service instances",
+                    type = "object",
+                    additionalProperties = {
+                        type = "array",
+                        items = {
+                            description = "candidate metadata value",
+                            type = "string",
+                        },
+                        uniqueItems = true,
+                    }
+                },
             }
         },
         pass_host = {

apisix/utils/discovery.lua

@@ -0,0 +1,66 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements.  See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License.  You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+local core   = require("apisix.core")
+local ipairs = ipairs
+local pairs  = pairs
+
+local _M = {}
+
+local function do_metadata_match(node, metadata_match)
+    local metadata = node.metadata
+    -- because metadata_match has already been checked in nodes_metadata_match,
+    -- there is at least one role, if there is no metadata in node, it's must not matched
+    if not metadata then
+        return false
+    end
+    for key, values in pairs(metadata_match) do
+        local matched = false
+        for _, value in ipairs(values) do
+            if metadata[key] == value then
+                matched = true
+                break
+            end
+        end
+        if not matched then
+            return false
+        end
+    end
+    return true
+end
+
+local function nodes_metadata_match(nodes, metadata_match)
+    if not nodes then
+        return nil
+    end
+
+    -- fast path: there is not metadata_match roles, all nodes are available,
+    -- and make a guarantee for do_metadata_match: at least one role
+    if not metadata_match then
+        return nodes
+    end
+
+    local result = {}
+    for _, node in ipairs(nodes) do
+        if do_metadata_match(node, metadata_match) then
+            core.table.insert(result, node)
+        end
+    end
+    return result
+end
+_M.nodes_metadata_match = nodes_metadata_match
+
+return _M

docs/en/latest/discovery/consul.md

@@ -232,6 +232,61 @@ $ curl http://127.0.0.1:9180/apisix/admin/stream_routes/1 -H "X-API-KEY: $admin_
 }'
 ```
 
+### discovery_args
+
+| Name           | Type   | Requirement | Default | Valid | Description                                                  |
+|----------------| ------ | ----------- | ------- | ----- | ------------------------------------------------------------ |
+| metadata_match | object | optional    | {}      |       | Filter service instances by metadata using containment matching |
+
+#### Metadata filtering
+
+APISIX supports filtering service instances based on metadata. When a route is configured with metadata conditions, only service instances whose metadata matched with roles specified in the route's `metadata_match` configuration will be selected.
+
+Example: If a service instance has metadata `{lane: "a", env: "prod", version: "1.0"}`, it will match routes configured with metadata `{lane: ["a"]}` or `{lane: ["a", "b"], env: "prod"}`, but not routes configured with `{lane: ["c"]}` or `{lane: "a", region: "us"}`.
+
+Example of routing a request with metadata filtering:
+
+```shell
+$ curl http://127.0.0.1:9180/apisix/admin/routes/5 -H "X-API-KEY: $admin_key" -X PUT -i -d '
+{
+    "uri": "/consulWithMetadata/*",
+    "upstream": {
+        "service_name": "APISIX-CONSUL",
+        "type": "roundrobin",
+        "discovery_type": "consul",
+        "discovery_args": {
+          "metadata_match": {
+            "version": ["v1", "v2"]
+          }
+        }
+    }
+}'
+```
+
+This route will only route traffic to service instances that have the metadata field `version` set to `v1` or `v2`.
+
+For multiple metadata criteria:
+
+```shell
+$ curl http://127.0.0.1:9180/apisix/admin/routes/6 -H "X-API-KEY: $admin_key" -X PUT -i -d '
+{
+    "uri": "/consulWithMultipleMetadata/*",
+    "upstream": {
+        "service_name": "APISIX-CONSUL",
+        "type": "roundrobin",
+        "discovery_type": "consul",
+        "discovery_args": {
+          "metadata_match": {
+            "lane": ["a"],
+            "env": ["prod"]
+          }
+        }
+    }
+}'
+```
+
+This route will only route traffic to service instances that have both `lane: "a"` and `env: "prod"` in their metadata.
+
 You could find more usage in the `apisix/t/discovery/stream/consul.t` file.
 
 ## Debugging API

t/discovery/consul.t

@@ -781,3 +781,71 @@ location /sleep {
     qr//
 ]
 --- ignore_error_log
+
+
+
+=== TEST 16: test metadata_match with consul discovery
+--- yaml_config eval: $::yaml_config
+--- apisix_yaml
+routes:
+  -
+    uri: /*
+    upstream:
+      service_name: service-a
+      type: roundrobin
+      discovery_type: consul
+      discovery_args:
+        metadata_match:
+          version:
+            - v2
+            - v3
+#END
+--- config
+location /v1/agent {
+    proxy_pass http://127.0.0.1:8500;
+}
+location /sleep {
+    content_by_lua_block {
+        local args = ngx.req.get_uri_args()
+        local sec = args.sec or "2"
+        ngx.sleep(tonumber(sec))
+        ngx.say("ok")
+    }
+}
+--- timeout: 5
+--- pipelined_requests eval
+[
+    "PUT /v1/agent/service/register\n" . "{\"ID\":\"service-a1\",\"Name\":\"service-a\",\"Address\":\"127.0.0.1\",\"Port\":30511,\"Meta\":{\"service_b_version\":\"4.1\",\"version\":\"v1\"},\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
+    "PUT /v1/agent/service/register\n" . "{\"ID\":\"service-a2\",\"Name\":\"service-a\",\"Address\":\"127.0.0.1\",\"Port\":30512,\"Meta\":{\"service_b_version\":\"4.1\",\"version\":\"v2\"},\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
+    "PUT /v1/agent/service/register\n" . "{\"ID\":\"service-a3\",\"Name\":\"service-a\",\"Address\":\"127.0.0.1\",\"Port\":30513,\"Meta\":{\"service_b_version\":\"4.1\",\"version\":\"v3\"},\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
+    "PUT /v1/agent/service/register\n" . "{\"ID\":\"service-a4\",\"Name\":\"service-a\",\"Address\":\"127.0.0.1\",\"Port\":30514,\"Meta\":{\"service_b_version\":\"4.1\",\"version\":\"v4\"},\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
+    "GET /sleep",
+
+    "GET /hello?run1",
+    "GET /hello?run2",
+    "GET /hello?run3",
+
+    "PUT /v1/agent/service/deregister/service-a1",
+    "PUT /v1/agent/service/deregister/service-a2",
+    "PUT /v1/agent/service/deregister/service-a3",
+    "PUT /v1/agent/service/deregister/service-a4",
+]
+--- response_body_like eval
+[
+    qr//,
+    qr//,
+    qr//,
+    qr//,
+    qr/ok\n/,
+
+    qr/[2-3]/,
+    qr/[2-3]/,
+    qr/[2-3]/,
+
+    qr//,
+    qr//,
+    qr//,
+    qr//
+]
+--- no_error_log
+[error]

t/discovery/consul2.t

@@ -189,7 +189,7 @@ discovery:
 --- request
 GET /t
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1},{"host":"127.0.0.1","port":30512,"weight":1},{"host":"localhost","port":30511,"weight":1},{"host":"localhost","port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","port":80,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","metadata":{"service_version":"1.0"},"port":80,"weight":1}]}
 
 
 
@@ -223,7 +223,7 @@ discovery:
 --- request
 GET /t
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1},{"host":"127.0.0.1","port":30512,"weight":1},{"host":"localhost","port":30511,"weight":1},{"host":"localhost","port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","port":80,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","metadata":{"service_version":"1.0"},"port":80,"weight":1}]}
 
 
 
@@ -257,7 +257,7 @@ discovery:
 --- request
 GET /t
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1},{"host":"localhost","port":30511,"weight":1},{"host":"127.0.0.1","port":30512,"weight":1},{"host":"localhost","port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","port":80,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","metadata":{"service_version":"1.0"},"port":80,"weight":1}]}
 
 
 
@@ -291,7 +291,7 @@ discovery:
 --- request
 GET /t
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1},{"host":"127.0.0.1","port":30512,"weight":1},{"host":"localhost","port":30511,"weight":1},{"host":"localhost","port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","port":80,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1},{"host":"localhost","metadata":{"service_a_version":"4.0"},"port":30512,"weight":1}],"service_no_port":[{"host":"127.0.0.1","metadata":{"service_version":"1.0"},"port":80,"weight":1}]}
 
 
 

t/discovery/consul_dump.t

@@ -57,12 +57,15 @@ location /v1/agent {
     "PUT /v1/agent/service/deregister/service_a2",
     "PUT /v1/agent/service/deregister/service_b1",
     "PUT /v1/agent/service/deregister/service_b2",
+    "PUT /v1/agent/service/deregister/service_c1",
+    "PUT /v1/agent/service/deregister/service_c2",
     "PUT /v1/agent/service/register\n" . "{\"ID\":\"service_a1\",\"Name\":\"service_a\",\"Tags\":[\"primary\",\"v1\"],\"Address\":\"127.0.0.1\",\"Port\":30511,\"Meta\":{\"service_a_version\":\"4.0\"},\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
     "PUT /v1/agent/service/register\n" . "{\"ID\":\"service_b1\",\"Name\":\"service_b\",\"Tags\":[\"primary\",\"v1\"],\"Address\":\"127.0.0.1\",\"Port\":8002,\"Meta\":{\"service_b_version\":\"4.1\"},\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
+    "PUT /v1/agent/service/register\n" . "{\"ID\":\"service_c1\",\"Name\":\"service_c\",\"Tags\":[\"primary\",\"v1\"],\"Address\":\"127.0.0.1\",\"Port\":8003,\"EnableTagOverride\":false,\"Weights\":{\"Passing\":10,\"Warning\":1}}",
 ]
 --- response_body eval
 --- error_code eval
-[200, 200, 200, 200, 200, 200]
+[200, 200, 200, 200, 200, 200, 200, 200, 200]
 
 
 
@@ -95,7 +98,7 @@ discovery:
 --- request
 GET /t
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1}],"service_b":[{"host":"127.0.0.1","port":8002,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1}],"service_b":[{"host":"127.0.0.1","metadata":{"service_b_version":"4.1"},"port":8002,"weight":1}],"service_c":[{"host":"127.0.0.1","port":8003,"weight":1}]}
 
 
 
@@ -141,9 +144,10 @@ location /v1/agent {
 [
     "PUT /v1/agent/service/deregister/service_a1",
     "PUT /v1/agent/service/deregister/service_b1",
+    "PUT /v1/agent/service/deregister/service_c1",
 ]
 --- error_code eval
-[200, 200]
+[200, 200, 200]
 
 
 
@@ -450,7 +454,7 @@ discovery:
 --- request
 GET /bonjour
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1}],"service_b":[{"host":"127.0.0.1","port":30517,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1}],"service_b":[{"host":"127.0.0.1","metadata":{"service_b_version":"4.1"},"port":30517,"weight":1}]}
 
 
 
@@ -508,4 +512,4 @@ discovery:
 --- request
 GET /t
 --- response_body
-{"service_a":[{"host":"127.0.0.1","port":30511,"weight":1}]}
+{"service_a":[{"host":"127.0.0.1","metadata":{"service_a_version":"4.0"},"port":30511,"weight":1}]}