opensource/graylog2-server
1
0
Fork 0
mirror of https://github.com/Graylog2/graylog2-server.git synced 2026-03-13 09:32:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

Publish and show data node state information (#17382)

Browse Source 
* publish data node status

* add paginated datanode service

* changelog

* adjust changelog

* adjust NodeImplTest

* refactor initial separation of server node and data node

* fix backward compatibility

* refactor separation of server node and data node

* add timezone

* add new service tests

* add entity tests

* inject DataNodeDto NodeService
This commit is contained in:
Matthias Oesterheld
2023-11-28 15:02:27 +01:00
committed by GitHub
parent 244cb5cdd1
commit 076db1a460
56 changed files with 1598 additions and 568 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
data-node/src/main/java/org/graylog/datanode/bindings/ServerBindings.java
View File

@@ -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<NodeService<DataNodeDto>>() {}).to(DataNodeClusterService.class);
}
private void bindDynamicFeatures() {

9
data-node/src/main/java/org/graylog/datanode/bootstrap/preflight/DataNodeConfigurationPeriodical.java
View File

@@ -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<DataNodeDto> 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<DataNodeDto> nodeService,
final NodeId nodeId,
final CsrMongoStorage csrStorage,
final CsrGenerator csrGenerator,

36
data-node/src/main/java/org/graylog/datanode/cluster/DataNodeServiceImpl.java
View File

@@ -1,36 +0,0 @@
/*
* 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
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
package org.graylog.datanode.cluster;
import org.graylog.datanode.Configuration;
import org.graylog2.cluster.Node;
import org.graylog2.cluster.NodeServiceImpl;
import org.graylog2.database.MongoConnection;
import javax.inject.Inject;
public class DataNodeServiceImpl extends NodeServiceImpl {
@Inject
public DataNodeServiceImpl(final MongoConnection mongoConnection, Configuration configuration) {
super(mongoConnection, configuration.getStaleLeaderTimeout());
}
@Override
public Node.Type type() {
return Node.Type.DATANODE;
}
}

18
data-node/src/main/java/org/graylog/datanode/commands/Server.java
View File

@@ -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<DataNodeDto> 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

10
data-node/src/main/java/org/graylog/datanode/configuration/OpensearchConfigurationProvider.java
View File

@@ -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<OpensearchConfi
private final InSecureConfiguration inSecureConfiguration;
private final DatanodeConfiguration datanodeConfiguration;
private final byte[] signingKey;
private final NodeService nodeService;
private final NodeService<DataNodeDto> nodeService;
private final PreflightConfigService preflightConfigService;
@Inject
@@ -59,7 +59,7 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
final UploadedCertFilesSecureConfiguration uploadedCertFilesSecureConfiguration,
final MongoCertSecureConfiguration mongoCertSecureConfiguration,
final InSecureConfiguration inSecureConfiguration,
final NodeService nodeService,
final NodeService<DataNodeDto> nodeService,
final PreflightConfigService preflightConfigService,
final @Named("password_secret") String passwordSecret) {
this.localConfiguration = localConfiguration;
@@ -98,7 +98,7 @@ public class OpensearchConfigurationProvider implements Provider<OpensearchConfi
if (localConfiguration.getInitialManagerNodes() != null && !localConfiguration.getInitialManagerNodes().isBlank()) {
opensearchProperties.put("cluster.initial_master_nodes", localConfiguration.getInitialManagerNodes());
} else if (isPreflight()) {
final var nodeList = String.join(",", nodeService.allActive(Node.Type.DATANODE).values().stream().map(Node::getHostname).collect(Collectors.toSet()));
final var nodeList = String.join(",", nodeService.allActive().values().stream().map(Node::getHostname).collect(Collectors.toSet()));
opensearchProperties.put("cluster.initial_master_nodes", nodeList);
}
opensearchProperties.putAll(commonOpensearchConfig(localConfiguration));

10
data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessImpl.java
View File

@@ -30,8 +30,8 @@ import org.graylog.datanode.process.ProcessState;
import org.graylog.datanode.process.ProcessStateMachine;
import org.graylog.datanode.process.StateMachineTracer;
import org.graylog.shaded.opensearch2.org.opensearch.client.RestHighLevelClient;
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.security.CustomCAX509TrustManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,11 +70,11 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
private final Queue<String> stdout;
private final Queue<String> stderr;
private final CustomCAX509TrustManager trustManager;
private final NodeService nodeService;
private final NodeService<DataNodeDto> 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<DataNodeDto> 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<String> current = nodeService.allActive(Node.Type.DATANODE).values().stream().map(Node::getClusterAddress).filter(Objects::nonNull).collect(Collectors.toSet());
final Set<String> 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());

5
data-node/src/main/java/org/graylog/datanode/management/OpensearchProcessService.java
View File

@@ -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<DataNodeDto> nodeService,
final Configuration configuration,
final NodeId nodeId) {
this.configurationProvider = configurationProvider;

52
data-node/src/main/java/org/graylog/datanode/periodicals/NodePingPeriodical.java
View File

@@ -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<DataNodeDto> nodeService;
private final NodeId nodeId;
private final Supplier<URI> opensearchBaseUri;
private final Supplier<String> opensearchClusterUri;
private final Supplier<Boolean> isLeader;
private final Configuration configuration;
private final Supplier<ProcessState> 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<DataNodeDto> nodeService, NodeId nodeId, Configuration configuration, OpensearchProcess managedOpenSearch) {
this(nodeService, nodeId, configuration, managedOpenSearch::getOpensearchBaseUrl, managedOpenSearch::getOpensearchClusterUrl, managedOpenSearch::isLeaderNode, () -> managedOpenSearch.processInfo().state());
}
NodePingPeriodical(
NodeService nodeService,
NodeService<DataNodeDto> nodeService,
NodeId nodeId,
Configuration configuration,
Supplier<URI> opensearchBaseUri,
Supplier<String> opensearchClusterUri,
Supplier<Boolean> isLeader
Supplier<Boolean> isLeader,
Supplier<ProcessState> 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());

31
data-node/src/main/java/org/graylog/datanode/process/ProcessState.java
View File

@@ -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;
}
}