增加LeastConnectionLoadBalancer负载均衡策略（config.json里面配置使用） (#35)

flomesh-io · Aug 31, 2023 · b669e13 · b669e13
1 parent d2da1b9
commit b669e13
Show file tree

Hide file tree

Showing 6 changed files with 222 additions and 12 deletions.
diff --git a/pjs/config.json b/pjs/config.json
@@ -17,14 +17,8 @@
       "Protocol": "HTTP",
       "Port": 8080,
       "AccessControlLists": {
-        "blacklist": [
-          "127.0.0.11",
-          "192.168.122.1/32"
-        ],
-        "whitelist": [
-          "192.168.122.18",
-          "127.0.0.1/32"
-        ]
+        "blacklist": [],
+        "whitelist": []
       }
     },
     {
@@ -93,7 +87,6 @@
       "HealthCheck": {
         "Interval": 10,
         "MaxFails": 3,
-        "FailTimeout": 30,
         "Uri": "/",
         "Matches": [
           {
@@ -105,7 +98,7 @@
           }
         ]
       },
-      "Algorithm": "LeastWorkLoadBalancer",
+      "Algorithm": "LeastConnectionLoadBalancer",
       "Endpoints": {
         "127.0.0.1:8081": {
           "Weight": 50
@@ -131,6 +124,8 @@
       "http/metrics.js",
       "http/tracing.js",
       "http/logging.js",
+      "http/access-control-domain.js",
+      "http/access-control-route.js",
       "http/circuit-breaker.js",
       "http/throttle-domain.js",
       "http/throttle-route.js",
@@ -152,6 +147,8 @@
       "http/metrics.js",
       "http/tracing.js",
       "http/logging.js",
+      "http/access-control-domain.js",
+      "http/access-control-route.js",
       "http/circuit-breaker.js",
       "http/throttle-domain.js",
       "http/throttle-route.js",

diff --git a/pjs/http/access-control-domain.js b/pjs/http/access-control-domain.js
@@ -0,0 +1,101 @@
+((
+  { isDebugEnabled } = pipy.solve('config.js'),
+
+  { aclCounter } = pipy.solve('lib/metrics.js'),
+  aclDomainCounter = aclCounter.withLabels('domain'),
+
+  parseIpList = ipList => (
+    (ips, ipRanges) => (
+      (ipList || []).forEach(
+        o => (
+          o.indexOf('/') > 0 ? (
+            !ipRanges && (ipRanges = []),
+            ipRanges.push(new Netmask(o))
+          ) : (
+            !ips && (ips = {}),
+            ips[o] = true
+          )
+        )
+      ),
+      (ips || ipRanges) ? ({ ips, ipRanges }) : undefined
+    )
+  )(),
+
+  aclsCache = new algo.Cache(
+    acls => (
+      {
+        blackList: parseIpList(acls?.blacklist),
+        whiteList: parseIpList(acls?.whitelist),
+      }
+    )
+  ),
+
+  checkACLs = ip => (
+    (
+      acls = aclsCache.get(__domain?.AccessControlLists),
+      white = acls?.whiteList,
+      black = acls?.blackList,
+      blackMode = true,
+      block = false,
+      pass = false,
+    ) => (
+      white && (
+        blackMode = false,
+        (white?.ips?.[ip] && (pass = true)) || (
+          pass = white?.ipRanges?.find?.(r => r.contains(ip))
+        )
+      ),
+      blackMode && (
+        (black?.ips?.[ip] && (block = true)) || (
+          block = black?.ipRanges?.find?.(r => r.contains(ip))
+        )
+      ),
+      blackMode ? Boolean(!block) : Boolean(pass)
+    )
+  )(),
+) => pipy({
+  _ips: null,
+  _pass: false,
+})
+
+.import({
+  __domain: 'route',
+})
+
+.pipeline()
+.handleMessageStart(
+  msg => (
+    __domain?.AccessControlLists?.enableXFF && (
+      _ips = msg.head?.headers['x-forwarded-for']
+    ),
+    _ips ? (
+      _pass = _ips.split(',').every(ip => checkACLs(ip.trim()))
+    ) : (
+      _pass = checkACLs(__inbound.remoteAddress)
+    )
+  )
+)
+.branch(
+  () => _pass, (
+    $=>$.chain()
+  ), (
+    $=>$
+    .branch(
+      isDebugEnabled, (
+        $=>$.handleStreamStart(
+          () => (
+            console.log('[access-control-domain] blocked XFF, IP address:', _ips, __inbound.remoteAddress, _pass)
+          )
+        )
+      )
+    )
+    .replaceMessage(
+      () => (
+        aclDomainCounter.increase(),
+        new Message({ status: __domain?.AccessControlLists?.status || 403 }, __domain?.AccessControlLists?.message || '')
+      )
+    )
+  )
+)
+
+)()
diff --git a/pjs/http/access-control-route.js b/pjs/http/access-control-route.js
@@ -0,0 +1,101 @@
+((
+  { isDebugEnabled } = pipy.solve('config.js'),
+
+  { aclCounter } = pipy.solve('lib/metrics.js'),
+  aclRouteCounter = aclCounter.withLabels('route'),
+
+  parseIpList = ipList => (
+    (ips, ipRanges) => (
+      (ipList || []).forEach(
+        o => (
+          o.indexOf('/') > 0 ? (
+            !ipRanges && (ipRanges = []),
+            ipRanges.push(new Netmask(o))
+          ) : (
+            !ips && (ips = {}),
+            ips[o] = true
+          )
+        )
+      ),
+      (ips || ipRanges) ? ({ ips, ipRanges }) : undefined
+    )
+  )(),
+
+  aclsCache = new algo.Cache(
+    acls => (
+      {
+        blackList: parseIpList(acls?.blacklist),
+        whiteList: parseIpList(acls?.whitelist),
+      }
+    )
+  ),
+
+  checkACLs = ip => (
+    (
+      acls = aclsCache.get(__route?.config?.AccessControlLists),
+      white = acls?.whiteList,
+      black = acls?.blackList,
+      blackMode = true,
+      block = false,
+      pass = false,
+    ) => (
+      white && (
+        blackMode = false,
+        (white?.ips?.[ip] && (pass = true)) || (
+          pass = white?.ipRanges?.find?.(r => r.contains(ip))
+        )
+      ),
+      blackMode && (
+        (black?.ips?.[ip] && (block = true)) || (
+          block = black?.ipRanges?.find?.(r => r.contains(ip))
+        )
+      ),
+      blackMode ? Boolean(!block) : Boolean(pass)
+    )
+  )(),
+) => pipy({
+  _ips: null,
+  _pass: false,
+})
+
+.import({
+  __route: 'route',
+})
+
+.pipeline()
+.handleMessageStart(
+  msg => (
+    __route?.config?.AccessControlLists?.enableXFF && (
+      _ips = msg.head?.headers['x-forwarded-for']
+    ),
+    _ips ? (
+      _pass = _ips.split(',').every(ip => checkACLs(ip.trim()))
+    ) : (
+      _pass = checkACLs(__inbound.remoteAddress)
+    )
+  )
+)
+.branch(
+  () => _pass, (
+    $=>$.chain()
+  ), (
+    $=>$
+    .branch(
+      isDebugEnabled, (
+        $=>$.handleStreamStart(
+          () => (
+            console.log('[access-control-route] blocked XFF, IP address:', _ips, __inbound.remoteAddress, _pass)
+          )
+        )
+      )
+    )
+    .replaceMessage(
+      () => (
+        aclRouteCounter.increase(),
+        new Message({ status: __route?.config?.AccessControlLists?.status || 403 }, __route?.config?.AccessControlLists?.message || '')
+      )
+    )
+  )
+)
+
+)()
diff --git a/pjs/http/forward.js b/pjs/http/forward.js
@@ -29,7 +29,11 @@
             (serviceConfig.Algorithm === 'HashingLoadBalancer') ? (
               new algo.HashingLoadBalancer(Object.keys(endpoints))
             ) : (
-              new algo[serviceConfig.Algorithm || 'RoundRobinLoadBalancer'](endpoints)
+              (serviceConfig.Algorithm === 'LeastConnectionLoadBalancer') ? (
+                new algo.LeastWorkLoadBalancer(Object.keys(endpoints))
+              ) : (
+                new algo[serviceConfig.Algorithm || 'RoundRobinLoadBalancer'](endpoints)
+              )
             )
           ),
           endpointAttributes,

diff --git a/pjs/lib/metrics.js b/pjs/lib/metrics.js
@@ -84,6 +84,9 @@
       rateLimitCounter: new stats.Counter('http_local_rate_limiter', [
         'http_local_rate_limit'
       ]),
+      aclCounter: new stats.Counter('access_control', [
+        'type'
+      ]),
     }
   )
 

diff --git a/pjs/tcp/forward.js b/pjs/tcp/forward.js
@@ -31,7 +31,11 @@
             (serviceConfig.Algorithm === 'HashingLoadBalancer') ? (
               new algo.HashingLoadBalancer(Object.keys(endpoints))
             ) : (
-              new algo[serviceConfig.Algorithm || 'RoundRobinLoadBalancer'](endpoints)
+              (serviceConfig.Algorithm === 'LeastConnectionLoadBalancer') ? (
+                new algo.LeastWorkLoadBalancer(Object.keys(endpoints))
+              ) : (
+                new algo[serviceConfig.Algorithm || 'RoundRobinLoadBalancer'](endpoints)
+              )
             )
           ),
           endpointAttributes,
-Original file line number
+Diff line change
@@ Expand Up / @@ -84,6 +84,9 @@ @@
           rateLimitCounter: new stats.Counter('http_local_rate_limiter', [
             'http_local_rate_limit'
           ]),
+          aclCounter: new stats.Counter('access_control', [
+            'type'
+          ]),
         }
       )
@@ Expand Down @@