Merge pull request #43 from hamishcoleman/main

Add pagination to management API

.github/workflows/tests.yml

@@ -140,7 +140,6 @@ jobs:
       fail-fast: true
       matrix:
         os:
-          - macos-11
           - macos-12
 
     steps:
@@ -151,7 +150,6 @@ jobs:
         run: |
           git fetch --force --tags
 
-
       - name: Install packages
         run: |
           brew install automake
@@ -362,6 +360,9 @@ jobs:
       fail-fast: true
       matrix:
         include:
+          - name: arm64
+            sdk_ver: 22.03.3
+            sdk: https://downloads.openwrt.org/releases/22.03.3/targets/armvirt/64/openwrt-sdk-22.03.3-armvirt-64_gcc-11.2.0_musl.Linux-x86_64.tar.xz
           - name: mips_24kc
             sdk_ver: 22.03.3
             sdk: https://downloads.openwrt.org/releases/22.03.3/targets/lantiq/xrx200/openwrt-sdk-22.03.3-lantiq-xrx200_gcc-11.2.0_musl.Linux-x86_64.tar.xz

VERSION

@@ -1 +1 @@
-3.3.4
+3.4.1

apps/n3n-supernode.c

@@ -54,8 +54,6 @@
 
 #define HASH_FIND_COMMUNITY(head, name, out) HASH_FIND_STR(head, name, out)
 
-static struct n3n_runtime_data sss_node;
-
 /* *************************************************** */
 
 #define GETOPTS "O:Vdhv"
@@ -383,6 +381,7 @@ static void term_handler (int sig)
 
 /** Main program entry point from kernel. */
 int main (int argc, char * argv[]) {
+    static struct n3n_runtime_data sss_node;
 
     // Do this early to register all internals
     n3n_initfuncs();

doc/ManagementAPI.md

@@ -96,3 +96,20 @@ will check for a standard HTTP Authorization header in the request.
 The authentication is a simple password that the client must provide. It
 defaults to 'n3n' and can either be set with the config option
 `management.password`
+
+## Pagination
+
+If the result of an API call will overrun the size of the internal buffer,
+the response will indicate an "overflow" condition by returning a 507 result
+and a JsonRPC error object.
+
+When an overflow condition is returned, the error object may contain the
+count of the number of items that could be added before the overflow occured.
+The request can be retried with pagination parameters to avoid the overflow.
+(Note that this also opens up a window for the internal data to change during
+the paginated request)
+
+Add the offset and limit values to the param dictionary.
+
+The n3nctl tool has an example on how to use this implemented in its
+JsonRPC.get() method

scripts/n3nctl

@@ -17,10 +17,6 @@ import urllib.parse
 import urllib.request
 
 
-class Unauthenticated(Exception):
-    pass
-
-
 class UnixHandler(urllib.request.BaseHandler):
     def __init__(self, basepath):
         self.basepath = basepath
@@ -102,33 +98,109 @@ class JsonRPC:
             req.add_header("Authorization", b"Basic " + encoded)
         return req
 
-    def get(self, method, params=None):
+    class Unauthenticated(Exception):
+        """Raised if the request needs authentication added"""
+        pass
+
+    class Overflow(Exception):
+        """Raised if the daemon overflows its internal buffer"""
+        def __init__(self, count):
+            self.count = count
+            super().__init__()
+
+    def get_nopagination(self, method, params=None):
+        """Makes the RPC request, with no automated pagination retry"""
+
         req = self._request_obj(method, params)
 
         try:
             r = urllib.request.urlopen(req, timeout=self.timeout)
         except urllib.error.HTTPError as e:
             if e.code == 401:
-                raise Unauthenticated
+                raise JsonRPC.Unauthenticated
             raise e
 
         if r.status == 401:
-            raise Unauthenticated
-        if r.status != 200:
-            raise ValueError(f"urllib request got {r.status} {r.reason}")
+            raise JsonRPC.Unauthenticated
 
         body = r.read()
 
         if self.debug:
             print("reply:", body)
-        r = json.loads(body)
+        body = json.loads(body)
+
+        if r.status == 507:
+            raise JsonRPC.Overflow(body["error"]["data"]["count"])
+        if r.status != 200:
+            raise ValueError(f"urllib request got {r.status} {r.reason}")
 
-        if "result" not in r:
+        if "result" not in body:
             raise ValueError("jsonrpc error")
 
-        assert (r['id'] == str(self.id))
+        assert (body['id'] == str(self.id))
+
+        return body['result']
+
+    def get(self, method, offset=None, limit=None, params=None):
+        if params is not None and len(params) == 0:
+            # This can happen with the args passed from CLI
+            params = None
+
+        # Populate the params if needed
+        if offset is not None:
+            if params is None:
+                params = dict()
+            params["offset"] = offset
+        if limit is not None:
+            if params is None:
+                params = dict()
+            params["limit"] = limit
+
+        # Try once, possibly without pagination, to see if we overflow. And
+        # if so, to detect a good size for the limit
+        try:
+            return self.get_nopagination(method, params)
+        except JsonRPC.Overflow as e:
+            count_max = e.count
+
+        if params is None:
+            params = dict()
+
+        if not isinstance(params, dict):
+            """Since we overflowed, params must be compatible"""
+            raise ValueError(f"Cannot use params={params} with autopagination")
+
+        params["limit"] = count_max
+        params["offset"] = 0
+
+        result = []
+
+        while True:
+            try:
+                partial = self.get_nopagination(method, params)
+            except JsonRPC.Overflow as e:
+                if count_max == e.count:
+                    # If the limit didnt get smaller, there is a problem with
+                    # the daemon, so we just bail out
+                    raise
+
+                # reduce our asking size and try again
+                count_max = e.count
+                params["limit"] = count_max
+                continue
+
+            if not isinstance(partial, list):
+                # The current API only returns lists, but there could be
+                # dicts in the future, catch this
+                raise NotImplementedError
+
+            result.extend(partial)
+            params["offset"] = len(result)
 
-        return r['result']
+            if len(partial) < params["limit"]:
+                # We fetched less than we asked for, so we must be at the end
+                # of the list
+                return result
 
 
 def str_table(rows, columns, orderby):
@@ -292,7 +364,7 @@ def subcmd_default(rpc, args):
     """Just pass command through to edge"""
     method = args.cmd
     params = args.args
-    rows = rpc.get(method, params)
+    rows = rpc.get(method, params=params)
     return json.dumps(rows, sort_keys=True, indent=4)
 
 
@@ -362,7 +434,7 @@ def main():
 
     try:
         result = func(rpc, args)
-    except Unauthenticated:
+    except JsonRPC.Unauthenticated:
         print("This request requires an authentication key")
         exit(1)
     except FileNotFoundError:

src/edge_utils.c

@@ -316,7 +316,7 @@ static int detect_local_ip_address (n2n_sock_t* out_sock, const struct n3n_runti
 
 // open socket, close it before if TCP
 // in case of TCP, 'connect()' is required
-int supernode_connect (struct n3n_runtime_data *eee) {
+void supernode_connect (struct n3n_runtime_data *eee) {
 
     int sockopt;
     struct sockaddr_in sn_sock;
@@ -338,7 +338,7 @@ int supernode_connect (struct n3n_runtime_data *eee) {
 
         if(eee->sock < 0) {
             traceEvent(TRACE_ERROR, "failed to bind main UDP port");
-            return -1;
+            return;
         }
 
         fill_sockaddr((struct sockaddr*)&sn_sock, sizeof(sn_sock), &eee->curr_sn->sock);
@@ -354,8 +354,9 @@ int supernode_connect (struct n3n_runtime_data *eee) {
 #endif
             if((connect(eee->sock, (struct sockaddr*)&(sn_sock), sizeof(struct sockaddr)) < 0)
                && (errno != EINPROGRESS)) {
+                traceEvent(TRACE_INFO, "Error connecting TCP: %i", errno);
                 eee->sock = -1;
-                return -1;
+                return;
             }
         }
 
@@ -426,7 +427,7 @@ int supernode_connect (struct n3n_runtime_data *eee) {
     // REVISIT: add mgmt port notification to listener for better mgmt port
     //          subscription support
 
-    return 0;
+    return;
 }
 
 
@@ -1026,6 +1027,7 @@ static void check_known_peer_sock_change (struct n3n_runtime_data *eee,
  * Confirm that we can send to this edge.
  * TODO: for the TCP case, this could cause a stall in the packet
  * send path, so this probably should be reworked to use a queue
+ * (and non blocking IO)
  */
 static bool check_sock_ready (struct n3n_runtime_data *eee) {
     if(!eee->conf.connect_tcp) {
@@ -1102,6 +1104,7 @@ static ssize_t sendto_fd (struct n3n_runtime_data *eee, const void *buf,
     if(eee->conf.connect_tcp) {
         supernode_disconnect(eee);
         eee->sn_wait = 1;
+        // Not true if eee->sock == -1
         traceEvent(TRACE_DEBUG, "error in sendto_fd");
     }
 
@@ -2886,6 +2889,9 @@ int fetch_and_eventually_process_data (struct n3n_runtime_data *eee, SOCKET sock
             return -1;
         }
 
+        // TODO: if bread > 64K, something is wrong
+        // but this case should not happen
+
         // we have a datagram to process...
         if(bread > 0) {
             // ...and the datagram has data (not just a header)