diff options
Diffstat (limited to 'qpid/java/bdbstore/src')
5 files changed, 399 insertions, 36 deletions
diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBSystemConfigImpl.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBSystemConfigImpl.java index 0fc44605fe..5d65d6e16d 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBSystemConfigImpl.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/BDBSystemConfigImpl.java @@ -26,6 +26,7 @@ import org.apache.qpid.server.logging.EventLogger; import org.apache.qpid.server.logging.LogRecorder; import org.apache.qpid.server.model.AbstractSystemConfig; import org.apache.qpid.server.model.Broker; +import org.apache.qpid.server.model.BrokerShutdownProvider; import org.apache.qpid.server.model.ManagedAttributeField; import org.apache.qpid.server.model.ManagedObject; import org.apache.qpid.server.model.SystemConfigFactoryConstructor; @@ -48,9 +49,10 @@ public class BDBSystemConfigImpl extends AbstractSystemConfig<BDBSystemConfigImp public BDBSystemConfigImpl(final TaskExecutor taskExecutor, final EventLogger eventLogger, final LogRecorder logRecorder, - final BrokerOptions brokerOptions) + final BrokerOptions brokerOptions, + final BrokerShutdownProvider brokerShutdownProvider) { - super(taskExecutor, eventLogger, logRecorder, brokerOptions); + super(taskExecutor, eventLogger, logRecorder, brokerOptions, brokerShutdownProvider); } @Override diff --git a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java index e6109954c0..98b9cc3cf0 100644 --- a/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java +++ b/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBHAVirtualHostNodeImpl.java @@ -20,7 +20,9 @@ */ package org.apache.qpid.server.virtualhostnode.berkeleydb; +import java.io.File; import java.net.InetSocketAddress; +import java.nio.file.Files; import java.security.PrivilegedAction; import java.text.MessageFormat; import java.util.ArrayList; @@ -74,6 +76,7 @@ import org.apache.qpid.server.store.berkeleydb.BDBConfigurationStore; import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacade; import org.apache.qpid.server.store.berkeleydb.replication.ReplicatedEnvironmentFacadeFactory; import org.apache.qpid.server.store.berkeleydb.replication.ReplicationGroupListener; +import org.apache.qpid.server.util.PortUtil; import org.apache.qpid.server.util.ServerScopedRuntimeException; import org.apache.qpid.server.virtualhost.berkeleydb.BDBHAVirtualHostImpl; import org.apache.qpid.server.virtualhostnode.AbstractVirtualHostNode; @@ -280,20 +283,6 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu public void onCreate() { super.onCreate(); - if (!isFirstNodeInAGroup()) - { - try - { - int dbPingSocketTimeout = getContextKeys(false).contains("qpid.bdb.ha.db_ping_socket_timeout") ? getContextValue(Integer.class, "qpid.bdb.ha.db_ping_socket_timeout") : 10000 /* JE's own default */; - Collection<String> permittedNodes = ReplicatedEnvironmentFacade.connectToHelperNodeAndCheckPermittedHosts(getName(), getAddress(), getGroupName(), getHelperNodeName(), getHelperAddress(), dbPingSocketTimeout); - setAttribute(PERMITTED_NODES, null, new ArrayList<String>(permittedNodes)); - } - catch(IllegalConfigurationException e) - { - deleted(); - throw e; - } - } getEventLogger().message(getVirtualHostNodeLogSubject(), HighAvailabilityMessages.CREATED()); } @@ -435,19 +424,102 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu public void onValidate() { super.onValidate(); + validatePermittedNodes(_permittedNodes); + } + + @Override + protected void postResolve() + { + super.postResolve(); _virtualHostNodeLogSubject = new BDBHAVirtualHostNodeLogSubject(getGroupName(), getName()); _groupLogSubject = new GroupLogSubject(getGroupName()); _virtualHostNodePrincipalName = MessageFormat.format(VIRTUAL_HOST_PRINCIPAL_NAME_FORMAT, getGroupName(), getName()); } @Override - public void onOpen() + public void validateOnCreate() { - super.onOpen(); + super.validateOnCreate(); - validatePermittedNodes(_permittedNodes); + validateAddress(); + + validateStorePath(); + + if (!isFirstNodeInAGroup()) + { + // validate that helper address points to valid node + // we need _permittedNodes for the further validation in onValidate + _permittedNodes = new ArrayList<>(getPermittedNodesFromHelper()); + } + } + + private Collection<String> getPermittedNodesFromHelper() + { + int dbPingSocketTimeout = getContextKeys(false).contains("qpid.bdb.ha.db_ping_socket_timeout") ? getContextValue(Integer.class, "qpid.bdb.ha.db_ping_socket_timeout") : 10000 /* JE's own default */; + return ReplicatedEnvironmentFacade.connectToHelperNodeAndCheckPermittedHosts(getName(), getAddress(), getGroupName(), getHelperNodeName(), getHelperAddress(), dbPingSocketTimeout); + } + + private void validateStorePath() + { + File storePath = new File(getStorePath()); + while (!storePath.exists()) + { + storePath = storePath.getParentFile(); + if (storePath == null) + { + throw new IllegalConfigurationException(String.format("Store path '%s' is invalid", getStorePath())); + } + } + + if (!storePath.isDirectory()) + { + throw new IllegalConfigurationException(String.format("Store path '%s' is not a folder", getStorePath())); + } + + if (!Files.isWritable(storePath.toPath())) + { + throw new IllegalConfigurationException(String.format("Store path '%s' is not writable", getStorePath())); + } } + private void validateAddress() + { + String address = getAddress(); + + if (address == null || "".equals(address)) + { + throw new IllegalConfigurationException("Node address is not set"); + } + + String[] tokens = address.split(":"); + if (tokens.length != 2) + { + throw new IllegalConfigurationException(String.format("Invalid address specified '%s'. ", address)); + } + + String hostName = tokens[0]; + if ("".equals(hostName.trim())) + { + throw new IllegalConfigurationException(String.format("Invalid address specified '%s'. ", address)); + } + + int port = -1; + try + { + port = Integer.parseInt(tokens[1]); + } + catch(Exception e) + { + throw new IllegalConfigurationException(String.format("Invalid port is specified in address '%s'. ", address)); + } + if (!PortUtil.isPortAvailable(hostName, port)) + { + throw new IllegalConfigurationException(String.format("Cannot bind to address '%s'. Address is already in use.", address)); + } + } + + + private void onMaster() { try @@ -761,7 +833,7 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu } } - private void validatePermittedNodes(List<String> proposedPermittedNodes) + private void validatePermittedNodes(Collection<String> proposedPermittedNodes) { if (getRemoteReplicationNodes().size() > 0 && getRole() != NodeRole.MASTER && !(getState() == State.STOPPED || getState() == State.ERRORED)) { @@ -772,28 +844,32 @@ public class BDBHAVirtualHostNodeImpl extends AbstractVirtualHostNode<BDBHAVirtu throw new IllegalArgumentException(String.format("Attribute '%s' is mandatory and must be set", PERMITTED_NODES)); } - String missingNodeAddress = null; - if (getPermittedNodes().contains(getAddress()) && !proposedPermittedNodes.contains(getAddress())) + if (_permittedNodes != null) { - missingNodeAddress = getAddress(); - } - else - { - for (final RemoteReplicationNode<?> node : getRemoteReplicationNodes()) + String missingNodeAddress = null; + + if (_permittedNodes.contains(getAddress()) && !proposedPermittedNodes.contains(getAddress())) { - final BDBHARemoteReplicationNode<?> bdbHaRemoteReplicationNode = (BDBHARemoteReplicationNode<?>) node; - final String remoteNodeAddress = bdbHaRemoteReplicationNode.getAddress(); - if (getPermittedNodes().contains(remoteNodeAddress) && !proposedPermittedNodes.contains(remoteNodeAddress)) + missingNodeAddress = getAddress(); + } + else + { + for (final RemoteReplicationNode<?> node : getRemoteReplicationNodes()) { - missingNodeAddress = remoteNodeAddress; - break; + final BDBHARemoteReplicationNode<?> bdbHaRemoteReplicationNode = (BDBHARemoteReplicationNode<?>) node; + final String remoteNodeAddress = bdbHaRemoteReplicationNode.getAddress(); + if (_permittedNodes.contains(remoteNodeAddress) && !proposedPermittedNodes.contains(remoteNodeAddress)) + { + missingNodeAddress = remoteNodeAddress; + break; + } } } - } - if (missingNodeAddress != null) - { - throw new IllegalArgumentException(String.format("The current group node '%s' cannot be removed from '%s' as its already a group member", missingNodeAddress, PERMITTED_NODES)); + if (missingNodeAddress != null) + { + throw new IllegalArgumentException(String.format("The current group node '%s' cannot be removed from '%s' as its already a group member", missingNodeAddress, PERMITTED_NODES)); + } } for (String permittedNode: proposedPermittedNodes) diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java index e9bcc5d754..16486e3564 100644 --- a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/store/berkeleydb/BDBHAVirtualHostNodeTest.java @@ -23,6 +23,8 @@ package org.apache.qpid.server.store.berkeleydb; import static org.mockito.Mockito.when; import java.io.File; +import java.net.InetSocketAddress; +import java.net.ServerSocket; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -53,6 +55,8 @@ import org.apache.qpid.server.virtualhostnode.berkeleydb.BDBHAVirtualHostNodeTes import org.apache.qpid.server.virtualhostnode.berkeleydb.NodeRole; import org.apache.qpid.test.utils.PortHelper; import org.apache.qpid.test.utils.QpidTestCase; +import org.apache.qpid.test.utils.TestFileUtils; +import org.apache.qpid.util.FileUtils; public class BDBHAVirtualHostNodeTest extends QpidTestCase { @@ -587,4 +591,87 @@ public class BDBHAVirtualHostNodeTest extends QpidTestCase assertTrue("Intruder protection was not triggered during expected timeout", stopLatch.await(20, TimeUnit.SECONDS)); } + public void testValidateOnCreateForNonExistingHelperNode() throws Exception + { + int node1PortNumber = findFreePort(); + int node2PortNumber = getNextAvailable(node1PortNumber + 1); + + + Map<String, Object> attributes = _helper.createNodeAttributes("node1", "group", "localhost:" + node1PortNumber, + "localhost:" + node2PortNumber, "node2", node1PortNumber, node1PortNumber, node2PortNumber); + try + { + _helper.createAndStartHaVHN(attributes); + fail("Node creation should fail because of invalid helper address"); + } + catch(IllegalConfigurationException e) + { + assertEquals("Unexpected exception on connection to non-existing helper address", + String.format("Cannot connect to '%s'", "localhost:" + node2PortNumber), e.getMessage()); + } + } + + public void testValidateOnCreateForAlreadyBoundAddress() throws Exception + { + int node1PortNumber = findFreePort(); + + ServerSocket serverSocket = null; + try + { + serverSocket = new ServerSocket(); + serverSocket.setReuseAddress(true); + serverSocket.bind(new InetSocketAddress("localhost", node1PortNumber)); + + + Map<String, Object> attributes = _helper.createNodeAttributes("node1", "group", "localhost:" + node1PortNumber, + "localhost:" + node1PortNumber, "node2", node1PortNumber, node1PortNumber); + try + { + _helper.createAndStartHaVHN(attributes); + fail("Node creation should fail because of invalid address"); + } + catch(IllegalConfigurationException e) + { + assertEquals("Unexpected exception on attempt to create node with already bound address", + String.format("Cannot bind to address '%s'. Address is already in use.", "localhost:" + node1PortNumber), e.getMessage()); + } + } + finally + { + if (serverSocket != null) + { + serverSocket.close(); + } + } + } + + public void testValidateOnCreateForInvalidStorePath() throws Exception + { + int node1PortNumber = findFreePort(); + + File storeBaseFolder = TestFileUtils.createTestDirectory(); + File file = new File(storeBaseFolder, getTestName()); + file.createNewFile(); + File storePath = new File(file, "test"); + try + { + Map<String, Object> attributes = _helper.createNodeAttributes("node1", "group", "localhost:" + node1PortNumber, + "localhost:" + node1PortNumber, "node2", node1PortNumber, node1PortNumber); + attributes.put(BDBHAVirtualHostNode.STORE_PATH, storePath.getAbsoluteFile()); + try + { + _helper.createAndStartHaVHN(attributes); + fail("Node creation should fail because of invalid store path"); + } + catch (IllegalConfigurationException e) + { + assertEquals("Unexpected exception on attempt to create environment in invalid location", + String.format("Store path '%s' is not a folder", storePath.getAbsoluteFile()), e.getMessage()); + } + } + finally + { + FileUtils.delete(storeBaseFolder, true); + } + } } diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHostImplTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHostImplTest.java new file mode 100644 index 0000000000..92da465dbd --- /dev/null +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBVirtualHostImplTest.java @@ -0,0 +1,106 @@ +/* + * + * 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. + * + */ +package org.apache.qpid.server.virtualhost.berkeleydb; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +import org.apache.qpid.server.configuration.IllegalConfigurationException; +import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor; +import org.apache.qpid.server.configuration.updater.TaskExecutor; +import org.apache.qpid.server.model.Broker; +import org.apache.qpid.server.model.BrokerModel; +import org.apache.qpid.server.model.VirtualHostNode; +import org.apache.qpid.server.store.DurableConfigurationStore; +import org.apache.qpid.server.util.BrokerTestHelper; +import org.apache.qpid.test.utils.QpidTestCase; +import org.apache.qpid.test.utils.TestFileUtils; +import org.apache.qpid.util.FileUtils; + +public class BDBVirtualHostImplTest extends QpidTestCase +{ + private File _storePath; + private VirtualHostNode<?> _node; + + @Override + public void setUp() throws Exception + { + super.setUp(); + Broker broker = BrokerTestHelper.createBrokerMock(); + + TaskExecutor taskExecutor = CurrentThreadTaskExecutor.newStartedInstance(); + when(broker.getTaskExecutor()).thenReturn(taskExecutor); + + _storePath = TestFileUtils.createTestDirectory(); + + _node = mock(VirtualHostNode.class); + when(_node.getParent(Broker.class)).thenReturn(broker); + when(_node.getModel()).thenReturn(BrokerModel.getInstance()); + when(_node.getTaskExecutor()).thenReturn(taskExecutor); + when(_node.getConfigurationStore()).thenReturn(mock(DurableConfigurationStore.class)); + when(_node.getId()).thenReturn(UUID.randomUUID()); + } + + @Override + public void tearDown() throws Exception + { + try + { + if (_storePath != null) + { + FileUtils.delete(_storePath, true); + } + } + finally + { + super.tearDown(); + } + } + + public void testValidateOnCreateForInvalidStorePath() throws Exception + { + String hostName = getTestName(); + File file = new File(_storePath + File.separator + hostName); + assertTrue("Empty file is not created", file.createNewFile()); + Map<String, Object> attributes = new HashMap<>(); + attributes.put(BDBVirtualHost.ID, UUID.randomUUID()); + attributes.put(BDBVirtualHost.TYPE, BDBVirtualHostImpl.VIRTUAL_HOST_TYPE); + attributes.put(BDBVirtualHost.NAME, hostName); + attributes.put(BDBVirtualHost.STORE_PATH, file.getAbsoluteFile()); + + BDBVirtualHostImpl host = new BDBVirtualHostImpl(attributes, _node); + try + { + host.create(); + fail("Cannot create DBD virtual host from existing empty file"); + } + catch (IllegalConfigurationException e) + { + assertTrue("Unexpected exception " + e.getMessage(), e.getMessage().startsWith("Cannot open virtual host message store")); + } + } + +} diff --git a/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBVirtualHostNodeTest.java b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBVirtualHostNodeTest.java new file mode 100644 index 0000000000..6608312088 --- /dev/null +++ b/qpid/java/bdbstore/src/test/java/org/apache/qpid/server/virtualhostnode/berkeleydb/BDBVirtualHostNodeTest.java @@ -0,0 +1,92 @@ +/* + * + * 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. + * + */ +package org.apache.qpid.server.virtualhostnode.berkeleydb; + +import static org.mockito.Mockito.when; + +import java.io.File; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +import org.apache.qpid.server.configuration.IllegalConfigurationException; +import org.apache.qpid.server.configuration.updater.CurrentThreadTaskExecutor; +import org.apache.qpid.server.model.Broker; +import org.apache.qpid.server.util.BrokerTestHelper; +import org.apache.qpid.test.utils.QpidTestCase; +import org.apache.qpid.test.utils.TestFileUtils; +import org.apache.qpid.util.FileUtils; + +public class BDBVirtualHostNodeTest extends QpidTestCase +{ + private Broker<?> _broker; + private File _storePath; + + @Override + public void setUp() throws Exception + { + super.setUp(); + _broker = BrokerTestHelper.createBrokerMock(); + when(_broker.getTaskExecutor()).thenReturn(CurrentThreadTaskExecutor.newStartedInstance()); + + _storePath = TestFileUtils.createTestDirectory(); + } + + @Override + public void tearDown() throws Exception + { + try + { + if (_storePath != null) + { + FileUtils.delete(_storePath, true); + } + } + finally + { + super.tearDown(); + } + } + + public void testValidateOnCreateForInvalidStorePath() throws Exception + { + String nodeName = getTestName(); + File file = new File(_storePath + File.separator + nodeName); + assertTrue("Empty file is not created", file.createNewFile()); + Map<String, Object> attributes = new HashMap<>(); + attributes.put(BDBVirtualHostNode.ID, UUID.randomUUID()); + attributes.put(BDBVirtualHostNode.TYPE, BDBVirtualHostNodeImpl.VIRTUAL_HOST_NODE_TYPE); + attributes.put(BDBVirtualHostNode.NAME, nodeName); + attributes.put(BDBVirtualHostNode.STORE_PATH, file.getAbsolutePath()); + + BDBVirtualHostNodeImpl node = new BDBVirtualHostNodeImpl(attributes, _broker); + try + { + node.create(); + fail("Cannot create DBD node from existing empty file"); + } + catch (IllegalConfigurationException e) + { + assertTrue("Unexpected exception " + e.getMessage(), e.getMessage().startsWith("Cannot open node configuration store")); + } + } + +} |