mirror of
https://github.com/Graylog2/graylog2-server.git
synced 2026-03-13 09:32:21 +08:00
Datanode: setting initial master nodes simplification (#16712)
* automatic initial manager nodes * fixing binding * removing unneccessary URI usage
This commit is contained in:
@@ -36,5 +36,5 @@ public interface OpensearchProcess extends ManagableProcess<OpensearchConfigurat
|
||||
List<String> stdErrLogs();
|
||||
|
||||
URI getOpensearchBaseUrl();
|
||||
URI getOpensearchClusterUrl();
|
||||
String getOpensearchClusterUrl();
|
||||
}
|
||||
|
||||
@@ -123,17 +123,9 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
|
||||
}
|
||||
|
||||
@Override
|
||||
public URI getOpensearchClusterUrl() {
|
||||
final String baseUrl = opensearchConfiguration.map(OpensearchConfiguration::getClusterBaseUrl)
|
||||
.map(httpHost -> new URIBuilder()
|
||||
.setHost(httpHost.getHostName())
|
||||
.setPort(httpHost.getPort())
|
||||
.setScheme(httpHost.getSchemeName()).toString())
|
||||
.orElseGet( () -> {
|
||||
final var hostname = DatanodeConfigurationProvider.getNodesFromConfig(configuration.getDatanodeNodeName());
|
||||
return URI.create("http://" + hostname + ":" + configuration.getOpensearchTransportPort()).toString();
|
||||
});
|
||||
return URI.create(baseUrl);
|
||||
public String getOpensearchClusterUrl() {
|
||||
final var hostname = DatanodeConfigurationProvider.getNodesFromConfig(configuration.getDatanodeNodeName());
|
||||
return URI.create(hostname + ":" + configuration.getOpensearchTransportPort()).toString();
|
||||
}
|
||||
|
||||
public void onEvent(ProcessEvent event) {
|
||||
@@ -167,15 +159,7 @@ class OpensearchProcessImpl implements OpensearchProcess, ProcessListener {
|
||||
|
||||
private void writeSeedHostsList() {
|
||||
try {
|
||||
final Set<String> current = nodeService.allActive(Node.Type.DATANODE).values().stream().map(node -> {
|
||||
try {
|
||||
return new URI(node.getClusterAddress()).getAuthority();
|
||||
} catch (URISyntaxException ex) {
|
||||
LOG.warn("Could not get host:port from {}", node);
|
||||
return null;
|
||||
}
|
||||
}).filter(Objects::nonNull).collect(Collectors.toSet());
|
||||
|
||||
final Set<String> current = nodeService.allActive(Node.Type.DATANODE).values().stream().map(Node::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: {} - {}", hostsfile, iox.getMessage());
|
||||
|
||||
@@ -36,7 +36,7 @@ public class NodePingPeriodical extends Periodical {
|
||||
private final NodeService nodeService;
|
||||
private final NodeId nodeId;
|
||||
private final Supplier<URI> opensearchBaseUri;
|
||||
private final Supplier<URI> opensearchClusterUri;
|
||||
private final Supplier<String> opensearchClusterUri;
|
||||
private final Supplier<Boolean> isLeader;
|
||||
|
||||
|
||||
@@ -49,7 +49,7 @@ public class NodePingPeriodical extends Periodical {
|
||||
NodeService nodeService,
|
||||
NodeId nodeId,
|
||||
Supplier<URI> opensearchBaseUri,
|
||||
Supplier<URI> opensearchClusterUri,
|
||||
Supplier<String> opensearchClusterUri,
|
||||
Supplier<Boolean> isLeader
|
||||
) {
|
||||
this.nodeService = nodeService;
|
||||
|
||||
@@ -31,7 +31,7 @@ class NodePingPeriodicalTest {
|
||||
|
||||
final SimpleNodeId nodeID = new SimpleNodeId("5ca1ab1e-0000-4000-a000-000000000000");
|
||||
final URI uri = URI.create("http://localhost:9200");
|
||||
final URI cluster = URI.create("http://localhost:9300");
|
||||
final String cluster = "localhost:9300";
|
||||
final NodeService nodeService = Mockito.mock(NodeService.class);
|
||||
|
||||
final NodePingPeriodical task = new NodePingPeriodical(
|
||||
@@ -57,7 +57,7 @@ class NodePingPeriodicalTest {
|
||||
|
||||
final SimpleNodeId nodeID = new SimpleNodeId("5ca1ab1e-0000-4000-a000-000000000000");
|
||||
final URI uri = URI.create("http://localhost:9200");
|
||||
final URI cluster = URI.create("http://localhost:9300");
|
||||
final String cluster = "localhost:9300";
|
||||
|
||||
final NodeService nodeService = Mockito.mock(NodeService.class);
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ import java.util.Map;
|
||||
public interface NodeService {
|
||||
Node.Type type();
|
||||
|
||||
boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, URI clusterUri, String hostname);
|
||||
boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, String clusterUri, String hostname);
|
||||
|
||||
default boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, String hostname) {
|
||||
return registerServer(nodeId, isLeader, httpPublishUri, null, hostname);
|
||||
@@ -49,7 +49,7 @@ public interface NodeService {
|
||||
|
||||
void dropOutdated();
|
||||
|
||||
void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, URI clusterAddress) throws NodeNotFoundException;
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -72,17 +72,17 @@ public class NodeServiceImpl extends PersistedServiceImpl implements NodeService
|
||||
return Node.Type.SERVER;
|
||||
}
|
||||
|
||||
private Map<String, Object> addClusterUriToMap(final Map<String, Object> orig, final URI clusterUri) {
|
||||
private Map<String, Object> addClusterUriToMap(final Map<String, Object> orig, final String clusterUri) {
|
||||
if(clusterUri == null) {
|
||||
return orig;
|
||||
}
|
||||
var newMap = new HashMap<>(orig);
|
||||
newMap.put("cluster_address", clusterUri.toString());
|
||||
newMap.put("cluster_address", clusterUri);
|
||||
return Map.copyOf(newMap);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, URI clusterUri, String hostname) {
|
||||
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(),
|
||||
@@ -181,7 +181,7 @@ public class NodeServiceImpl extends PersistedServiceImpl implements NodeService
|
||||
/**
|
||||
* Mark this node as alive and probably update some settings that may have changed since last server boot.
|
||||
*/
|
||||
public void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, URI clusterAddress) throws NodeNotFoundException {
|
||||
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,
|
||||
|
||||
@@ -48,8 +48,8 @@ public class TestNodeService implements NodeService {
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean registerServer(String nodeId, boolean isLeader, URI httpPublishUri, URI clusterUri, String hostname) {
|
||||
return nodes.add(new NodeRecord(type, nodeId, isLeader, httpPublishUri.toString(), clusterUri != null ? clusterUri.toString() : null, hostname, DateTime.now(DateTimeZone.UTC)));
|
||||
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)));
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -94,7 +94,7 @@ public class TestNodeService implements NodeService {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, URI clusterAddress) {
|
||||
public void markAsAlive(NodeId node, boolean isLeader, URI restTransportAddress, String clusterAddress) {
|
||||
throw new UnsupportedOperationException("Unsupported operation");
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user