From a619b6ed6f4e4c43bd7de1c9e2d4cf5c27f97205 Mon Sep 17 00:00:00 2001
From: Markus Hilger <markus.hilger@megware.com>
Date: Mon, 27 Jul 2026 06:15:32 +0200
Subject: [PATCH] Bound the sessions the nodediscover CSV import opens

Replacing the forked children with a gather kept their fan-out: every
row of the import file gets a session of its own and they all start at
once, so a large file opens a local socket and a server side session
task per node simultaneously.

Hold a semaphore for the duration of each node's assignment instead, so
a finished node's session is dropped before the next one starts.  Also
build that session once per node rather than once per MAC, and say why
the caller's session is not reused, which was self evident while this
ran in a forked child.
---
 confluent_client/bin/nodediscover | 33 +++++++++++++++++++------------
 1 file changed, 20 insertions(+), 13 deletions(-)

diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover
index ea93f298..533a983e 100755
--- a/confluent_client/bin/nodediscover
+++ b/confluent_client/bin/nodediscover
@@ -180,6 +180,9 @@ def datum_to_attrib(datum):
 
 unique_fields = frozenset(['serial', 'mac', 'uuid'])
 
+# Cap how many nodes hold a discovery session at once while importing
+maxconcurrentassign = 128
+
 async def import_csv(options, session):
     nodedata = []
     unique_data = {}
@@ -225,6 +228,7 @@ async def import_csv(options, session):
     if broken:
         sys.exit(1)
     assignments = []
+    assignlimit = asyncio.Semaphore(maxconcurrentassign)
     for datum in nodedata:
         maclist = await search_record(datum, options, session)
         datum = datum_to_attrib(datum)
@@ -238,27 +242,30 @@ async def import_csv(options, session):
                 print('Defined ' + res['created'])
             else:
                 print(repr(res))
-        assignments.append(assign_macs(maclist, nodename))
+        assignments.append(assign_macs(maclist, nodename, assignlimit))
     for rcode in await asyncio.gather(*assignments):
         exitcode |= rcode
     if exitcode:
         sys.exit(exitcode)
 
 
-async def assign_macs(maclist, nodename):
+async def assign_macs(maclist, nodename, assignlimit):
     exitcode = 0
-    for mac in maclist:
+    async with assignlimit:
+        # A session of our own, since the connection carries one request at a
+        # time and the caller's is busy defining the remaining nodes
         mysess = client.Command()
-        async for res in mysess.update('/discovery/by-mac/{0}'.format(mac),
-                                       {'node': nodename}):
-            if 'error' in res:
-                sys.stderr.write(res['error'] + '\n')
-                exitcode |= res.get('errorcode', 1)
-                continue
-            elif 'assigned' in res:
-                print('Discovered ' + res['assigned'])
-            else:
-                print(repr(res))
+        for mac in maclist:
+            async for res in mysess.update('/discovery/by-mac/{0}'.format(mac),
+                                           {'node': nodename}):
+                if 'error' in res:
+                    sys.stderr.write(res['error'] + '\n')
+                    exitcode |= res.get('errorcode', 1)
+                    continue
+                elif 'assigned' in res:
+                    print('Discovered ' + res['assigned'])
+                else:
+                    print(repr(res))
     return exitcode