From c64a431832224798189f43ffa55b24c2ddbd480e Mon Sep 17 00:00:00 2001 From: Tigran Mkrtchyan Date: Mon, 12 Oct 2026 15:16:51 +0200 Subject: [PATCH] cells: announce correct domain role on outbound core tunnels Motivation: Since commit 5021255d1d ("cells: simplify initialization of LocationManagerConnector") the connector is created without cell arguments. However, those arguments were forwarded to LocationMgrTunnel, which reads the `role` option passed from batch files. Thus, it always fall-backed to SATELLITE. As a result, a core domain that connects to another core domain announces itself as a satellite. The connecting core installs a default route to its peer, and the accepting core allows forwarding of messages received through that tunnel. Messages published to a topic are then delivered by every core domain along the default route chain, e.g. billing records show up two or three times in setups with multiple core domains. Modification: Pass the domain role explicitly to LocationManagerConnector and from there to LocationMgrTunnel through a dedicated constructor, instead of relying on forwarded cell arguments. Add unit test to cover the regression. Result: Core domains no longer install default routes to other core domains and messages sent to a topic are delivered only once. No more duplicate billing records in setups with multiple core domains. Issue: #8008 Acked-by: Marina Sahakyan Target: master, 12.0, 11.2, 11.1, 11.0 Require-book: no Require-notes: yes (cherry picked from commit 3c3ff60890209e805ef8ec62dd79bc623ba8fead) Signed-off-by: Tigran Mkrtchyan --- .../network/LocationManagerConnector.java | 12 ++- .../dmg/cells/network/LocationMgrTunnel.java | 24 ++++- .../dmg/cells/services/LocationManager.java | 3 +- .../network/LocationManagerConnectorTest.java | 72 ++++++++++++++ .../cells/network/LocationMgrTunnelTest.java | 93 +++++++++++++++++++ .../dmg/cells/network/TestCellDomain.java | 28 ++++++ 6 files changed, 223 insertions(+), 9 deletions(-) create mode 100644 modules/cells/src/test/java/dmg/cells/network/LocationManagerConnectorTest.java create mode 100644 modules/cells/src/test/java/dmg/cells/network/LocationMgrTunnelTest.java create mode 100644 modules/cells/src/test/java/dmg/cells/network/TestCellDomain.java diff --git a/modules/cells/src/main/java/dmg/cells/network/LocationManagerConnector.java b/modules/cells/src/main/java/dmg/cells/network/LocationManagerConnector.java index c204123622b..ba534e3dab0 100644 --- a/modules/cells/src/main/java/dmg/cells/network/LocationManagerConnector.java +++ b/modules/cells/src/main/java/dmg/cells/network/LocationManagerConnector.java @@ -6,6 +6,7 @@ import com.google.common.net.HostAndPort; import com.google.common.util.concurrent.Uninterruptibles; import dmg.cells.nucleus.CellAdapter; +import dmg.cells.nucleus.CellDomainRole; import dmg.util.DummyStreamEngine; import dmg.util.Exceptions; import dmg.util.StreamEngine; @@ -22,7 +23,6 @@ import javax.net.SocketFactory; import org.dcache.alarms.AlarmMarkerFactory; import org.dcache.alarms.PredefinedAlarm; -import org.dcache.util.Args; import org.dcache.util.NDC; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,6 +37,7 @@ public class LocationManagerConnector private final String _domain; private final SocketFactory _ssf; private final InetSocketAddress _address; + private final CellDomainRole _role; private Thread _thread; private volatile String _status = "disconnected"; private volatile int _retries; @@ -48,12 +49,14 @@ public class LocationManagerConnector * @param socketFactory the socket factory to use for creating connections * @param remoteDomain the domain name of the location manager * @param endpoint the address of the location manager + * @param role the role of the local domain, announced to the remote domain during tunnel handshake */ - public LocationManagerConnector(String cellName, SocketFactory socketFactory, String remoteDomain, InetSocketAddress endpoint) { - super(cellName, "System"); + public LocationManagerConnector(String cellName, SocketFactory socketFactory, String remoteDomain, InetSocketAddress endpoint, CellDomainRole role) { + super(cellName, "System", ""); _domain = remoteDomain; _ssf = requireNonNull(socketFactory); _address = endpoint; + _role = requireNonNull(role); } @Override @@ -89,7 +92,6 @@ public void run() { * 4 to 20 seconds between connection attempts. The thread is * terminated by interrupting it. */ - Args args = getArgs(); String name = getCellName() + '*'; Random random = new Random(); NDC.push(_address.toString()); @@ -98,7 +100,7 @@ public void run() { try { _retries++; - LocationMgrTunnel tunnel = new LocationMgrTunnel(name, connect(), args); + LocationMgrTunnel tunnel = new LocationMgrTunnel(name, connect(), _role); try { tunnel.start().get(); _retries = 0; diff --git a/modules/cells/src/main/java/dmg/cells/network/LocationMgrTunnel.java b/modules/cells/src/main/java/dmg/cells/network/LocationMgrTunnel.java index 09b62789542..aed8f77ef0c 100644 --- a/modules/cells/src/main/java/dmg/cells/network/LocationMgrTunnel.java +++ b/modules/cells/src/main/java/dmg/cells/network/LocationMgrTunnel.java @@ -1,6 +1,6 @@ /* dCache - http://www.dcache.org/ * - * Copyright (C) 2001 - 2018 Deutsches Elektronen-Synchrotron + * Copyright (C) 2001 - 2026 Deutsches Elektronen-Synchrotron * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as @@ -17,6 +17,8 @@ */ package dmg.cells.network; +import static java.util.Objects.requireNonNull; + import dmg.cells.nucleus.CellAdapter; import dmg.cells.nucleus.CellDomainInfo; import dmg.cells.nucleus.CellDomainRole; @@ -91,13 +93,29 @@ public class LocationMgrTunnel private LongAdder _messagesToSystem = new LongAdder(); public LocationMgrTunnel(String cellName, StreamEngine engine, Args args) { + this(cellName, engine, args, CellDomainRole.valueOf( + args.getOption("role", CellDomainRole.SATELLITE.name()).toUpperCase())); + } + + /** + * Create a new tunnel. + * @param cellName the name of this tunnel as a cell + * @param engine the stream engine of the established connection + * @param role the role of the local domain, announced to the remote domain during handshake + */ + public LocationMgrTunnel(String cellName, StreamEngine engine, CellDomainRole role) { + this(cellName, engine, new Args(""), role); + } + + private LocationMgrTunnel(String cellName, StreamEngine engine, Args args, + CellDomainRole role) { super(cellName, "System", args); + + requireNonNull(role); _nucleus = getNucleus(); _socket = engine.getSocket(); _rawOut = new BufferedOutputStream(engine.getOutputStream()); _rawIn = new BufferedInputStream(engine.getInputStream()); - CellDomainRole role = args.hasOption("role") ? CellDomainRole.valueOf( - args.getOption("role").toUpperCase()) : CellDomainRole.SATELLITE; _localDomainInfo = new CellDomainInfo(_nucleus.getCellDomainName(), Version.of(LocationMgrTunnel.class).getVersion(), role, diff --git a/modules/cells/src/main/java/dmg/cells/services/LocationManager.java b/modules/cells/src/main/java/dmg/cells/services/LocationManager.java index 4cd7ce4793b..6651ac7b7ef 100644 --- a/modules/cells/src/main/java/dmg/cells/services/LocationManager.java +++ b/modules/cells/src/main/java/dmg/cells/services/LocationManager.java @@ -719,7 +719,8 @@ private String startConnector(String remoteDomain, CoreDomainInfo domainInfo) String cellName = "c-" + remoteDomain + '*'; LOGGER.info("Starting connector {} to {} ({})", cellName, remoteDomain, where); InetSocketAddress tunnelEndpoint = new InetSocketAddress(where.getHost(), where.getPort()); - LocationManagerConnector c = new LocationManagerConnector(cellName, socketFactory, remoteDomain, tunnelEndpoint); + LocationManagerConnector c = new LocationManagerConnector(cellName, socketFactory, remoteDomain, + tunnelEndpoint, role); c.start().get(); return c.getCellName(); } diff --git a/modules/cells/src/test/java/dmg/cells/network/LocationManagerConnectorTest.java b/modules/cells/src/test/java/dmg/cells/network/LocationManagerConnectorTest.java new file mode 100644 index 00000000000..2bfef27d8eb --- /dev/null +++ b/modules/cells/src/test/java/dmg/cells/network/LocationManagerConnectorTest.java @@ -0,0 +1,72 @@ +package dmg.cells.network; + +import static org.junit.Assert.assertEquals; + +import dmg.cells.nucleus.CellDomainInfo; +import dmg.cells.nucleus.CellDomainRole; +import java.io.ObjectInputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import javax.net.SocketFactory; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class LocationManagerConnectorTest { + + private static final String REMOTE_DOMAIN = "remoteDomain"; + + private ServerSocket _remote; + private LocationManagerConnector _connector; + private TestCellDomain _domain; + + @Before + public void setUp() throws Exception { + _domain = TestCellDomain.LOCAL; + _remote = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + _remote.setSoTimeout(10_000); + } + + @After + public void tearDown() throws Exception { + if (_connector != null) { + _connector.getNucleus().kill(_connector.getCellName()); + } + _remote.close(); + } + + @Test + public void shouldAnnounceCoreRoleToRemoteDomain() throws Exception { + assertEquals(CellDomainRole.CORE, announcedDomainInfo(CellDomainRole.CORE).getRole()); + } + + @Test + public void shouldAnnounceSatelliteRoleToRemoteDomain() throws Exception { + assertEquals(CellDomainRole.SATELLITE, + announcedDomainInfo(CellDomainRole.SATELLITE).getRole()); + } + + @Test + public void shouldAnnounceLocalDomainNameToRemoteDomain() throws Exception { + assertEquals(_domain.getCellDomainName(), + announcedDomainInfo(CellDomainRole.CORE).getCellDomainName()); + } + + /** + * Starts a connector with the given role and returns the domain info the tunnel presents to + * the remote domain during the handshake. + */ + private CellDomainInfo announcedDomainInfo(CellDomainRole role) throws Exception { + _connector = new LocationManagerConnector("c-" + REMOTE_DOMAIN + '*', + SocketFactory.getDefault(), REMOTE_DOMAIN, + new InetSocketAddress(_remote.getInetAddress(), _remote.getLocalPort()), role); + _connector.start().get(); + + try (Socket socket = _remote.accept()) { + socket.setSoTimeout(10_000); + return (CellDomainInfo) new ObjectInputStream(socket.getInputStream()).readObject(); + } + } +} diff --git a/modules/cells/src/test/java/dmg/cells/network/LocationMgrTunnelTest.java b/modules/cells/src/test/java/dmg/cells/network/LocationMgrTunnelTest.java new file mode 100644 index 00000000000..87568cc895e --- /dev/null +++ b/modules/cells/src/test/java/dmg/cells/network/LocationMgrTunnelTest.java @@ -0,0 +1,93 @@ +package dmg.cells.network; + +import static org.junit.Assert.assertEquals; + +import dmg.cells.nucleus.CellDomainInfo; +import dmg.cells.nucleus.CellDomainRole; +import dmg.util.DummyStreamEngine; +import java.io.ObjectInputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import org.dcache.util.Args; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class LocationMgrTunnelTest { + + private ServerSocket _remote; + private Socket _local; + private TestCellDomain _domain; + + @Before + public void setUp() throws Exception { + _domain = TestCellDomain.LOCAL; + _remote = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); + _remote.setSoTimeout(10_000); + _local = new Socket(_remote.getInetAddress(), _remote.getLocalPort()); + } + + @After + public void tearDown() throws Exception { + _local.close(); + _remote.close(); + } + + @Test + public void shouldAnnounceCoreRolePassedAsArgument() throws Exception { + assertEquals(CellDomainRole.CORE, announcedDomainInfo(new Args("-role=core")).getRole()); + } + + @Test + public void shouldAnnounceSatelliteRolePassedAsArgument() throws Exception { + assertEquals(CellDomainRole.SATELLITE, + announcedDomainInfo(new Args("-role=satellite")).getRole()); + } + + @Test + public void shouldIgnoreCaseOfRolePassedAsArgument() throws Exception { + assertEquals(CellDomainRole.CORE, announcedDomainInfo(new Args("-role=CoRe")).getRole()); + } + + @Test + public void shouldAnnounceRoleWhenOtherArgumentsArePresent() throws Exception { + assertEquals(CellDomainRole.CORE, + announcedDomainInfo(new Args("-prot=raw -role=core -lm=lm")).getRole()); + } + + @Test + public void shouldAnnounceSatelliteRoleIfNoRoleArgument() throws Exception { + assertEquals(CellDomainRole.SATELLITE, + announcedDomainInfo(new Args("-prot=raw")).getRole()); + } + + @Test(expected = IllegalArgumentException.class) + public void shouldRejectUnknownRolePassedAsArgument() throws Exception { + new LocationMgrTunnel("tunnel*", new DummyStreamEngine(_local), new Args("-role=bogus")); + } + + @Test + public void shouldAnnounceRolePassedExplicitly() throws Exception { + LocationMgrTunnel tunnel = new LocationMgrTunnel("tunnel*", new DummyStreamEngine(_local), + CellDomainRole.CORE); + assertEquals(CellDomainRole.CORE, announcedDomainInfo(tunnel).getRole()); + } + + private CellDomainInfo announcedDomainInfo(Args args) throws Exception { + return announcedDomainInfo( + new LocationMgrTunnel("tunnel*", new DummyStreamEngine(_local), args)); + } + + /** + * Starts the tunnel and returns the domain info it presents to the remote domain during the + * handshake. + */ + private CellDomainInfo announcedDomainInfo(LocationMgrTunnel tunnel) throws Exception { + tunnel.start(); + try (Socket socket = _remote.accept()) { + socket.setSoTimeout(10_000); + return (CellDomainInfo) new ObjectInputStream(socket.getInputStream()).readObject(); + } + } +} diff --git a/modules/cells/src/test/java/dmg/cells/network/TestCellDomain.java b/modules/cells/src/test/java/dmg/cells/network/TestCellDomain.java new file mode 100644 index 00000000000..3fb46192796 --- /dev/null +++ b/modules/cells/src/test/java/dmg/cells/network/TestCellDomain.java @@ -0,0 +1,28 @@ +package dmg.cells.network; + +import static org.mockito.Mockito.mock; + +import dmg.cells.nucleus.SerializationHandler; +import dmg.cells.nucleus.SystemCell; +import java.util.Optional; +import org.apache.curator.framework.imps.CuratorFrameworkBase; + +/** + * The cell domain used for testing. The cell domain is created when the enum is loaded. + */ +enum TestCellDomain { + + LOCAL("localDomain"); + + private final String name; + + TestCellDomain(String name) { + this.name = name; + SystemCell.create(name, mock(CuratorFrameworkBase.class), Optional.empty(), + SerializationHandler.Serializer.JOS); + } + + String getCellDomainName() { + return name; + } +}