-
Notifications
You must be signed in to change notification settings - Fork 554
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
10354: Allow AtomixCluster to be autowired everywhere r=npepinpe a=npepinpe ## Description This PR extracts building the `AtomixCluster` from the broker into the `dist` module. Unfortunately, because many of our testing utilities break all kinds of abstraction, and our tests still heavily rely on modifying the cluster via the `BrokerCfg`, mapping the `BrokerCfg` to the `AtomixCluster` is still done in the `broker` module. However, everything is prepared to eventually move this out to the dist module itself, such that our configuration would simply instantiate the Atomix `ClusterConfig` type, and later inject a `AtomixCluster` in the broker. See the `StandaloneGateway` as an example; there is plenty of opportunity to DRY it up once both are simply configuring `ClusterConfig` directly, and it will remove intermediate layers of builders/config. In order to achieve that at the moment, since the network configuration relies on being initialized with the right broker base path, I also had to pull up the working directory into its own configuration, such that it can be autowired as well. This is a little more abstraction than I'd like, but it's due to how the `BrokerCfg` is still the central point for all the components, even those not specific to the broker. The broker configuration is thus not initialized in the system context or broker anymore, but is given to the broker already initialized. This is done via the Spring configuration `BrokerConfiguration`. It uses two beans to do this, which I dislike, but I'm not sure what's the best way to do this. I generally dislike that the configuration must be initialized anyway, it's very brittle and easy to forget, so I'd rather figure a way to avoid this in the future. I would propose a follow up which removes this, and instead enforces that we create the broker config either passing the base already, or using a factory method which does this. I can also do it here, but it will increase how much changes are here. I think in the future the cluster services should be started before the broker itself, but we can figure this out later. For now, it means that if the actuator wants to make use of them before the broker finishes starting, it will fail - this is something we can handle semi-gracefully, so I would leave it as is. ## Related issues related to #9996 Co-authored-by: Nicolas Pepin-Perreault <nicolas.pepin-perreault@camunda.com>
- Loading branch information
Showing
37 changed files
with
562 additions
and
439 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
92 changes: 0 additions & 92 deletions
92
broker/src/main/java/io/camunda/zeebe/broker/clustering/AtomixClusterFactory.java
This file was deleted.
Oops, something went wrong.
92 changes: 92 additions & 0 deletions
92
broker/src/main/java/io/camunda/zeebe/broker/clustering/ClusterConfigFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| /* | ||
| * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under | ||
| * one or more contributor license agreements. See the NOTICE file distributed | ||
| * with this work for additional information regarding copyright ownership. | ||
| * Licensed under the Zeebe Community License 1.1. You may not use this file | ||
| * except in compliance with the Zeebe Community License 1.1. | ||
| */ | ||
| package io.camunda.zeebe.broker.clustering; | ||
|
|
||
| import io.atomix.cluster.ClusterConfig; | ||
| import io.atomix.cluster.MemberConfig; | ||
| import io.atomix.cluster.NodeConfig; | ||
| import io.atomix.cluster.discovery.BootstrapDiscoveryConfig; | ||
| import io.atomix.cluster.messaging.MessagingConfig; | ||
| import io.atomix.cluster.protocol.SwimMembershipProtocolConfig; | ||
| import io.atomix.utils.net.Address; | ||
| import io.camunda.zeebe.broker.system.configuration.BrokerCfg; | ||
| import io.camunda.zeebe.broker.system.configuration.ClusterCfg; | ||
| import io.camunda.zeebe.broker.system.configuration.MembershipCfg; | ||
| import io.camunda.zeebe.broker.system.configuration.NetworkCfg; | ||
| import io.camunda.zeebe.broker.system.configuration.SocketBindingCfg; | ||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| // TODO: move this to BrokerClusterConfiguration in the dist module | ||
| public final class ClusterConfigFactory { | ||
| public ClusterConfig mapConfiguration(final BrokerCfg config) { | ||
| final var cluster = config.getCluster(); | ||
| final var name = cluster.getClusterName(); | ||
| final var discovery = discoveryConfig(cluster.getInitialContactPoints()); | ||
| final var membership = membershipConfig(cluster.getMembership()); | ||
| final var network = config.getNetwork(); | ||
|
|
||
| final var messaging = messagingConfig(cluster, network); | ||
| final var member = memberConfig(network.getInternalApi(), cluster.getNodeId()); | ||
|
|
||
| return new ClusterConfig() | ||
| .setClusterId(name) | ||
| .setNodeConfig(member) | ||
| .setDiscoveryConfig(discovery) | ||
| .setMessagingConfig(messaging) | ||
| .setProtocolConfig(membership); | ||
| } | ||
|
|
||
| private MemberConfig memberConfig(final SocketBindingCfg network, final int nodeId) { | ||
| final var advertisedAddress = | ||
| Address.from(network.getAdvertisedHost(), network.getAdvertisedPort()); | ||
|
|
||
| return new MemberConfig().setAddress(advertisedAddress).setId(String.valueOf(nodeId)); | ||
| } | ||
|
|
||
| private SwimMembershipProtocolConfig membershipConfig(final MembershipCfg config) { | ||
| return new SwimMembershipProtocolConfig() | ||
| .setBroadcastDisputes(config.isBroadcastDisputes()) | ||
| .setBroadcastUpdates(config.isBroadcastUpdates()) | ||
| .setFailureTimeout(config.getFailureTimeout()) | ||
| .setGossipFanout(config.getGossipFanout()) | ||
| .setGossipInterval(config.getGossipInterval()) | ||
| .setNotifySuspect(config.isNotifySuspect()) | ||
| .setProbeInterval(config.getProbeInterval()) | ||
| .setProbeTimeout(config.getProbeTimeout()) | ||
| .setSuspectProbes(config.getSuspectProbes()) | ||
| .setSyncInterval(config.getSyncInterval()); | ||
| } | ||
|
|
||
| private BootstrapDiscoveryConfig discoveryConfig(final Collection<String> contactPoints) { | ||
| final var nodes = | ||
| contactPoints.stream() | ||
| .map(Address::from) | ||
| .map(address -> new NodeConfig().setAddress(address)) | ||
| .collect(Collectors.toSet()); | ||
|
|
||
| return new BootstrapDiscoveryConfig().setNodes(nodes); | ||
| } | ||
|
|
||
| private MessagingConfig messagingConfig(final ClusterCfg cluster, final NetworkCfg network) { | ||
| final var messaging = | ||
| new MessagingConfig() | ||
| .setCompressionAlgorithm(cluster.getMessageCompression()) | ||
| .setInterfaces(Collections.singletonList(network.getInternalApi().getHost())) | ||
| .setPort(network.getInternalApi().getPort()); | ||
|
|
||
| if (network.getSecurity().isEnabled()) { | ||
| messaging | ||
| .setTlsEnabled(true) | ||
| .setCertificateChain(network.getSecurity().getCertificateChainPath()) | ||
| .setPrivateKey(network.getSecurity().getPrivateKeyPath()); | ||
| } | ||
| return messaging; | ||
| } | ||
| } |
Oops, something went wrong.