From f35629bd9af5510c5bdf26ff731d5e8a30737db2 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 23 Oct 2026 22:38:21 +0200 Subject: [PATCH] Bound cron-init jobs and scheduler I/O on 0.9.0 --- deploy/scheduler/cron.yaml | 14 +- openserverless/actionexecutor.py | 154 ++++++++---- openserverless/couchdb_util.py | 31 ++- openserverless/templates/cron-init.yaml | 14 +- tests/test_actionexecutor_bounds.py | 268 +++++++++++++++++++++ tests/test_actionexecutor_cron_security.py | 5 +- 6 files changed, 425 insertions(+), 61 deletions(-) create mode 100644 tests/test_actionexecutor_bounds.py diff --git a/deploy/scheduler/cron.yaml b/deploy/scheduler/cron.yaml index d002475..2799664 100644 --- a/deploy/scheduler/cron.yaml +++ b/deploy/scheduler/cron.yaml @@ -24,14 +24,26 @@ metadata: namespace: openserverless spec: schedule: "* * * * *" + concurrencyPolicy: Allow + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 1 jobTemplate: spec: + activeDeadlineSeconds: 120 + backoffLimit: 0 template: spec: containers: - name: cron-init image: "ghcr.io/nuvolaris/nuvolaris-operator:0.2.1-trinity.22080920" imagePullPolicy: IfNotPresent + resources: + requests: + cpu: 25m + memory: 64Mi + limits: + cpu: 250m + memory: 256Mi command: - /bin/sh - -c @@ -39,5 +51,5 @@ spec: env: - name: "SCHEDULER_CONFIG" value: "{}" - restartPolicy: OnFailure + restartPolicy: Never diff --git a/openserverless/actionexecutor.py b/openserverless/actionexecutor.py index e4fdcca..e97f782 100644 --- a/openserverless/actionexecutor.py +++ b/openserverless/actionexecutor.py @@ -15,7 +15,7 @@ # specific language governing permissions and limitations # under the License. # -import os, logging, json +import os, logging, json, time, math import openserverless.couchdb_util as cu import openserverless.kube as kube import croniter as cn @@ -25,6 +25,23 @@ # Bound tenant-controlled input before passing it to the cron parser. MAX_CRON_EXPRESSION_LENGTH = 4096 +# Leave time for process shutdown before the Job's 120 second deadline. +HTTP_TIMEOUT = (3.0, 10.0) +RUN_BUDGET_SECONDS = 100.0 +MAX_QUERY_PAGES = 1000 +MAX_QUERY_DOCUMENTS = 25000 +MAX_RESPONSE_BYTES = 4 * 1024 * 1024 +MAX_QUERY_BYTES = 16 * 1024 * 1024 + + +def request_timeout(deadline=None): + if deadline is None: + return HTTP_TIMEOUT + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError('Scheduler time budget exceeded') + return tuple(min(value, remaining) for value in HTTP_TIMEOUT) + def check(f, what, res): if f: logging.info(f"OK: {what}") @@ -94,42 +111,49 @@ def action_should_trigger(currentDate, executionInterval, actionCronExpression): # # query the dbn database using the specified selecto # -def find_docs(db, dbn, selector, username, password): +def find_docs(db, dbn, selector, username, password, deadline=None): documents = [] query = json.loads(selector) logging.info(f"Querying couchdb {dbn} for documents") - #CouchDB returns no more than 25 records. We iterate to get all the cron enabled actions. - while(True): - logging.info(f"select query param {json.dumps(query)}") - res = db.find_doc(dbn, json.dumps(query), username, password) - - if(res == None): - break - - if(res['docs']): - docs = list(res['docs']) - if(len(docs) > 0): - documents.extend(docs) - if(res['bookmark']): - query['bookmark']=res['bookmark'] - else: - logging.info('docs item is an emtpy list. No more documents found') - break - else: - logging.info('docs items not present. no more documents found') - break - - return list(documents) + if deadline is None: + deadline = time.monotonic() + RUN_BUDGET_SECONDS + seen_bookmarks = set() + total_bytes = 0 + for _ in range(MAX_QUERY_PAGES): + res = db.find_doc(dbn, json.dumps(query), username, password, + timeout=request_timeout(deadline), raise_on_error=True, + max_response_bytes=MAX_RESPONSE_BYTES, deadline=deadline) + request_timeout(deadline) + if not isinstance(res, dict) or not isinstance(res.get('docs'), list): + raise RuntimeError('Invalid CouchDB pagination response') + total_bytes += len(json.dumps(res).encode('utf-8')) + if total_bytes > MAX_QUERY_BYTES: + raise RuntimeError('CouchDB query byte limit exceeded') + docs = res['docs'] + if not docs: + return documents + if not all(isinstance(doc, dict) for doc in docs): + raise RuntimeError('Invalid CouchDB document') + bookmark = res.get('bookmark') + if (not isinstance(bookmark, str) or not bookmark + or bookmark == query.get('bookmark') or bookmark in seen_bookmarks): + raise RuntimeError('CouchDB pagination did not advance') + if len(documents) + len(docs) > MAX_QUERY_DOCUMENTS: + raise RuntimeError('CouchDB query document limit exceeded') + seen_bookmarks.add(bookmark) + documents.extend(docs) + query['bookmark'] = bookmark + raise RuntimeError('CouchDB pagination page limit exceeded') # # Get subject from openserverless_subjects db # #TODO need to find a way to build a dictionary with dynamic key or a hashmap -def get_subjects(db, username, password): +def get_subjects(db, username, password, deadline=None): subjects = [] selector = '{"selector":{"subject": {"$exists": true}},"fields":["namespaces"]}' - namespaces = find_docs(db, "subjects", selector, username, password) + namespaces = find_docs(db, "subjects", selector, username, password, deadline) for entry in namespaces: currentNamespaceList = list(entry['namespaces']) @@ -142,13 +166,13 @@ def get_subjects(db, username, password): # get actions from the openserverless_whisks db # having an annotation with key=cron or key=autoexec and value=true # -def get_cron_aware_actions(db, username, password): +def get_cron_aware_actions(db, username, password, deadline=None): selector = '{"selector":{"entityType":"action", "$or":[{"annotations": {"$elemMatch": {"key": "cron"}}},{"annotations": {"$elemMatch": {"$and":[{"key":"autoexec"},{"value":true}]}}}] }, "fields": ["_id", "annotations", "name", "_rev","namespace","parameters","entityType"]}' - return find_docs(db, "whisks", selector, username, password) + return find_docs(db, "whisks", selector, username, password, deadline) # # POST a request to invoke the ow action # -def call_ow_action(url, parameters, ow_auth): +def call_ow_action(url, parameters, ow_auth, deadline=None): logging.info(f"POST request to {url}") headers = {'Content-Type': 'application/json'} @@ -158,37 +182,44 @@ def call_ow_action(url, parameters, ow_auth): logging.info("calling an action with %s parameters",len(parameters)) # OpenWhisk passes automatically the registered parameters, so this cron executor does not need to pass them. - response = req.post(url, auth=(ow_auth['username'],ow_auth['password'])) + response = req.post(url, auth=(ow_auth['username'],ow_auth['password']), + timeout=request_timeout(deadline), stream=True, + allow_redirects=False) if (response.status_code in [200,202]): - logging.info(f"call to {url} succeeded with {response.status_code}. Body {response.text}") + logging.info(f"call to {url} succeeded with {response.status_code}") return True - logging.warn(f"query to {url} failed with {response.status_code}. Body {response.text}") + logging.warning(f"call to {url} failed with {response.status_code}") return False except Exception as inst: - logging.warn(f"Failed to invoke action {type(inst)}") - logging.warn(inst) + logging.warning("Invocation request failed; acceptance may be uncertain; not retrying: %s", inst) return False + finally: + if response is not None: + response.close() # # Update the action annotations to disable the cron execution # by setting {"autoexec":false} # -def unschedule_autoexec_action(action_url, ow_auth): +def unschedule_autoexec_action(action_url, ow_auth, deadline=None): logging.info(f"Purging cron details from action {action_url}") headers = {'Content-Type': 'application/json'} + response = None try: updating_data = {} updated_annotation = [{"key":"autoexec","value":False}] updating_data["annotations"]=updated_annotation logging.info(f"updating with {json.dumps(updating_data)}") - response = req.put(f"{action_url}?overwrite=true", auth=(ow_auth['username'],ow_auth['password']), headers=headers, data=json.dumps(updating_data)) + response = req.put(f"{action_url}?overwrite=true", auth=(ow_auth['username'],ow_auth['password']), headers=headers, data=json.dumps(updating_data), + timeout=request_timeout(deadline), stream=True, + allow_redirects=False) if response.status_code != 200: - logging.warn(f"PUT call to {action_url}?overwrite=true failed with {response.status_code}. Body {response.text}") + logging.warning(f"PUT call to {action_url}?overwrite=true failed with {response.status_code}") return False logging.info(f"PUT call to {action_url}?overwrite=true succeeded with {response.status_code}. Action cron policy removed") @@ -196,7 +227,10 @@ def unschedule_autoexec_action(action_url, ow_auth): except Exception as inst: logging.warn(f"Failed to invoke action {type(inst)}") logging.warn(inst) - return False + return False + finally: + if response is not None: + response.close() # # Extract the cron expression from the given annotations list @@ -320,7 +354,7 @@ def get_subject(action_namespace): # dAction input is a json Object with similar structure. # # dAction = '{"_id":"openserverless/hello-cron-action","annotations":[{"key":"cron","value":"*/2 * * * *"},{"key":"provide-api-key","value":false},{"key":"exec","value":"nodejs:14"}],"name":"hello-cron-action","_rev":"1-19f424e1fec1c02a2ecccf6f90978e31","namespace":"openserverless","parameters":[],"entityType":"action"}' -def handle_action(baseurl, currentDate, executionInterval, dAction, subjects): +def handle_action(baseurl, currentDate, executionInterval, dAction, subjects, deadline=None): actionName = dAction['name'] entityType = dAction['entityType'] actionNamespace = dAction['namespace'] @@ -338,7 +372,7 @@ def handle_action(baseurl, currentDate, executionInterval, dAction, subjects): execute = should_trigger(actionNamespace,actionName,actionCronExpression,currentDate,executionInterval) if "no_execution" == execute: - return None + return True namespaceSubject = get_subject(actionNamespace) auth = get_auth(subjects, namespaceSubject) @@ -346,13 +380,14 @@ def handle_action(baseurl, currentDate, executionInterval, dAction, subjects): package = get_package_from_namespace(actionNamespace) base_action_url = build_action_url(baseurl,namespaceSubject, package, actionName) - ret = call_ow_action(f"{base_action_url}?blocking=false&result=false", actionParameters, auth) + ret = call_ow_action(f"{base_action_url}?blocking=false&result=false", actionParameters, auth, deadline) if ret and "execute_once" == execute: - unschedule_autoexec_action(base_action_url, auth) + return unschedule_autoexec_action(base_action_url, auth, deadline) + return ret else: - logging.warn('No subject {subjectName} credentials found!') - return None + logging.warning('No subject %s credentials found!', namespaceSubject) + return False # # Search and return a {'username':'xxx','passowrd':'xxx'} dictionary @@ -372,14 +407,26 @@ def get_auth(subjects, subjectName): # Implement the logic to query for actions and evaluate how to execute them # def start(): + deadline = time.monotonic() + RUN_BUDGET_SECONDS + logging.basicConfig(level=logging.INFO) + try: + return run(deadline) + except Exception: + logging.exception('Scheduler run failed; no automatic replay') + return 1 + + +def run(deadline): # load scheduler config from the os environment cfg = os.environ.get("SCHEDULER_CONFIG") - if cfg: - logging.basicConfig(level=logging.INFO) - config = json.loads(cfg) + if not cfg: + raise ValueError('SCHEDULER_CONFIG is required') + config = json.loads(cfg) currentDate = datetime.now() interval = from_cron_to_seconds(currentDate, config['scheduler.schedule']) + if not math.isfinite(interval) or interval <= 0: + raise ValueError('Scheduler interval must be finite and positive') logging.info(f"interval in seconds between 2 execution is {interval} seconds") db = cu.CouchDB() @@ -390,16 +437,23 @@ def start(): ow_host = config['controller.host'] ow_port = config['controller.port'] baseurl = f"{ow_protocol}://{ow_host}:{ow_port}/api/v1/namespaces/" - actions = get_cron_aware_actions(db, config['couchdb.controller.user'],config['couchdb.controller.password']) + actions = get_cron_aware_actions(db, config['couchdb.controller.user'],config['couchdb.controller.password'], deadline) if(len(actions) > 0): - subjects = get_subjects(db, config['couchdb.controller.user'],config['couchdb.controller.password']) + subjects = get_subjects(db, config['couchdb.controller.user'],config['couchdb.controller.password'], deadline) + failed = False for action in actions: + request_timeout(deadline) try: - handle_action(baseurl, currentDate, interval, action, subjects) + if not handle_action(baseurl, currentDate, interval, action, subjects, deadline): + failed = True except Exception: + failed = True logging.exception("Failed to process scheduled action; continuing with remaining actions") + return 1 if failed else 0 else: logging.info('No cron aware action extracted. Exiting....') + return 0 else: logging.warn("CouchDB it is not available. Exiting....") + return 1 diff --git a/openserverless/couchdb_util.py b/openserverless/couchdb_util.py index 49cc0ac..2049534 100644 --- a/openserverless/couchdb_util.py +++ b/openserverless/couchdb_util.py @@ -149,7 +149,9 @@ def add_role(self, database: str, members = [], admins =[]): # # Submit a POST request to the _find endpoint using the specified selector # - def find_doc(self, database, selector, user=None, password="", no_auth=False): + def find_doc(self, database, selector, user=None, password="", no_auth=False, + *, timeout=None, raise_on_error=False, max_response_bytes=None, + deadline=None): url = f"{self.db_base}{database}/_find" headers = {'Content-Type': 'application/json'} session = req.Session() @@ -162,9 +164,24 @@ def find_doc(self, database, selector, user=None, password="", no_auth=False): db_auth = self.db_auth session.auth=db_auth - r = session.post(url, headers=headers, data=selector) - if r.status_code == 200: - return json.loads(r.text) - - logging.warn(f"query to {url} failed with {r.status_code}. Body {r.text}") - return None \ No newline at end of file + try: + with session.post(url, headers=headers, data=selector, timeout=timeout, + stream=max_response_bytes is not None, + allow_redirects=not raise_on_error) as r: + if raise_on_error and r.status_code != 200: + raise RuntimeError(f'CouchDB query failed with HTTP {r.status_code}') + if r.status_code == 200: + if max_response_bytes is None: + return json.loads(r.text) + body = bytearray() + for chunk in r.iter_content(chunk_size=4096): + if deadline is not None and time.monotonic() >= deadline: + raise RuntimeError('Scheduler time budget exceeded') + if len(body) + len(chunk) > max_response_bytes: + raise RuntimeError('CouchDB response byte limit exceeded') + body.extend(chunk) + return json.loads(body) + logging.warning(f"query to {url} failed with {r.status_code}") + return None + finally: + session.close() diff --git a/openserverless/templates/cron-init.yaml b/openserverless/templates/cron-init.yaml index 2090e11..8deeff3 100644 --- a/openserverless/templates/cron-init.yaml +++ b/openserverless/templates/cron-init.yaml @@ -22,14 +22,26 @@ metadata: namespace: openserverless spec: schedule: "{{schedule}}" + concurrencyPolicy: Allow + successfulJobsHistoryLimit: 3 + failedJobsHistoryLimit: 1 jobTemplate: spec: + activeDeadlineSeconds: 120 + backoffLimit: 0 template: spec: containers: - name: cron-init image: "{{image}}" imagePullPolicy: IfNotPresent + resources: + requests: + cpu: 25m + memory: 64Mi + limits: + cpu: 250m + memory: 256Mi command: - /bin/sh - -c @@ -38,4 +50,4 @@ spec: - name: "SCHEDULER_CONFIG" value: > {{config}} - restartPolicy: OnFailure \ No newline at end of file + restartPolicy: Never diff --git a/tests/test_actionexecutor_bounds.py b/tests/test_actionexecutor_bounds.py new file mode 100644 index 0000000..8452f80 --- /dev/null +++ b/tests/test_actionexecutor_bounds.py @@ -0,0 +1,268 @@ +# 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 json +import threading +import time +import unittest +from datetime import datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest.mock import Mock, patch + +import requests +import yaml +from jinja2 import Template + +import openserverless.actionexecutor as executor +from openserverless.couchdb_util import CouchDB + + +class PaginationTest(unittest.TestCase): + def find(self, db, **kwargs): + return executor.find_docs(db, 'whisks', '{"selector":{}}', 'u', 'p', **kwargs) + + def test_multiple_pages_and_empty_final_page(self): + db = Mock() + db.find_doc.side_effect = [ + {'docs': [{'_id': 'a'}], 'bookmark': 'a'}, + {'docs': [{'_id': 'b'}], 'bookmark': 'b'}, {'docs': []}] + self.assertEqual(self.find(db), [{'_id': 'a'}, {'_id': 'b'}]) + self.assertEqual(json.loads(db.find_doc.call_args_list[1].args[1])['bookmark'], 'a') + self.assertTrue(db.find_doc.call_args.kwargs['raise_on_error']) + + def test_invalid_pages_and_bookmarks_fail(self): + for page in (None, [], {}, {'docs': {}}, {'docs': [1]}, + {'docs': [{}]}, {'docs': [{}], 'bookmark': ''}, + {'docs': [{}], 'bookmark': False}): + with self.subTest(page=page): + db = Mock(); db.find_doc.return_value = page + with self.assertRaises(RuntimeError): self.find(db) + + def test_repeated_bookmark_and_cycle_fail(self): + for bookmarks in (('a', 'a'), ('a', 'b', 'a')): + db = Mock() + db.find_doc.side_effect = [{'docs': [{}], 'bookmark': b} for b in bookmarks] + with self.assertRaisesRegex(RuntimeError, 'did not advance'): self.find(db) + self.assertEqual(db.find_doc.call_count, len(bookmarks)) + + def test_forever_advancing_bookmarks_are_bounded(self): + db = Mock() + db.find_doc.side_effect = [{'docs': [{}], 'bookmark': str(i)} for i in range(5)] + with patch.object(executor, 'MAX_QUERY_PAGES', 3): + with self.assertRaisesRegex(RuntimeError, 'page limit'): self.find(db) + self.assertEqual(db.find_doc.call_count, 3) + + def test_document_and_byte_limits_fail_without_truncation(self): + for limit in ('MAX_QUERY_DOCUMENTS', 'MAX_QUERY_BYTES'): + db = Mock(); db.find_doc.return_value = {'docs': [{}], 'bookmark': 'a'} + with patch.object(executor, limit, 0): + with self.assertRaisesRegex(RuntimeError, 'limit'): self.find(db) + + def test_expired_budget_prevents_request(self): + db = Mock() + with self.assertRaisesRegex(RuntimeError, 'time budget'): + self.find(db, deadline=time.monotonic() - 1) + db.find_doc.assert_not_called() + + +class RealHttpTest(unittest.TestCase): + """Exercise real Requests timeouts/streaming through the production helpers.""" + @classmethod + def setUpClass(cls): + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): pass + def do_POST(self): self.respond() + def do_PUT(self): self.respond() + def respond(self): + self.server.calls.append((self.command, self.path)) + try: + self.rfile.read(int(self.headers.get('Content-Length', '0'))) + if self.server.mode == 'headers_stall': time.sleep(.3) + status = 500 if self.server.mode == 'error' else 200 + self.send_response(status) + self.end_headers() + if self.server.mode == 'body_stall': time.sleep(.3) + body = {'docs': [], 'bookmark': ''} + if self.server.mode == 'large': body['padding'] = 'x' * 10000 + self.wfile.write(json.dumps(body).encode()) + except (BrokenPipeError, ConnectionResetError): pass + cls.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + cls.url = 'http://127.0.0.1:' + str(cls.server.server_port) + + @classmethod + def tearDownClass(cls): + cls.server.shutdown(); cls.server.server_close(); cls.thread.join() + + def setUp(self): + self.server.calls = []; self.server.mode = 'healthy' + self.db = CouchDB.__new__(CouchDB) + self.db.db_base = self.url + '/openserverless_' + self.db.db_auth = None + + def query(self, **kwargs): + return self.db.find_doc('fixture', '{}', timeout=(.05, .05), **kwargs) + + def test_successful_bounded_response(self): + self.assertEqual(self.query(raise_on_error=True, max_response_bytes=1024)['docs'], []) + + def test_http_error_strict_mode_is_opt_in(self): + self.server.mode = 'error' + self.assertIsNone(self.query()) + with self.assertRaisesRegex(RuntimeError, 'HTTP 500'): + self.query(raise_on_error=True, max_response_bytes=1024) + + def test_headers_and_body_stalls_time_out(self): + for mode in ('headers_stall', 'body_stall'): + with self.subTest(mode=mode): + self.server.mode = mode + with self.assertRaises(requests.RequestException): + self.query(raise_on_error=True, max_response_bytes=1024) + + def test_oversized_response_fails(self): + self.server.mode = 'large' + with self.assertRaisesRegex(RuntimeError, 'response byte limit'): + self.query(raise_on_error=True, max_response_bytes=100) + + def test_expired_body_budget_fails(self): + with self.assertRaisesRegex(RuntimeError, 'time budget'): + self.query(max_response_bytes=1024, deadline=time.monotonic() - 1) + + def test_invocation_timeout_is_not_retried(self): + self.server.mode = 'headers_stall' + with patch.object(executor, 'HTTP_TIMEOUT', (.05, .05)): + self.assertFalse(executor.call_ow_action(self.url, [], {'username': 'u', 'password': 'p'})) + self.assertEqual(len(self.server.calls), 1) + + def test_invocation_does_not_wait_for_unneeded_body(self): + self.server.mode = 'body_stall' + with patch.object(executor, 'HTTP_TIMEOUT', (.05, .05)): + self.assertTrue(executor.call_ow_action(self.url, [], {'username': 'u', 'password': 'p'})) + self.assertTrue(executor.unschedule_autoexec_action(self.url, {'username': 'u', 'password': 'p'})) + + def test_rejected_post_and_put_fail_once(self): + self.server.mode = 'error' + self.assertFalse(executor.call_ow_action(self.url, [], {'username': 'u', 'password': 'p'})) + self.assertFalse(executor.unschedule_autoexec_action(self.url, {'username': 'u', 'password': 'p'})) + self.assertEqual([m for m, _ in self.server.calls], ['POST', 'PUT']) + + +class ExecutionTest(unittest.TestCase): + def setUp(self): + self.config = { + 'scheduler.schedule': '* * * * *', 'controller.protocol': 'http', + 'controller.host': 'controller', 'controller.port': 3233, + 'couchdb.controller.user': 'u', 'couchdb.controller.password': 'p'} + self.db = Mock(); self.db.wait_db_ready.return_value = True + self.action = {'name': 'fixture', 'namespace': 'tenant', 'entityType': 'action', + 'parameters': [], 'annotations': [{'key': 'autoexec', 'value': True}]} + self.subjects = [{'name': 'tenant', 'uuid': 'u', 'key': 'p'}] + + def start(self): + with (patch.dict(executor.os.environ, {'SCHEDULER_CONFIG': json.dumps(self.config)}), + patch.object(executor.cu, 'CouchDB', return_value=self.db)): + return executor.start() + + def test_readiness_failure_exits_nonzero(self): + self.db.wait_db_ready.return_value = False + self.assertEqual(self.start(), 1) + self.db.find_doc.assert_not_called() + + def test_empty_catalog_succeeds(self): + self.db.find_doc.return_value = {'docs': []} + self.assertEqual(self.start(), 0) + + def test_later_action_page_failure_never_dispatches_partial_list(self): + self.db.find_doc.side_effect = [{'docs': [self.action], 'bookmark': 'a'}, None] + with patch.object(executor.req, 'post') as post: + self.assertEqual(self.start(), 1) + post.assert_not_called() + + def test_later_subject_page_failure_never_dispatches(self): + self.db.find_doc.side_effect = [ + {'docs': [self.action], 'bookmark': 'a'}, {'docs': []}, + {'docs': [{'namespaces': self.subjects}], 'bookmark': 's'}, None] + with patch.object(executor.req, 'post') as post: + self.assertEqual(self.start(), 1) + post.assert_not_called() + + def test_not_due_is_success_without_requests(self): + self.action['annotations'] = [{'key': 'cron', 'value': '0 0 1 1 *'}] + with patch.object(executor.req, 'post') as post: + self.assertTrue(executor.handle_action('http://controller/', datetime(2026, 10, 23, 12), + 60, self.action, self.subjects)) + post.assert_not_called() + + def test_missing_credentials_fail_without_request(self): + with patch.object(executor.req, 'post') as post: + self.assertFalse(executor.handle_action('http://controller/', datetime.now(), + 60, self.action, [])) + post.assert_not_called() + + def test_autoexec_put_failure_is_reported_without_post_replay(self): + with (patch.object(executor.req, 'post', return_value=Mock(status_code=202)) as post, + patch.object(executor.req, 'put', return_value=Mock(status_code=500)) as put): + self.assertFalse(executor.handle_action('http://controller/', datetime.now(), + 60, self.action, self.subjects)) + self.assertEqual((post.call_count, put.call_count), (1, 1)) + + def test_failed_invocation_does_not_stop_other_actions_or_replay(self): + other = dict(self.action, name='other', annotations=[{'key': 'cron', 'value': '* * * * *'}]) + self.db.find_doc.side_effect = [ + {'docs': [self.action, other], 'bookmark': 'a'}, {'docs': []}, + {'docs': [{'namespaces': self.subjects}], 'bookmark': 's'}, {'docs': []}] + with patch.object(executor.req, 'post', side_effect=[Mock(status_code=500), Mock(status_code=202)]) as post: + self.assertEqual(self.start(), 1) + self.assertEqual(post.call_count, 2) + + def test_total_budget_expiry_stops_before_next_action(self): + with (patch.object(executor.time, 'monotonic', return_value=10), + patch.dict(executor.os.environ, {'SCHEDULER_CONFIG': json.dumps(self.config)}), + patch.object(executor.cu, 'CouchDB', return_value=self.db)): + with self.assertRaisesRegex(RuntimeError, 'time budget'): + executor.run(deadline=9) + + def test_missing_invalid_and_nonfinite_configuration_fails(self): + for raw in ('', 'null', '{}', 'not-json'): + with patch.dict(executor.os.environ, {'SCHEDULER_CONFIG': raw}): + self.assertEqual(executor.start(), 1) + for interval in (float('nan'), float('inf'), 0, -1): + with patch.object(executor, 'from_cron_to_seconds', return_value=interval): + self.assertEqual(self.start(), 1) + + +class ManifestTest(unittest.TestCase): + def test_base_and_rendered_template_have_same_containment(self): + root = Path(__file__).resolve().parents[1] + paths = ('deploy/scheduler/cron.yaml', 'openserverless/templates/cron-init.yaml') + for path in paths: + text = Template((root / path).read_text()).render(schedule='* * * * *', + image='fixture:local', config='{}') + spec = next(doc for doc in yaml.safe_load_all(text) if doc)['spec'] + self.assertEqual(spec['concurrencyPolicy'], 'Allow') + job = spec['jobTemplate']['spec'] + self.assertEqual(job['activeDeadlineSeconds'], 120) + self.assertEqual(job['backoffLimit'], 0) + pod = job['template']['spec'] + self.assertEqual(pod['restartPolicy'], 'Never') + self.assertIn('limits', pod['containers'][0]['resources']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_actionexecutor_cron_security.py b/tests/test_actionexecutor_cron_security.py index 94721a1..db3b246 100644 --- a/tests/test_actionexecutor_cron_security.py +++ b/tests/test_actionexecutor_cron_security.py @@ -75,9 +75,10 @@ def test_malformed_actions_do_not_stop_other_tenants(self): ], } - def find_doc(database, selector, username, password): + def find_doc(database, selector, username, password, **kwargs): self.assertEqual((username, password), ("controller", "secret")) self.assertIn("selector", json.loads(selector)) + self.assertTrue(kwargs["raise_on_error"]) return batches[database].pop(0) db.find_doc.side_effect = find_doc @@ -89,7 +90,7 @@ def find_doc(database, selector, username, password): patch.object(executor.req, "put", return_value=Mock(status_code=200, text="")) as unschedule, self.assertLogs(level="WARNING") as logs, ): - executor.start() + self.assertEqual(executor.start(), 1) self.assertTrue(any("Failed to process scheduled action" in line for line in logs.output)) self.assertEqual(invoke.call_count, 2)