diff --git a/changelog/unreleased/issue-17247.toml b/changelog/unreleased/issue-17247.toml new file mode 100644 index 0000000000..7f68f37fd7 --- /dev/null +++ b/changelog/unreleased/issue-17247.toml @@ -0,0 +1,5 @@ +type = "a" +message = "Data nodes now publish state machine state. Provides a backend for data node table including state information." +issues = ["17247"] +pulls = ["17382"] + diff --git a/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java b/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java index bc42581033..ee22ce8690 100644 --- a/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java +++ b/data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java @@ -17,24 +17,24 @@ package org.graylog.datanode.bindings; import com.google.common.eventbus.EventBus; +import com.google.inject.TypeLiteral; import com.google.inject.multibindings.Multibinder; import com.google.inject.multibindings.OptionalBinder; import org.graylog.datanode.Configuration; -import org.graylog.datanode.cluster.DataNodeServiceImpl; import org.graylog.datanode.shared.system.activities.DataNodeActivityWriter; import org.graylog.security.certutil.keystore.storage.KeystoreContentMover; import org.graylog.security.certutil.keystore.storage.SinglePasswordKeystoreContentMover; import org.graylog2.bindings.providers.ClusterEventBusProvider; import org.graylog2.cluster.ClusterConfigServiceImpl; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeClusterService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.events.ClusterEventBus; import org.graylog2.jackson.InputConfigurationBeanDeserializerModifier; import org.graylog2.plugin.cluster.ClusterConfigService; import org.graylog2.plugin.cluster.ClusterIdFactory; import org.graylog2.plugin.cluster.RandomUUIDClusterIdFactory; import org.graylog2.plugin.inject.Graylog2Module; -import org.graylog2.plugin.system.FilePersistedNodeIdProvider; -import org.graylog2.plugin.system.NodeId; import org.graylog2.shared.bindings.providers.EventBusProvider; import org.graylog2.shared.system.activities.ActivityWriter; @@ -83,7 +83,7 @@ public class ServerBindings extends Graylog2Module { private void bindInterfaces() { bind(ActivityWriter.class).to(DataNodeActivityWriter.class); OptionalBinder.newOptionalBinder(binder(), ClusterIdFactory.class).setDefault().to(RandomUUIDClusterIdFactory.class); - bind(NodeService.class).to(DataNodeServiceImpl.class); + bind(new TypeLiteral>() {}).to(DataNodeClusterService.class); } private void bindDynamicFeatures() { diff --git a/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodeConfigurationPeriodical.java b/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodeConfigurationPeriodical.java index ca08fed62d..ccbab43b89 100644 --- a/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodeConfigurationPeriodical.java +++ b/data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodeConfigurationPeriodical.java @@ -32,7 +32,8 @@ import org.graylog.security.certutil.keystore.storage.location.KeystoreMongoColl import org.graylog.security.certutil.keystore.storage.location.KeystoreMongoLocation; import org.graylog.security.certutil.privatekey.PrivateKeyEncryptedFileStorage; import org.graylog2.cluster.NodeNotFoundException; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningConfig; import org.graylog2.cluster.preflight.DataNodeProvisioningService; import org.graylog2.plugin.periodical.Periodical; @@ -45,8 +46,8 @@ import javax.inject.Inject; import javax.inject.Named; import javax.inject.Singleton; import java.io.IOException; -import java.nio.file.Path; import java.net.InetAddress; +import java.nio.file.Path; import java.security.KeyStore; import java.util.Collections; import java.util.Objects; @@ -59,7 +60,7 @@ public class DataNodeConfigurationPeriodical extends Periodical { private static final Logger LOG = LoggerFactory.getLogger(DataNodeConfigurationPeriodical.class); private final DataNodeProvisioningService dataNodeProvisioningService; - private final NodeService nodeService; + private final NodeService nodeService; private final NodeId nodeId; private final PrivateKeyEncryptedFileStorage privateKeyEncryptedStorage; private final CsrMongoStorage csrStorage; @@ -71,7 +72,7 @@ public class DataNodeConfigurationPeriodical extends Periodical { @Inject public DataNodeConfigurationPeriodical(final DataNodeProvisioningService dataNodeProvisioningService, - final NodeService nodeService, + final NodeService nodeService, final NodeId nodeId, final CsrMongoStorage csrStorage, final CsrGenerator csrGenerator, diff --git a/data-node/src/main/java/org/graylog/datanode/commands/Server.java b/data-node/src/main/java/org/graylog/datanode/commands/Server.java index d9ffcce676..4fb23c8af4 100644 --- a/data-node/src/main/java/org/graylog/datanode/commands/Server.java +++ b/data-node/src/main/java/org/graylog/datanode/commands/Server.java @@ -21,6 +21,7 @@ import com.github.rvesse.airline.annotations.Option; import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.ServiceManager; import com.google.inject.Injector; +import com.google.inject.Key; import com.google.inject.Module; import com.google.inject.spi.Message; import com.mongodb.MongoException; @@ -34,7 +35,9 @@ import org.graylog.datanode.rest.RestBindings; import org.graylog.datanode.shutdown.GracefulShutdown; import org.graylog2.bindings.MongoDBModule; import org.graylog2.bindings.PasswordAlgorithmBindings; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodeStatus; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningBindings; import org.graylog2.configuration.MongoDbConfiguration; import org.graylog2.configuration.TLSProtocolsConfiguration; @@ -127,13 +130,16 @@ public class Server extends ServerBootstrap { @Override protected void startNodeRegistration(Injector injector) { - final NodeService nodeService = injector.getInstance(NodeService.class); + final NodeService nodeService = injector.getInstance(new Key<>() {}); final NodeId nodeId = injector.getInstance(NodeId.class); // always set leader to "false" on startup and let the NodePingPeriodical take care of it later - nodeService.registerServer(nodeId.getNodeId(), - false, - configuration.getHttpPublishUri(), - Tools.getLocalCanonicalHostname()); + nodeService.registerServer(DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(false) + .setTransportAddress(configuration.getHttpPublishUri().toString()) + .setHostname(Tools.getLocalCanonicalHostname()) + .setDataNodeStatus(DataNodeStatus.STARTING) + .build()); } @Override diff --git a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java index e7ffec88bd..eb90730b0a 100644 --- a/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java +++ b/data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java @@ -28,7 +28,8 @@ import org.graylog.security.certutil.ca.exceptions.KeyStoreStorageException; import org.graylog2.bootstrap.preflight.PreflightConfigResult; import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import javax.inject.Inject; import javax.inject.Named; @@ -36,7 +37,6 @@ import javax.inject.Provider; import javax.inject.Singleton; import java.io.IOException; import java.nio.charset.StandardCharsets; -import java.nio.file.Path; import java.security.GeneralSecurityException; import java.util.List; import java.util.Optional; @@ -50,7 +50,7 @@ public class OpensearchConfigurationProvider implements Provider nodeService; private final PreflightConfigService preflightConfigService; @Inject @@ -59,7 +59,7 @@ public class OpensearchConfigurationProvider implements Provider nodeService, final PreflightConfigService preflightConfigService, final @Named("password_secret") String passwordSecret) { this.localConfiguration = localConfiguration; @@ -98,7 +98,7 @@ public class OpensearchConfigurationProvider implements Provider stdout; private final Queue stderr; private final CustomCAX509TrustManager trustManager; - private final NodeService nodeService; + private final NodeService nodeService; private final Configuration configuration; - OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService nodeService) { + OpensearchProcessImpl(DatanodeConfiguration datanodeConfiguration, int logsCacheSize, final CustomCAX509TrustManager trustManager, final Configuration configuration, final NodeService nodeService) { this.datanodeConfiguration = datanodeConfiguration; this.processState = ProcessStateMachine.createNew(); tracerAggregator = new StateMachineTracerAggregator(); @@ -156,7 +156,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener { private void writeSeedHostsList() { try { final Path hostsfile = datanodeConfiguration.datanodeDirectories().createOpensearchProcessConfigurationFile(UNICAST_HOSTS_FILE); - final Set current = nodeService.allActive(Node.Type.DATANODE).values().stream().map(Node::getClusterAddress).filter(Objects::nonNull).collect(Collectors.toSet()); + final Set current = nodeService.allActive().values().stream().map(DataNodeDto::getClusterAddress).filter(Objects::nonNull).collect(Collectors.toSet()); Files.write(hostsfile, current, Charset.defaultCharset(), StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.CREATE); } catch (IOException iox) { LOG.error("Could not write to file: {} - {}", UNICAST_HOSTS_FILE, iox.getMessage()); diff --git a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java index 9b561e2405..1f109e6572 100644 --- a/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java +++ b/data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java @@ -22,7 +22,8 @@ import com.google.common.util.concurrent.AbstractIdleService; import org.graylog.datanode.Configuration; import org.graylog.datanode.configuration.DatanodeConfiguration; import org.graylog.datanode.process.OpensearchConfiguration; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningStateChangeEvent; import org.graylog2.datanode.DataNodeLifecycleEvent; import org.graylog2.events.ClusterEventBus; @@ -54,7 +55,7 @@ public class OpensearchProcessService extends AbstractIdleService implements Pro final EventBus eventBus, final ClusterEventBus clusterEventBus, final CustomCAX509TrustManager trustManager, - final NodeService nodeService, + final NodeService nodeService, final Configuration configuration, final NodeId nodeId) { this.configurationProvider = configurationProvider; 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 index 1bc566291d..800526dd80 100644 --- a/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java +++ b/data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java @@ -18,44 +18,45 @@ 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.graylog.datanode.process.ProcessState; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodeStatus; +import org.graylog2.cluster.nodes.NodeService; 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.annotation.Nullable; import javax.inject.Inject; -import javax.inject.Named; 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 NodeService nodeService; private final NodeId nodeId; private final Supplier opensearchBaseUri; private final Supplier opensearchClusterUri; private final Supplier isLeader; private final Configuration configuration; + private final Supplier processState; @Inject - public NodePingPeriodical(NodeService nodeService, NodeId nodeId, Configuration configuration, OpensearchProcess managedOpenSearch) { - this(nodeService, nodeId, configuration, managedOpenSearch::getOpensearchBaseUrl, managedOpenSearch::getOpensearchClusterUrl, managedOpenSearch::isLeaderNode); + public NodePingPeriodical(NodeService nodeService, NodeId nodeId, Configuration configuration, OpensearchProcess managedOpenSearch) { + this(nodeService, nodeId, configuration, managedOpenSearch::getOpensearchBaseUrl, managedOpenSearch::getOpensearchClusterUrl, managedOpenSearch::isLeaderNode, () -> managedOpenSearch.processInfo().state()); } NodePingPeriodical( - NodeService nodeService, + NodeService nodeService, NodeId nodeId, Configuration configuration, Supplier opensearchBaseUri, Supplier opensearchClusterUri, - Supplier isLeader + Supplier isLeader, + Supplier processState ) { this.nodeService = nodeService; this.nodeId = nodeId; @@ -63,6 +64,7 @@ public class NodePingPeriodical extends Periodical { this.opensearchClusterUri = opensearchClusterUri; this.isLeader = isLeader; this.configuration = configuration; + this.processState = processState; } @Override @@ -108,20 +110,28 @@ public class NodePingPeriodical extends Periodical { @Override public void doRun() { - try { - nodeService.markAsAlive(nodeId, isLeader.get(), opensearchBaseUri.get(), opensearchClusterUri.get()); - } catch (NodeNotFoundException e) { - LOG.warn("Did not find meta info of this node. Re-registering."); - registerServer(); - } + final DataNodeDto dto = DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(isLeader.get()) + .setTransportAddress(opensearchBaseUri.get().toString()) + .setClusterAddress(opensearchClusterUri.get()) + .setDataNodeStatus(processState.get().getDataNodeStatus()) + .setHostname(configuration.getHostname()) + .build(); + + nodeService.ping(dto); + } private void registerServer() { - final boolean registrationSucceeded = nodeService.registerServer(nodeId.getNodeId(), - isLeader.get(), - opensearchBaseUri.get(), - opensearchClusterUri.get(), - configuration.getHostname()); + final boolean registrationSucceeded = nodeService.registerServer(DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(isLeader.get()) + .setTransportAddress(opensearchBaseUri.get().toString()) + .setClusterAddress(opensearchClusterUri.get()) + .setHostname(configuration.getHostname()) + .setDataNodeStatus(DataNodeStatus.STARTING) + .build()); if (!registrationSucceeded) { LOG.error("Failed to register node {} for heartbeats.", nodeId.getNodeId()); diff --git a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java b/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java index ffd632b537..c01e67f5b3 100644 --- a/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java +++ b/data-node/src/main/java/org/graylog/datanode/process/ProcessState.java @@ -16,41 +16,54 @@ */ package org.graylog.datanode.process; +import org.graylog2.cluster.nodes.DataNodeStatus; + public enum ProcessState { /** * Fresh created process, not started yet */ - WAITING_FOR_CONFIGURATION, + WAITING_FOR_CONFIGURATION(DataNodeStatus.UNCONFIGURED), /** * The process is running on the underlying OS and has a process ID. It's not responding to the REST API yet */ - STARTING, + STARTING(DataNodeStatus.STARTING), /** * Opensearch is now available on the default port with GREEN cluster status, all good */ - AVAILABLE, + AVAILABLE(DataNodeStatus.AVAILABLE), /** * There are problems in the REST communication with the opensearch. We'll retry several times before * we go to the FAILED state, giving up. */ - NOT_RESPONDING, + NOT_RESPONDING(DataNodeStatus.UNAVAILABLE), /** * Failed to reach the opensearch REST API, but the underlying process is still alive */ - FAILED, + FAILED(DataNodeStatus.UNAVAILABLE), /** * Removal of node from Opensearch cluster requested */ - REMOVING, + REMOVING(DataNodeStatus.REMOVING), /** * Removal of node from Opensearch cluster completed */ - REMOVED, - + REMOVED(DataNodeStatus.REMOVED), + /** * The OS process is not running anymore on the underlying system, it has been terminated */ - TERMINATED + TERMINATED(DataNodeStatus.UNAVAILABLE); + + + private final DataNodeStatus dataNodeStatus; + + ProcessState(DataNodeStatus dataNodeStatus) { + this.dataNodeStatus = dataNodeStatus; + } + + public DataNodeStatus getDataNodeStatus() { + return dataNodeStatus; + } } 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 index 754b912605..9f4418e708 100644 --- a/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java +++ b/data-node/src/test/java/org/graylog/datanode/periodicals/NodePingPeriodicalTest.java @@ -17,8 +17,10 @@ package org.graylog.datanode.periodicals; import org.graylog.datanode.Configuration; +import org.graylog.datanode.process.ProcessState; import org.graylog2.cluster.NodeNotFoundException; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.plugin.system.SimpleNodeId; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -33,7 +35,8 @@ class NodePingPeriodicalTest { final SimpleNodeId nodeID = new SimpleNodeId("5ca1ab1e-0000-4000-a000-000000000000"); final URI uri = URI.create("http://localhost:9200"); final String cluster = "localhost:9300"; - final NodeService nodeService = Mockito.mock(NodeService.class); + @SuppressWarnings("unchecked") + final NodeService nodeService = (NodeService) Mockito.mock(NodeService.class); Configuration configuration = Mockito.mock(Configuration.class); Mockito.when(configuration.getHostname()).thenReturn("localhost"); @@ -44,50 +47,21 @@ class NodePingPeriodicalTest { configuration, () -> uri, () -> cluster, - () -> true + () -> true, + () -> ProcessState.AVAILABLE ); task.doRun(); - Mockito.verify(nodeService).markAsAlive( - Mockito.eq(nodeID), - Mockito.eq(true), - Mockito.eq(uri), - Mockito.eq(cluster)); - } - - - @Test - void doRunWithRegister() throws NodeNotFoundException { - - final SimpleNodeId nodeID = new SimpleNodeId("5ca1ab1e-0000-4000-a000-000000000000"); - final URI uri = URI.create("http://localhost:9200"); - final String cluster = "localhost:9300"; - - final NodeService nodeService = Mockito.mock(NodeService.class); - - Mockito.doThrow(new NodeNotFoundException("Node not found")).when(nodeService).markAsAlive(nodeID, true, uri, cluster); - - Configuration configuration = Mockito.mock(Configuration.class); - Mockito.when(configuration.getHostname()).thenReturn("hostname.setting.from.config"); - - final NodePingPeriodical task = new NodePingPeriodical( - nodeService, - nodeID, - configuration, - () -> uri, - () -> cluster, - () -> true - ); - - task.doRun(); - - Mockito.verify(nodeService).registerServer( - Mockito.eq(nodeID.getNodeId()), - Mockito.eq(true), - Mockito.eq(uri), - Mockito.eq(cluster), - Mockito.eq("hostname.setting.from.config") - ); + Mockito.verify(nodeService).ping(Mockito.eq(DataNodeDto.Builder.builder() + .setId(nodeID.getNodeId()) + .setLeader(true) + .setTransportAddress(uri.toString()) + .setClusterAddress(cluster) + .setDataNodeStatus(ProcessState.AVAILABLE.getDataNodeStatus()) + .setHostname("localhost") + .build() + )); } + } diff --git a/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalService.java b/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalService.java index e7b34f2679..fee11a2944 100644 --- a/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalService.java +++ b/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalService.java @@ -16,16 +16,31 @@ */ package org.graylog.security.certutil; -import org.graylog2.cluster.Node; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodeStatus; import org.graylog2.cluster.preflight.DataNodeProvisioningConfig; import java.time.LocalDateTime; +import java.util.Collection; import java.util.List; public interface CertRenewalService { - record DataNode(String nodeId, Node.Type type, String transportAddress, DataNodeProvisioningConfig.State status, String errorMsg, String hostname, String shortNodeId, LocalDateTime certValidUntil) {} + + record ProvisioningInformation(DataNodeProvisioningConfig.State status, String errorMsg, + LocalDateTime certValidUntil) { + } + + record DataNode(String nodeId, DataNodeStatus dataNodeStatus, String transportAddress, + DataNodeProvisioningConfig.State status, String errorMsg, String hostname, String shortNodeId, + LocalDateTime certValidUntil) { + public String getId() { + return nodeId; + } + } void checkCertificatesForRenewal(); void initiateRenewalForNode(String nodeId); List findNodes(); + + List addProvisioningInformation(Collection nodes); } diff --git a/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalServiceImpl.java b/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalServiceImpl.java index c26155f4c3..680a3153d3 100644 --- a/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalServiceImpl.java +++ b/graylog2-server/src/main/java/org/graylog/security/certutil/CertRenewalServiceImpl.java @@ -25,7 +25,8 @@ import org.graylog.security.certutil.keystore.storage.KeystoreMongoStorage; import org.graylog.security.certutil.keystore.storage.location.KeystoreMongoCollections; import org.graylog.security.certutil.keystore.storage.location.KeystoreMongoLocation; import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningConfig; import org.graylog2.cluster.preflight.DataNodeProvisioningService; import org.graylog2.notifications.Notification; @@ -49,6 +50,7 @@ import java.security.cert.X509Certificate; import java.time.Duration; import java.time.LocalDateTime; import java.time.ZoneId; +import java.util.Collection; import java.util.Date; import java.util.List; import java.util.Map; @@ -64,7 +66,7 @@ public class CertRenewalServiceImpl implements CertRenewalService { private final ClusterConfigService clusterConfigService; private final KeystoreMongoStorage keystoreMongoStorage; - private final NodeService nodeService; + private final NodeService nodeService; private final DataNodeProvisioningService dataNodeProvisioningService; private final NotificationService notificationService; private final DBJobTriggerService jobTriggerService; @@ -78,7 +80,7 @@ public class CertRenewalServiceImpl implements CertRenewalService { @Inject public CertRenewalServiceImpl(final ClusterConfigService clusterConfigService, final KeystoreMongoStorage keystoreMongoStorage, - final NodeService nodeService, + final NodeService nodeService, final DataNodeProvisioningService dataNodeProvisioningService, final NotificationService notificationService, final DBJobTriggerService jobTriggerService, @@ -178,9 +180,9 @@ public class CertRenewalServiceImpl implements CertRenewalService { } } - protected List findNodesThatNeedCertificateRenewal(final RenewalPolicy renewalPolicy) { + protected List findNodesThatNeedCertificateRenewal(final RenewalPolicy renewalPolicy) { final var nextRenewal = getNextRenewal(); - final Map activeDataNodes = nodeService.allActive(Node.Type.DATANODE); + final Map activeDataNodes = nodeService.allActive(); return activeDataNodes.values().stream().map(node -> { final var keystore = loadKeyStoreForNode(node); final var certificate = keystore.flatMap(this::getCertificateForNode); @@ -202,14 +204,14 @@ public class CertRenewalServiceImpl implements CertRenewalService { @Override public List findNodes() { - final Map activeDataNodes = nodeService.allActive(Node.Type.DATANODE); + final Map activeDataNodes = nodeService.allActive(); return activeDataNodes.values().stream().map(node -> { final var keystore = loadKeyStoreForNode(node); final var certificate = keystore.flatMap(this::getCertificateForNode); final var certValidUntil = certificate.map(cert -> cert.getNotAfter().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime()); final var config = getDataNodeProvisioningConfig(node).orElseThrow(() -> new IllegalStateException("No config found for data node " + node.getNodeId())); return new DataNode(node.getNodeId(), - node.getType(), + node.getDataNodeStatus(), node.getTransportAddress(), config.state(), config.errorMsg(), @@ -219,11 +221,24 @@ public class CertRenewalServiceImpl implements CertRenewalService { }).toList(); } + @Override + public List addProvisioningInformation(Collection nodes) { + return nodes.stream().map(node -> { + final var keystore = loadKeyStoreForNode(node); + final var certificate = keystore.flatMap(this::getCertificateForNode); + final var certValidUntil = certificate.map(cert -> cert.getNotAfter().toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime()); + final var config = getDataNodeProvisioningConfig(node).orElseThrow(() -> new IllegalStateException("No config found for data node " + node.getNodeId())); + return node.toBuilder().setProvisioningInformation(new CertRenewalService.ProvisioningInformation( + config.state(), config.errorMsg(), certValidUntil.orElse(null) + )).build(); + }).toList(); + } + private Optional getDataNodeProvisioningConfig(final Node node) { return dataNodeProvisioningService.getPreflightConfigFor(node.getNodeId()); } - private void notifyManualRenewalForNode(final List nodes) { + private void notifyManualRenewalForNode(final List nodes) { final var key = String.join(",", nodes.stream().map(Node::getNodeId).toList()); if(!notificationService.isFirst(Notification.Type.CERTIFICATE_NEEDS_RENEWAL)) { notificationService.fixed(Notification.Type.CERTIFICATE_NEEDS_RENEWAL); diff --git a/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java b/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java index 4419b01f85..439b246559 100644 --- a/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java +++ b/graylog2-server/src/main/java/org/graylog2/bindings/PersistenceServicesBindings.java @@ -17,9 +17,15 @@ package org.graylog2.bindings; import com.google.inject.AbstractModule; +import com.google.inject.TypeLiteral; import com.google.inject.multibindings.OptionalBinder; -import org.graylog2.cluster.NodeService; import org.graylog2.cluster.NodeServiceImpl; +import org.graylog2.cluster.nodes.DataNodeClusterService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodePaginatedService; +import org.graylog2.cluster.nodes.NodeService; +import org.graylog2.cluster.nodes.ServerNodeClusterService; +import org.graylog2.cluster.nodes.ServerNodeDto; import org.graylog2.database.suggestions.EntitySuggestionService; import org.graylog2.database.suggestions.MongoEntitySuggestionService; import org.graylog2.indexer.IndexFailureService; @@ -53,7 +59,10 @@ public class PersistenceServicesBindings extends AbstractModule { bind(SystemMessageService.class).to(SystemMessageServiceImpl.class).asEagerSingleton(); bind(NotificationService.class).to(NotificationServiceImpl.class).asEagerSingleton(); bind(IndexFailureService.class).to(IndexFailureServiceImpl.class).asEagerSingleton(); - bind(NodeService.class).to(NodeServiceImpl.class); + bind(org.graylog2.cluster.NodeService.class).to(NodeServiceImpl.class); + bind(new TypeLiteral>() {}).to(ServerNodeClusterService.class); + bind(new TypeLiteral>() {}).to(DataNodeClusterService.class); + bind(DataNodePaginatedService.class).asEagerSingleton(); bind(IndexRangeService.class).to(MongoIndexRangeService.class).asEagerSingleton(); bind(InputService.class).to(InputServiceImpl.class); bind(UserService.class).to(UserServiceImpl.class).asEagerSingleton(); diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogCertificateProvisioningPeriodical.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogCertificateProvisioningPeriodical.java index b1bfea3867..c17ac174bd 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogCertificateProvisioningPeriodical.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/GraylogCertificateProvisioningPeriodical.java @@ -36,7 +36,8 @@ import org.graylog.security.certutil.cert.storage.CertChainStorage; import org.graylog.security.certutil.csr.CsrSigner; import org.graylog.security.certutil.csr.storage.CsrMongoStorage; import org.graylog2.Configuration; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningConfig; import org.graylog2.cluster.preflight.DataNodeProvisioningService; import org.graylog2.notifications.Notification; @@ -87,7 +88,7 @@ public class GraylogCertificateProvisioningPeriodical extends Periodical { private static final String ERROR_MESSAGE_PREFIX = "Error trying to connect to data node "; private final DataNodeProvisioningService dataNodeProvisioningService; - private final NodeService nodeService; + private final NodeService nodeService; private final CaConfiguration configuration; private final CsrMongoStorage csrStorage; @@ -108,7 +109,7 @@ public class GraylogCertificateProvisioningPeriodical extends Periodical { final CertChainMongoStorage certMongoStorage, final CaService caService, final Configuration configuration, - final NodeService nodeService, + final NodeService nodeService, final CsrSigner csrSigner, final ClusterConfigService clusterConfigService, final @Named("password_secret") String passwordSecret, diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java index 69aa2f4078..e0f8242cdc 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/PreflightWebModule.java @@ -35,10 +35,11 @@ import org.graylog2.bootstrap.preflight.web.resources.PreflightAssetsResource; import org.graylog2.bootstrap.preflight.web.resources.PreflightResource; import org.graylog2.bootstrap.preflight.web.resources.PreflightStatusResource; import org.graylog2.cluster.ClusterConfigServiceImpl; -import org.graylog2.cluster.NodeService; -import org.graylog2.cluster.NodeServiceImpl; import org.graylog2.cluster.leader.LeaderElectionModule; import org.graylog2.cluster.lock.LockServiceModule; +import org.graylog2.cluster.nodes.DataNodeClusterService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.database.MongoConnection; import org.graylog2.events.ClusterEventBus; import org.graylog2.events.ClusterEventCleanupPeriodical; @@ -70,7 +71,7 @@ public class PreflightWebModule extends Graylog2Module { bind(ServiceManager.class).toProvider(ServiceManagerProvider.class).asEagerSingleton(); bind(MongoConnection.class).toProvider(MongoConnectionProvider.class); - bind(NodeService.class).to(NodeServiceImpl.class); + bind(new TypeLiteral>() {}).to(DataNodeClusterService.class); bind(KeystoreContentMover.class).to(SinglePasswordKeystoreContentMover.class).asEagerSingleton(); bind(ClusterConfigService.class).to(ClusterConfigServiceImpl.class); bind(CaService.class).to(CaServiceImpl.class); diff --git a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java index 6bbda39865..409e4f6dc5 100644 --- a/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java +++ b/graylog2-server/src/main/java/org/graylog2/bootstrap/preflight/web/resources/PreflightResource.java @@ -26,8 +26,9 @@ import org.graylog2.bootstrap.preflight.PreflightConstants; import org.graylog2.bootstrap.preflight.web.resources.model.CA; import org.graylog2.bootstrap.preflight.web.resources.model.CertParameters; import org.graylog2.bootstrap.preflight.web.resources.model.CreateCARequest; -import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodeEntity; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.cluster.preflight.DataNodeProvisioningConfig; import org.graylog2.cluster.preflight.DataNodeProvisioningService; import org.graylog2.plugin.certificates.RenewalPolicy; @@ -58,14 +59,14 @@ import java.util.stream.Collectors; @Produces(MediaType.APPLICATION_JSON) public class PreflightResource { - private final NodeService nodeService; + private final NodeService nodeService; private final DataNodeProvisioningService dataNodeProvisioningService; private final CaService caService; private final ClusterConfigService clusterConfigService; private final String passwordSecret; @Inject - public PreflightResource(final NodeService nodeService, + public PreflightResource(final NodeService nodeService, final DataNodeProvisioningService dataNodeProvisioningService, final CaService caService, final ClusterConfigService clusterConfigService, @@ -77,18 +78,18 @@ public class PreflightResource { this.passwordSecret = passwordSecret; } - record DataNode(String nodeId, Node.Type type, String transportAddress, DataNodeProvisioningConfig.State status, String errorMsg, String hostname, String shortNodeId) {} + record DataNode(String nodeId, String transportAddress, DataNodeProvisioningConfig.State status, String errorMsg, + String hostname, String shortNodeId) {} @GET @Path("/data_nodes") public List listDataNodes() { - final Map activeDataNodes = nodeService.allActive(Node.Type.DATANODE); + final Map activeDataNodes = nodeService.allActive(); final var preflightDataNodes = dataNodeProvisioningService.streamAll().collect(Collectors.toMap(DataNodeProvisioningConfig::nodeId, Function.identity())); return activeDataNodes.values().stream().map(n -> { final var preflight = preflightDataNodes.get(n.getNodeId()); return new DataNode(n.getNodeId(), - n.getType(), n.getTransportAddress(), preflight != null ? preflight.state() : null, preflight != null ? preflight.errorMsg() : null, n.getHostname(), @@ -144,7 +145,7 @@ public class PreflightResource { @Path("/generate") @NoAuditEvent("No Audit Event needed") public void generate() { - final Map activeDataNodes = nodeService.allActive(Node.Type.DATANODE); + final Map activeDataNodes = nodeService.allActive(); activeDataNodes.values().forEach(node -> dataNodeProvisioningService.changeState(node.getNodeId(), DataNodeProvisioningConfig.State.CONFIGURED)); } diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/Node.java b/graylog2-server/src/main/java/org/graylog2/cluster/Node.java index 5ccc6a2c0e..f0e5ab9a86 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/Node.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/Node.java @@ -23,9 +23,6 @@ import org.graylog2.shared.utilities.StringUtils; import org.joda.time.DateTime; public interface Node { - enum Type { - SERVER, DATANODE - } String getNodeId(); @@ -45,12 +42,8 @@ public interface Node { boolean isLeader(); String getTransportAddress(); - String getClusterAddress(); - DateTime getLastSeen(); - Node.Type getType(); - String getHostname(); default String getShortNodeId() { @@ -61,4 +54,5 @@ public interface Node { default String getTitle() { return StringUtils.f("%s / %s", getShortNodeId(), getHostname()); } + } 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 e8d640f040..e77c049bd6 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodeService.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/NodeService.java @@ -16,15 +16,19 @@ */ package org.graylog2.cluster; -import org.graylog2.plugin.database.PersistedService; + import org.graylog2.plugin.system.NodeId; import java.net.URI; import java.util.Collection; import java.util.Map; +/** + * @deprecated Please use the generic org.graylog2.cluster.nodes.NodeService specifying the type of node + * to be returned (either server or data node). + */ +@Deprecated(since = "6.0") public interface NodeService { - Node.Type type(); boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, String clusterUri, String hostname); @@ -38,23 +42,8 @@ public interface NodeService { Map byNodeIds(Collection nodeIds); - Map allActive(Node.Type type); - - /** - * Please use the {@link #allActive(Node.Type)} method and provide explicit type of the node. Otherwise, - * the implementation will fall back to {@link #type()} and provide only nodes of this type. - */ - @Deprecated Map allActive(); - void dropOutdated(); - - void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, String clusterAddress) throws NodeNotFoundException; - default void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress) throws NodeNotFoundException { - markAsAlive(node, isLeader, restTransportAddress, null); - } - - boolean isOnlyLeader(NodeId nodeIde); - boolean isAnyLeaderPresent(); + } 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 f779bc7d18..c9cacb3a22 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodeServiceImpl.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/NodeServiceImpl.java @@ -16,219 +16,71 @@ */ package org.graylog2.cluster; -import com.google.common.collect.ImmutableList; -import com.mongodb.AggregationOptions; -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; -import org.graylog2.database.PersistedServiceImpl; +import org.graylog2.cluster.nodes.ServerNodeDto; import org.graylog2.plugin.system.NodeId; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.inject.Inject; import java.net.URI; import java.util.Collection; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; import java.util.Map; -import java.util.Spliterator; -import java.util.Spliterators; -import java.util.function.Function; import java.util.stream.Collectors; -import java.util.stream.Stream; -import java.util.stream.StreamSupport; -public class NodeServiceImpl extends PersistedServiceImpl implements NodeService { +/** + * @deprecated This is left for compatibility reasons, + * delegating calls to org.graylog2.cluster.nodes.NodeService<ServerNodeEntity> + */ +@Deprecated(since = "6.0") +public class NodeServiceImpl implements NodeService { - private static final Logger LOG = LoggerFactory.getLogger(NodeServiceImpl.class); - - public static final String LAST_SEEN_FIELD = "$last_seen"; - private final long pingTimeout; - private final static Map lastSeenFieldDefinition = Map.of("last_seen", Map.of("$type", "timestamp")); - private final static DBObject addLastSeenFieldAsDate = new BasicDBObject("$addFields", Map.of("last_seen_date", Map.of("$cond", - Map.of( - "if", Map.of("$isNumber", LAST_SEEN_FIELD), - "then", Map.of("$toDate", Map.of("$toLong", LAST_SEEN_FIELD)), - "else", Map.of("$toDate", Map.of("$dateToString", Map.of("date", LAST_SEEN_FIELD))) - ) - ))); + private final org.graylog2.cluster.nodes.NodeService delegate; @Inject - public NodeServiceImpl(final MongoConnection mongoConnection, Configuration configuration) { - this(mongoConnection, configuration.getStaleLeaderTimeout()); - } - - public NodeServiceImpl(final MongoConnection mongoConnection, final int staleLeaderTimeout) { - super(mongoConnection); - this.pingTimeout = staleLeaderTimeout; - } - - public Node.Type type() { - return Node.Type.SERVER; - } - - private Map addClusterUriToMap(final Map orig, final String clusterUri) { - if(clusterUri == null) { - return orig; - } - var newMap = new HashMap<>(orig); - newMap.put("cluster_address", clusterUri); - return Map.copyOf(newMap); + public NodeServiceImpl(org.graylog2.cluster.nodes.NodeService delegate) { + this.delegate = delegate; } @Override public boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, String clusterUri, String hostname) { - final var params = addClusterUriToMap(Map.of( - "node_id", nodeId, - "type", type().toString(), - "is_leader", isLeader, - "transport_address", httpPublishUri.toString(), - "hostname", hostname - ), clusterUri); - - final Map fields = Map.of( - "$set", params, - "$currentDate", lastSeenFieldDefinition - ); - - final WriteResult result = this.collection(NodeImpl.class).update( - new BasicDBObject("node_id", nodeId), - new BasicDBObject(fields), - true, - false - ); - return result.getN() == 1; + ServerNodeDto dto = ServerNodeDto.Builder.builder() + .setId(nodeId) + .setLeader(isLeader) + .setTransportAddress(httpPublishUri.toString()) + .setHostname(hostname) + .build(); + return delegate.registerServer(dto); } @Override public Node byNodeId(String nodeId) throws NodeNotFoundException { - DBObject query = new BasicDBObject("node_id", nodeId); - DBObject o = findOne(NodeImpl.class, query); - - if (o == null || !o.containsField("node_id")) { - throw new NodeNotFoundException("Unable to find node " + nodeId); - } - - return new NodeImpl((ObjectId) o.get("_id"), o.toMap()); + return delegate.byNodeId(nodeId); } @Override public Node byNodeId(NodeId nodeId) throws NodeNotFoundException { - return byNodeId(nodeId.getNodeId()); + return delegate.byNodeId(nodeId); } @Override public Map byNodeIds(Collection nodeIds) { - return query(NodeImpl.class, new BasicDBObject("node_id", new BasicDBObject("$in", nodeIds))) - .stream() - .map(o -> new NodeImpl((ObjectId) o.get("_id"), o.toMap())) - .collect(Collectors.toMap(NodeImpl::getNodeId, Function.identity())); + return transformMap(delegate.byNodeIds(nodeIds)); } - private Stream aggregate(List pipeline) { - return cursorToStream(this.collection(NodeImpl.class).aggregate(pipeline, AggregationOptions.builder().build())); - } - - @Override - public Map allActive(Node.Type type) { - return aggregate(recentHeartbeat(List.of(Map.of("type", type.toString())))) - .collect(Collectors.toMap(obj -> (String) obj.get("node_id"), obj -> new NodeImpl((ObjectId) obj.get("_id"), obj.toMap()))); - } - - @Deprecated @Override public Map allActive() { - return allActive(type()); - } - - private List recentHeartbeat(List> additionalMatches) { - var match = ImmutableList.builder() - .add(Map.of("$expr", Map.of("$gte", List.of("$last_seen_date", Map.of("$subtract", List.of("$$NOW", this.pingTimeout)))))) - .addAll(additionalMatches) - .build(); - return List.of( - addLastSeenFieldAsDate, - new BasicDBObject("$match", Map.of("$and", match)), - new BasicDBObject("$unset", "last_seen_date") - ); - } - - @Override - public void dropOutdated() { - var outdatedIds = aggregate(List.of( - addLastSeenFieldAsDate, - new BasicDBObject("$match", Map.of("$expr", Map.of("$lt", List.of("$last_seen_date", Map.of("$subtract", List.of("$$NOW", this.pingTimeout)))))), - new BasicDBObject("$project", Map.of("_id", "$_id")) - )) - .map(obj -> obj.get("_id")) - .toList(); - - if (!outdatedIds.isEmpty()) { - final BasicDBObject query = new BasicDBObject("_id", Map.of("$in", outdatedIds)); - destroyAll(NodeImpl.class, query); - } - } - - private Stream cursorToStream(Iterator cursor) { - return StreamSupport.stream(Spliterators.spliteratorUnknownSize(cursor, Spliterator.ORDERED), false); - } - - /** - * Mark this node as alive and probably update some settings that may have changed since last server boot. - */ - public void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, String clusterAddress) throws NodeNotFoundException { - BasicDBObject query = new BasicDBObject("node_id", node.getNodeId()); - final var params = addClusterUriToMap(Map.of( - "is_leader", isLeader, - "transport_address", restTransportAddress.toString() - ), clusterAddress); - - final BasicDBObject update = new BasicDBObject(Map.of( - "$set", params, - "$currentDate", lastSeenFieldDefinition - )); - - 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 - public boolean isOnlyLeader(NodeId nodeId) { - - if(type() != Node.Type.SERVER) { - LOG.warn("Caution, isOnlyLeader called in the {} context, but returning only results of type {}", type(), Node.Type.SERVER); - } - - return aggregate(recentHeartbeat(List.of( - Map.of( - "type", Node.Type.SERVER.toString(), - "node_id", new BasicDBObject("$ne", nodeId.getNodeId()), - "is_leader", true - ) - ))).findAny().isEmpty(); + return transformMap(delegate.allActive()); } @Override public boolean isAnyLeaderPresent() { + return delegate.isAnyLeaderPresent(); + } - if(type() != Node.Type.SERVER) { - LOG.warn("Caution, isOnlyLeader called in the {} context, but returning only results of type {}", type(), Node.Type.SERVER); - } - - return aggregate(recentHeartbeat(List.of( - Map.of( - "type", Node.Type.SERVER.toString(), - "is_leader", true - ) - ))).findAny().isPresent(); + private Map transformMap(Map nodes) { + return nodes.entrySet().stream() + .collect( + Collectors.toMap( + Map.Entry::getKey, + Map.Entry::getValue + )); } } diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/NodeImpl.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/AbstractNode.java similarity index 76% rename from graylog2-server/src/main/java/org/graylog2/cluster/NodeImpl.java rename to graylog2-server/src/main/java/org/graylog2/cluster/nodes/AbstractNode.java index ebc86dff55..c9a13b2af5 100644 --- a/graylog2-server/src/main/java/org/graylog2/cluster/NodeImpl.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/AbstractNode.java @@ -14,31 +14,29 @@ * along with this program. If not, see * . */ -package org.graylog2.cluster; +package org.graylog2.cluster.nodes; import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonProperty; +import de.undercouch.bson4jackson.types.Timestamp; import org.bson.types.BSONTimestamp; import org.bson.types.ObjectId; -import org.graylog2.database.DbEntity; +import org.graylog2.cluster.Node; import org.graylog2.database.PersistedImpl; import org.graylog2.plugin.database.validators.Validator; -import org.graylog2.shared.utilities.StringUtils; import org.joda.time.DateTime; import org.joda.time.DateTimeZone; import java.util.Collections; -import java.util.Locale; import java.util.Map; -@DbEntity(collection = "nodes", titleField = "node_id") -public class NodeImpl extends PersistedImpl implements Node { +public abstract class AbstractNode extends PersistedImpl implements Node { - public NodeImpl(Map fields) { + public AbstractNode(Map fields) { super(fields); } - public NodeImpl(ObjectId id, Map fields) { + public AbstractNode(ObjectId id, Map fields) { super(id, fields); } @@ -58,11 +56,6 @@ public class NodeImpl extends PersistedImpl implements Node { return (String) fields.get("transport_address"); } - @Override - public String getClusterAddress() { - return (String) fields.get("cluster_address"); - } - @Override public DateTime getLastSeen() { final Object rawLastSeen = fields.get("last_seen"); @@ -72,22 +65,15 @@ public class NodeImpl extends PersistedImpl implements Node { if (rawLastSeen instanceof BSONTimestamp) { return new DateTime(((BSONTimestamp) rawLastSeen).getTime() * 1000L, DateTimeZone.UTC); } - return new DateTime(((Integer) rawLastSeen) * 1000L, DateTimeZone.UTC); - } - - - @Override - public Type getType() { - if (!fields.containsKey("type")) { - return Type.SERVER; + if (rawLastSeen instanceof Timestamp ts) { + return new DateTime(ts.getTime() * 1000L, DateTimeZone.UTC); } - - return Type.valueOf(fields.get("type").toString().toUpperCase(Locale.ENGLISH)); + return new DateTime(((Integer) rawLastSeen) * 1000L, DateTimeZone.UTC); } @Override public String getHostname() { - return (String)fields.get("hostname"); + return (String) fields.get("hostname"); } @Override @@ -106,4 +92,7 @@ public class NodeImpl extends PersistedImpl implements Node { public String toString() { return getTitle(); } + + public abstract DTO toDto(); + } diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/AbstractNodeService.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/AbstractNodeService.java new file mode 100644 index 0000000000..964c30762c --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/AbstractNodeService.java @@ -0,0 +1,230 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.google.common.collect.ImmutableList; +import com.mongodb.AggregationOptions; +import com.mongodb.BasicDBObject; +import com.mongodb.DBObject; +import com.mongodb.WriteResult; +import org.bson.types.ObjectId; +import org.graylog2.Configuration; +import org.graylog2.cluster.Node; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.database.MongoConnection; +import org.graylog2.database.PersistedServiceImpl; +import org.graylog2.plugin.system.NodeId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.inject.Inject; +import java.lang.reflect.InvocationTargetException; +import java.util.Collection; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Spliterator; +import java.util.Spliterators; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import java.util.stream.StreamSupport; + +public abstract class AbstractNodeService, DTO extends NodeDto> extends PersistedServiceImpl implements NodeService { + + private static final Logger LOG = LoggerFactory.getLogger(AbstractNodeService.class); + + public static final String LAST_SEEN_FIELD = "$last_seen"; + private final long pingTimeout; + private final static Map lastSeenFieldDefinition = Map.of("last_seen", Map.of("$type", "timestamp")); + private final static DBObject addLastSeenFieldAsDate = new BasicDBObject("$addFields", Map.of("last_seen_date", Map.of("$cond", + Map.of( + "if", Map.of("$isNumber", LAST_SEEN_FIELD), + "then", Map.of("$toDate", Map.of("$toLong", LAST_SEEN_FIELD)), + "else", Map.of("$toDate", Map.of("$dateToString", Map.of("date", LAST_SEEN_FIELD))) + ) + ))); + + private final Class nodeClass; + + @Inject + public AbstractNodeService(final MongoConnection mongoConnection, Configuration configuration, Class nodeClass) { + this(mongoConnection, configuration.getStaleLeaderTimeout(), nodeClass); + } + + private AbstractNodeService(final MongoConnection mongoConnection, final int staleLeaderTimeout, Class nodeClass) { + super(mongoConnection); + this.pingTimeout = staleLeaderTimeout; + this.nodeClass = nodeClass; + } + + @Override + public boolean registerServer(NodeDto dto) { + final var params = dto.toEntityParameters(); + + final Map fields = Map.of( + "$set", params, + "$currentDate", lastSeenFieldDefinition + ); + + final WriteResult result = this.collection(nodeClass).update( + new BasicDBObject("node_id", dto.getId()), + new BasicDBObject(fields), + true, + false + ); + return result.getN() == 1; + } + + @SuppressWarnings("unchecked") + @Override + public DTO byNodeId(String nodeId) throws NodeNotFoundException { + DBObject query = new BasicDBObject("node_id", nodeId); + DBObject o = findOne(nodeClass, query); + + if (o == null || !o.containsField("node_id")) { + throw new NodeNotFoundException("Unable to find node " + nodeId); + } + + return (DTO) construct((ObjectId) o.get("_id"), o.toMap()).toDto(); + } + + protected T construct(ObjectId id, Map map) { + try { + return nodeClass + .getDeclaredConstructor(ObjectId.class, Map.class) + .newInstance(id, map); + } catch (InstantiationException | IllegalAccessException | InvocationTargetException | + NoSuchMethodException e) { + LOG.error("Could not construct node {}", nodeClass.getName(), e); + throw new RuntimeException("Could not construct node"); + } + } + + @Override + public DTO byNodeId(NodeId nodeId) throws NodeNotFoundException { + return byNodeId(nodeId.getNodeId()); + } + + @SuppressWarnings("unchecked") + @Override + public Map byNodeIds(Collection nodeIds) { + return query(nodeClass, new BasicDBObject("node_id", new BasicDBObject("$in", nodeIds))) + .stream() + .map(o -> (DTO) construct((ObjectId) o.get("_id"), o.toMap()).toDto()) + .collect(Collectors.toMap(Node::getNodeId, Function.identity())); + } + + private Stream aggregate(List pipeline) { + return cursorToStream(this.collection(nodeClass).aggregate(pipeline, AggregationOptions.builder().build())); + } + + @SuppressWarnings("unchecked") + @Override + public Map allActive() { + return aggregate(recentHeartbeat(List.of(Map.of()))) + .collect(Collectors.toMap(obj -> (String) obj.get("node_id"), obj -> (DTO) construct((ObjectId) obj.get("_id"), obj.toMap()).toDto())); + } + + private List recentHeartbeat(List> additionalMatches) { + var match = ImmutableList.builder() + .add(Map.of("$expr", Map.of("$gte", List.of("$last_seen_date", Map.of("$subtract", List.of("$$NOW", this.pingTimeout)))))) + .addAll(additionalMatches) + .build(); + return List.of( + addLastSeenFieldAsDate, + new BasicDBObject("$match", Map.of("$and", match)), + new BasicDBObject("$unset", "last_seen_date") + ); + } + + @Override + public void dropOutdated() { + var outdatedIds = aggregate(List.of( + addLastSeenFieldAsDate, + new BasicDBObject("$match", Map.of("$expr", Map.of("$lt", List.of("$last_seen_date", Map.of("$subtract", List.of("$$NOW", this.pingTimeout)))))), + new BasicDBObject("$project", Map.of("_id", "$_id")) + )) + .map(obj -> obj.get("_id")) + .toList(); + + if (!outdatedIds.isEmpty()) { + final BasicDBObject query = new BasicDBObject("_id", Map.of("$in", outdatedIds)); + destroyAll(nodeClass, query); + } + } + + private Stream cursorToStream(Iterator cursor) { + return StreamSupport.stream(Spliterators.spliteratorUnknownSize(cursor, Spliterator.ORDERED), false); + } + + /** + * Mark this node as alive and probably update some settings that may have changed since last server boot. + */ + @Override + public void markAsAlive(NodeDto dto) throws NodeNotFoundException { + BasicDBObject query = new BasicDBObject("node_id", dto.getId()); + final var params = dto.toEntityParameters(); + + final BasicDBObject update = new BasicDBObject(Map.of( + "$set", params, + "$currentDate", lastSeenFieldDefinition + )); + + final WriteResult result = super.collection(nodeClass).update(query, update); + + final int updatedDocumentsCount = result.getN(); + if (updatedDocumentsCount != 1) { + throw new NodeNotFoundException("Unable to find node " + dto.getId()); + } + } + + @Override + public boolean isOnlyLeader(NodeId nodeId) { + return aggregate(recentHeartbeat(List.of( + Map.of( + "node_id", new BasicDBObject("$ne", nodeId.getNodeId()), + "is_leader", true + ) + ))).findAny().isEmpty(); + } + + @Override + public boolean isAnyLeaderPresent() { + return aggregate(recentHeartbeat(List.of( + Map.of( + "is_leader", true + ) + ))).findAny().isPresent(); + } + + @Override + public void ping(NodeDto dto) { + try { + markAsAlive(dto); + } catch (NodeNotFoundException e) { + LOG.warn("Did not find meta info of this node. Re-registering."); + registerServer(dto); + } + try { + // Remove old nodes that are no longer running. (Just some housekeeping) + dropOutdated(); + } catch (Exception e) { + LOG.warn("Caught exception during node ping.", e); + } + } +} diff --git a/data-node/src/main/java/org/graylog/datanode/cluster/DataNodeServiceImpl.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeClusterService.java similarity index 60% rename from data-node/src/main/java/org/graylog/datanode/cluster/DataNodeServiceImpl.java rename to graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeClusterService.java index 4f6f90361b..b1fc299573 100644 --- a/data-node/src/main/java/org/graylog/datanode/cluster/DataNodeServiceImpl.java +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeClusterService.java @@ -14,23 +14,18 @@ * along with this program. If not, see * . */ -package org.graylog.datanode.cluster; +package org.graylog2.cluster.nodes; -import org.graylog.datanode.Configuration; -import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeServiceImpl; +import org.graylog2.Configuration; import org.graylog2.database.MongoConnection; import javax.inject.Inject; -public class DataNodeServiceImpl extends NodeServiceImpl { +public class DataNodeClusterService extends AbstractNodeService { + @Inject - public DataNodeServiceImpl(final MongoConnection mongoConnection, Configuration configuration) { - super(mongoConnection, configuration.getStaleLeaderTimeout()); + public DataNodeClusterService(MongoConnection mongoConnection, Configuration configuration) { + super(mongoConnection, configuration, DataNodeEntity.class); } - @Override - public Node.Type type() { - return Node.Type.DATANODE; - } } diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeDto.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeDto.java new file mode 100644 index 0000000000..d126f420a5 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeDto.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.graylog2.cluster.nodes; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonUnwrapped; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.google.auto.value.AutoValue; +import org.graylog.security.certutil.CertRenewalService; + +import javax.annotation.Nullable; +import java.util.Map; + +@AutoValue +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonSerialize(as = DataNodeDto.class) +@JsonDeserialize(builder = DataNodeDto.Builder.class) +public abstract class DataNodeDto extends NodeDto { + + @Nullable + @JsonProperty("cluster_address") + public abstract String getClusterAddress(); + + @JsonProperty("data_node_status") + public abstract DataNodeStatus getDataNodeStatus(); + + @Nullable + @JsonUnwrapped + public abstract CertRenewalService.ProvisioningInformation getProvisioningInformation(); + + @Override + public Map toEntityParameters() { + final Map params = super.toEntityParameters(); + params.put("cluster_address", getClusterAddress()); + params.put("datanode_status", getDataNodeStatus()); + return params; + } + + public abstract Builder toBuilder(); + + @AutoValue.Builder + @JsonIgnoreProperties(ignoreUnknown = true) + public abstract static class Builder extends NodeDto.Builder { + + @JsonCreator + public static Builder builder() { + return new AutoValue_DataNodeDto.Builder(); + } + + @JsonProperty("cluster_address") + public abstract Builder setClusterAddress(String clusterAddress); + + @JsonProperty("datanode_status") + public abstract Builder setDataNodeStatus(DataNodeStatus dataNodeStatus); + + public abstract Builder setProvisioningInformation(CertRenewalService.ProvisioningInformation provisioningInformation); + + public abstract DataNodeDto build(); + + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeEntity.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeEntity.java new file mode 100644 index 0000000000..ac65f2e057 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeEntity.java @@ -0,0 +1,62 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.fasterxml.jackson.annotation.JsonCreator; +import org.bson.types.ObjectId; +import org.graylog2.database.DbEntity; + +import java.util.Map; + +@DbEntity(collection = "datanodes", titleField = "node_id") +public class DataNodeEntity extends AbstractNode { + + @JsonCreator + public DataNodeEntity(Map fields) { + super(fields); + } + + public DataNodeEntity(ObjectId id, Map fields) { + super(id, fields); + } + + public String getClusterAddress() { + return (String) fields.get("cluster_address"); + } + + public DataNodeStatus getDataNodeStatus() { + if (!fields.containsKey("datanode_status")) { + return null; + } + return DataNodeStatus.valueOf(fields.get("datanode_status").toString()); + } + + @Override + public DataNodeDto toDto() { + return DataNodeDto.Builder.builder() + .setObjectId(this.getObjectId().toHexString()) + .setId(this.getNodeId()) + .setTransportAddress(this.getTransportAddress()) + .setLastSeen(this.getLastSeen()) + .setHostname(this.getHostname()) + .setLeader(this.isLeader()) + .setClusterAddress(this.getClusterAddress()) + .setDataNodeStatus(this.getDataNodeStatus()) + .build(); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodePaginatedService.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodePaginatedService.java new file mode 100644 index 0000000000..4cc758a5db --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodePaginatedService.java @@ -0,0 +1,43 @@ +/* + * 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.graylog2.cluster.nodes; + +import org.graylog2.bindings.providers.MongoJackObjectMapperProvider; +import org.graylog2.database.MongoConnection; +import org.graylog2.database.PaginatedDbService; +import org.graylog2.database.PaginatedList; +import org.graylog2.search.SearchQuery; + +import javax.inject.Inject; +import javax.inject.Singleton; + +@Singleton +public class DataNodePaginatedService extends PaginatedDbService { + private static final String NODE_COLLECTION_NAME = "datanodes"; + + @Inject + public DataNodePaginatedService(MongoConnection mongoConnection, MongoJackObjectMapperProvider mapper) { + super(mongoConnection, mapper, DataNodeDto.class, NODE_COLLECTION_NAME); + } + + public PaginatedList searchPaginated(SearchQuery query, + String sortByField, String sortOrder, int page, int perPage) { + return findPaginatedWithQueryFilterAndSort(query.toDBQuery(), node -> true, + getSortBuilder(sortOrder, sortByField), page, perPage); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeStatus.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeStatus.java new file mode 100644 index 0000000000..3a565bf13f --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/DataNodeStatus.java @@ -0,0 +1,21 @@ +/* + * 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.graylog2.cluster.nodes; + +public enum DataNodeStatus { + UNCONFIGURED, STARTING, AVAILABLE, UNAVAILABLE, REMOVING, REMOVED +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/NodeDto.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/NodeDto.java new file mode 100644 index 0000000000..aaeddd84ba --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/NodeDto.java @@ -0,0 +1,96 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.fasterxml.jackson.annotation.JsonProperty; +import de.undercouch.bson4jackson.types.Timestamp; +import org.graylog2.cluster.Node; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; + +import javax.annotation.Nullable; +import java.util.HashMap; +import java.util.Map; + +public abstract class NodeDto implements Node { + + @JsonProperty("object_id") + @Nullable + public abstract String getObjectId(); + + @JsonProperty("id") + public abstract String getId(); + + @JsonProperty("node_id") + public String getNodeId() { + return getId(); + } + + @JsonProperty("transport_address") + public abstract String getTransportAddress(); + + @JsonProperty("last_seen") + @Nullable + public abstract DateTime getLastSeen(); + + @JsonProperty("hostname") + @Nullable + public abstract String getHostname(); + + @JsonProperty("is_leader") + public abstract boolean isLeader(); + + @JsonProperty("short_node_id") + public String getShortNodeId() { + return getId().split("-")[0]; + } + + public Map toEntityParameters() { + HashMap params = new HashMap<>(); + params.put("node_id", getNodeId()); + params.put("transport_address", getTransportAddress()); + params.put("is_leader", isLeader()); + params.put("hostname", getHostname()); + return params; + } + + public abstract static class Builder> { + @JsonProperty("_id") + public abstract B setObjectId(String id); + + @JsonProperty("node_id") + public abstract B setId(String id); + + @JsonProperty("transport_address") + public abstract B setTransportAddress(String transportAddress); + + @JsonProperty("last_seen") + public B setLastSeen(Timestamp timestamp) { + return setLastSeen(new DateTime(timestamp.getTime() * 1000L, DateTimeZone.UTC)); + } + + public abstract B setLastSeen(DateTime lastSeen); + + @JsonProperty("hostname") + public abstract B setHostname(String hostname); + + @JsonProperty("is_leader") + public abstract B setLeader(boolean leader); + + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/NodeService.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/NodeService.java new file mode 100644 index 0000000000..8f45492a57 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/NodeService.java @@ -0,0 +1,51 @@ +/* + * 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.graylog2.cluster.nodes; + +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.plugin.system.NodeId; + +import java.util.Collection; +import java.util.Map; + +public interface NodeService { + boolean registerServer(NodeDto dto); + + T byNodeId(String nodeId) throws NodeNotFoundException; + + T byNodeId(NodeId nodeId) throws NodeNotFoundException; + + Map byNodeIds(Collection nodeIds); + + Map allActive(); + + void dropOutdated(); + + void markAsAlive(NodeDto dto) throws NodeNotFoundException; + + boolean isOnlyLeader(NodeId nodeIde); + + boolean isAnyLeaderPresent(); + + /** + * Sets the last seen date of the node, additionally updating the settings. + * If the node hasn't been registered yet, it will be registered. + * + * @param dto Dto of the node to be marked as alive + */ + void ping(NodeDto dto); +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeClusterService.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeClusterService.java new file mode 100644 index 0000000000..2158411ba5 --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeClusterService.java @@ -0,0 +1,32 @@ +/* + * 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.graylog2.cluster.nodes; + +import org.graylog2.Configuration; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.database.MongoConnection; + +import javax.inject.Inject; + +public class ServerNodeClusterService extends AbstractNodeService { + + @Inject + public ServerNodeClusterService(MongoConnection mongoConnection, Configuration configuration) { + super(mongoConnection, configuration, ServerNodeEntity.class); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeDto.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeDto.java new file mode 100644 index 0000000000..050857d14c --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeDto.java @@ -0,0 +1,46 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.google.auto.value.AutoValue; + +@AutoValue +@JsonIgnoreProperties(ignoreUnknown = true) +@JsonSerialize(as = ServerNodeDto.class) +@JsonDeserialize(builder = ServerNodeDto.Builder.class) +public abstract class ServerNodeDto extends NodeDto { + + public abstract Builder toBuilder(); + + @AutoValue.Builder + @JsonIgnoreProperties(ignoreUnknown = true) + public abstract static class Builder extends NodeDto.Builder { + + @JsonCreator + public static Builder builder() { + return new AutoValue_ServerNodeDto.Builder(); + } + + public abstract ServerNodeDto build(); + + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeEntity.java b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeEntity.java new file mode 100644 index 0000000000..6e35b85dca --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/cluster/nodes/ServerNodeEntity.java @@ -0,0 +1,49 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.fasterxml.jackson.annotation.JsonCreator; +import org.bson.types.ObjectId; +import org.graylog2.database.DbEntity; + +import java.util.Map; + +@DbEntity(collection = "nodes", titleField = "node_id") +public class ServerNodeEntity extends AbstractNode { + + @JsonCreator + public ServerNodeEntity(Map fields) { + super(fields); + } + + public ServerNodeEntity(ObjectId id, Map fields) { + super(id, fields); + } + + @Override + public ServerNodeDto toDto() { + return ServerNodeDto.Builder.builder() + .setObjectId(this.getObjectId().toHexString()) + .setId(this.getNodeId()) + .setTransportAddress(this.getTransportAddress()) + .setLastSeen(this.getLastSeen()) + .setHostname(this.getHostname()) + .setLeader(this.isLeader()) + .build(); + } + +} diff --git a/graylog2-server/src/main/java/org/graylog2/commands/Server.java b/graylog2-server/src/main/java/org/graylog2/commands/Server.java index df5b1cf6ae..f588de91ac 100644 --- a/graylog2-server/src/main/java/org/graylog2/commands/Server.java +++ b/graylog2-server/src/main/java/org/graylog2/commands/Server.java @@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableList; import com.google.common.util.concurrent.Service; import com.google.common.util.concurrent.ServiceManager; import com.google.inject.Injector; +import com.google.inject.Key; import com.google.inject.Module; import com.google.inject.spi.Message; import com.mongodb.MongoException; @@ -63,8 +64,9 @@ import org.graylog2.bindings.PersistenceServicesBindings; import org.graylog2.bindings.ServerBindings; import org.graylog2.bootstrap.Main; import org.graylog2.bootstrap.ServerBootstrap; -import org.graylog2.cluster.NodeService; import org.graylog2.cluster.leader.LeaderElectionService; +import org.graylog2.cluster.nodes.NodeService; +import org.graylog2.cluster.nodes.ServerNodeDto; import org.graylog2.configuration.ContentStreamConfiguration; import org.graylog2.configuration.ElasticsearchClientConfiguration; import org.graylog2.configuration.ElasticsearchConfiguration; @@ -240,14 +242,17 @@ public class Server extends ServerBootstrap { @Override protected void startNodeRegistration(Injector injector) { // Register this node. - final NodeService nodeService = injector.getInstance(NodeService.class); + final NodeService nodeService = injector.getInstance(new Key<>() {}); final ServerStatus serverStatus = injector.getInstance(ServerStatus.class); final ActivityWriter activityWriter = injector.getInstance(ActivityWriter.class); final LeaderElectionService leaderElectionService = injector.getInstance(LeaderElectionService.class); - nodeService.registerServer(serverStatus.getNodeId().toString(), - leaderElectionService.isLeader(), - httpConfiguration.getHttpPublishUri(), - Tools.getLocalCanonicalHostname()); + nodeService.registerServer( + ServerNodeDto.Builder.builder() + .setId(serverStatus.getNodeId().toString()) + .setLeader(leaderElectionService.isLeader()) + .setTransportAddress(httpConfiguration.getHttpPublishUri().toString()) + .setHostname(Tools.getLocalCanonicalHostname()) + .build()); serverStatus.setLocalMode(isLocal()); if (leaderElectionService.isLeader() && !nodeService.isOnlyLeader(serverStatus.getNodeId())) { LOG.warn("Detected another leader in the cluster. Retrying in {} seconds to make sure it is not " diff --git a/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryModule.java b/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryModule.java index aff100da3b..58251bbb4a 100644 --- a/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryModule.java +++ b/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryModule.java @@ -19,10 +19,11 @@ package org.graylog2.configuration; import com.google.inject.AbstractModule; import com.google.inject.TypeLiteral; import org.graylog2.bindings.providers.MongoConnectionProvider; - import org.graylog2.bootstrap.preflight.PreflightConfigService; +import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.bootstrap.preflight.PreflightConfigServiceImpl; -import org.graylog2.cluster.NodeService; -import org.graylog2.cluster.NodeServiceImpl; +import org.graylog2.cluster.nodes.DataNodeClusterService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.database.MongoConnection; import java.net.URI; @@ -33,7 +34,7 @@ public class IndexerDiscoveryModule extends AbstractModule { protected void configure() { bind(new TypeLiteral>() {}).annotatedWith(IndexerHosts.class).toProvider(IndexerDiscoveryProvider.class).asEagerSingleton(); bind(Boolean.class).annotatedWith(RunsWithDataNode.class).toProvider(RunsWithDataNodeDiscoveryProvider.class).asEagerSingleton(); - bind(NodeService.class).to(NodeServiceImpl.class); + bind(new TypeLiteral>() {}).to(DataNodeClusterService.class); bind(PreflightConfigService.class).to(PreflightConfigServiceImpl.class); bind(MongoConnection.class).toProvider(MongoConnectionProvider.class); } diff --git a/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryProvider.java b/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryProvider.java index d3d3d8c353..2af86dcc39 100644 --- a/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryProvider.java +++ b/graylog2-server/src/main/java/org/graylog2/configuration/IndexerDiscoveryProvider.java @@ -17,11 +17,11 @@ package org.graylog2.configuration; import com.google.common.base.Suppliers; -import org.graylog2.bootstrap.preflight.PreflightConfig; import org.graylog2.bootstrap.preflight.PreflightConfigResult; import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,7 +43,7 @@ public class IndexerDiscoveryProvider implements Provider> { private final List hosts; private final PreflightConfigService preflightConfigService; - private final NodeService nodeService; + private final NodeService nodeService; private final Supplier> resultsCachingSupplier; @@ -51,7 +51,7 @@ public class IndexerDiscoveryProvider implements Provider> { public IndexerDiscoveryProvider( @Named("elasticsearch_hosts") List hosts, PreflightConfigService preflightConfigService, - NodeService nodeService) { + NodeService nodeService) { this.hosts = hosts; this.preflightConfigService = preflightConfigService; this.nodeService = nodeService; @@ -91,7 +91,7 @@ public class IndexerDiscoveryProvider implements Provider> { } private List discover() { - return nodeService.allActive(Node.Type.DATANODE).values().stream() + return nodeService.allActive().values().stream() .map(Node::getTransportAddress) .map(URI::create) .collect(Collectors.toList()); diff --git a/graylog2-server/src/main/java/org/graylog2/configuration/RunsWithDataNodeDiscoveryProvider.java b/graylog2-server/src/main/java/org/graylog2/configuration/RunsWithDataNodeDiscoveryProvider.java index 0c8ba6a5e9..c86aabfdf1 100644 --- a/graylog2-server/src/main/java/org/graylog2/configuration/RunsWithDataNodeDiscoveryProvider.java +++ b/graylog2-server/src/main/java/org/graylog2/configuration/RunsWithDataNodeDiscoveryProvider.java @@ -17,19 +17,17 @@ package org.graylog2.configuration; import com.google.common.base.Suppliers; -import org.graylog2.bootstrap.preflight.PreflightConfig; import org.graylog2.bootstrap.preflight.PreflightConfigResult; import org.graylog2.bootstrap.preflight.PreflightConfigService; import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import javax.inject.Inject; -import javax.inject.Named; import javax.inject.Provider; import java.net.URI; -import java.util.Collections; import java.util.List; import java.util.function.Supplier; import java.util.stream.Collectors; @@ -39,14 +37,14 @@ public class RunsWithDataNodeDiscoveryProvider implements Provider { private static final Logger LOG = LoggerFactory.getLogger(RunsWithDataNodeDiscoveryProvider.class); private final PreflightConfigService preflightConfigService; - private final NodeService nodeService; + private final NodeService nodeService; private final Supplier resultsCachingSupplier; @Inject public RunsWithDataNodeDiscoveryProvider( PreflightConfigService preflightConfigService, - NodeService nodeService) { + NodeService nodeService) { this.preflightConfigService = preflightConfigService; this.nodeService = nodeService; this.resultsCachingSupplier = Suppliers.memoize(this::doGet); @@ -73,7 +71,7 @@ public class RunsWithDataNodeDiscoveryProvider implements Provider { } private List discover() { - return nodeService.allActive(Node.Type.DATANODE).values().stream() + return nodeService.allActive().values().stream() .map(Node::getTransportAddress) .map(URI::create) .collect(Collectors.toList()); diff --git a/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java index 64fac741f9..cff338395c 100644 --- a/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java +++ b/graylog2-server/src/main/java/org/graylog2/datanode/DataNodeServiceImpl.java @@ -18,7 +18,8 @@ package org.graylog2.datanode; import org.graylog2.cluster.Node; import org.graylog2.cluster.NodeNotFoundException; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.events.ClusterEventBus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,10 +31,10 @@ public class DataNodeServiceImpl implements DataNodeService { private static final Logger LOG = LoggerFactory.getLogger(DataNodeServiceImpl.class); private final ClusterEventBus clusterEventBus; - private final NodeService nodeService; + private final NodeService nodeService; @Inject - public DataNodeServiceImpl(ClusterEventBus clusterEventBus, NodeService nodeService) { + public DataNodeServiceImpl(ClusterEventBus clusterEventBus, NodeService nodeService) { this.clusterEventBus = clusterEventBus; this.nodeService = nodeService; } 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 5697573f21..0be6ff48f7 100644 --- a/graylog2-server/src/main/java/org/graylog2/periodical/NodePingThread.java +++ b/graylog2-server/src/main/java/org/graylog2/periodical/NodePingThread.java @@ -16,9 +16,9 @@ */ package org.graylog2.periodical; -import org.graylog2.cluster.NodeNotFoundException; -import org.graylog2.cluster.NodeService; import org.graylog2.cluster.leader.LeaderElectionService; +import org.graylog2.cluster.nodes.NodeService; +import org.graylog2.cluster.nodes.ServerNodeDto; import org.graylog2.configuration.HttpConfiguration; import org.graylog2.plugin.ServerStatus; import org.graylog2.plugin.Tools; @@ -34,13 +34,13 @@ import javax.inject.Singleton; public class NodePingThread extends Periodical { private static final Logger LOG = LoggerFactory.getLogger(NodePingThread.class); - private final NodeService nodeService; + private final NodeService nodeService; private final HttpConfiguration httpConfiguration; private final ServerStatus serverStatus; private final LeaderElectionService leaderElectionService; @Inject - public NodePingThread(NodeService nodeService, + public NodePingThread(NodeService nodeService, HttpConfiguration httpConfiguration, ServerStatus serverStatus, LeaderElectionService leaderElectionService) { this.nodeService = nodeService; @@ -53,21 +53,13 @@ public class NodePingThread extends Periodical { // This method is "synchronized" because we are also calling it directly in AutomaticLeaderElectionService public synchronized void doRun() { final boolean isLeader = leaderElectionService.isLeader(); - try { - 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(), - isLeader, - httpConfiguration.getHttpPublishUri().resolve(HttpConfiguration.PATH_API), - Tools.getLocalCanonicalHostname()); - } - try { - // Remove old nodes that are no longer running. (Just some housekeeping) - nodeService.dropOutdated(); - } catch (Exception e) { - LOG.warn("Caught exception during node ping.", e); - } + ServerNodeDto dto = ServerNodeDto.Builder.builder() + .setId(serverStatus.getNodeId().getNodeId()) + .setLeader(isLeader) + .setTransportAddress(httpConfiguration.getHttpPublishUri().resolve(HttpConfiguration.PATH_API).toString()) + .setHostname(Tools.getLocalCanonicalHostname()) + .build(); + nodeService.ping(dto); } @Override diff --git a/graylog2-server/src/main/java/org/graylog2/rest/models/system/cluster/responses/NodeSummary.java b/graylog2-server/src/main/java/org/graylog2/rest/models/system/cluster/responses/NodeSummary.java index 7e968cf268..af31e441cc 100644 --- a/graylog2-server/src/main/java/org/graylog2/rest/models/system/cluster/responses/NodeSummary.java +++ b/graylog2-server/src/main/java/org/graylog2/rest/models/system/cluster/responses/NodeSummary.java @@ -32,9 +32,6 @@ public abstract class NodeSummary { @JsonProperty public abstract String nodeId(); - @JsonProperty - public abstract String type(); - @Deprecated @JsonProperty("is_master") public boolean isMaster() { @@ -59,12 +56,11 @@ public abstract class NodeSummary { @JsonCreator public static NodeSummary create(@JsonProperty("cluster_id") String clusterId, @JsonProperty("node_id") String nodeId, - @JsonProperty("type") String type, @JsonProperty("is_leader") boolean isLeader, @JsonProperty("transport_address") String transportAddress, @JsonProperty("last_seen") String lastSeen, @JsonProperty("short_node_id") String shortNodeId, @JsonProperty("hostname") String hostname) { - return new AutoValue_NodeSummary(clusterId, nodeId, type, isLeader, transportAddress, lastSeen, shortNodeId, hostname); + return new AutoValue_NodeSummary(clusterId, nodeId, isLeader, transportAddress, lastSeen, shortNodeId, hostname); } } diff --git a/graylog2-server/src/main/java/org/graylog2/rest/resources/system/ClusterResource.java b/graylog2-server/src/main/java/org/graylog2/rest/resources/system/ClusterResource.java index 9805a0f654..4d6d56b856 100644 --- a/graylog2-server/src/main/java/org/graylog2/rest/resources/system/ClusterResource.java +++ b/graylog2-server/src/main/java/org/graylog2/rest/resources/system/ClusterResource.java @@ -18,29 +18,44 @@ package org.graylog2.rest.resources.system; import com.codahale.metrics.annotation.Timed; import com.eaio.uuid.UUID; +import com.google.common.collect.ImmutableMap; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import io.swagger.annotations.ApiParam; import io.swagger.annotations.ApiResponse; import io.swagger.annotations.ApiResponses; import org.apache.shiro.authz.annotation.RequiresAuthentication; +import org.graylog.events.notifications.NotificationDto; +import org.graylog.security.certutil.CertRenewalService; import org.graylog2.cluster.Node; import org.graylog2.cluster.NodeNotFoundException; import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodePaginatedService; +import org.graylog2.database.PaginatedList; import org.graylog2.plugin.Tools; import org.graylog2.plugin.cluster.ClusterConfigService; import org.graylog2.plugin.cluster.ClusterId; import org.graylog2.plugin.system.NodeId; import org.graylog2.rest.models.system.cluster.responses.NodeSummary; import org.graylog2.rest.models.system.cluster.responses.NodeSummaryList; +import org.graylog2.rest.models.tools.responses.PageListResponse; +import org.graylog2.rest.resources.entities.EntityAttribute; +import org.graylog2.rest.resources.entities.EntityDefaults; +import org.graylog2.rest.resources.entities.Sorting; +import org.graylog2.search.SearchQuery; +import org.graylog2.search.SearchQueryField; +import org.graylog2.search.SearchQueryParser; import org.graylog2.shared.rest.resources.RestResource; import javax.inject.Inject; import javax.validation.constraints.NotEmpty; +import javax.ws.rs.DefaultValue; import javax.ws.rs.GET; import javax.ws.rs.Path; import javax.ws.rs.PathParam; import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; import javax.ws.rs.core.MediaType; import java.util.ArrayList; import java.util.List; @@ -53,14 +68,39 @@ import java.util.Map; @Produces(MediaType.APPLICATION_JSON) public class ClusterResource extends RestResource { private final NodeService nodeService; + private final DataNodePaginatedService dataNodePaginatedService; + private final SearchQueryParser searchQueryParser; + private final CertRenewalService certRenewalService; + private final NodeId nodeId; private final ClusterId clusterId; + private static final ImmutableMap SEARCH_FIELD_MAPPING = ImmutableMap.builder() + .put("id", SearchQueryField.create("_id", SearchQueryField.Type.OBJECT_ID)) + .put("title", SearchQueryField.create(NotificationDto.FIELD_TITLE)) + .put("description", SearchQueryField.create(NotificationDto.FIELD_DESCRIPTION)) + .build(); + + private static final String DEFAULT_SORT_FIELD = "title"; + private static final String DEFAULT_SORT_DIRECTION = "asc"; + private static final List attributes = List.of( + EntityAttribute.builder().id("title").title("Title").build(), + EntityAttribute.builder().id("description").title("Description").build(), + EntityAttribute.builder().id("type").title("Type").build() + ); + private static final EntityDefaults settings = EntityDefaults.builder() + .sort(Sorting.create(DEFAULT_SORT_FIELD, Sorting.Direction.valueOf(DEFAULT_SORT_DIRECTION.toUpperCase(Locale.ROOT)))) + .build(); + @Inject public ClusterResource(final NodeService nodeService, - final ClusterConfigService clusterConfigService, + final DataNodePaginatedService dataNodePaginatedService, + CertRenewalService certRenewalService, final ClusterConfigService clusterConfigService, final NodeId nodeId) { this.nodeService = nodeService; + this.dataNodePaginatedService = dataNodePaginatedService; + this.certRenewalService = certRenewalService; + this.searchQueryParser = new SearchQueryParser(NotificationDto.FIELD_TITLE, SEARCH_FIELD_MAPPING); this.nodeId = nodeId; this.clusterId = clusterConfigService.getOrDefault(ClusterId.class, ClusterId.create(UUID.nilUUID().toString())); } @@ -70,7 +110,7 @@ public class ClusterResource extends RestResource { @Path("/nodes") @ApiOperation(value = "List all active nodes in this cluster.") public NodeSummaryList nodes() { - final Map nodes = nodeService.allActive(Node.Type.SERVER); + final Map nodes = nodeService.allActive(); final List nodeList = new ArrayList<>(nodes.size()); for (Node node : nodes.values()) { nodeList.add(nodeSummary(node)); @@ -79,12 +119,37 @@ public class ClusterResource extends RestResource { return NodeSummaryList.create(nodeList); } + @GET + @Timed + @Path("/datanodes") + @ApiOperation(value = "Get a paginated list of all datanodes in this cluster") + public PageListResponse dataNodes(@ApiParam(name = "page") @QueryParam("page") @DefaultValue("1") int page, + @ApiParam(name = "per_page") @QueryParam("per_page") @DefaultValue("50") int perPage, + @ApiParam(name = "query") @QueryParam("query") @DefaultValue("") String query, + @ApiParam(name = "sort", + value = "The field to sort the result on", + required = true, + allowableValues = "title,description,type") + @DefaultValue(DEFAULT_SORT_FIELD) @QueryParam("sort") String sort, + @ApiParam(name = "order", value = "The sort direction", allowableValues = "asc, desc") + @DefaultValue(DEFAULT_SORT_DIRECTION) @QueryParam("order") String order + + ) { + final SearchQuery searchQuery = searchQueryParser.parse(query); + final PaginatedList result = dataNodePaginatedService.searchPaginated(searchQuery, sort, order, page, perPage); + + final List dataNodes = certRenewalService.addProvisioningInformation(result.delegate()); + + return PageListResponse.create(query, result.pagination(), + result.grandTotal().orElse(0L), sort, order, dataNodes, attributes, settings); + } + @GET @Timed @Path("/node") @ApiOperation(value = "Information about this node.", - notes = "This is returning information of this node in context to its state in the cluster. " + - "Use the system API of the node itself to get system information.") + notes = "This is returning information of this node in context to its state in the cluster. " + + "Use the system API of the node itself to get system information.") public NodeSummary node() throws NodeNotFoundException { return nodeSummary(nodeService.byNodeId(nodeId)); } @@ -93,8 +158,8 @@ public class ClusterResource extends RestResource { @Timed @Path("/nodes/{nodeId}") @ApiOperation(value = "Information about a node.", - notes = "This is returning information of a node in context to its state in the cluster. " + - "Use the system API of the node itself to get system information.") + notes = "This is returning information of a node in context to its state in the cluster. " + + "Use the system API of the node itself to get system information.") @ApiResponses(value = { @ApiResponse(code = 404, message = "Node not found.") }) @@ -106,7 +171,6 @@ public class ClusterResource extends RestResource { return NodeSummary.create( clusterId.clusterId(), node.getNodeId(), - node.getType().toString().toLowerCase(Locale.ENGLISH), node.isLeader(), node.getTransportAddress(), Tools.getISO8601String(node.getLastSeen()), diff --git a/graylog2-server/src/main/java/org/graylog2/telemetry/rest/TelemetryService.java b/graylog2-server/src/main/java/org/graylog2/telemetry/rest/TelemetryService.java index 7c80f3c5d9..b823820164 100644 --- a/graylog2-server/src/main/java/org/graylog2/telemetry/rest/TelemetryService.java +++ b/graylog2-server/src/main/java/org/graylog2/telemetry/rest/TelemetryService.java @@ -19,8 +19,8 @@ package org.graylog2.telemetry.rest; import com.google.common.eventbus.EventBus; import com.google.common.eventbus.Subscribe; import com.google.common.hash.HashCode; -import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.indexer.cluster.ClusterAdapter; import org.graylog2.plugin.PluginMetaData; import org.graylog2.plugin.database.users.User; @@ -67,7 +67,7 @@ public class TelemetryService { private final boolean isTelemetryEnabled; private final TelemetryClusterService telemetryClusterService; private final String installationSource; - private final NodeService nodeService; + private final NodeService nodeService; @Inject public TelemetryService( @@ -83,7 +83,7 @@ public class TelemetryService { EventBus eventBus, TelemetryClusterService telemetryClusterService, @Named("installation_source") String installationSource, - NodeService nodeService) { + NodeService nodeService) { this.isTelemetryEnabled = isTelemetryEnabled; this.trafficCounterService = trafficCounterService; this.enterpriseDataProvider = enterpriseDataProvider; @@ -213,7 +213,7 @@ public class TelemetryService { } private Map getDataNodeInfo() { - final var dataNodes = nodeService.allActive(Node.Type.DATANODE); + final var dataNodes = nodeService.allActive(); return Map.of( "data_nodes_count", dataNodes.size() ); diff --git a/graylog2-server/src/test/java/org/graylog/plugins/views/search/rest/scriptingapi/response/decorators/NodeTitleDecoratorTest.java b/graylog2-server/src/test/java/org/graylog/plugins/views/search/rest/scriptingapi/response/decorators/NodeTitleDecoratorTest.java index 44b29a18db..c5ca3bd9cb 100644 --- a/graylog2-server/src/test/java/org/graylog/plugins/views/search/rest/scriptingapi/response/decorators/NodeTitleDecoratorTest.java +++ b/graylog2-server/src/test/java/org/graylog/plugins/views/search/rest/scriptingapi/response/decorators/NodeTitleDecoratorTest.java @@ -19,7 +19,6 @@ package org.graylog.plugins.views.search.rest.scriptingapi.response.decorators; import org.assertj.core.api.Assertions; import org.graylog.plugins.views.search.rest.TestSearchUser; import org.graylog.plugins.views.search.rest.scriptingapi.request.RequestedField; -import org.graylog2.cluster.Node; import org.graylog2.cluster.NodeService; import org.graylog2.cluster.TestNodeService; import org.graylog2.plugin.Message; @@ -34,7 +33,7 @@ class NodeTitleDecoratorTest { @BeforeEach void setUp() { - final NodeService nodeService = new TestNodeService(Node.Type.SERVER); + final NodeService nodeService = new TestNodeService(); nodeService.registerServer("5ca1ab1e-0000-4000-a000-000000000000", false, URI.create("http://my-host.example.com"), "my-host.example.com"); decorator = new NodeTitleDecorator(nodeService); 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 0f0b2998d4..1615dd58cf 100644 --- a/graylog2-server/src/test/java/org/graylog2/cluster/NodeServiceImplTest.java +++ b/graylog2-server/src/test/java/org/graylog2/cluster/NodeServiceImplTest.java @@ -17,15 +17,13 @@ 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.cluster.nodes.ServerNodeClusterService; import org.graylog2.plugin.Tools; -import org.graylog2.plugin.database.ValidationException; 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; @@ -35,17 +33,6 @@ import org.mockito.junit.MockitoJUnit; import org.mockito.junit.MockitoRule; import java.net.URI; -import java.sql.Timestamp; -import java.time.Duration; -import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneOffset; -import java.time.temporal.ChronoUnit; -import java.util.AbstractMap; -import java.util.Calendar; -import java.util.Locale; -import java.util.Map; -import java.util.TimeZone; import static org.assertj.core.api.Assertions.assertThat; @@ -65,12 +52,13 @@ public class NodeServiceImplTest { private Configuration configuration; private final NodeId nodeId = new SimpleNodeId(NODE_ID); - private NodeServiceImpl nodeService; + private NodeService nodeService; @Before public void setUp() throws Exception { Mockito.when(configuration.getStaleLeaderTimeout()).thenReturn(STALE_LEADER_TIMEOUT_MS); - this.nodeService = new NodeServiceImpl(mongodb.mongoConnection(), configuration); + this.nodeService = new NodeServiceImpl( + new ServerNodeClusterService(mongodb.mongoConnection(), configuration)); } @Test @@ -116,54 +104,12 @@ public class NodeServiceImplTest { 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); - - assertThat(node.isLeader()).isFalse(); - assertThat(node.getTransportAddress()).isEqualTo("http://10.0.0.1:12901"); - assertThat(node.getLastSeen()).isNotNull().isInstanceOf(DateTime.class); - } - @Test public void testAllActive() throws NodeNotFoundException { assertThat(nodeService.allActive().keySet()).isEmpty(); nodeService.registerServer(nodeId.getNodeId(), true, TRANSPORT_URI, LOCAL_CANONICAL_HOSTNAME); assertThat(nodeService.allActive().keySet()).containsExactly(nodeId.getNodeId()); - nodeService.markAsAlive(nodeId, false, TRANSPORT_URI); - assertThat(nodeService.allActive().keySet()).containsExactly(nodeId.getNodeId()); - } - - @Test - public void testLastSeenBackwardsCompatibility() throws NodeNotFoundException, ValidationException { - nodeService.registerServer(nodeId.getNodeId(), true, TRANSPORT_URI, LOCAL_CANONICAL_HOSTNAME); - final NodeImpl node = (NodeImpl) nodeService.byNodeId(nodeId); - - final long lastSeenMs = System.currentTimeMillis() - 2 * STALE_LEADER_TIMEOUT_MS; - node.getFields().put("last_seen", (int)(lastSeenMs / 1000)); - nodeService.save(node); - - final Node nodeAfterUpdate = nodeService.byNodeId(nodeId); - final long lastSeenFromDb = nodeAfterUpdate.getLastSeen().toInstant().getMillis(); - Assertions.assertThat(lastSeenMs - lastSeenFromDb).isLessThan(1000); // make sure that our lastSeen from int is the same valid date - - final Map activeNodes = nodeService.allActive(); - - // the node is stale, should not be present here - Assertions.assertThat(activeNodes).isEmpty(); - - // this should drop the node with the int timestamp, as it's at least 2xstale_delay outdated. - nodeService.dropOutdated(); - - Assertions.assertThatThrownBy(() -> nodeService.byNodeId(nodeId)) - .isInstanceOf(NodeNotFoundException.class); } + } diff --git a/graylog2-server/src/test/java/org/graylog2/cluster/TestNodeService.java b/graylog2-server/src/test/java/org/graylog2/cluster/TestNodeService.java index d2476f2369..b38cdf0cb4 100644 --- a/graylog2-server/src/test/java/org/graylog2/cluster/TestNodeService.java +++ b/graylog2-server/src/test/java/org/graylog2/cluster/TestNodeService.java @@ -16,6 +16,7 @@ */ package org.graylog2.cluster; +import org.graylog2.cluster.nodes.DataNodeStatus; import org.graylog2.plugin.system.NodeId; import org.joda.time.DateTime; import org.joda.time.DateTimeZone; @@ -35,21 +36,11 @@ import java.util.stream.Collectors; */ public class TestNodeService implements NodeService { - private final Node.Type type; private final List nodes = new LinkedList<>(); - public TestNodeService(Node.Type type) { - this.type = type; - } - - @Override - public Node.Type type() { - return type; - } - @Override public boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, String clusterUri, String hostname) { - return nodes.add(new NodeRecord(type, nodeId, isLeader, httpPublishUri.toString(), clusterUri, hostname, DateTime.now(DateTimeZone.UTC))); + return nodes.add(new NodeRecord(nodeId, isLeader, httpPublishUri.toString(), hostname, DateTime.now(DateTimeZone.UTC))); } @Override @@ -67,15 +58,9 @@ public class TestNodeService implements NodeService { return nodes.stream().filter(n -> nodeIds.contains(n.getNodeId())).collect(Collectors.toMap(Node::getNodeId, Function.identity())); } - @Override - public Map allActive(Node.Type type) { - return nodes.stream().filter(n -> n.getType().equals(type)).collect(Collectors.toMap(Node::getNodeId, Function.identity())); - } - - @Deprecated @Override public Map allActive() { - return allActive(type); + return nodes.stream().collect(Collectors.toMap(Node::getNodeId, Function.identity())); } @Override @@ -83,22 +68,8 @@ public class TestNodeService implements NodeService { return nodes.stream().anyMatch(Node::isLeader); } - @Override - public boolean isOnlyLeader(NodeId nodeId) { - return nodes.stream().filter(n -> !Objects.equals(n.getNodeId(), nodeId.getNodeId())).noneMatch(Node::isLeader); - } - - @Override - public void dropOutdated() { - throw new UnsupportedOperationException("Unsupported operation"); - } - - @Override - public void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, String clusterAddress) { - throw new UnsupportedOperationException("Unsupported operation"); - } - - record NodeRecord(Type type, String nodeId, boolean isLeader, String transportAddress, String clusterAddress, String hostname, DateTime lastSeen) implements Node { + record NodeRecord(String nodeId, boolean isLeader, String transportAddress, + String hostname, DateTime lastSeen) implements Node { @Override public String getNodeId() { @@ -115,24 +86,15 @@ public class TestNodeService implements NodeService { return transportAddress; } - @Override - public String getClusterAddress() { - return clusterAddress; - } - @Override public DateTime getLastSeen() { return lastSeen; } - @Override - public Type getType() { - return type; - } - @Override public String getHostname() { return hostname; } + } } diff --git a/graylog2-server/src/test/java/org/graylog2/cluster/nodes/DataNodeClusterServiceTest.java b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/DataNodeClusterServiceTest.java new file mode 100644 index 0000000000..6d3ab2d67b --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/DataNodeClusterServiceTest.java @@ -0,0 +1,174 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.mongodb.DBCollection; +import org.assertj.core.api.Assertions; +import org.bson.types.ObjectId; +import org.graylog.testing.mongodb.MongoDBFixtures; +import org.graylog.testing.mongodb.MongoDBInstance; +import org.graylog2.Configuration; +import org.graylog2.cluster.Node; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.plugin.Tools; +import org.graylog2.plugin.database.ValidationException; +import org.graylog2.plugin.system.NodeId; +import org.graylog2.plugin.system.SimpleNodeId; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +import java.net.URI; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +public class DataNodeClusterServiceTest { + public static final int STALE_LEADER_TIMEOUT_MS = 2000; + @Rule + public final MongoDBInstance mongodb = MongoDBInstance.createForClass(); + + private static final URI TRANSPORT_URI = URI.create("http://10.0.0.1:12900"); + private static final String LOCAL_CANONICAL_HOSTNAME = Tools.getLocalCanonicalHostname(); + private static final String NODE_ID = "28164cbe-4ad9-4c9c-a76e-088655aa7889"; + + @Rule + public final MockitoRule mockitoRule = MockitoJUnit.rule(); + + @Mock + private Configuration configuration; + private final NodeId nodeId = new SimpleNodeId(NODE_ID); + + private DataNodeClusterService nodeService; + + @Before + public void setUp() throws Exception { + Mockito.when(configuration.getStaleLeaderTimeout()).thenReturn(STALE_LEADER_TIMEOUT_MS); + this.nodeService = + new DataNodeClusterService(mongodb.mongoConnection(), configuration); + } + + @Test + @MongoDBFixtures("DataNodeClusterTest-empty.json") + public void testRegisterServer() throws Exception { + assertThat(nodeService.allActive()) + .describedAs("The collection should be empty") + .isEmpty(); + + nodeService.registerServer(DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .setDataNodeStatus(DataNodeStatus.STARTING) + .build()); + + final Node node = nodeService.byNodeId(nodeId); + + assertThat(node).isNotNull(); + assertThat(node.getHostname()).isEqualTo(LOCAL_CANONICAL_HOSTNAME); + assertThat(node.getTransportAddress()).isEqualTo(TRANSPORT_URI.toString()); + assertThat(node.isLeader()).isTrue(); + } + + @Test + @MongoDBFixtures("DataNodeClusterTest-one-node.json") + public void testRegisterServerWithExistingNode() throws Exception { + final Node node1 = nodeService.byNodeId(nodeId); + + assertThat(node1.getNodeId()) + .describedAs("There should be one existing node") + .isEqualTo(NODE_ID); + + nodeService.registerServer(DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .setDataNodeStatus(DataNodeStatus.STARTING) + .build()); + + @SuppressWarnings("deprecation") + final DBCollection collection = mongodb.mongoConnection().getDatabase().getCollection("datanodes"); + + assertThat(collection.count()) + .describedAs("There should only be one node") + .isEqualTo(1); + + final Node node2 = nodeService.byNodeId(nodeId); + + assertThat(node2).isNotNull(); + assertThat(node2.getHostname()).isEqualTo(LOCAL_CANONICAL_HOSTNAME); + assertThat(node2.getTransportAddress()).isEqualTo(TRANSPORT_URI.toString()); + assertThat(node2.isLeader()).isTrue(); + } + + @Test + public void testAllActive() throws NodeNotFoundException { + assertThat(nodeService.allActive().keySet()).isEmpty(); + nodeService.registerServer(DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .setDataNodeStatus(DataNodeStatus.STARTING) + .build()); + assertThat(nodeService.allActive().keySet()).containsExactly(nodeId.getNodeId()); + + } + + @Test + public void testLastSeenBackwardsCompatibility() throws NodeNotFoundException, ValidationException { + nodeService.registerServer(DataNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .setDataNodeStatus(DataNodeStatus.STARTING) + .build()); + + final DataNodeDto node = nodeService.byNodeId(nodeId); + + final long lastSeenMs = System.currentTimeMillis() - 2 * STALE_LEADER_TIMEOUT_MS; + + final Map fields = node.toEntityParameters(); + fields.put("last_seen", (int) (lastSeenMs / 1000)); + DataNodeEntity nodeEntity = new DataNodeEntity(new ObjectId(node.getObjectId()), fields); + + nodeService.save(nodeEntity); + + final Node nodeAfterUpdate = nodeService.byNodeId(nodeId); + final long lastSeenFromDb = nodeAfterUpdate.getLastSeen().toInstant().getMillis(); + Assertions.assertThat(lastSeenMs - lastSeenFromDb).isLessThan(1000); // make sure that our lastSeen from int is the same valid date + + final Map activeNodes = nodeService.allActive(); + + // the node is stale, should not be present here + Assertions.assertThat(activeNodes).isEmpty(); + + // this should drop the node with the int timestamp, as it's at least 2xstale_delay outdated. + nodeService.dropOutdated(); + + Assertions.assertThatThrownBy(() -> nodeService.byNodeId(nodeId)) + .isInstanceOf(NodeNotFoundException.class); + } + +} diff --git a/graylog2-server/src/test/java/org/graylog2/cluster/nodes/DataNodeEntityTest.java b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/DataNodeEntityTest.java new file mode 100644 index 0000000000..8e22dfd092 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/DataNodeEntityTest.java @@ -0,0 +1,83 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.Maps; +import org.bson.types.ObjectId; +import org.graylog2.shared.bindings.providers.ObjectMapperProvider; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.ZonedDateTime; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class DataNodeEntityTest { + private final ObjectMapper mapper = new ObjectMapperProvider().get(); + + @Test + void serialize() throws Exception { + final ZonedDateTime lastSeen = Instant.ofEpochSecond(1).atZone(ZoneOffset.UTC); + final String nodeId = "2d4cff7a-b9c4-440c-9c62-89ba1fb06211"; + final String transportAddress = "http://127.0.0.1:9200"; + final String clusterAddress = "http://127.0.0.1:9300"; + final String hostname = "graylog.local"; + final Map fields = Maps.newHashMap(); + fields.put("last_seen", (int) lastSeen.toEpochSecond()); + fields.put("node_id", nodeId); + fields.put("is_leader", true); + fields.put("transport_address", transportAddress); + fields.put("hostname", hostname); + fields.put("datanode_status", DataNodeStatus.AVAILABLE); + fields.put("cluster_address", clusterAddress); + + final String id = "61b9c2861448530c3e061283"; + final DataNodeEntity node = new DataNodeEntity(new ObjectId(id), fields); + + final JsonNode jsonNode = mapper.readTree(mapper.writeValueAsString(node)); + + assertThat(jsonNode.size()).isEqualTo(10); + + assertThat(ZonedDateTime.parse(jsonNode.path("last_seen").asText())).isEqualTo(lastSeen); + assertThat(jsonNode.path("node_id").asText()).isEqualTo(nodeId); + assertThat(jsonNode.path("is_leader").asBoolean()).isEqualTo(true); + assertThat(jsonNode.path("transport_address").asText()).isEqualTo(transportAddress); + assertThat(jsonNode.path("hostname").asText()).isEqualTo(hostname); + + assertThat(jsonNode.path("id").asText()).isEqualTo(id); + assertThat(jsonNode.path("is_master").asBoolean()).isEqualTo(true); + assertThat(jsonNode.path("short_node_id").asText()).isEqualTo("2d4cff7a"); + assertThat(jsonNode.path("data_node_status").asText()).isEqualTo(DataNodeStatus.AVAILABLE.name()); + + assertThat(node.toDto()).isEqualTo(DataNodeDto.Builder.builder() + .setLastSeen(new DateTime(lastSeen.toEpochSecond() * 1000, DateTimeZone.UTC)) + .setId(nodeId) + .setLeader(true) + .setTransportAddress(transportAddress) + .setHostname(hostname) + .setDataNodeStatus(DataNodeStatus.AVAILABLE) + .setClusterAddress(clusterAddress) + .setObjectId(id) + .build()); + } +} diff --git a/graylog2-server/src/test/java/org/graylog2/cluster/nodes/ServerNodeClusterServiceTest.java b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/ServerNodeClusterServiceTest.java new file mode 100644 index 0000000000..abcd285fd0 --- /dev/null +++ b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/ServerNodeClusterServiceTest.java @@ -0,0 +1,170 @@ +/* + * 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.graylog2.cluster.nodes; + +import com.mongodb.DBCollection; +import org.assertj.core.api.Assertions; +import org.bson.types.ObjectId; +import org.graylog.testing.mongodb.MongoDBFixtures; +import org.graylog.testing.mongodb.MongoDBInstance; +import org.graylog2.Configuration; +import org.graylog2.cluster.Node; +import org.graylog2.cluster.NodeNotFoundException; +import org.graylog2.plugin.Tools; +import org.graylog2.plugin.database.ValidationException; +import org.graylog2.plugin.system.NodeId; +import org.graylog2.plugin.system.SimpleNodeId; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +import java.net.URI; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +public class ServerNodeClusterServiceTest { + public static final int STALE_LEADER_TIMEOUT_MS = 2000; + @Rule + public final MongoDBInstance mongodb = MongoDBInstance.createForClass(); + + private static final URI TRANSPORT_URI = URI.create("http://10.0.0.1:12900"); + private static final String LOCAL_CANONICAL_HOSTNAME = Tools.getLocalCanonicalHostname(); + private static final String NODE_ID = "28164cbe-4ad9-4c9c-a76e-088655aa7889"; + + @Rule + public final MockitoRule mockitoRule = MockitoJUnit.rule(); + + @Mock + private Configuration configuration; + private final NodeId nodeId = new SimpleNodeId(NODE_ID); + + private ServerNodeClusterService nodeService; + + @Before + public void setUp() throws Exception { + Mockito.when(configuration.getStaleLeaderTimeout()).thenReturn(STALE_LEADER_TIMEOUT_MS); + this.nodeService = + new ServerNodeClusterService(mongodb.mongoConnection(), configuration); + } + + @Test + @MongoDBFixtures("ServerNodeClusterTest-empty.json") + public void testRegisterServer() throws Exception { + assertThat(nodeService.allActive()) + .describedAs("The collection should be empty") + .isEmpty(); + + nodeService.registerServer(ServerNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .build()); + + final Node node = nodeService.byNodeId(nodeId); + + assertThat(node).isNotNull(); + assertThat(node.getHostname()).isEqualTo(LOCAL_CANONICAL_HOSTNAME); + assertThat(node.getTransportAddress()).isEqualTo(TRANSPORT_URI.toString()); + assertThat(node.isLeader()).isTrue(); + } + + @Test + @MongoDBFixtures("ServerNodeClusterTest-one-node.json") + public void testRegisterServerWithExistingNode() throws Exception { + final Node node1 = nodeService.byNodeId(nodeId); + + assertThat(node1.getNodeId()) + .describedAs("There should be one existing node") + .isEqualTo(NODE_ID); + + nodeService.registerServer(ServerNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .build()); + + @SuppressWarnings("deprecation") + final DBCollection collection = mongodb.mongoConnection().getDatabase().getCollection("nodes"); + + assertThat(collection.count()) + .describedAs("There should only be one node") + .isEqualTo(1); + + final Node node2 = nodeService.byNodeId(nodeId); + + assertThat(node2).isNotNull(); + assertThat(node2.getHostname()).isEqualTo(LOCAL_CANONICAL_HOSTNAME); + assertThat(node2.getTransportAddress()).isEqualTo(TRANSPORT_URI.toString()); + assertThat(node2.isLeader()).isTrue(); + } + + @Test + public void testAllActive() throws NodeNotFoundException { + assertThat(nodeService.allActive().keySet()).isEmpty(); + nodeService.registerServer(ServerNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .build()); + assertThat(nodeService.allActive().keySet()).containsExactly(nodeId.getNodeId()); + + } + + @Test + public void testLastSeenBackwardsCompatibility() throws NodeNotFoundException, ValidationException { + nodeService.registerServer(ServerNodeDto.Builder.builder() + .setId(nodeId.getNodeId()) + .setLeader(true) + .setTransportAddress(TRANSPORT_URI.toString()) + .setHostname(LOCAL_CANONICAL_HOSTNAME) + .build()); + + final ServerNodeDto node = nodeService.byNodeId(nodeId); + + final long lastSeenMs = System.currentTimeMillis() - 2 * STALE_LEADER_TIMEOUT_MS; + + final Map fields = node.toEntityParameters(); + fields.put("last_seen", (int) (lastSeenMs / 1000)); + ServerNodeEntity nodeEntity = new ServerNodeEntity(new ObjectId(node.getObjectId()), fields); + + nodeService.save(nodeEntity); + + final Node nodeAfterUpdate = nodeService.byNodeId(nodeId); + final long lastSeenFromDb = nodeAfterUpdate.getLastSeen().toInstant().getMillis(); + Assertions.assertThat(lastSeenMs - lastSeenFromDb).isLessThan(1000); // make sure that our lastSeen from int is the same valid date + + final Map activeNodes = nodeService.allActive(); + + // the node is stale, should not be present here + Assertions.assertThat(activeNodes).isEmpty(); + + // this should drop the node with the int timestamp, as it's at least 2xstale_delay outdated. + nodeService.dropOutdated(); + + Assertions.assertThatThrownBy(() -> nodeService.byNodeId(nodeId)) + .isInstanceOf(NodeNotFoundException.class); + } + +} diff --git a/graylog2-server/src/test/java/org/graylog2/cluster/NodeImplTest.java b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/ServerNodeEntityTest.java similarity index 64% rename from graylog2-server/src/test/java/org/graylog2/cluster/NodeImplTest.java rename to graylog2-server/src/test/java/org/graylog2/cluster/nodes/ServerNodeEntityTest.java index fd02c99bb8..06fb0b5865 100644 --- a/graylog2-server/src/test/java/org/graylog2/cluster/NodeImplTest.java +++ b/graylog2-server/src/test/java/org/graylog2/cluster/nodes/ServerNodeEntityTest.java @@ -14,13 +14,15 @@ * along with this program. If not, see * . */ -package org.graylog2.cluster; +package org.graylog2.cluster.nodes; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.collect.Maps; import org.bson.types.ObjectId; import org.graylog2.shared.bindings.providers.ObjectMapperProvider; +import org.joda.time.DateTime; +import org.joda.time.DateTimeZone; import org.junit.jupiter.api.Test; import java.time.Instant; @@ -30,38 +32,47 @@ import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; -class NodeImplTest { +class ServerNodeEntityTest { private final ObjectMapper mapper = new ObjectMapperProvider().get(); @Test void serialize() throws Exception { final ZonedDateTime lastSeen = Instant.ofEpochSecond(1).atZone(ZoneOffset.UTC); + final String nodeId = "2d4cff7a-b9c4-440c-9c62-89ba1fb06211"; + final String transportAddress = "http://127.0.0.1:9000/api/"; + final String hostname = "graylog.local"; final Map fields = Maps.newHashMap(); fields.put("last_seen", (int) lastSeen.toEpochSecond()); - fields.put("node_id", "2d4cff7a-b9c4-440c-9c62-89ba1fb06211"); - fields.put("type", Node.Type.SERVER.toString()); + fields.put("node_id", nodeId); fields.put("is_leader", true); - fields.put("transport_address", "http://127.0.0.1:9000/api/"); - fields.put("cluster_address", "http://127.0.0.1:9300"); - fields.put("hostname", "graylog.local"); + fields.put("transport_address", transportAddress); + fields.put("hostname", hostname); - final NodeImpl node = new NodeImpl(new ObjectId("61b9c2861448530c3e061283"), fields); + final String id = "61b9c2861448530c3e061283"; + final ServerNodeEntity node = new ServerNodeEntity(new ObjectId(id), fields); final JsonNode jsonNode = mapper.readTree(mapper.writeValueAsString(node)); - assertThat(jsonNode.size()).isEqualTo(10); + assertThat(jsonNode.size()).isEqualTo(8); assertThat(ZonedDateTime.parse(jsonNode.path("last_seen").asText())).isEqualTo(lastSeen); - assertThat(jsonNode.path("node_id").asText()).isEqualTo("2d4cff7a-b9c4-440c-9c62-89ba1fb06211"); - assertThat(jsonNode.path("type").asText()).isEqualTo(Node.Type.SERVER.toString()); + assertThat(jsonNode.path("node_id").asText()).isEqualTo(nodeId); assertThat(jsonNode.path("is_leader").asBoolean()).isEqualTo(true); - assertThat(jsonNode.path("transport_address").asText()).isEqualTo("http://127.0.0.1:9000/api/"); - assertThat(jsonNode.path("cluster_address").asText()).isEqualTo("http://127.0.0.1:9300"); - assertThat(jsonNode.path("hostname").asText()).isEqualTo("graylog.local"); + assertThat(jsonNode.path("transport_address").asText()).isEqualTo(transportAddress); + assertThat(jsonNode.path("hostname").asText()).isEqualTo(hostname); - assertThat(jsonNode.path("id").asText()).isEqualTo("61b9c2861448530c3e061283"); + assertThat(jsonNode.path("id").asText()).isEqualTo(id); assertThat(jsonNode.path("is_master").asBoolean()).isEqualTo(true); assertThat(jsonNode.path("short_node_id").asText()).isEqualTo("2d4cff7a"); + + assertThat(node.toDto()).isEqualTo(ServerNodeDto.Builder.builder() + .setLastSeen(new DateTime(lastSeen.toEpochSecond() * 1000, DateTimeZone.UTC)) + .setId(nodeId) + .setLeader(true) + .setTransportAddress(transportAddress) + .setHostname(hostname) + .setObjectId(id) + .build()); } } diff --git a/graylog2-server/src/test/java/org/graylog2/configuration/IndexerDiscoveryProviderTest.java b/graylog2-server/src/test/java/org/graylog2/configuration/IndexerDiscoveryProviderTest.java index e1bf641ea1..196415c094 100644 --- a/graylog2-server/src/test/java/org/graylog2/configuration/IndexerDiscoveryProviderTest.java +++ b/graylog2-server/src/test/java/org/graylog2/configuration/IndexerDiscoveryProviderTest.java @@ -16,25 +16,25 @@ */ package org.graylog2.configuration; -import org.apache.commons.lang3.RandomStringUtils; import org.assertj.core.api.Assertions; -import org.graylog2.bootstrap.preflight.PreflightConfig; import org.graylog2.bootstrap.preflight.PreflightConfigResult; import org.graylog2.bootstrap.preflight.PreflightConfigService; -import org.graylog2.cluster.Node; -import org.graylog2.cluster.NodeService; -import org.graylog2.cluster.TestNodeService; -import org.graylog2.plugin.database.ValidationException; -import org.jetbrains.annotations.NotNull; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.DataNodeStatus; +import org.graylog2.cluster.nodes.NodeService; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; import javax.annotation.Nullable; import java.net.URI; import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.Optional; +import java.util.Map; import java.util.UUID; +import java.util.stream.Collectors; + +import static org.mockito.Mockito.when; class IndexerDiscoveryProviderTest { @@ -94,11 +94,23 @@ class IndexerDiscoveryProviderTest { .hasMessageStartingWith("No Datanode available"); } - private NodeService nodes(String... transportAddress) { - final NodeService service = new TestNodeService(Node.Type.DATANODE); - Arrays.stream(transportAddress) - .map(URI::create) - .forEach(address -> service.registerServer(UUID.randomUUID().toString(), false, address, "localhost")); + private NodeService nodes(String... transportAddress) { + + final NodeService service = Mockito.mock(NodeService.class); + final Map map = Arrays.stream(transportAddress) + .map(address -> DataNodeDto.Builder.builder() + .setId(UUID.randomUUID().toString()) + .setLeader(false) + .setTransportAddress(address) + .setHostname("localhost") + .setDataNodeStatus(DataNodeStatus.AVAILABLE) + .build()) + .collect(Collectors.toMap( + DataNodeDto::getId, + obj -> obj + )); + + when(service.allActive()).thenReturn(map); return service; } diff --git a/graylog2-server/src/test/java/org/graylog2/database/dbcatalog/DbEntitiesScannerTest.java b/graylog2-server/src/test/java/org/graylog2/database/dbcatalog/DbEntitiesScannerTest.java index f581f2e1b0..f6aa039f5f 100644 --- a/graylog2-server/src/test/java/org/graylog2/database/dbcatalog/DbEntitiesScannerTest.java +++ b/graylog2-server/src/test/java/org/graylog2/database/dbcatalog/DbEntitiesScannerTest.java @@ -17,7 +17,7 @@ package org.graylog2.database.dbcatalog; -import org.graylog2.cluster.NodeImpl; +import org.graylog2.cluster.nodes.ServerNodeEntity; import org.graylog2.indexer.indexset.IndexSetConfig; import org.graylog2.users.UserImpl; import org.junit.jupiter.api.Test; @@ -68,11 +68,11 @@ class DbEntitiesScannerTest { final DbEntitiesCatalog dbEntitiesCatalog = scanner.get(); final DbEntityCatalogEntry entryByCollectionName = dbEntitiesCatalog.getByCollectionName("nodes").get(); - final DbEntityCatalogEntry entryByModelClass = dbEntitiesCatalog.getByModelClass(NodeImpl.class).get(); + final DbEntityCatalogEntry entryByModelClass = dbEntitiesCatalog.getByModelClass(ServerNodeEntity.class).get(); assertSame(entryByCollectionName, entryByModelClass); - assertEquals(new DbEntityCatalogEntry("nodes", "node_id", NodeImpl.class, NOBODY_ALLOWED), entryByCollectionName); + assertEquals(new DbEntityCatalogEntry("nodes", "node_id", ServerNodeEntity.class, NOBODY_ALLOWED), entryByCollectionName); } @Test diff --git a/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java b/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java index f79ffc2437..fb2d8dbe9d 100644 --- a/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java +++ b/graylog2-server/src/test/java/org/graylog2/datanode/DataNodeServiceImplTest.java @@ -16,9 +16,9 @@ */ package org.graylog2.datanode; -import org.graylog2.cluster.Node; import org.graylog2.cluster.NodeNotFoundException; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.DataNodeDto; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.events.ClusterEventBus; import org.junit.Before; import org.junit.Test; @@ -36,7 +36,7 @@ public class DataNodeServiceImplTest { @Mock private ClusterEventBus clusterEventBus; @Mock - private NodeService nodeService; + private NodeService nodeService; private DataNodeServiceImpl classUnderTest; @@ -47,7 +47,7 @@ public class DataNodeServiceImplTest { @Test public void removeNodePublishesClusterEvent() throws NodeNotFoundException { - Node node = mock(Node.class); + DataNodeDto node = mock(DataNodeDto.class); when(node.getNodeId()).thenReturn("returnedid"); // would be the same in real life when(nodeService.byNodeId("nodeid")).thenReturn(node); classUnderTest.removeNode("nodeid"); @@ -56,7 +56,7 @@ public class DataNodeServiceImplTest { @Test public void resetNodePublishesClusterEvent() throws NodeNotFoundException { - Node node = mock(Node.class); + DataNodeDto node = mock(DataNodeDto.class); when(node.getNodeId()).thenReturn("returnedid"); // would be the same in real life when(nodeService.byNodeId("nodeid")).thenReturn(node); classUnderTest.resetNode("nodeid"); diff --git a/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceTest.java b/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceTest.java index 9975c7790d..ffc0a08605 100644 --- a/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceTest.java +++ b/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceTest.java @@ -17,7 +17,7 @@ package org.graylog2.telemetry.rest; import com.google.common.eventbus.EventBus; -import org.graylog2.cluster.NodeService; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.indexer.cluster.ClusterAdapter; import org.graylog2.plugin.PluginMetaData; import org.graylog2.plugin.database.users.User; diff --git a/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceWithDbTest.java b/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceWithDbTest.java index 1eedd8d96e..a026e5564a 100644 --- a/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceWithDbTest.java +++ b/graylog2-server/src/test/java/org/graylog2/telemetry/rest/TelemetryServiceWithDbTest.java @@ -20,8 +20,8 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.eventbus.EventBus; import org.graylog.testing.mongodb.MongoDBInstance; import org.graylog2.bindings.providers.MongoJackObjectMapperProvider; -import org.graylog2.cluster.NodeService; import org.graylog2.cluster.leader.LeaderElectionService; +import org.graylog2.cluster.nodes.NodeService; import org.graylog2.indexer.cluster.ClusterAdapter; import org.graylog2.plugin.PluginMetaData; import org.graylog2.plugin.ServerStatus; diff --git a/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/DataNodeClusterTest-empty.json b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/DataNodeClusterTest-empty.json new file mode 100644 index 0000000000..8e14a61e83 --- /dev/null +++ b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/DataNodeClusterTest-empty.json @@ -0,0 +1,4 @@ +{ + "datanodes": [ + ] +} diff --git a/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/DataNodeClusterTest-one-node.json b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/DataNodeClusterTest-one-node.json new file mode 100644 index 0000000000..1aae44a52c --- /dev/null +++ b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/DataNodeClusterTest-one-node.json @@ -0,0 +1,16 @@ +{ + "datanodes": [ + { + "_id": { + "$oid": "569e0f60c83026b93fb2ecb6" + }, + "is_leader": true, + "last_seen": 1453209426, + "transport_address": "http://10.0.0.1:12900", + "type": "SERVER", + "hostname": "host.example.com", + "node_id": "28164cbe-4ad9-4c9c-a76e-088655aa7889", + "datanode_status": "AVAILABLE" + } + ] +} diff --git a/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/ServerNodeClusterTest-empty.json b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/ServerNodeClusterTest-empty.json new file mode 100644 index 0000000000..69959ae8c9 --- /dev/null +++ b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/ServerNodeClusterTest-empty.json @@ -0,0 +1,4 @@ +{ + "nodes": [ + ] +} diff --git a/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/ServerNodeClusterTest-one-node.json b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/ServerNodeClusterTest-one-node.json new file mode 100644 index 0000000000..99b216a0b6 --- /dev/null +++ b/graylog2-server/src/test/resources/org/graylog2/cluster/nodes/ServerNodeClusterTest-one-node.json @@ -0,0 +1,13 @@ +{ + "nodes": [ + { + "_id" : { "$oid": "569e0f60c83026b93fb2ecb6"}, + "is_leader" : true, + "last_seen" : 1453209426, + "transport_address" : "http://10.0.0.1:12900", + "type" : "SERVER", + "hostname": "host.example.com", + "node_id" : "28164cbe-4ad9-4c9c-a76e-088655aa7889" + } + ] +}