-
Notifications
You must be signed in to change notification settings - Fork 565
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
9572: Enable gateway to use reverse proxy for cluster communication r=npepinpe a=npepinpe ## Description This PR adds the advertised host/port configuration to the cluster configuration of the standalone gateway. This enables the use case where the gateway and broker must communicate with a reverse proxy in between, such as when one is using a sidecar-contained based service mesh (e.g. Istio or Linkerd). To improve test-ability, I extracted the creation of the `AtomixCluster` into a Spring component which is injected when the standalone gateway starts. That way we can test the component by itself. There's still an integration test which ensures it works end-to-end of course. ## Related issues closes #9342 Co-authored-by: Nicolas Pepin-Perreault <nicolas.pepin-perreault@camunda.com>
- Loading branch information
Showing
12 changed files
with
275 additions
and
128 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,6 +10,3 @@ zeebe: | |
|
|
||
| internalApi: | ||
| host: internalHost | ||
|
|
||
| monitoringApi: | ||
| host: monitoringHost | ||
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
113 changes: 113 additions & 0 deletions
113
dist/src/main/java/io/camunda/zeebe/gateway/AtomixComponent.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,113 @@ | ||
| /* | ||
| * 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.gateway; | ||
|
|
||
| import io.atomix.cluster.AtomixCluster; | ||
| import io.atomix.cluster.AtomixClusterBuilder; | ||
| import io.atomix.cluster.discovery.BootstrapDiscoveryProvider; | ||
| import io.atomix.cluster.protocol.GroupMembershipProtocol; | ||
| import io.atomix.cluster.protocol.SwimMembershipProtocol; | ||
| import io.atomix.utils.net.Address; | ||
| import io.camunda.zeebe.gateway.impl.configuration.ClusterCfg; | ||
| import io.camunda.zeebe.gateway.impl.configuration.GatewayCfg; | ||
| import io.camunda.zeebe.gateway.impl.configuration.MembershipCfg; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.ScopedProxyMode; | ||
| import org.springframework.stereotype.Component; | ||
| import org.springframework.web.context.annotation.ApplicationScope; | ||
|
|
||
| @Component | ||
| final class AtomixComponent { | ||
| private final GatewayCfg config; | ||
|
|
||
| @Autowired | ||
| AtomixComponent(final GatewayCfg config) { | ||
| this.config = config; | ||
| } | ||
|
|
||
| @Bean("atomixCluster") | ||
| @ApplicationScope(proxyMode = ScopedProxyMode.NO) | ||
| AtomixCluster createAtomixCluster() { | ||
| final var clusterConfig = config.getCluster(); | ||
| final var membershipProtocol = createMembershipProtocol(clusterConfig.getMembership()); | ||
|
|
||
| final var builder = | ||
| AtomixCluster.builder() | ||
| .withMemberId(clusterConfig.getMemberId()) | ||
| .withMessagingInterface(clusterConfig.getHost()) | ||
| .withMessagingPort(clusterConfig.getPort()) | ||
| .withAddress( | ||
| Address.from(clusterConfig.getAdvertisedHost(), clusterConfig.getAdvertisedPort())) | ||
| .withClusterId(clusterConfig.getClusterName()) | ||
| .withMembershipProvider( | ||
| BootstrapDiscoveryProvider.builder() | ||
| .withNodes(Address.from(clusterConfig.getContactPoint())) | ||
| .build()) | ||
| .withMembershipProtocol(membershipProtocol) | ||
| .withMessageCompression(clusterConfig.getMessageCompression()); | ||
|
|
||
| if (clusterConfig.getSecurity().isEnabled()) { | ||
| applyClusterSecurityConfig(clusterConfig, builder); | ||
| } | ||
|
|
||
| return builder.build(); | ||
| } | ||
|
|
||
| private GroupMembershipProtocol createMembershipProtocol(final MembershipCfg config) { | ||
| return SwimMembershipProtocol.builder() | ||
| .withFailureTimeout(config.getFailureTimeout()) | ||
| .withGossipInterval(config.getGossipInterval()) | ||
| .withProbeInterval(config.getProbeInterval()) | ||
| .withProbeTimeout(config.getProbeTimeout()) | ||
| .withBroadcastDisputes(config.isBroadcastDisputes()) | ||
| .withBroadcastUpdates(config.isBroadcastUpdates()) | ||
| .withGossipFanout(config.getGossipFanout()) | ||
| .withNotifySuspect(config.isNotifySuspect()) | ||
| .withSuspectProbes(config.getSuspectProbes()) | ||
| .withSyncInterval(config.getSyncInterval()) | ||
| .build(); | ||
| } | ||
|
|
||
| private void applyClusterSecurityConfig( | ||
| final ClusterCfg config, final AtomixClusterBuilder builder) { | ||
| final var security = config.getSecurity(); | ||
| final var certificateChainPath = security.getCertificateChainPath(); | ||
| final var privateKeyPath = security.getPrivateKeyPath(); | ||
|
|
||
| if (certificateChainPath == null) { | ||
| throw new IllegalArgumentException( | ||
| "Expected to have a valid certificate chain path for cluster security, but none " | ||
| + "configured"); | ||
| } | ||
|
|
||
| if (privateKeyPath == null) { | ||
| throw new IllegalArgumentException( | ||
| "Expected to have a valid private key path for cluster security, but none was " | ||
| + "configured"); | ||
| } | ||
|
|
||
| if (!certificateChainPath.canRead()) { | ||
| throw new IllegalArgumentException( | ||
| String.format( | ||
| "Expected the configured cluster security certificate chain path '%s' to point to a" | ||
| + " readable file, but it does not", | ||
| certificateChainPath)); | ||
| } | ||
|
|
||
| if (!privateKeyPath.canRead()) { | ||
| throw new IllegalArgumentException( | ||
| String.format( | ||
| "Expected the configured cluster security private key path '%s' to point to a " | ||
| + "readable file, but it does not", | ||
| privateKeyPath)); | ||
| } | ||
|
|
||
| builder.withSecurity(certificateChainPath, privateKeyPath); | ||
| } | ||
| } |
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
51 changes: 51 additions & 0 deletions
51
dist/src/test/java/io/camunda/zeebe/gateway/AtomixComponentTest.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,51 @@ | ||
| /* | ||
| * 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.gateway; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
|
|
||
| import io.atomix.utils.net.Address; | ||
| import io.camunda.zeebe.gateway.impl.configuration.GatewayCfg; | ||
| import org.junit.jupiter.api.Nested; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| final class AtomixComponentTest { | ||
| private final GatewayCfg config = new GatewayCfg(); | ||
| private final AtomixComponent component = new AtomixComponent(config); | ||
|
|
||
| @Nested | ||
| final class MessagingServiceTest { | ||
| @Test | ||
| void shouldAdvertiseConfiguredAddress() { | ||
| // given | ||
| config.getCluster().setAdvertisedHost("foo").setAdvertisedPort(5); | ||
|
|
||
| // when | ||
| final var cluster = component.createAtomixCluster(); | ||
|
|
||
| // then | ||
| assertThat(cluster.getMessagingService().address()).isEqualTo(Address.from("foo", 5)); | ||
| assertThat(cluster.getMessagingService().bindingAddresses()) | ||
| .doesNotContain(Address.from("foo", 5)); | ||
| } | ||
|
|
||
| @Test | ||
| void shouldBindToCorrectAddress() { | ||
| // given | ||
| config.getCluster().setHost("foo").setPort(5).setAdvertisedPort(6).setAdvertisedHost("bar"); | ||
|
|
||
| // when | ||
| final var cluster = component.createAtomixCluster(); | ||
|
|
||
| // then | ||
| assertThat(cluster.getMessagingService().address()).isNotEqualTo(Address.from("foo", 5)); | ||
| assertThat(cluster.getMessagingService().bindingAddresses()) | ||
| .containsExactly(Address.from("foo", 5)); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.