diff --git a/data-node/pom.xml b/data-node/pom.xml index 479516e232..acb99a86fe 100644 --- a/data-node/pom.xml +++ b/data-node/pom.xml @@ -413,6 +413,12 @@ test + + org.mockito + mockito-core + ${mockito.version} + test + io.rest-assured diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java index 8227898471..b258be9392 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/PeriodicalBindings.java @@ -19,11 +19,10 @@ package org.graylog.datanode.bindings; import com.google.inject.AbstractModule; import com.google.inject.multibindings.Multibinder; import org.graylog.datanode.periodicals.ClusterManagerDiscovery; +import org.graylog.datanode.periodicals.NodePingPeriodical; import org.graylog.datanode.periodicals.OpensearchNodeHeartbeat; import org.graylog2.events.ClusterEventCleanupPeriodical; import org.graylog2.events.ClusterEventPeriodical; -import org.graylog2.periodical.NodePingThread; -import org.graylog2.periodical.UserSessionTerminationPeriodical; import org.graylog2.plugin.periodical.Periodical; public class PeriodicalBindings extends AbstractModule { @@ -35,6 +34,6 @@ public class PeriodicalBindings extends AbstractModule { periodicalBinder.addBinding().to(OpensearchNodeHeartbeat.class); periodicalBinder.addBinding().to(ClusterManagerDiscovery.class); // periodicalBinder.addBinding().to(UserSessionTerminationPeriodical.class); -// periodicalBinder.addBinding().to(NodePingThread.class); + periodicalBinder.addBinding().to(NodePingPeriodical.class); } } diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java index 606564d621..1972a5196e 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcess.java @@ -29,6 +29,7 @@ public interface OpensearchProcess extends ManagableProcess { Object nodeName(); + boolean isLeaderNode(); void setLeaderNode(boolean isManagerNode); String opensearchVersion(); diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java index 6e5b9a1b58..2c1494109b 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java @@ -133,6 +133,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { this.isLeaderNode = isLeaderNode; } + @Override public boolean isLeaderNode() { return isLeaderNode; } diff --git a/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java new file mode 100644 index 0000000000..1b19740aa4 --- /dev/null +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java @@ -0,0 +1,108 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.periodicals; + +import org.graylog.datanode.Configuration; +import org.graylog.datanode.management.OpensearchProcess; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.cluster.NodeService; +import org.graylog2.plugin.Tools; +import org.graylog2.plugin.periodical.Periodical; +import org.graylog2.plugin.system.NodeId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; +import javax.inject.Inject; +import java.net.URI; +import java.util.function.Supplier; + +public class NodePingPeriodical extends Periodical { + + private static final Logger LOG = LoggerFactory.getLogger(NodePingPeriodical.class); + private final NodeService nodeService; + private final NodeId nodeId; + private final URI httpPublishUri; + private final Supplier isLeader; + + + @Inject + public NodePingPeriodical(NodeService nodeService, NodeId nodeId, Configuration configuration, OpensearchProcess managedOpenSearch) { + this(nodeService, nodeId, configuration.getHttpPublishUri(), managedOpenSearch::isLeaderNode); + } + + NodePingPeriodical( + NodeService nodeService, + NodeId nodeId, + URI httpPublishUri, + Supplier isLeader + ) { + this.nodeService = nodeService; + this.nodeId = nodeId; + this.httpPublishUri = httpPublishUri; + this.isLeader = isLeader; + } + + @Override + public boolean runsForever() { + return false; + } + + @Override + public boolean stopOnGracefulShutdown() { + return false; + } + + @Override + public boolean startOnThisNode() { + return true; + } + + @Override + public boolean isDaemon() { + return true; + } + + @Override + public int getInitialDelaySeconds() { + return 0; + } + + @Override + public int getPeriodSeconds() { + return 1; + } + + @Nonnull + @Override + protected Logger getLogger() { + return LOG; + } + + @Override + public void doRun() { + try { + nodeService.markAsAlive(nodeId, isLeader.get(), httpPublishUri); + } catch (NodeNotFoundException e) { + LOG.warn("Did not find meta info of this node. Re-registering."); + nodeService.registerServer(nodeId.getNodeId(), + isLeader.get(), + httpPublishUri, + Tools.getLocalCanonicalHostname()); + } + } +} diff --git a/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java new file mode 100644 index 0000000000..ffed830bb3 --- /dev/null +++ b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java @@ -0,0 +1,78 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.datanode.periodicals; + +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.cluster.NodeService; +import org.graylog2.plugin.system.SimpleNodeId; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.net.URI; + +class NodePingPeriodicalTest { + + @Test + void doRun() throws NodeNotFoundException { + + final SimpleNodeId nodeID = new SimpleNodeId("5ca1ab1e-0000-4000-a000-000000000000"); + final URI uri = URI.create("http://localhost:8999"); + final NodeService nodeService = Mockito.mock(NodeService.class); + + final NodePingPeriodical task = new NodePingPeriodical( + nodeService, + nodeID, + uri, + () -> true + ); + + task.doRun(); + + Mockito.verify(nodeService).markAsAlive( + Mockito.eq(nodeID), + Mockito.eq(true), + Mockito.eq(uri)); + } + + + @Test + void doRunWithRegister() throws NodeNotFoundException { + + final SimpleNodeId nodeID = new SimpleNodeId("5ca1ab1e-0000-4000-a000-000000000000"); + final URI uri = URI.create("http://localhost:8999"); + + final NodeService nodeService = Mockito.mock(NodeService.class); + + Mockito.doThrow(new NodeNotFoundException("Node not found")).when(nodeService).markAsAlive(nodeID, true, uri); + + final NodePingPeriodical task = new NodePingPeriodical( + nodeService, + nodeID, + uri, + () -> true + ); + + task.doRun(); + + Mockito.verify(nodeService).registerServer( + Mockito.eq(nodeID.getNodeId()), + Mockito.eq(true), + Mockito.eq(uri), + Mockito.any() + ); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/NodeService.java b/graylog2-server/src/main/java/org/graylog2/cluster/NodeService.java index 554bb1e67d..76bf1e7ed0 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodeService.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/NodeService.java @@ -37,9 +37,7 @@ public interface NodeService extends PersistedService { void dropOutdated(); - void markAsAlive(Node node, boolean isLeader, String restTransportAddress); - - void markAsAlive(Node node, boolean isLeader, URI restTransportAddress); + void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress) throws NodeNotFoundException; boolean isOnlyLeader(NodeId nodeIde); diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/NodeServiceImpl.java b/graylog2-server/src/main/java/org/graylog2/cluster/NodeServiceImpl.java index 6131712af3..908b9fc192 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodeServiceImpl.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/NodeServiceImpl.java @@ -19,6 +19,7 @@ package org.graylog2.cluster; import com.google.common.collect.Maps; import com.mongodb.BasicDBObject; import com.mongodb.DBObject; +import com.mongodb.WriteResult; import org.bson.types.ObjectId; import org.graylog2.Configuration; import org.graylog2.database.MongoConnection; @@ -132,21 +133,24 @@ public class NodeServiceImpl extends PersistedServiceImpl implements NodeService /** * Mark this node as alive and probably update some settings that may have changed since last server boot. */ - @Override - public void markAsAlive(Node node, boolean isLeader, String restTransportAddress) { - node.getFields().put("last_seen", Tools.getUTCTimestamp()); - node.getFields().put("is_leader", isLeader); - node.getFields().put("transport_address", restTransportAddress); - try { - save(node); - } catch (ValidationException e) { - throw new RuntimeException("Validation failed.", e); - } - } + public void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress) throws NodeNotFoundException { + BasicDBObject query = new BasicDBObject("node_id", node.getNodeId()); - @Override - public void markAsAlive(Node node, boolean isLeader, URI restTransportAddress) { - markAsAlive(node, isLeader, restTransportAddress.toString()); + final BasicDBObject fields = new BasicDBObject(); + fields.put("is_leader", isLeader); + fields.put("transport_address", restTransportAddress.toString()); + fields.put("last_seen", Tools.getUTCTimestamp()); // TODO: deprecate and remove in later releases and use last_seen_date only! + final BasicDBObject update = new BasicDBObject("$set", fields); + + // using $currentDate will persist the database date, avoiding time sync problems across different nodes + //update.put("$currentDate", new BasicDBObject("last_seen_date", new BasicDBObject("$type", "date"))); + + final WriteResult result = super.collection(NodeImpl.class).update(query, update); + + final int updatedDocumentsCount = result.getN(); + if (updatedDocumentsCount != 1) { + throw new NodeNotFoundException("Unable to find node " + node.getNodeId()); + } } @Override diff --git a/graylog2-server/src/main/java/org/graylog2/periodical/NodePingThread.java b/graylog2-server/src/main/java/org/graylog2/periodical/NodePingThread.java index 779295285f..966f613185 100644 --- a/graylog2-server/src/main/java/org/graylog2/periodical/NodePingThread.java +++ b/graylog2-server/src/main/java/org/graylog2/periodical/NodePingThread.java @@ -65,8 +65,7 @@ public class NodePingThread extends Periodical { public synchronized void doRun() { final boolean isLeader = leaderElectionService.isLeader(); try { - Node node = nodeService.byNodeId(serverStatus.getNodeId()); - nodeService.markAsAlive(node, isLeader, httpConfiguration.getHttpPublishUri().resolve(HttpConfiguration.PATH_API)); + nodeService.markAsAlive(serverStatus.getNodeId(), isLeader, httpConfiguration.getHttpPublishUri().resolve(HttpConfiguration.PATH_API)); } catch (NodeNotFoundException e) { LOG.warn("Did not find meta info of this node. Re-registering."); nodeService.registerServer(serverStatus.getNodeId().toString(), diff --git a/graylog2-server/src/test/java/org/graylog2/cluster/NodeServiceImplTest.java b/graylog2-server/src/test/java/org/graylog2/cluster/NodeServiceImplTest.java index 37a6dc8d3b..c34168e4fa 100644 --- a/graylog2-server/src/test/java/org/graylog2/cluster/NodeServiceImplTest.java +++ b/graylog2-server/src/test/java/org/graylog2/cluster/NodeServiceImplTest.java @@ -17,12 +17,14 @@ package org.graylog2.cluster; import com.mongodb.DBCollection; +import org.assertj.core.api.Assertions; import org.graylog.testing.mongodb.MongoDBFixtures; import org.graylog.testing.mongodb.MongoDBInstance; import org.graylog2.Configuration; import org.graylog2.plugin.Tools; import org.graylog2.plugin.system.NodeId; import org.graylog2.plugin.system.SimpleNodeId; +import org.joda.time.DateTime; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -98,4 +100,21 @@ public class NodeServiceImplTest { assertThat(node2.getTransportAddress()).isEqualTo(TRANSPORT_URI.toString()); assertThat(node2.isLeader()).isTrue(); } + + @Test + public void testMarkAsAlive() throws NodeNotFoundException { + assertThat(nodeService.allActive()) + .describedAs("The collection should be empty") + .isEmpty(); + + nodeService.registerServer(nodeId.getNodeId(), true, TRANSPORT_URI, LOCAL_CANONICAL_HOSTNAME); + nodeService.markAsAlive(nodeId, false, URI.create("http://10.0.0.1:12901")); + + final Node node = nodeService.byNodeId(nodeId); + + Assertions.assertThat(node.isLeader()).isFalse(); + Assertions.assertThat(node.getTransportAddress()).isEqualTo("http://10.0.0.1:12901"); + Assertions.assertThat(node.getLastSeen()).isNotNull().isInstanceOf(DateTime.class); + + } }