Is asyncssh the solution of my problem? #475
-
|
Hello everyone, I created a script where I send 15 000 commands to 70 switchs, so around 200 commands per switch to fetch data from them. My script is taking 30 minutes to run and to fetch all the data I want from the switches. Do you think asyncssh is a good tool for what I'm doing and will the performance of asyncssh be better than what I'm using today? Thank you, |
Beta Was this translation helpful? Give feedback.
Replies: 5 comments 15 replies
-
|
The exact performance will depend on how quickly the switches respond to your commands, but I did a quick test here of opening 75 connections to "localhost", with each connection sequentially starting 200 sessions running "ls" and collecting up all the results. To avoid running into simultaneous authentication limits on localhost, I added a semaphore that limited the code to starting at most 10 connections at once, but you probably wouldn't need that if you were connecting to different target systems. Here's some example code: import asyncio, asyncssh
sem = asyncio.Semaphore(10)
targets = 75*[{'host': 'localhost', 'commands': 200*['ls']}]
async def run_client(host, commands):
async with sem:
conn = await asyncssh.connect(host)
async with conn:
return [await conn.run(command) for command in commands]
async def run_multiple_clients():
tasks = (run_client(target['host'], target['commands']) for target in targets)
results = await asyncio.gather(*tasks, return_exceptions=True)
asyncio.run(run_multiple_clients())Even when running all of this in a single event loop (and thus only being able to take advantage of a single core), my test finished in just a little over 1 minute. Here's a variation that uses multiple threads, with an event loop per thread. Interestingly, it's actually a bit slower than the single event loop version, even though it did take advantage of more than one CPU core. import asyncio, asyncssh
from concurrent.futures import ThreadPoolExecutor
targets = 75*[{'host': 'localhost', 'commands': 200*['echo foo']}]
async def run_client(host, commands):
async with asyncssh.connect(host) as conn:
return [await conn.run(command) for command in commands]
def run_async(target):
return asyncio.run(run_client(target['host'], target['commands']))
if __name__ == '__main__':
with ThreadPoolExecutor(10) as executor:
results = executor.map(run_async, targets)To get real benefit here, it might be necessary to use multiple processes instead, so the Python global interpreter lock doesn't get in the way of concurrency. However, attempting to switch to ProcessPoolExecutor and processing results locally in each process to avoid having to return them to the main process only shaved off 10 seconds or so, and it used a lot more total CPU. By far, the first version above was the most efficient. It's hard to know with your actual target hosts whether AsyncSSH will be faster than your current solution or not, but it's certainly capable of handling this kind of workload without a problem. The code is also pretty compact. |
Beta Was this translation helpful? Give feedback.
-
|
Hello Ron, Thank you so much for your prompt reply. I had a bit of trouble implementing a jump server connection at first, but I finally figured out how to do it. I will let you know as soon as I finish it. Thank you |
Beta Was this translation helpful? Give feedback.
-
|
Hello Ron, I managed to make it work for the first step of my script. I'm iterating over 70 switches and I send only one command per switch. All of this, connecting via a jump server. It took 9 seconds to proceed. Now, I'm trying to make the second step. I want now to iterate over each host to launch another command on the switch it is attached, this to get another information: The port on which the host is attached to the switch. In order to do this, here what I am doing: The problem is that I get result for only 97 hosts over 5000. For the others, I get two kind of errors:
I think I did a mistake in the way I wrote my script. Thank you so much |
Beta Was this translation helpful? Give feedback.
-
|
Regarding jump hosts, doing a connect() to the jump host and then doing connect_ssh() on that tunnel is one open, but in that case you'll probably want to move the "async with sem" to only be used on the inner connect_ssh() connections, as you'll only have one connection to the jump host that all of the other connections are sharing. Also, I don't know if the Python asyncio semaphore is designed to be recursive -- acquiring it on the inner calls while you're already holding it at the outer level could be a problem. There's a more fundamental issue here, though, which is that you are waiting for the calls to connect_ssh() to complete in your for loop over the leaf_ips, and you probably don't want to do that. You want to allow those connections to go in parallel, but with a limit enforced by the semaphore to avoid too many connections being opened at once. Note how in my example the "async with sem" is only around the connect() call, and the semaphore is released before the "async with conn" which uses the connection to send commands. Once they are opened, you can then open sessions for your commands. You may not want to do those in parallel, though -- that'll depend whether the switches you are contacting can handle parallel commands or not. In my example, I did the initial connects in parallel (while holding the semaphore), but then sent the commands to each target serially, to avoid overloading the target with too many sessions at once. You could possible do this in parallel with another set of semaphore, but I'd suggest starting with sending the commands serially and getting that to work before you attempt creating a task per command like you're doing now. Instead, just have a single task which does the connect_ssh() to a leaf and then does a for loop to serially send commands to that switch. In your current code, it looks like you are creating a task per command with no limit on the number of those you open in parallel, so that may explain some of the errors you are seeing. Between that and reusing the semaphore, I'm not really sure that's going to do any limiting at all. Regarding getting the command results, the SSHCompletedProcess returned by run() has a "stdout" member with the command output. It also has an exit_status, but my guess is that switches probably won't set that. So, you'll have to probably need to parse the output to determine the success of the command and extract any result you want from it. As for the number of connections you can make in parallel, that depends on the server you are connecting to. There is usually one limit on number of simultaneous authentications allowed (MaxStartups in OpenSSH), and there may be some limit on the total number of connections, but that's usually much higher. There is also generally a limit on the number of sessions any single connection can support (MaxSessions in OpenSSH), so you may run into that limit on the jump host. If you hit that, you may need to make multiple independent connections to the jump host to stay under that limit. By default, you need to stay under 10 connections in authentication at once to avoid OpenSSH returning failures, and stay under 10 simultaneous sessions. The switches you are talking to may limit this even more, so you'll simply have to experiment (or check the switch documentation). |
Beta Was this translation helpful? Give feedback.
-
|
Hello Ron, Thank you for your reply. Yes you completely right, actually I'm working switch per switch whereas I should send commands at the same time to several switches in parallel. The issue I have is that each command is different because each command has the mac of the host like that: You said: You mean I should create one task for each switch? Then I do a for loop to add inside the different commands? Do you mean something like that? Thank you, |
Beta Was this translation helpful? Give feedback.
-
|
That's getting closer, but you'll want to move the connect_ssh() call and the "async with sem" around it to be the first lines in get_port_one(). That way, that code runs as part of the tasks you are creating and running in parallel, instead of the connects running serially in the for loop in get_port_all() as the tasks are being created. Once you've done that, I think the main remaining issue will be that you may hit the maximum number of allowed sessions on the jump host. To fix that, you'd need to call asyncssh.connect() multiple times, limiting the number of calls to connect_ssh() made on any given tunnel. You can still reuse a tunnel object for multiple connect_ssh() calls, but just not ALL of the calls if you have a lot of switches. Moving the asyncssh.connect() inside the for loop and keeping a counter for the number of sessions you've opened on it would probably be the easiest thing. When the count hits 10, do a new call to asyncssh.connect(), replacing the previous tunnel object. The previous sessions you created should be enough to keep the previous tunnel connection from being garbage-collected even though you don't have an explicit reference to it. You'd be relying on garbage collection to clean it up later on when you are done, but since your program exits at that point, that's not a big concern. You could also keep an explicit list of all the tunnel objects you open in this way, and call close() and await wait_closed() on them at the very end, if you wanted to cleanly free them. |
Beta Was this translation helpful? Give feedback.
-
|
Hello Ron and thank you for your time. So I've done all what you said and it all makes sense but I still receive errors in results: Here is my code: Maybe is it the fact that each task I'm appending to the tasks list is not just one task but multiple conn.run tasks? Thank you very much, |
Beta Was this translation helpful? Give feedback.
-
|
This definitely looks better. Having a single task for each switch which runs the commands sequentially is a good first step. You could try and parallelize it further if the switch can handle multiple parallel sessions on a single connection, but this I'd try to get this version to work before you attempt parallelizing the conn.run() calls. The semaphore looks good here now, but you may be running into limits on the ability for the jump host to handle more than 10 sessions on a single connection. Right now, the semaphore just protects against MaxStartups on a given switch but not MaxSessions on the jump host. To deal with that, you may need to open multiple jump host connections, to limit each connection to at most 10 sessions. Either that, or most the "for" loop in get_port_one() inside the "async with sem" block so that you only allow at most 10 sessions total to be running at once, delaying the remaining sessions until some earlier session completes running all of its commands and closes the jump host session. |
Beta Was this translation helpful? Give feedback.
-
|
Hello Ron, I am sorry for the delay in my answer. I was in vacation for a few weeks. I really would like to thank you for your time and your patience with me. David |
Beta Was this translation helpful? Give feedback.
-
|
That's terrific, David -- glad to hear you got it working, and got such an improvement! I'm happy to help - thanks for letting me know how you made out... |
Beta Was this translation helpful? Give feedback.
-
|
Hello @ronf, I hope you are doing well. I'm trying to use again asyncSSH to send several commands on the same equipment this time, but i'm having some trouble to do it. I want to send the command When I'm running this script, it seems like an infinite loop and nothing happens. Thank you, |
Beta Was this translation helpful? Give feedback.
-
|
Adding the semaphore may speed things up, as asking the target system to handle 15,000 simultaneous sessions may be contributing to the slowness. Running fewer of those commands in parallel may allow them to complete more quickly overall, even if that means some of the commands will be a bit delayed in starting up. If there are algorithms in the server that do things like a linear search to map channel numbers to channels, running with the semaphore should take less time to do the search. Do you really need to be running this IP reverse lookup on the target devices? Generally speaking, that's just going to trigger the target to send queries to its DNS servers. If you know what those DNS servers are, perhaps you could do the reverse lookup from your client machine without getting the target machine involved. You wouldn't even have to use "dig" here -- you could use something like Python's Also, one of the things which may be contributing to the slowness is that some DNS servers don't respond to DNS reverse lookup requests, causing the lookup to eventually time out. Are you seeing any examples of that, where the "dig" command returns no output for some of the lookups? Unfortunately, unless you own the DNS servers for all of the IPs you are looking up, this may not totally be under your control. |
Beta Was this translation helpful? Give feedback.
-
|
Hello @ronf, Thank you for your reply.
And it doesn't display the total time. Here is my final script: Thank you, |
Beta Was this translation helpful? Give feedback.
-
|
I'm a bit surprised this would work as-is, given that os.system() is not an async function. I'd expect it to fail when you did the "await task", if you waited long enough for it to finish appending all the tasks. That first part is going to be very slow, though, as it'll be doing all the os.system() calls in a blocking manner, one at a time. You can also make it faster by not launching a separate process for each lookup, using the asyncio version of the getnameinfo() function to do reverse lookups. Here's an example: import asyncio
import os
from time import time
start = time()
ips = [] # we assume the ips list is full of IP addresses
async def lookup_host(loop, ip):
return (await loop.getnameinfo((ip, 0), 0))[0]
sem = asyncio.Semaphore(20)
async def hostname(ips):
loop = asyncio.get_event_loop()
tasks = []
for ip in ips:
tasks.append(lookup_host(loop, ip))
for task in asyncio.as_completed(tasks):
async with sem:
result = await task
print(result)
asyncio.run(hostname(ips))
print(f'Temps total: {time() - start}')On my machine here, 15,000 lookups of names owned by a local resolver complete in a little under 2 seconds, but there might be some caching involved in that. So, with a larger number of unique IPs, it might take a bit longer. It should be much faster than using AsycnSSH or os.system(), though, even if you were to fix the os.system() calls to happen in parallel. |
Beta Was this translation helpful? Give feedback.
-
|
Hi Ron, It's very weird, the script is working but taking very long time at the beginning before displaying all the results in one second. Thanks |
Beta Was this translation helpful? Give feedback.
-
|
In cases where no reverse lookup exists for a given IP address, this script will return back the original IP, unlike "dig" which ends up returning an empty string. If the script stops generating output, it may be because it is waiting for a DNS response. It's common for some DNS servers to not respond to reverse lookups, leading to clients waiting a long time and then eventually timing out. That could be part of why you were seeing slowness in the first place. Looking at the positioning of the semaphore, I also think that's in the wrong place now that as_completed() is being used. It should probably be more like: import asyncio
import os
from time import time
start = time()
ips = [] # we assume the ips list is full of IP addresses
sem = asyncio.Semaphore(20)
async def lookup_host(loop, ip):
async with sem:
return (await loop.getnameinfo((ip, 0), 0))[0]
async def hostname(ips):
loop = asyncio.get_event_loop()
tasks = []
for ip in ips:
tasks.append(lookup_host(loop, ip))
for task in asyncio.as_completed(tasks):
result = await task
print(result)
asyncio.run(hostname(ips))
print(f'Temps total: {time() - start}')However, this won't necessarily help if some of the DNS servers you are going to take a long time to respond (or never respond at all). Are all of the addresses you are looking up "internal" addresses, with their reverse lookup owned by local DNS servers? |
Beta Was this translation helpful? Give feedback.
That's terrific, David -- glad to hear you got it working, and got such an improvement!
I'm happy to help - thanks for letting me know how you made out...