From cd13b271ba76b9dd4c694d502c16b518b8fc119a Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Fri, 15 Aug 2025 08:55:42 +0200 Subject: [PATCH] Implement external scheduler call --- cinder/opts.py | 2 + cinder/scheduler/external.py | 141 +++++++++ cinder/scheduler/filter_scheduler.py | 8 + cinder/tests/unit/scheduler/test_external.py | 281 ++++++++++++++++++ .../unit/scheduler/test_filter_scheduler.py | 25 ++ 5 files changed, 457 insertions(+) create mode 100644 cinder/scheduler/external.py create mode 100644 cinder/tests/unit/scheduler/test_external.py diff --git a/cinder/opts.py b/cinder/opts.py index 53d4636317a..0e0c061e4b9 100644 --- a/cinder/opts.py +++ b/cinder/opts.py @@ -57,6 +57,7 @@ from cinder.message import api as cinder_message_api from cinder import quota as cinder_quota from cinder.scheduler import driver as cinder_scheduler_driver +from cinder.scheduler import external as cinder_scheduler_external from cinder.scheduler.filters import shard_filter as \ cinder_scheduler_filters_shardfilter from cinder.scheduler import host_manager as cinder_scheduler_hostmanager @@ -274,6 +275,7 @@ def list_opts(): cinder_message_api.messages_opts, cinder_quota.quota_opts, cinder_scheduler_driver.scheduler_driver_opts, + cinder_scheduler_external.scheduler_external_opts, cinder_scheduler_hostmanager.host_manager_opts, cinder_scheduler_manager.scheduler_manager_opts, [cinder_scheduler_scheduleroptions. diff --git a/cinder/scheduler/external.py b/cinder/scheduler/external.py new file mode 100644 index 00000000000..a2b03d6970e --- /dev/null +++ b/cinder/scheduler/external.py @@ -0,0 +1,141 @@ +# Copyright (c) 2025 SAP SE +# All Rights Reserved. +# +# Licensed 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. +""" +This module provides functionality to interact with an external scheduler API +to reorder and filter hosts based on additional criteria. + +The external scheduler API is expected to take a list of weighed hosts and +their weights, along with the request specification, and return a reordered +and filtered list of host names. +""" +import jsonschema +from oslo_config import cfg +from oslo_log import log as logging +import requests + + +scheduler_external_opts = [ + cfg.StrOpt( + "external_scheduler_api_url", + default="", + help=""" +The API URL of the external scheduler. + +If this URL is provided, Cinder will call an external service after filters +and weighers have been applied. This service can reorder and filter the +list of hosts before Cinder attempts to place the volume. + +If not provided, this step will be skipped. +"""), + cfg.IntOpt( + "external_scheduler_timeout", + default=5, + min=1, + help=""" +The timeout in seconds for the external scheduler. + +If external_scheduler_api_url is configured, Cinder will call and wait for the +external scheduler to respond for this long. If the external scheduler does not +respond within this time, the request will be aborted. In this case, the +scheduler will continue with the original host selection and weights. +"""), +] + +CONF = cfg.CONF +CONF.register_opts(scheduler_external_opts) + +LOG = logging.getLogger(__name__) + +# The expected response schema from the external scheduler api. +# The response should contain a list of ordered host names. +RESPONSE_SCHEMA = { + "type": "object", + "properties": { + "hosts": { + "type": "array", + "items": { + "type": "string" + } + } + }, + "required": ["hosts"], + "additionalProperties": False, +} + + +def call_external_scheduler_api(context, weighed_hosts, spec_dict): + """Reorder and filter hosts using an external scheduler service. + + :param context: The RequestContext object containing request id, user, etc. + :param weighed_hosts: List of w. hosts to send to the external scheduler. + :param spec_dict: The RequestSpec object with the share specification. + """ + if not weighed_hosts: + return weighed_hosts + if not (url := CONF.external_scheduler_api_url): + LOG.debug("External scheduler API is not enabled.") + return weighed_hosts + timeout = CONF.external_scheduler_timeout + # We shouldn't pass and log the auth token. Thus, we delete it here. + ctx_dict = context.to_dict() + if "auth_token" in ctx_dict: + del ctx_dict["auth_token"] + json_data = { + "spec": spec_dict, + # Also serialize the request context, which contains the global request + # id and other information helpful for logging and request tracing. + "context": ctx_dict, + # Only provide basic information for the hosts for now. + # The external scheduler is expected to fetch statistics + # about the hosts separately, so we don't need to pass + # them here. + "hosts": [ + { + "host": h.obj.host, + } for h in weighed_hosts + ], + # Also pass previous weights from the Cinder weigher pipeline. + # The external scheduler api is expected to take these weights + # into account if provided. + "weights": {h.obj.host: h.weight for h in weighed_hosts}, + } + LOG.debug("Calling external scheduler API with %s", json_data) + try: + response = requests.post(url, json=json_data, timeout=timeout) + response.raise_for_status() + # If the JSON parsing fails, this will also raise a RequestException. + response_json = response.json() + except requests.RequestException as e: + LOG.error("Failed to call external scheduler API: %s", e) + return weighed_hosts + + # The external scheduler api is expected to return a json with + # a sorted list of host names. Note that no weights are returned. + try: + jsonschema.validate(response_json, RESPONSE_SCHEMA) + except jsonschema.ValidationError as e: + LOG.error("External scheduler response is invalid: %s", e) + return weighed_hosts + + # The list of host names can also be empty. In this case, we trust + # the external scheduler decision and return an empty list. + if not (host_names := response_json["hosts"]): + # If this case happens often, it may indicate an issue. + LOG.warning("External scheduler filtered out all hosts.") + + # Reorder the weighed hosts based on the list of host names returned + # by the external scheduler api. + weighed_hosts_dict = {h.obj.host: h for h in weighed_hosts} + return [weighed_hosts_dict[h] for h in host_names] diff --git a/cinder/scheduler/filter_scheduler.py b/cinder/scheduler/filter_scheduler.py index 6c1b070844b..56b578f33f2 100644 --- a/cinder/scheduler/filter_scheduler.py +++ b/cinder/scheduler/filter_scheduler.py @@ -33,6 +33,7 @@ from cinder.i18n import _ from cinder import objects from cinder.scheduler import driver +from cinder.scheduler.external import call_external_scheduler_api from cinder.scheduler.host_manager import BackendState from cinder.scheduler import scheduler_options from cinder.scheduler.weights import WeighedHost @@ -405,6 +406,13 @@ def _get_weighted_candidates( # backend for the job. weighed_backends = self.host_manager.get_weighed_backends( backends, filter_properties) + + # Call an external service that can modify `weighed_hosts` once more. + # This service may filter out some hosts, or it may re-order them. + # Note: the result can also be empty. + weighed_backends = call_external_scheduler_api( + context, weighed_backends, request_spec) + return weighed_backends def _get_weighted_candidates_generic_group( diff --git a/cinder/tests/unit/scheduler/test_external.py b/cinder/tests/unit/scheduler/test_external.py new file mode 100644 index 00000000000..3f57d97bc5e --- /dev/null +++ b/cinder/tests/unit/scheduler/test_external.py @@ -0,0 +1,281 @@ +# Copyright 2025 SAP SE or an SAP affiliate company. +# +# Licensed 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. +""" +Tests for the external scheduler api call. +""" + +from unittest.mock import MagicMock +from unittest.mock import patch + +import jsonschema +import requests + +from cinder import context +from cinder.scheduler.base_weight import WeighedObject +from cinder.scheduler.external import call_external_scheduler_api +from cinder.tests.unit.scheduler import fakes +from cinder.tests.unit.scheduler import test_scheduler + +# The expected request schema for the external scheduler API. +# It should contain the spec, hosts, and weights. +REQUEST_SCHEMA = { + "type": "object", + "properties": { + "spec": {"type": "object"}, + "context": {"type": "object"}, + "hosts": { + "type": "array", + "items": { + "type": "object", + "properties": { + "host": {"type": "string"}, + }, + "required": ["host"], + }, + }, + "weights": {"type": "object"}, + }, + "required": ["spec", "context", "hosts", "weights"], + "additionalProperties": False, +} + + +class ExternalSchedulerAPITestCase(test_scheduler.SchedulerTestCase): + def setUp(self): + super(ExternalSchedulerAPITestCase, self).setUp() + self.flags(external_scheduler_api_url='http://127.0.0.1:1234') + self.flags(external_scheduler_timeout=5) + self.h1 = fakes.FakeBackendState('host1', {}) + self.h2 = fakes.FakeBackendState('host2', {}) + self.h3 = fakes.FakeBackendState('host3', {}) + self.example_weighed_hosts = [ + WeighedObject(self.h1, 1.0), + WeighedObject(self.h2, 0.5), + WeighedObject(self.h3, 0.0), + ] + self.example_spec = { + 'volume_properties': {'size': 1234}, + } + self.example_ctx = context.RequestContext( + user_id='fake_user', + project_id='fake_project', + is_admin=True, + read_deleted='no', + global_request_id='fake_global_request_id', + ) + + def _check_request(self, response=None): + """Utility to check the request for validity.""" + def wrapped(url, json, timeout): + self.assertEqual(timeout, 5) # should be the default timeout + self.assertEqual(url, 'http://127.0.0.1:1234') + try: + jsonschema.validate(json, REQUEST_SCHEMA) + except jsonschema.ValidationError as e: + msg = f"Request JSON schema validation failed: {e.message}" + self.fail(msg) + return response or MagicMock() + return wrapped + + @patch('requests.post') + def test_context_included_in_request(self, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'hosts': ['host1', 'host3']} + mock_post.side_effect = self._check_request(mock_response) + + call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + + # Check that the context is serialized and included in the request + _, kwargs = mock_post.call_args + self.assertIn( + 'context', kwargs['json'], + 'Context should be included in the request' + ) + self.assertIn( + 'global_request_id', kwargs['json']['context'], + 'Global request ID should be included in the context' + ) + # The auth_token should be excluded from the context + self.assertNotIn( + 'auth_token', kwargs['json']['context'], + 'Auth token should not be included in the context' + ) + expected_dict = self.example_ctx.to_dict() + del expected_dict['auth_token'] + # Check that the context is serialized correctly + self.assertEqual( + expected_dict, + kwargs['json']['context'], + 'Context should be serialized correctly' + ) + + @patch('requests.post') + @patch('cinder.scheduler.external.LOG.debug') + def test_enabled_api_success(self, mock_debug_log, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'hosts': ['host1', 'host3']} + mock_post.side_effect = self._check_request(mock_response) + + log = "" + + def append_log(msg, data): + nonlocal log + log += msg % data + mock_debug_log.side_effect = append_log + + hosts = call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + self.assertEqual( + ['host1', 'host3'], + [h.obj.host for h in hosts] + ) + self.assertIn('Calling external scheduler API with ', log) + + @patch('requests.post') + @patch('cinder.scheduler.external.LOG.warning') + def test_enabled_api_empty_response(self, mock_warn_log, mock_post): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {'hosts': []} + mock_post.side_effect = self._check_request(mock_response) + + hosts = call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + self.assertEqual([], hosts) + mock_warn_log.assert_called_with( + 'External scheduler filtered out all hosts.' + ) + + @patch('requests.post') + @patch('cinder.scheduler.external.LOG.error') + def test_enabled_api_timeout(self, mock_err_log, mock_post): + mock_post.side_effect = requests.exceptions.Timeout + + log = "" + + def append_log(msg, data): + nonlocal log + log += msg % data + mock_err_log.side_effect = append_log + + hosts = call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + # Should fallback to the original host list. + self.assertEqual( + ['host1', 'host2', 'host3'], + [h.obj.host for h in hosts] + ) + self.assertIn('Failed to call external scheduler API: ', log) + + @patch('requests.post') + @patch('cinder.scheduler.external.LOG.error') + def test_enabled_api_invalid_response(self, mock_err_log, mock_post): + invalid_response_dicts = [ + {}, + {"hosts": "not a list"}, + {"hosts": [1, 2, "host1"]}, + {"hosts": [{"name": "host1", "status": "up"}]}, + ] + + log = "" + + def append_log(msg, data): + nonlocal log + log += msg % data + mock_err_log.side_effect = append_log + + for response_dict in invalid_response_dicts: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = response_dict + mock_post.side_effect = self._check_request(mock_response) + + hosts = call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + # Should fallback to the original host list. + self.assertEqual( + ['host1', 'host2', 'host3'], + [h.obj.host for h in hosts] + ) + self.assertIn('External scheduler response is invalid: ', log) + + @patch('requests.post') + @patch('cinder.scheduler.external.LOG.error') + def test_enabled_api_json_decode_err(self, mock_err_log, mock_post): + log = "" + + def append_log(msg, data): + nonlocal log + log += msg % data + mock_err_log.side_effect = append_log + + mock_response = MagicMock() + mock_response.status_code = 200 + # Note: requests.exceptions.InvalidJSONError is also a RequestException + mock_response.json.side_effect = requests.exceptions.InvalidJSONError + mock_post.side_effect = self._check_request(mock_response) + + hosts = call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + # Should fallback to the original host list. + self.assertEqual( + ['host1', 'host2', 'host3'], + [h.obj.host for h in hosts] + ) + self.assertIn('Failed to call external scheduler API: ', log) + + @patch('requests.post') + @patch('cinder.scheduler.external.LOG.error') + def test_enabled_api_error_reply(self, mock_err_log, mock_post): + mock_post.side_effect = requests.exceptions.HTTPError + + log = "" + + def append_log(msg, data): + nonlocal log + log += msg % data + mock_err_log.side_effect = append_log + + hosts = call_external_scheduler_api( + self.example_ctx, + self.example_weighed_hosts, + self.example_spec, + ) + # Should fallback to the original host list. + self.assertEqual( + ['host1', 'host2', 'host3'], + [h.obj.host for h in hosts] + ) + self.assertIn('Failed to call external scheduler API: ', log) diff --git a/cinder/tests/unit/scheduler/test_filter_scheduler.py b/cinder/tests/unit/scheduler/test_filter_scheduler.py index f1e2432a3d6..5ec77ed00f7 100644 --- a/cinder/tests/unit/scheduler/test_filter_scheduler.py +++ b/cinder/tests/unit/scheduler/test_filter_scheduler.py @@ -21,6 +21,7 @@ from cinder import context from cinder import exception from cinder import objects +from cinder.scheduler.base_weight import WeighedObject from cinder.scheduler import filter_scheduler from cinder.scheduler import host_manager from cinder.tests.unit import fake_constants as fake @@ -57,6 +58,30 @@ def test_create_group_no_hosts(self): fake_context, 'faki-id1', group_spec, request_spec_list, {}, []) + @mock.patch('requests.post') + def test_external_scheduler_disabled(self, mock_post): + """Tests that the external scheduler is not called when disabled.""" + sched = fakes.FakeFilterScheduler() + h1 = fakes.FakeBackendState('host1', {}) + h2 = fakes.FakeBackendState('host2', {}) + self.mock_object( + sched.host_manager, 'get_filtered_backends', + mock.Mock(return_value=[h1, h2]), + ) + wh1 = WeighedObject(h1, 1.0) + wh2 = WeighedObject(h2, 1.0) + self.mock_object( + sched.host_manager, 'get_weighed_backends', + mock.Mock(return_value=[wh1, wh2]), + ) + ctx = context.RequestContext('user', 'project', is_admin=True) + spec = { + 'volume_properties': {'size': 1234}, + } + self.flags(external_scheduler_api_url='') + _ = sched._get_weighted_candidates(ctx, spec, {}) + mock_post.assert_not_called() + @ddt.data( {'capabilities:consistent_group_snapshot_enabled': ' True'}, {'consistent_group_snapshot_enabled': ' True'}