diff --git a/.github/kind.yaml b/.github/kind.yaml index 7576197..624e326 100644 --- a/.github/kind.yaml +++ b/.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 diff --git a/openserverless/seaweedfs_deploy.py b/openserverless/seaweedfs_deploy.py index 28dfc02..765b015 100644 --- a/openserverless/seaweedfs_deploy.py +++ b/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: diff --git a/openserverless/seaweedfs_util.py b/openserverless/seaweedfs_util.py index df91413..ede373c 100644 --- a/openserverless/seaweedfs_util.py +++ b/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"): diff --git a/openserverless/util.py b/openserverless/util.py index 356dd63..ebfdd70 100644 --- a/openserverless/util.py +++ b/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 : and return diff --git a/tests/kind/milvus_standalone_slim_test.ipy b/tests/kind/milvus_standalone_slim_test.ipy index cf0c297..4633b27 100644 --- a/tests/kind/milvus_standalone_slim_test.ipy +++ b/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()) diff --git a/tests/kind/milvus_standalone_test.ipy b/tests/kind/milvus_standalone_test.ipy index 2931c0f..47aa5ff 100644 --- a/tests/kind/milvus_standalone_test.ipy +++ b/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()) diff --git a/tests/kind/seaweedfs_test.ipy b/tests/kind/seaweedfs_test.ipy index 55f0c28..57b961e 100644 --- a/tests/kind/seaweedfs_test.ipy +++ b/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()) \ No newline at end of file +assert(seaweedfs_deploy.delete()) diff --git a/tests/kind/seaweedfs_util_test.ipy b/tests/kind/seaweedfs_util_test.ipy index b56dd7a..c118034 100644 --- a/tests/kind/seaweedfs_util_test.ipy +++ b/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()) \ No newline at end of file +assert(seaweedfs.delete()) diff --git a/tests/kind/storage_static_test.ipy b/tests/kind/storage_static_test.ipy index 7737da2..e2356d6 100644 --- a/tests/kind/storage_static_test.ipy +++ b/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()) \ No newline at end of file +assert(seaweedfs.delete()) diff --git a/tests/test_datastore_security.py b/tests/test_datastore_security.py index 4688432..282520b 100644 --- a/tests/test_datastore_security.py +++ b/tests/test_datastore_security.py @@ -16,6 +16,7 @@ # under the License. import base64 +import contextlib import json import logging from pathlib import Path @@ -23,7 +24,7 @@ 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