Fix SeaweedFS integration access after NodePort removal

.github/kind.yaml

@@ -69,9 +69,6 @@ nodes:
   - containerPort: 32090
     hostPort: 9000
     protocol: TCP 
-  - containerPort: 32091
-    hostPort: 9090
-    protocol: TCP
   - containerPort: 32530
     hostPort: 19530
     protocol: TCP 

openserverless/seaweedfs_deploy.py

@@ -25,6 +25,7 @@
 import openserverless.operator_util as operator_util
 import openserverless.datastore_security as datastore_security
 
+from openserverless.datastore_connection import service_endpoint
 from openserverless.user_config import UserConfig
 from openserverless.user_metadata import UserMetadata
 from openserverless.seaweedfs_util import SeaweedfsClient
@@ -68,6 +69,37 @@ def _add_seaweedfs_user_metadata(ucfg: UserConfig, user_metadata:UserMetadata):
         logging.error(f"failed to build SEAWEEDFS metadata for {ucfg.get('namespace')}: {e}")
         return None 
 
+def _bootstrap_private_filer(data, timeout=30):
+    """Wait for the private filer, recreating a tunnel lost during startup."""
+    deadline = time.monotonic() + timeout
+    last_error = None
+
+    while time.monotonic() < deadline:
+        ready = False
+        try:
+            with service_endpoint(data['name'], 9090) as (host, port):
+                filer_url = f"http://{host}:{port}"
+                try:
+                    # A port-forward can exit after its first connection if the
+                    # target socket is not listening yet. Keep each probe short
+                    # so the next attempt can establish a fresh tunnel.
+                    util.wait_for_http(filer_url, up_statuses=[200, 401], timeout=3)
+                except TimeoutError as error:
+                    last_error = error
+                else:
+                    ready = True
+                    create_seaweedfs_nuv_storage(data, filer_url)
+                    return
+        except RuntimeError as error:
+            if ready:
+                raise
+            last_error = error
+
+        if time.monotonic() < deadline:
+            time.sleep(1)
+
+    raise TimeoutError("SeaweedFS filer did not become ready before timeout") from last_error
+
 def create(owner=None):
     logging.info("*** configuring seaweedfs standalone")
 
@@ -90,10 +122,10 @@ def create(owner=None):
     util.wait_for_pod_ready("{.items[?(@.metadata.labels.app == 'seaweedfs')].metadata.name}")
     seaweedfs_ingress.create_seaweedfs_ingresses(data, owner)
 
-    logging.info("*** waiting for seaweedfs filer api to be available")    
-    util.wait_for_http(util.get_seaweedds_filer_host(), up_statuses=[200,401], timeout=30)
-
-    create_seaweedfs_nuv_storage(data)  
+    # The filer API is private. In-cluster this resolves to the Service;
+    # integration runs use an authenticated, loopback-only port-forward.
+    logging.info("*** waiting for seaweedfs filer api to be available")
+    _bootstrap_private_filer(data)
 
     logging.info("*** configured seaweedfs standalone")
     return res
@@ -172,12 +204,12 @@ def reconcile_anonymous_web_access(client):
     if web_buckets & private_buckets:
         raise ValueError("Public and private S3 bucket names overlap")
 
-def create_seaweedfs_nuv_storage(data):
+def create_seaweedfs_nuv_storage(data, filer_url=None):
     """
     Creates openserverless SEAWEEDFS custom resources
     """
     logging.info("*** configuring SEAWEEDFS storage for openserverless")
-    seaweedsfsClient = SeaweedfsClient()
+    seaweedsfsClient = SeaweedfsClient(filer_url)
 
     res = seaweedsfsClient.make_bucket("openserverless-data",data["default_bucket_quota"])
     if res:

openserverless/seaweedfs_util.py

@@ -46,8 +46,8 @@ def __init__(self):
 
 class SeaweedfsClient:
 
-    def __init__(self):
-        self.filer_url = util.get_seaweedds_filer_host()
+    def __init__(self, filer_url=None):
+        self.filer_url = filer_url or util.get_seaweedds_filer_host()
         self.pod_name = util.get_pod_name_by_selector("app=seaweedfs","{.items[?(@.metadata.labels.app == 'seaweedfs')].metadata.name}")
 
     def _request(self, endpoint: str, json=None, method: str = "POST"):

openserverless/util.py

@@ -302,6 +302,8 @@ def wait_for_http(url: str, timeout: int = 60, up_statuses: List[Union[int, str]
                 logging.error(f"Protocol error: {e}")
         time.sleep(1)
 
+    raise TimeoutError("Service did not become ready before timeout")
+
 def parse_image(img):
     """
     Parse a string representing a pod image in the form <image>:<tag> and return

tests/kind/milvus_standalone_slim_test.ipy

@@ -49,10 +49,6 @@ is_seaweedfs = cfg.get('components.seaweedfs') or False
 assert(is_seaweedfs)
 
 
-# for this test seaweedfs and Milvus should see this env variable
-os.environ['SEAWEEDFS_API_HOST']='localhost'
-os.environ['SEAWEEDFS_API_PORT']='9090'
-
 os.environ['MILVUS_API_HOST']='localhost'
 
 assert(etcd.create())

tests/kind/milvus_standalone_test.ipy

@@ -49,9 +49,6 @@ is_seaweedfs = cfg.get('components.seaweedfs') or False
 assert(is_seaweedfs)
 
 
-# for this test seaweedfs and Milvus should see this env variable
-os.environ['SEAWEEDFS_API_HOST']='localhost'
-os.environ['SEAWEEDFS_API_PORT']='9090'
 os.environ['MILVUS_API_HOST']='localhost'
 
 assert(etcd.create())

tests/kind/seaweedfs_test.ipy

@@ -15,8 +15,6 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-import os
-
 import openserverless.config as cfg
 import openserverless.kube as kube
 import openserverless.seaweedfs_deploy as seaweedfs_deploy
@@ -29,18 +27,13 @@ logging.basicConfig(level=logging.DEBUG)
 tu.run_proc("kubectl -n openserverless delete all --all")
 tu.run_proc("kubectl -n openserverless delete pvc --all")
 
-# for this test seaweedfsClient should see this env variable
-os.environ['SEAWEEDFS_API_HOST']='localhost'
-os.environ['SEAWEEDFS_API_PORT']='9090'
-
 # test
 assert(cfg.configure(tu.load_sample_config()))
 assert(cfg.detect_storage()["openserverless.storageclass"])
 
-# for this test the seaweedfs filer should see this env variable
 assert(seaweedfs_deploy.create())
 
 pod_name = util.get_pod_name("{.items[?(@.metadata.labels.app == 'seaweedfs')].metadata.name}")
 assert(pod_name)
 
-assert(seaweedfs_deploy.delete())
+assert(seaweedfs_deploy.delete())

tests/kind/seaweedfs_util_test.ipy

@@ -15,13 +15,12 @@
 # specific language governing permissions and limitations
 # under the License.
 #
-import os
-
 import openserverless.config as cfg
 import openserverless.seaweedfs_deploy as seaweedfs
 import openserverless.seaweedfs_util as sus
 import openserverless.testutil as tu
 import openserverless.util as util
+from openserverless.datastore_connection import service_endpoint
 
 tu.enable_debug_logging()
 
@@ -32,18 +31,16 @@ tu.run_proc("kubectl -n openserverless delete pvc --all")
 assert(cfg.configure(tu.load_sample_config()))
 assert(cfg.detect_storage()["openserverless.storageclass"])
 
-# for this test the seaweedfsClient should see these env variables
-os.environ['SEAWEEDFS_API_HOST']='localhost'
-os.environ['SEAWEEDFS_API_PORT']='9090'
 assert(seaweedfs.create())
 
 pod_name = util.get_pod_name("{.items[?(@.metadata.labels.app == 'seaweedfs')].metadata.name}")
 assert(pod_name)
 
-seaweedfsClient = sus.SeaweedfsClient()
-assert(seaweedfsClient.make_bucket("ftt-data"))
-assert(seaweedfsClient.make_public_bucket("ftt-web"))
-assert(seaweedfsClient.add_user("ftt","jgfkjsgcasgfjgdsafgsdkfgkaj","jgfkjsgcasgfjgdsafgsdkfgkaj","ftt-web,ftt-data"))
-assert(seaweedfsClient.delete_user("ftt"))
+with service_endpoint('seaweedfs', 9090) as (host, port):
+    seaweedfsClient = sus.SeaweedfsClient(f'http://{host}:{port}')
+    assert(seaweedfsClient.make_bucket("ftt-data"))
+    assert(seaweedfsClient.make_public_bucket("ftt-web"))
+    assert(seaweedfsClient.add_user("ftt","jgfkjsgcasgfjgdsafgsdkfgkaj","jgfkjsgcasgfjgdsafgsdkfgkaj","ftt-web,ftt-data"))
+    assert(seaweedfsClient.delete_user("ftt"))
 
-assert(seaweedfs.delete())
+assert(seaweedfs.delete())

tests/kind/storage_static_test.ipy

@@ -16,8 +16,6 @@
 # under the License.
 #
 
-import os
-
 import openserverless.config as cfg
 import openserverless.seaweedfs_deploy as seaweedfs
 import openserverless.storage_static as static
@@ -31,12 +29,8 @@ tu.run_proc("kubectl -n openserverless delete pvc --all")
 assert(cfg.configure(tu.load_sample_config()))
 assert(cfg.detect_storage()["openserverless.storageclass"])
 
-# for this test the seaweedfs filer should see these env variables
-os.environ['SEAWEEDFS_API_HOST']='localhost'
-os.environ['SEAWEEDFS_API_PORT']='9090'
-
 assert(seaweedfs.create())
 assert(static.create())
 
 assert(static.delete())
-assert(seaweedfs.delete())
+assert(seaweedfs.delete())

tests/test_datastore_security.py

@@ -16,14 +16,15 @@
 # under the License.
 
 import base64
+import contextlib
 import json
 import logging
 from pathlib import Path
 import subprocess
 import unittest
 from unittest.mock import patch
 
-from openserverless import config, datastore_security as security
+from openserverless import config, datastore_security as security, util
 
 
 class SecurityTests(unittest.TestCase):
@@ -134,20 +135,57 @@ def test_seaweedfs_action_cannot_impersonate_trusted_peers(self):
     def test_seaweedfs_create_applies_isolation_policy(self):
         from openserverless import seaweedfs_deploy
         spec = {'apiVersion': 'v1', 'kind': 'List', 'items': []}
-        with patch.object(seaweedfs_deploy.util, 'get_seaweedfs_config_data', return_value={
-                'affinity': False, 'tolerations': False}), \
+        data = {'name': 'seaweedfs', 'affinity': False, 'tolerations': False}
+        with patch.object(seaweedfs_deploy.util, 'get_seaweedfs_config_data', return_value=data), \
                 patch.object(seaweedfs_deploy.kus, 'patchTemplates'), \
                 patch.object(seaweedfs_deploy.kus, 'kustom_list', return_value=spec), \
                 patch.object(seaweedfs_deploy.cfg, 'put'), \
                 patch.object(seaweedfs_deploy.kube, 'apply') as apply, \
                 patch.object(seaweedfs_deploy.util, 'wait_for_pod_ready'), \
                 patch.object(seaweedfs_deploy.seaweedfs_ingress, 'create_seaweedfs_ingresses'), \
-                patch.object(seaweedfs_deploy.util, 'wait_for_http'), \
-                patch.object(seaweedfs_deploy, 'create_seaweedfs_nuv_storage'):
+                patch.object(seaweedfs_deploy, 'service_endpoint', return_value=
+                             contextlib.nullcontext(('127.0.0.1', 41237))) as endpoint, \
+                patch.object(seaweedfs_deploy.util, 'wait_for_http') as wait, \
+                patch.object(seaweedfs_deploy, 'create_seaweedfs_nuv_storage') as bootstrap:
             seaweedfs_deploy.create()
         self.assertIn(security.policy('seaweedfs'), spec['items'])
         self.assertEqual(apply.call_args_list[0].args[0]['items'],
                          [security.policy('seaweedfs')])
+        endpoint.assert_called_once_with('seaweedfs', 9090)
+        wait.assert_called_once_with('http://127.0.0.1:41237', up_statuses=[200, 401], timeout=3)
+        bootstrap.assert_called_once_with(data, 'http://127.0.0.1:41237')
+
+    def test_seaweedfs_bootstrap_reopens_tunnel_after_startup_race(self):
+        from openserverless import seaweedfs_deploy
+        data = {'name': 'seaweedfs'}
+        tunnels = [contextlib.nullcontext(('127.0.0.1', 41237)),
+                   contextlib.nullcontext(('127.0.0.1', 41238))]
+        with patch.object(seaweedfs_deploy, 'service_endpoint', side_effect=tunnels) as endpoint, \
+                patch.object(seaweedfs_deploy.util, 'wait_for_http',
+                             side_effect=[TimeoutError('startup race'), None]) as wait, \
+                patch.object(seaweedfs_deploy, 'create_seaweedfs_nuv_storage') as bootstrap, \
+                patch.object(seaweedfs_deploy.time, 'sleep'):
+            seaweedfs_deploy._bootstrap_private_filer(data)
+        self.assertEqual(endpoint.call_count, 2)
+        self.assertEqual(wait.call_count, 2)
+        bootstrap.assert_called_once_with(data, 'http://127.0.0.1:41238')
+
+    def test_wait_for_http_raises_after_deadline(self):
+        with self.assertRaisesRegex(TimeoutError, 'did not become ready'):
+            util.wait_for_http('http://127.0.0.1:1', timeout=0)
+
+    def test_wait_for_http_keeps_accepted_status_behavior(self):
+        with patch.object(util.urllib3.connectionpool, 'HTTPConnectionPool') as pool:
+            pool.return_value.request.return_value.status = 401
+            util.wait_for_http('http://seaweedfs:9090', timeout=1, up_statuses=[200, 401])
+        pool.return_value.request.assert_called_once_with('GET', '/')
+
+    def test_seaweedfs_client_uses_explicit_private_endpoint(self):
+        from openserverless import seaweedfs_util
+        with patch.object(seaweedfs_util.util, 'get_pod_name_by_selector',
+                          return_value='seaweedfs-0'):
+            client = seaweedfs_util.SeaweedfsClient('http://127.0.0.1:41237')
+        self.assertEqual(client.filer_url, 'http://127.0.0.1:41237')
 
     def test_existing_cluster_reconcile_secures_before_service_update(self):
         from openserverless import seaweedfs_deploy