If the ingress service is configured with a placement count lower than the number of hosts,
then when one node goes down, Keepalived is configured on the remaining hosts.
When the failed host comes back up, if it was previously configured as the master,
the VIP may become active on two hosts. To prevent this, Keepalived is disabled by default
and started within the service loop.
Fixes: https://tracker.ceph.com/issues/76304
Signed-off-by: Shweta Bhosale <Shweta.Bhosale1@ibm.com>
cephadm_agent.deploy_daemon_unit(config_js)
else:
if c:
- # Disable automatic startup for NFS daemons
- enable_daemon = daemon_type != 'nfs'
+ # Disable automatic systemd enable for NFS and keepalived; the mgr
+ # starts them when appropriate (see cephadm serve / DISABLED_SERVICES).
+ enable_daemon = daemon_type not in ('nfs', 'keepalived')
deploy_daemon_units(
ctx,
ident,
from mgr_util import format_bytes
from cephadm.services.service_registry import service_registry
from cephadm.services.nfs import NFSService
+from cephadm.services.ingress import IngressService
from . import utils
from . import exchange
logger = logging.getLogger(__name__)
REQUIRES_POST_ACTIONS = ['grafana', 'iscsi', 'prometheus', 'alertmanager', 'rgw', 'nvmeof', 'mgmt-gateway']
-DISABLED_SERVICES = ['nfs']
+# Daemons not systemd-enabled on deploy, mgr start them when not user stopped
+DISABLED_SERVICES = ['nfs', 'keepalived']
CEPHADM_EXE = ssh.RemoteExecutable('/usr/bin/cephadm')
highest_gen_daemon = max(same_rank_daemons,
key=lambda d: d.rank_generation or 0)
should_start = (dd == highest_gen_daemon)
+ elif dd.daemon_type == 'keepalived' and spec is not None:
+ should_start = IngressService.keepalived_should_auto_start(
+ self.mgr, dd, spec
+ )
else:
should_start = True
if should_start:
from ceph.deployment.utils import is_ipv6
from mgr_util import build_url
from cephadm import utils
-from orchestrator import OrchestratorError, DaemonDescription
+from orchestrator import OrchestratorError, DaemonDescription, DaemonDescriptionStatus
from cephadm.services.cephadmservice import CephadmDaemonDeploySpec, CephService, CephadmService
from .service_registry import register_cephadm_service
from cephadm.tlsobject_types import TLSCredentials
return daemon_spec
+ @staticmethod
+ def _ingress_keepalived_required_count(
+ mgr: "CephadmOrchestrator",
+ ispec: IngressSpec,
+ ) -> int:
+ """
+ get keepalived slot count from ingress placement only
+ """
+ return ispec.placement.get_target_count(mgr.cache.get_schedulable_hosts())
+
+ @staticmethod
+ def keepalived_should_auto_start(
+ mgr: "CephadmOrchestrator",
+ dd: DaemonDescription,
+ spec: ServiceSpec,
+ ) -> bool:
+ """
+ Whether a stopped keepalived unit should be started by the serve loop.
+ Does not auto-start while more keepalived daemons exist than the placement target
+ """
+ ispec = cast(IngressSpec, spec)
+ ingress_daemons = mgr.cache.get_daemons_by_service(ispec.service_name())
+ keepalived_total = len(
+ [d for d in ingress_daemons if d.daemon_type == 'keepalived']
+ )
+ required = IngressService._ingress_keepalived_required_count(mgr, ispec)
+ if keepalived_total > required:
+ return False
+ if ispec.keepalive_only:
+ if not ispec.backend_service:
+ return False
+ for d in mgr.cache.get_daemons_by_service(ispec.backend_service):
+ if d.hostname != dd.hostname:
+ continue
+ if d.status in (
+ DaemonDescriptionStatus.running,
+ DaemonDescriptionStatus.starting,
+ ):
+ return True
+ return False
+ for d in mgr.cache.get_daemons_by_service(spec.service_name()):
+ if d.daemon_type != 'haproxy' or d.hostname != dd.hostname:
+ continue
+ if d.status in (
+ DaemonDescriptionStatus.running,
+ DaemonDescriptionStatus.starting,
+ ):
+ return True
+ return False
+
@staticmethod
def get_keepalived_dependencies(mgr: "CephadmOrchestrator", spec: Optional[ServiceSpec]) -> List[str]:
# because cephadm creates new daemon instances whenever
from cephadm.tests.fixtures import with_host, with_service, async_side_effect
from ceph.utils import datetime_now
-from orchestrator._interface import DaemonDescription
+from orchestrator._interface import DaemonDescription, DaemonDescriptionStatus
+from orchestrator import HostSpec
+from cephadm.services.ingress import IngressService
class TestIngressService:
)
ganesha_conf = nfs_generated_conf['files']['ganesha.conf']
assert "Bind_addr = 10.10.2.20" in ganesha_conf
+
+
+def test_keepalived_should_auto_start_colocated_haproxy():
+ mgr = MagicMock()
+ mgr.cache.get_schedulable_hosts.return_value = [HostSpec('host-a')]
+ spec = IngressSpec(
+ service_id='test',
+ backend_service='rgw.foo',
+ frontend_port=8080,
+ monitor_port=8999,
+ virtual_ip='192.168.1.50/24',
+ placement=PlacementSpec(hosts=['host-a']),
+ )
+ k = DaemonDescription(
+ daemon_type='keepalived',
+ daemon_id='test',
+ hostname='host-a',
+ service_name='ingress.test',
+ )
+ hp = DaemonDescription(
+ daemon_type='haproxy',
+ daemon_id='test',
+ hostname='host-a',
+ service_name='ingress.test',
+ status=DaemonDescriptionStatus.running,
+ ports=[8080, 8999],
+ )
+ hp_stopped = DaemonDescription(
+ daemon_type='haproxy',
+ daemon_id='test',
+ hostname='host-a',
+ service_name='ingress.test',
+ status=DaemonDescriptionStatus.stopped,
+ ports=[8080, 8999],
+ )
+
+ mgr.cache.get_daemons_by_service.side_effect = lambda n: [] if n == 'ingress.test' else []
+ assert not IngressService.keepalived_should_auto_start(mgr, k, spec)
+
+ mgr.cache.get_daemons_by_service.side_effect = lambda n: [hp, k] if n == 'ingress.test' else []
+ assert IngressService.keepalived_should_auto_start(mgr, k, spec)
+
+ mgr.cache.get_daemons_by_service.side_effect = (
+ lambda n: [hp_stopped, k] if n == 'ingress.test' else []
+ )
+ assert not IngressService.keepalived_should_auto_start(mgr, k, spec)
+
+
+def test_keepalived_should_auto_start_false_when_more_kv_than_required():
+ mgr = MagicMock()
+ mgr.cache.get_schedulable_hosts.return_value = [
+ HostSpec('host-a'),
+ HostSpec('host-b'),
+ HostSpec('host-c'),
+ ]
+ spec = IngressSpec(
+ service_id='test',
+ backend_service='rgw.foo',
+ frontend_port=8080,
+ monitor_port=8999,
+ virtual_ip='192.168.1.50/24',
+ placement=PlacementSpec(hosts=['host-a', 'host-b']),
+ )
+ hp_a = DaemonDescription(
+ daemon_type='haproxy',
+ daemon_id='test',
+ hostname='host-a',
+ service_name='ingress.test',
+ status=DaemonDescriptionStatus.running,
+ ports=[8080, 8999],
+ )
+ hp_b = DaemonDescription(
+ daemon_type='haproxy',
+ daemon_id='test',
+ hostname='host-b',
+ service_name='ingress.test',
+ status=DaemonDescriptionStatus.running,
+ ports=[8080, 8999],
+ )
+ kv_a = DaemonDescription(
+ daemon_type='keepalived',
+ daemon_id='test',
+ hostname='host-a',
+ service_name='ingress.test',
+ )
+ kv_b = DaemonDescription(
+ daemon_type='keepalived',
+ daemon_id='test',
+ hostname='host-b',
+ service_name='ingress.test',
+ )
+ kv_extra = DaemonDescription(
+ daemon_type='keepalived',
+ daemon_id='test',
+ hostname='host-c',
+ service_name='ingress.test',
+ )
+ mgr.cache.get_daemons_by_service.side_effect = (
+ lambda n: [hp_a, hp_b, kv_a, kv_b, kv_extra] if n == 'ingress.test' else []
+ )
+ assert not IngressService.keepalived_should_auto_start(mgr, kv_extra, spec)
+
+
+def test_keepalived_should_auto_start_keepalive_only_backend():
+ mgr = MagicMock()
+ mgr.cache.get_schedulable_hosts.return_value = [HostSpec('host1')]
+ spec = IngressSpec(
+ service_id='test',
+ backend_service='nfs.foo',
+ frontend_port=2049,
+ monitor_port=9049,
+ virtual_ip='192.168.122.100/24',
+ keepalive_only=True,
+ placement=PlacementSpec(hosts=['host1']),
+ )
+ k = DaemonDescription(
+ daemon_type='keepalived',
+ daemon_id='keepalived.test',
+ hostname='host1',
+ service_name='ingress.test',
+ )
+ nfs_d = DaemonDescription(
+ daemon_type='nfs',
+ daemon_id='foo.host1',
+ hostname='host1',
+ service_name='nfs.foo',
+ status=DaemonDescriptionStatus.starting,
+ )
+
+ mgr.cache.get_daemons_by_service.side_effect = lambda n: []
+ assert not IngressService.keepalived_should_auto_start(mgr, k, spec)
+
+ def _with_nfs(name: str):
+ if name == 'ingress.test':
+ return [k]
+ if name == 'nfs.foo':
+ return [nfs_d]
+ return []
+
+ mgr.cache.get_daemons_by_service.side_effect = _with_nfs
+ assert IngressService.keepalived_should_auto_start(mgr, k, spec)
import contextlib
+from typing import cast
from unittest.mock import MagicMock, patch, ANY
import pytest
+from ceph.utils import datetime_now
+from orchestrator import DaemonDescriptionStatus
+
+from cephadm.serve import CephadmServe
from cephadm.services.service_registry import service_registry
from cephadm.services.cephadmservice import CephadmDaemonDeploySpec
from cephadm.module import CephadmOrchestrator
# _daemon_action so we can check what action was chosen
mock_cephadm.serve(cephadm_module)._check_daemons()
mock_cephadm._daemon_action.assert_called_with(ANY, action="redeploy")
+
+
+@patch("cephadm.services.nfs.NFSService.run_grace_tool", MagicMock())
+@patch("cephadm.services.nfs.NFSService.purge", MagicMock())
+@patch("cephadm.services.nfs.NFSService.create_rados_config_obj", MagicMock())
+def test_check_daemons_starts_keepalived_when_stopped_and_haproxy_running(
+ cephadm_module, mock_cephadm
+):
+ """Regression: serve loop must issue 'start' for keepalived when deps are ok
+ and colocated haproxy is running (keepalived is not systemd-enabled)."""
+ nfs_spec = NFSServiceSpec(
+ service_id="foo",
+ placement=PlacementSpec(hosts=['test']),
+ port=8765,
+ )
+ ingress_spec = IngressSpec(
+ service_id='bar',
+ backend_service='nfs.foo',
+ frontend_port=2468,
+ monitor_port=8642,
+ virtual_ip='1.2.3.0/24',
+ placement=PlacementSpec(hosts=['test']),
+ keepalived_password='abcde',
+ )
+ with contextlib.ExitStack() as stack:
+ stack.enter_context(with_host(cephadm_module, "test"))
+ cephadm_module.cache.update_host_networks(
+ 'test',
+ {
+ '1.2.3.0/24': {
+ 'if0': [
+ '1.2.3.4',
+ '1.2.3.1',
+ ]
+ }
+ },
+ )
+ stack.enter_context(with_service(cephadm_module, nfs_spec))
+ stack.enter_context(with_service(cephadm_module, ingress_spec))
+
+ ingress_svc = service_registry.get_service('ingress')
+ ispec = cast(IngressSpec, cephadm_module.spec_store[ingress_spec.service_name()].spec)
+
+ for dd in list(cephadm_module.cache.get_daemons_by_service(ingress_spec.service_name())):
+ if dd.daemon_type == 'haproxy' and dd.hostname == 'test':
+ dd.status = DaemonDescriptionStatus.running
+ elif dd.daemon_type == 'keepalived' and dd.hostname == 'test':
+ dd.status = DaemonDescriptionStatus.stopped
+ assert dd.hostname is not None
+ deps = ingress_svc.sorted_dependencies(
+ cephadm_module, ispec, 'keepalived'
+ )
+ cephadm_module.cache.update_daemon_config_deps(
+ dd.hostname, dd.name(), deps, datetime_now()
+ )
+
+ mock_cephadm._daemon_action.reset_mock()
+ CephadmServe(cephadm_module)._check_daemons()
+
+ keepalived_started = any(
+ getattr(c.args[0], 'daemon_type', None) == 'keepalived'
+ and (len(c.args) > 1 and c.args[1] == 'start' or c.kwargs.get('action') == 'start')
+ for c in mock_cephadm._daemon_action.call_args_list
+ )
+ assert keepalived_started
projects / ceph.git / commitdiff