-
Notifications
You must be signed in to change notification settings - Fork 217
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Propagate readiness status to all datacenters (#1826)
This is part of the feature of falling back to remote datacenters when the local one is unhealthy during publishing. Considering that hermes-frontend has access only to ZooKeeper in its local datacenter, it needs to have information there on the readiness status of all datacenters to select only those that are ready. Before this change, only the readiness status of the local datacenter was stored in the local ZooKeeper.
- Loading branch information
1 parent
f29ae97
commit f2cef4c
Showing
32 changed files
with
294 additions
and
244 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 |
|---|---|---|
|
|
@@ -52,7 +52,6 @@ public int hashCode() { | |
|
|
||
| public enum ReadinessStatus { | ||
| READY, | ||
| NOT_READY, | ||
| UNDEFINED | ||
| NOT_READY | ||
| } | ||
| } | ||
2 changes: 1 addition & 1 deletion
2
...k/src/jmh/java/pl/allegro/tech/hermes/benchmark/environment/DisabledReadinessChecker.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
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
14 changes: 14 additions & 0 deletions
14
...common/src/main/java/pl/allegro/tech/hermes/domain/readiness/DatacenterReadinessList.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,14 @@ | ||
| package pl.allegro.tech.hermes.domain.readiness; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonCreator; | ||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import pl.allegro.tech.hermes.api.DatacenterReadiness; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public record DatacenterReadinessList(List<DatacenterReadiness> datacenters) { | ||
| @JsonCreator | ||
| public DatacenterReadinessList(@JsonProperty("datacenters") List<DatacenterReadiness> datacenters) { | ||
| this.datacenters = datacenters; | ||
| } | ||
| } |
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -128,7 +128,7 @@ | |
| }, | ||
| { | ||
| "datacenter": "DC3", | ||
| "status": "UNDEFINED" | ||
| "status": "READY" | ||
| } | ||
| ], | ||
| "constraints": { | ||
|
|
||
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
45 changes: 45 additions & 0 deletions
45
...frontend/src/main/java/pl/allegro/tech/hermes/frontend/config/ReadinessConfiguration.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,45 @@ | ||
| package pl.allegro.tech.hermes.frontend.config; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.apache.curator.framework.CuratorFramework; | ||
| import org.springframework.boot.context.properties.EnableConfigurationProperties; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
| import pl.allegro.tech.hermes.frontend.readiness.AdminReadinessService; | ||
| import pl.allegro.tech.hermes.frontend.readiness.DefaultReadinessChecker; | ||
| import pl.allegro.tech.hermes.frontend.readiness.HealthCheckService; | ||
| import pl.allegro.tech.hermes.frontend.server.TopicMetadataLoadingRunner; | ||
| import pl.allegro.tech.hermes.infrastructure.dc.DatacenterNameProvider; | ||
| import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperPaths; | ||
|
|
||
| @Configuration | ||
| @EnableConfigurationProperties({ReadinessCheckProperties.class}) | ||
| public class ReadinessConfiguration { | ||
|
|
||
| @Bean | ||
| public DefaultReadinessChecker readinessChecker(ReadinessCheckProperties readinessCheckProperties, | ||
| TopicMetadataLoadingRunner topicMetadataLoadingRunner, | ||
| AdminReadinessService adminReadinessService) { | ||
| return new DefaultReadinessChecker( | ||
| topicMetadataLoadingRunner, | ||
| adminReadinessService, | ||
| readinessCheckProperties.isEnabled(), | ||
| readinessCheckProperties.isKafkaCheckEnabled(), | ||
| readinessCheckProperties.getInterval() | ||
| ); | ||
| } | ||
|
|
||
| @Bean(initMethod = "start", destroyMethod = "stop") | ||
| public AdminReadinessService adminReadinessService(ObjectMapper mapper, | ||
| CuratorFramework zookeeper, | ||
| ZookeeperPaths paths, | ||
| DatacenterNameProvider datacenterNameProvider) { | ||
| String localDatacenterName = datacenterNameProvider.getDatacenterName(); | ||
| return new AdminReadinessService(mapper, zookeeper, paths, localDatacenterName); | ||
| } | ||
|
|
||
| @Bean(initMethod = "startup") | ||
| public HealthCheckService healthCheckService() { | ||
| return new HealthCheckService(); | ||
| } | ||
| } |
86 changes: 86 additions & 0 deletions
86
...ontend/src/main/java/pl/allegro/tech/hermes/frontend/readiness/AdminReadinessService.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,86 @@ | ||
| package pl.allegro.tech.hermes.frontend.readiness; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.apache.curator.framework.CuratorFramework; | ||
| import org.apache.curator.framework.recipes.cache.ChildData; | ||
| import org.apache.curator.framework.recipes.cache.NodeCache; | ||
| import org.apache.curator.framework.recipes.cache.NodeCacheListener; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import pl.allegro.tech.hermes.api.DatacenterReadiness; | ||
| import pl.allegro.tech.hermes.common.exception.InternalProcessingException; | ||
| import pl.allegro.tech.hermes.domain.readiness.DatacenterReadinessList; | ||
| import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperPaths; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static pl.allegro.tech.hermes.api.DatacenterReadiness.ReadinessStatus.READY; | ||
|
|
||
| public class AdminReadinessService implements NodeCacheListener { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(AdminReadinessService.class); | ||
|
|
||
| private final NodeCache cache; | ||
| private final ObjectMapper mapper; | ||
| private final String localDatacenterName; | ||
|
|
||
| private volatile Map<String, Boolean> readinessPerDatacenter; | ||
|
|
||
| public AdminReadinessService(ObjectMapper mapper, | ||
| CuratorFramework curator, | ||
| ZookeeperPaths paths, | ||
| String localDatacenterName) { | ||
| this.mapper = mapper; | ||
| this.localDatacenterName = localDatacenterName; | ||
| this.cache = new NodeCache(curator, paths.datacenterReadinessPath()); | ||
| cache.getListenable().addListener(this); | ||
| try { | ||
| cache.start(true); | ||
| } catch (Exception e) { | ||
| throw new InternalProcessingException("Readiness cache cannot start.", e); | ||
| } | ||
| } | ||
|
|
||
| public void start() { | ||
| refreshAdminReady(); | ||
| } | ||
|
|
||
| public void stop() { | ||
| try { | ||
| cache.close(); | ||
| } catch (Exception e) { | ||
| logger.warn("Failed to stop readiness cache", e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void nodeChanged() { | ||
| refreshAdminReady(); | ||
| } | ||
|
|
||
| private void refreshAdminReady() { | ||
| try { | ||
| ChildData nodeData = cache.getCurrentData(); | ||
| if (nodeData != null) { | ||
| byte[] data = nodeData.getData(); | ||
| DatacenterReadinessList readiness = mapper.readValue(data, DatacenterReadinessList.class); | ||
| readinessPerDatacenter = readiness.datacenters().stream() | ||
| .collect(Collectors.toMap(DatacenterReadiness::getDatacenter, e -> e.getStatus() == READY)); | ||
| } else { | ||
| readinessPerDatacenter = Collections.emptyMap(); | ||
| } | ||
| } catch (Exception e) { | ||
| logger.error("Failed reloading readiness cache.", e); | ||
| } | ||
| } | ||
|
|
||
| public boolean isLocalDatacenterReady() { | ||
| return isDatacenterReady(localDatacenterName); | ||
| } | ||
|
|
||
| public boolean isDatacenterReady(String datacenter) { | ||
| return readinessPerDatacenter.getOrDefault(datacenter, true); | ||
| } | ||
| } |
Oops, something went wrong.