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

## Closes #2418

Conceptual problem - `smbprotocol` has a global connection cache
[here](https://github.com/jborean93/smbprotocol/blob/master/src/smbclient/_pool.py#L33)
(see usages). Hash key is host/port:

```python
connection_key = f"{server.lower()}:{port}"
```

Library is not natively async, that means that it can cause fun stuff
when used in asynchronous/threaded environment. In our case one
connector configuration can spawn at least 2 connections for the same
server: 1 for running `get_docs` method while the sync is running and 1
for `ping` method that gets spawned every 30 seconds by default.

Things get worse if advanced filtering rules are used and several
connectors point to the same Network Drive.

What happens is:

1. `get_docs` is called and a connection is established
2. `ping` is triggered, connection is re-used from the cache
3. `ping` finishes and closes the connection. That causes the loss of
the connection by `get_docs`
4. `get_docs` sometimes can re-establish it (see [this
code](https://github.com/elastic/connectors/blob/main/app/connectors_service/connectors/sources/network_drive/datasource.py#L276)).
Sometimes it just fails

This change attempts to address it by providing connection cache that's
local to each connector instead of using a global one. This way
connector instances would not attempt to close each other's connections
if they connect to the same resource.

#### Pre-Review Checklist
- [x] this PR does NOT contain credentials of any kind, such as API keys
or username/passwords (double check `config.yml.example`)
- [x] this PR has a meaningful title
- [x] this PR links to all relevant github issues that it fixes or
partially addresses
- [x] if there is no GH issue, please create it. Each PR should have a
link to an issue
- [x] this PR has a thorough description
- [ ] Covered the changes with automated tests
- [x] Tested the changes locally
- [x] Added a label for each target release version (example: `v7.13.2`,
`v7.14.0`, `v8.0.0`)
- [ ] For bugfixes: backport safely to all minor branches still
receiving patch releases
- [ ] Considered corresponding documentation changes
- [ ] Contributed any configuration settings changes to the
configuration reference
- [ ] if you added or changed Rich Configurable Fields for a Native
Connector, you made a corresponding PR in
[Kibana](https://github.com/elastic/kibana/blob/main/packages/kbn-search-connectors/types/native_connectors.ts)

## Release Note

Fix the bug with Network Drive connector closing connections to SMB
servers prematurely.

---------

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

Author: artem-shelkovnikov

app/connectors_service/connectors/sources/network_drive/datasource.py

@@ -75,13 +75,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):
@@ -203,6 +210,7 @@ async def traverse_diretory(self, path):
                             username=self.username,
                             password=self.password,
                             port=self.port,
+                            connection_cache=self._connection_cache,
                         ),
                     )
                 )
@@ -257,6 +265,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:
@@ -337,9 +348,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):
@@ -350,7 +361,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,
             ),
         )
 
@@ -615,7 +629,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):

app/connectors_service/connectors/sources/network_drive/netdrive.py

@@ -222,12 +222,13 @@ 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._connection_cache = connection_cache
         self._logger = logger
 
     def create_connection(self):
@@ -238,6 +239,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)