[8.19] [Network Drive] Switch to usage of local connection cache (#3868) (#3871)

# Backport

This will backport the following commits from `main` to `8.19`:
- [[Network Drive] Switch to usage of local connection cache
(#3868)](#3868)

<!--- Backport version: 9.6.6 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sorenlouv/backport)

---------

Co-authored-by: Elastic Machine <[email protected]>

Author: artem-shelkovnikov

connectors/sources/network_drive.py

@@ -261,13 +261,14 @@ def fetch_members(self, group_name):
 class SMBSession:
     _connection = None
 
-    def __init__(self, server_ip, username, password, port):
+    def __init__(self, server_ip, username, password, port, connection_cache):
         self.server_ip = server_ip
         self.username = username
         self.password = password
         self.port = port
         self.session = None
         self._logger = logger
+        self._connection_cache = connection_cache
 
     def create_connection(self):
         """Creates an SMB session to the shared drive."""
@@ -277,6 +278,7 @@ def create_connection(self):
                 username=self.username,
                 password=self.password,
                 port=self.port,
+                connection_cache=self._connection_cache,
             )
         except SMBResponseException as exception:
             self.handle_smb_response_errors(exception=exception)
@@ -336,13 +338,20 @@ def __init__(self, configuration):
         self.drive_type = self.configuration["drive_type"]
         self.identity_mappings = self.configuration["identity_mappings"]
         self.security_info = SecurityInfo(self.username, self.password, self.server_ip)
+        self._connection_cache = {}
 
     def advanced_rules_validators(self):
         return [NetworkDriveAdvancedRulesValidator(self)]
 
     @cached_property
     def smb_connection(self):
-        return SMBSession(self.server_ip, self.username, self.password, self.port)
+        return SMBSession(
+            self.server_ip,
+            self.username,
+            self.password,
+            self.port,
+            self._connection_cache,
+        )
 
     @classmethod
     def get_default_configuration(cls):
@@ -464,6 +473,7 @@ async def traverse_diretory(self, path):
                             username=self.username,
                             password=self.password,
                             port=self.port,
+                            connection_cache=self._connection_cache,
                         ),
                     )
                 )
@@ -518,6 +528,9 @@ async def traverse_directory_for_syncrule(self, path, glob_pattern, indexed_rule
                         smbclient.scandir,
                         path=current_path,
                         port=self.port,
+                        username=self.username,
+                        password=self.password,
+                        connection_cache=self._connection_cache,
                     ),
                 )
                 for file in directory_info:
@@ -598,9 +611,9 @@ async def validate_config(self):
 
     async def ping(self):
         """Verify the connection with Network Drive"""
+        if self.smb_connection.session is None:
+            await asyncio.to_thread(self.smb_connection.create_connection)
 
-        await asyncio.to_thread(self.smb_connection.create_connection)
-        await self.close()
         self._logger.info("Successfully connected to the Network Drive")
 
     async def close(self):
@@ -611,7 +624,10 @@ async def close(self):
         await loop.run_in_executor(
             executor=None,
             func=partial(
-                smbclient.delete_session, server=self.server_ip, port=self.port
+                smbclient.delete_session,
+                server=self.server_ip,
+                port=self.port,
+                connection_cache=self._connection_cache,
             ),
         )
 
@@ -876,7 +892,9 @@ async def get_docs(self, filtering=None):
         Yields:
             dictionary: Dictionary containing the Network Drive files and folders as documents
         """
-        await asyncio.to_thread(self.smb_connection.create_connection)
+        if self.smb_connection.session is None:
+            await asyncio.to_thread(self.smb_connection.create_connection)
+
         if filtering and filtering.has_advanced_rules():
             advanced_rules = filtering.get_advanced_rules()
             async for document in self.fetch_filtered_directory(advanced_rules):