diff --git a/deploy/seaweedfs/seaweedfs-sts.yaml b/deploy/seaweedfs/seaweedfs-sts.yaml index 8fed6e9..d09b6cb 100644 --- a/deploy/seaweedfs/seaweedfs-sts.yaml +++ b/deploy/seaweedfs/seaweedfs-sts.yaml @@ -34,6 +34,20 @@ spec: containers: - name: seaweedfs image: chrislusf/seaweedfs:3.97 + startupProbe: + httpGet: + path: / + port: 9090 + periodSeconds: 2 + timeoutSeconds: 2 + failureThreshold: 90 + readinessProbe: + httpGet: + path: / + port: 9090 + periodSeconds: 5 + timeoutSeconds: 2 + failureThreshold: 3 ports: - containerPort: 9000 # S3 - containerPort: 9090 # Filer diff --git a/docs/seaweedfs-filer-readiness.md b/docs/seaweedfs-filer-readiness.md new file mode 100644 index 0000000..5b5bac3 --- /dev/null +++ b/docs/seaweedfs-filer-readiness.md @@ -0,0 +1,76 @@ + + +# SeaweedFS Filer readiness on 0.9.0 + +An existing installation can report its SeaweedFS pod Ready before the private +Filer listener starts. Resume-time IAM reconciliation must wait for the actual +private HTTP endpoint. See [OpenServerless #286](https://github.com/apache/openserverless/issues/286). + +Bootstrap and resume share the private endpoint readiness helper. Probe timeouts, +including urllib3 connection/read timeouts, are retried within the readiness +window. A failed startup port-forward is recreated. Once readiness succeeds, +bootstrap and IAM errors propagate; an incorrect persisted grant is never +accepted or retried as readiness. + +New SeaweedFS StatefulSets receive HTTP startup and readiness probes on the +private Filer port 9090. Resume fills missing probes on existing StatefulSets +using a strategic merge patch to the named container. Existing custom probes, +image, arguments, environment, storage and sidecars are preserved. The resulting +rollout is awaited before IAM operations and ingress publication. + +Installing missing probes causes a rolling restart of the single SeaweedFS pod +and a temporary storage interruption. Repeating reconciliation without missing +probes does not patch or restart the pod. An existing OnDelete workload requiring +new probes is rejected with an explicit operator-managed rollout requirement. +The operator does not delete its pod or change its rollout policy automatically. + +The Filer root probe establishes HTTP listener readiness. It does not prove S3 +authorization, backend durability, application acceptance or high availability. +Persisted IAM readback remains a separate mandatory check. Filer/master ports +remain private; Service exposure and NetworkPolicy restrictions are unchanged. + +## Validation + +- All 166 Python unit tests pass with the dependencies of the exact 0.9.0 image + used for the WSL qualification. Twelve dedicated readiness tests cover timeout + retries, tunnel recovery, error propagation, probe preservation and ordering. +- An isolated copy of the 0.9.0 laboratory using SeaweedFS 3.97 passed resume and + coordinated SeaweedFS/operator restart checks. Repeated resume preserved the + SeaweedFS pod UID. Only probe fields changed in its StatefulSet; all 22 PVC + UIDs were preserved. +- The final laboratory candidate passed 12 synthetic S3 checks across Alpha and + Beta: authenticated data/web round trips, anonymous scope and peer denial. + Four owned synthetic objects were removed after the test. Both frontends + returned HTTP 200. These are focused storage checks, not a full application + regression suite. +- WSL-specific runtime evidence and candidate identities are retained separately + from the earlier clean laboratory and historical startup logs. A first + candidate exposed the inherited urllib3 connect-timeout gap; the final change + adds coverage for it without weakening the IAM verifier. +- The operator-authorized WSL2 test on K3s 1.37.2 and SeaweedFS 3.97 passed both + final-candidate resume and coordinated SeaweedFS/operator restart. The latter + waited approximately six seconds for Filer HTTP availability and completed + resume without a handler failure. All 18 persistent pods were Ready afterwards; + all 11 PVC UIDs and SeaweedFS settings other than the missing probe fields were + preserved. The final image was bound to the reviewed source hash. This WSL + check did not read or modify application documents and is not full application + acceptance. + +Validation was performed by one agent; no independent review is claimed. diff --git a/openserverless/seaweedfs_deploy.py b/openserverless/seaweedfs_deploy.py index ce08001..87eada9 100644 --- a/openserverless/seaweedfs_deploy.py +++ b/openserverless/seaweedfs_deploy.py @@ -16,6 +16,8 @@ # under the License. # import kopf, logging, time, os, json +from contextlib import contextmanager +from urllib3.exceptions import TimeoutError as HTTPTimeoutError import openserverless.kube as kube import openserverless.kustomize as kus import openserverless.config as cfg @@ -69,7 +71,8 @@ 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): +@contextmanager +def _ready_private_filer(data, timeout=30): """Wait for the private filer, recreating a tunnel lost during startup.""" deadline = time.monotonic() + timeout last_error = None @@ -84,11 +87,11 @@ def _bootstrap_private_filer(data, timeout=30): # 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: + except (TimeoutError, HTTPTimeoutError) as error: last_error = error else: ready = True - create_seaweedfs_nuv_storage(data, filer_url) + yield filer_url return except RuntimeError as error: if ready: @@ -100,6 +103,33 @@ def _bootstrap_private_filer(data, timeout=30): raise TimeoutError("SeaweedFS filer did not become ready before timeout") from last_error +def _bootstrap_private_filer(data, timeout=30): + with _ready_private_filer(data, timeout) as filer_url: + create_seaweedfs_nuv_storage(data, filer_url) + +def _reconcile_seaweedfs_probes(rendered): + """Fill missing probes without replacing live image, storage or custom probes.""" + desired = [item for item in rendered['items'] + if item['kind'] == 'StatefulSet' and item['metadata']['name'] == 'seaweedfs'] + if len(desired) != 1: + raise RuntimeError('Could not render the SeaweedFS StatefulSet') + defaults = next(container for container in desired[0]['spec']['template']['spec']['containers'] + if container['name'] == 'seaweedfs') + live = json.loads(kube.kubectl('get', 'statefulset/seaweedfs', '-o', 'json')) + containers = [container for container in live['spec']['template']['spec']['containers'] + if container['name'] == 'seaweedfs'] + if len(containers) != 1: + raise RuntimeError('Could not identify the SeaweedFS container') + missing = {key: defaults[key] for key in ('startupProbe', 'readinessProbe') + if not containers[0].get(key)} + if missing: + if live['spec'].get('updateStrategy', {}).get('type', 'RollingUpdate') != 'RollingUpdate': + raise RuntimeError('SeaweedFS missing probes requires a RollingUpdate or an operator-managed rollout') + kube.patch('statefulset/seaweedfs', {'spec': {'template': {'spec': { + 'containers': [{'name': 'seaweedfs', **missing}]}}}}, tpe='strategic') + # Wait for the updated pod, rather than probing a still-running old revision. + kube.kubectl('rollout', 'status', 'statefulset/seaweedfs', '--timeout=240s', timeout=250) + def create(owner=None): logging.info("*** configuring seaweedfs standalone") @@ -149,10 +179,12 @@ def reconcile(owner=None): kopf.append_owner_reference(resources, owner) datastore_security.apply(policy) result = kube.apply(services[0]) + _reconcile_seaweedfs_probes(rendered) # Existing deployments and restored filer data can still carry global Read. # Normalize IAM before publishing ingress, without bootstrapping buckets. - with service_endpoint(data['name'], 9090) as (host, port): - reconcile_anonymous_web_access(SeaweedfsClient(f"http://{host}:{port}")) + logging.info("*** waiting for seaweedfs filer api before IAM reconciliation") + with _ready_private_filer(data) as filer_url: + reconcile_anonymous_web_access(SeaweedfsClient(filer_url)) seaweedfs_ingress.create_seaweedfs_ingresses(data, owner) return result diff --git a/tests/test_datastore_security.py b/tests/test_datastore_security.py index bc18c47..161407a 100644 --- a/tests/test_datastore_security.py +++ b/tests/test_datastore_security.py @@ -211,6 +211,10 @@ def test_existing_cluster_reconcile_secures_before_service_update(self): side_effect=lambda obj: events.append(('policy', obj))), \ patch.object(seaweedfs_deploy.kube, 'apply', side_effect=lambda obj: events.append(('service', obj))), \ + patch.object(seaweedfs_deploy, '_reconcile_seaweedfs_probes', + side_effect=lambda obj: events.append(('probes', obj))), \ + patch.object(seaweedfs_deploy.util, 'wait_for_http', + side_effect=lambda *args, **kwargs: events.append(('ready', None))), \ patch.object(seaweedfs_deploy, 'service_endpoint', return_value= contextlib.nullcontext(('127.0.0.1', 41237))), \ patch.object(seaweedfs_deploy, 'SeaweedfsClient') as client, \ @@ -220,7 +224,7 @@ def test_existing_cluster_reconcile_secures_before_service_update(self): 'create_seaweedfs_ingresses', side_effect=lambda data, obj: events.append(('ingress', obj))): seaweedfs_deploy.reconcile(owner) - self.assertEqual([event[0] for event in events], ['policy', 'service', 'iam', 'ingress']) + self.assertEqual([event[0] for event in events], ['policy', 'service', 'probes', 'ready', 'iam', 'ingress']) client.assert_called_once_with('http://127.0.0.1:41237') self.assertEqual(events[0][1]['kind'], 'NetworkPolicy') self.assertEqual(events[1][1]['spec']['type'], 'ClusterIP') diff --git a/tests/test_s3_anonymous_access.py b/tests/test_s3_anonymous_access.py index a272e6d..011772d 100644 --- a/tests/test_s3_anonymous_access.py +++ b/tests/test_s3_anonymous_access.py @@ -60,6 +60,8 @@ def test_failed_iam_reconciliation_does_not_publish_ingress_or_bootstrap_storage {"kind": "Service", "metadata": {"name": "seaweedfs"}}]}), \ patch.object(seaweedfs_deploy.datastore_security, "apply"), \ patch.object(seaweedfs_deploy.kube, "apply"), \ + patch.object(seaweedfs_deploy, "_reconcile_seaweedfs_probes"), \ + patch.object(seaweedfs_deploy.util, "wait_for_http"), \ patch.object(seaweedfs_deploy, "service_endpoint", return_value=contextlib.nullcontext(("seaweedfs", 9090))), \ patch.object(seaweedfs_deploy, "SeaweedfsClient"), \ patch.object(seaweedfs_deploy, "reconcile_anonymous_web_access", side_effect=RuntimeError("IAM unavailable")), \ diff --git a/tests/test_seaweedfs_readiness.py b/tests/test_seaweedfs_readiness.py new file mode 100644 index 0000000..12ce202 --- /dev/null +++ b/tests/test_seaweedfs_readiness.py @@ -0,0 +1,175 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import contextlib +import copy +import json +from pathlib import Path +import unittest +from unittest.mock import patch +from urllib3.exceptions import ConnectTimeoutError, ReadTimeoutError + +import yaml +from openserverless import seaweedfs_deploy as deploy + + +class SeaweedFSReadinessTests(unittest.TestCase): + def setUp(self): + self.sts = yaml.safe_load((Path(__file__).resolve().parents[1] / + 'deploy/seaweedfs/seaweedfs-sts.yaml').read_text()) + self.rendered = {'items': [self.sts]} + + def test_probe_patch_contains_only_missing_probes_and_waits_for_rollout(self): + live = copy.deepcopy(self.sts) + container = live['spec']['template']['spec']['containers'][0] + del container['startupProbe'] + del container['readinessProbe'] + container.update(image='private/image:pinned', env=[{'name': 'CUSTOM', 'value': 'keep'}]) + before = copy.deepcopy(live) + with patch.object(deploy.kube, 'kubectl', side_effect=[json.dumps(live), 'ready']) as kubectl, \ + patch.object(deploy.kube, 'patch') as apply: + deploy._reconcile_seaweedfs_probes(self.rendered) + name, payload = apply.call_args.args + self.assertEqual(name, 'statefulset/seaweedfs') + self.assertEqual(apply.call_args.kwargs, {'tpe': 'strategic'}) + self.assertEqual(set(payload), {'spec'}) + self.assertEqual(set(payload['spec']), {'template'}) + self.assertEqual(set(payload['spec']['template']), {'spec'}) + self.assertEqual(set(payload['spec']['template']['spec']), {'containers'}) + self.assertEqual(set(payload['spec']['template']['spec']['containers'][0]), + {'name', 'startupProbe', 'readinessProbe'}) + self.assertEqual(live, before) + self.assertEqual(kubectl.call_args.args, + ('rollout', 'status', 'statefulset/seaweedfs', '--timeout=240s')) + self.assertEqual(kubectl.call_args.kwargs, {'timeout': 250}) + + def test_present_custom_probes_are_not_overwritten_or_restarted(self): + self.sts['spec']['template']['spec']['containers'][0]['readinessProbe'] = { + 'tcpSocket': {'port': 9000}, 'periodSeconds': 15} + with patch.object(deploy.kube, 'kubectl', side_effect=[json.dumps(self.sts), 'ready']), \ + patch.object(deploy.kube, 'patch') as apply: + deploy._reconcile_seaweedfs_probes(self.rendered) + apply.assert_not_called() + + def test_partial_probe_patch_keeps_existing_custom_probe(self): + live = copy.deepcopy(self.sts) + del live['spec']['template']['spec']['containers'][0]['startupProbe'] + with patch.object(deploy.kube, 'kubectl', side_effect=[json.dumps(live), 'ready']), \ + patch.object(deploy.kube, 'patch') as apply: + deploy._reconcile_seaweedfs_probes(self.rendered) + self.assertEqual(set(apply.call_args.args[1]['spec']['template']['spec']['containers'][0]), + {'name', 'startupProbe'}) + + def test_lookup_failure_never_becomes_a_replacement_workload(self): + with patch.object(deploy.kube, 'kubectl', side_effect=RuntimeError('unavailable')), \ + patch.object(deploy.kube, 'patch') as apply: + with self.assertRaisesRegex(RuntimeError, 'unavailable'): + deploy._reconcile_seaweedfs_probes(self.rendered) + apply.assert_not_called() + + def test_missing_container_is_rejected_without_patch(self): + live = copy.deepcopy(self.sts) + live['spec']['template']['spec']['containers'][0]['name'] = 'unexpected' + with patch.object(deploy.kube, 'kubectl', return_value=json.dumps(live)), \ + patch.object(deploy.kube, 'patch') as apply: + with self.assertRaisesRegex(RuntimeError, 'identify'): + deploy._reconcile_seaweedfs_probes(self.rendered) + apply.assert_not_called() + + def test_ondelete_does_not_silently_leave_old_pod_without_probes(self): + live = copy.deepcopy(self.sts) + live['spec']['updateStrategy'] = {'type': 'OnDelete'} + del live['spec']['template']['spec']['containers'][0]['startupProbe'] + with patch.object(deploy.kube, 'kubectl', return_value=json.dumps(live)), \ + patch.object(deploy.kube, 'patch') as apply: + with self.assertRaisesRegex(RuntimeError, 'operator-managed rollout'): + deploy._reconcile_seaweedfs_probes(self.rendered) + apply.assert_not_called() + + def test_iam_failure_after_readiness_is_not_retried(self): + with patch.object(deploy, 'service_endpoint', + side_effect=lambda *args: contextlib.nullcontext(('seaweedfs', 9090))) as endpoint, \ + patch.object(deploy.util, 'wait_for_http') as wait: + with self.assertRaisesRegex(RuntimeError, 'IAM mismatch'): + with deploy._ready_private_filer({'name': 'seaweedfs'}): + raise RuntimeError('IAM mismatch') + self.assertEqual(endpoint.call_count, 1) + self.assertEqual(wait.call_count, 1) + + def test_probe_connection_and_read_timeouts_are_retried(self): + with patch.object(deploy, 'service_endpoint', side_effect=lambda *args: + contextlib.nullcontext(('seaweedfs', 9090))) as endpoint, \ + patch.object(deploy.util, 'wait_for_http', side_effect=[ + ConnectTimeoutError('delayed service routing'), + ReadTimeoutError(None, '/', 'starting filer'), None]) as wait, \ + patch.object(deploy.time, 'sleep'): + with deploy._ready_private_filer({'name': 'seaweedfs'}) as url: + self.assertEqual(url, 'http://seaweedfs:9090') + self.assertEqual(endpoint.call_count, 3) + self.assertEqual(wait.call_count, 3) + + def test_timeout_in_iam_body_is_not_retried_as_readiness(self): + with patch.object(deploy, 'service_endpoint', side_effect=lambda *args: + contextlib.nullcontext(('seaweedfs', 9090))) as endpoint, \ + patch.object(deploy.util, 'wait_for_http'): + with self.assertRaises(ConnectTimeoutError): + with deploy._ready_private_filer({'name': 'seaweedfs'}): + raise ConnectTimeoutError('IAM transport failure') + self.assertEqual(endpoint.call_count, 1) + + def test_lost_tunnel_is_recreated_before_caller_runs(self): + def endpoint(*args): + if endpoint_mock.call_count == 1: + raise RuntimeError('lost port-forward') + return contextlib.nullcontext(('127.0.0.1', 41238)) + with patch.object(deploy, 'service_endpoint', side_effect=endpoint) as endpoint_mock, \ + patch.object(deploy.util, 'wait_for_http'), \ + patch.object(deploy.time, 'sleep'): + with deploy._ready_private_filer({'name': 'seaweedfs'}) as url: + self.assertEqual(url, 'http://127.0.0.1:41238') + self.assertEqual(endpoint_mock.call_count, 2) + + def test_deadline_prevents_iam_or_ingress(self): + deploy.cfg.configure({'components': {'seaweedfs': True}}) + self.addCleanup(deploy.cfg.clean) + service = {'kind': 'Service', 'metadata': {'name': 'seaweedfs'}} + with patch.object(deploy.util, 'get_seaweedfs_config_data', return_value={'name': 'seaweedfs'}), \ + patch.object(deploy, '_render_seaweedfs_spec', return_value={'items': [service]}), \ + patch.object(deploy.datastore_security, 'apply'), \ + patch.object(deploy.kube, 'apply'), \ + patch.object(deploy, '_reconcile_seaweedfs_probes'), \ + patch.object(deploy, 'service_endpoint', side_effect=lambda *args: + contextlib.nullcontext(('seaweedfs', 9090))), \ + patch.object(deploy.util, 'wait_for_http', side_effect=TimeoutError('not ready')), \ + patch.object(deploy.time, 'monotonic', side_effect=[0, 0, 31, 31]), \ + patch.object(deploy, 'reconcile_anonymous_web_access') as iam, \ + patch.object(deploy.seaweedfs_ingress, 'create_seaweedfs_ingresses') as ingress: + with self.assertRaisesRegex(TimeoutError, 'Filer|filer'): + deploy.reconcile({'metadata': {'uid': 'synthetic'}}) + iam.assert_not_called() + ingress.assert_not_called() + + def test_rollout_failure_prevents_iam(self): + with patch.object(deploy.kube, 'kubectl', side_effect=[json.dumps(self.sts), TimeoutError('rollout')]), \ + patch.object(deploy, 'reconcile_anonymous_web_access') as iam: + with self.assertRaisesRegex(TimeoutError, 'rollout'): + deploy._reconcile_seaweedfs_probes(self.rendered) + iam.assert_not_called() + + +if __name__ == '__main__': + unittest.main()