Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

HH-242822 : Filtering upstreams #269

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

HH-242822 : Filtering upstreams #269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8b59014
HH-242822 : Add contraints for upstreams. Now it can represent only a…
vbazhmin Dec 19, 2024
1eac0f1
HH-242822 : Migrated from List to Set in ServerStore
vbazhmin Dec 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions balancing/src/main/java/ru/hh/jclient/common/balancing/BalancingUpstreamManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package ru.hh.jclient.common.balancing;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import static java.util.Objects.requireNonNull;
import java.util.Set;
Expand Down Expand Up @@ -45,7 +44,7 @@ public void updateUpstreams(Collection<String> upstreams) {
}

private void updateUpstream(@Nonnull String upstreamName) {
List<Server> servers = serverStore.getServers(upstreamName);
Set<Server> servers = serverStore.getServers(upstreamName);

if (servers.isEmpty() && serverStore.getInitialSize(upstreamName).filter(val -> val > 0).isPresent()) {
monitoring.forEach(m -> m.countUpdateIgnore(upstreamName, datacenter));
Expand All @@ -62,14 +61,16 @@ private void updateUpstream(@Nonnull String upstreamName) {
if (upstream == null) {
upstream = createUpstream(upstreamName, newConfig, servers);
} else {
upstream.update(newConfig, servers);
// TODO REPLACE servers.stream().toList() WITH SET
upstream.update(newConfig, servers.stream().toList());
}
return upstream;
});
}

private Upstream createUpstream(String upstreamName, UpstreamConfigs upstreamConfigs, List<Server> servers) {
return new Upstream(upstreamName, upstreamConfigs, servers, datacenter, allowCrossDCRequests);
private Upstream createUpstream(String upstreamName, UpstreamConfigs upstreamConfigs, Set<Server> servers) {
// TODO REPLACE servers.stream().toList() WITH SET
return new Upstream(upstreamName, upstreamConfigs, servers.stream().toList(), datacenter, allowCrossDCRequests);
}

@Override
Expand Down
4 changes: 2 additions & 2 deletions balancing/src/main/java/ru/hh/jclient/common/balancing/ServerStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
package ru.hh.jclient.common.balancing;

import java.util.Collection;
import java.util.List;
import java.util.Optional;
import java.util.Set;

public interface ServerStore {
List<Server> getServers(String serviceName);
Set<Server> getServers(String upstreamName);
Optional<Integer> getInitialSize(String serviceName);
void updateServers(String serviceName, Collection<Server> aliveServers, Collection<Server> deadServers);
}
10 changes: 5 additions & 5 deletions balancing/src/main/java/ru/hh/jclient/common/balancing/ServerStoreImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
package ru.hh.jclient.common.balancing;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
Expand All @@ -14,11 +14,11 @@ public class ServerStoreImpl implements ServerStore {
private final Map<String, Integer> initialCapacities = new HashMap<>();

@Override
public List<Server> getServers(String serviceName) {
public Set<Server> getServers(String upstreamName) {
return Optional
.ofNullable(serverList.get(serviceName))
.map(List::copyOf)
.orElseGet(List::of);
.ofNullable(serverList.get(upstreamName))
.map(Set::copyOf)
.orElseGet(Collections::emptySet);
}

@Override
Expand Down
37 changes: 21 additions & 16 deletions balancing/src/main/java/ru/hh/jclient/consul/UpstreamServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
package ru.hh.jclient.consul;

import io.netty.util.NetUtil;
import java.math.BigInteger;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
Expand Down Expand Up @@ -231,12 +233,13 @@ private void checkServersForAllUpstreamsExist(boolean throwIfError, List<String>
}
}

void updateUpstreams(Map<ServiceHealthKey, ServiceHealth> upstreams, String serviceName, String datacenter) {
Set<Server> currentServers = serverStore
.getServers(serviceName)
void updateUpstreams(Map<ServiceHealthKey, ServiceHealth> upstreams, String upstreamName, String datacenter) {
// ServerStore is backed by a SET which MUST have ordering guarantees
LinkedHashSet<Server> currentServers = serverStore
.getServers(upstreamName)
.stream()
.filter(server -> datacenter.equals(server.getDatacenter()))
.collect(Collectors.toSet());
.collect(Collectors.toCollection(LinkedHashSet::new));

Map<String, Server> serverToRemoveByAddress = currentServers.stream().collect(toMap(Server::getAddress, Function.identity()));

Expand All @@ -249,7 +252,14 @@ void updateUpstreams(Map<ServiceHealthKey, ServiceHealth> upstreams, String serv

Service service = serviceHealth.getService();

String address = Server.addressFromHostPort(getAddress(serviceHealth), service.getPort());
// A known constraint. We do not allow upstream names because it floods DNS server with resolve requests.
String ipAddress = serviceHealth.getService().getAddress();
if (!isValidIpAddress(ipAddress)) {
LOGGER.warn("Invalid ip address supplied {}", ipAddress);
continue;
}

String address = Server.addressFromHostPort(ipAddress, service.getPort());
String nodeDatacenter = serviceHealth.getNode().getDatacenter().map(this::restoreOriginalDataCenterName).orElse(null);
int serverWeight = service.getWeights().orElse(defaultWeight).getPassing();

Expand All @@ -261,16 +271,20 @@ void updateUpstreams(Map<ServiceHealthKey, ServiceHealth> upstreams, String serv
}
server.update(serverWeight, service.getMeta(), service.getTags());
}
serverStore.updateServers(serviceName, currentServers, serverToRemoveByAddress.values());
serverStore.updateServers(upstreamName, currentServers, serverToRemoveByAddress.values());
LOGGER.info(
"upstreams for {} were updated in DC {}; alive servers: {}, dead servers: {}",
serviceName,
upstreamName,
datacenter,
LOGGER.isDebugEnabled() ? currentServers : currentServers.size(),
LOGGER.isDebugEnabled() ? serverToRemoveByAddress.values() : serverToRemoveByAddress.values().size()
);
}

private static boolean isValidIpAddress(String address) {
return NetUtil.isValidIpV4Address(address) || NetUtil.isValidIpV6Address(address);
}

private boolean notSameNode(String nodeName) {
return !StringUtils.isBlank(currentNode) && !currentNode.equalsIgnoreCase(nodeName);
}
Expand All @@ -284,15 +298,6 @@ private String restoreOriginalDataCenterName(String lowerCasedDcName) {
return restoredDc;
}

private static String getAddress(ServiceHealth serviceHealth) {
String address = serviceHealth.getService().getAddress();
if (!StringUtils.isBlank(address)) {
return address;
}

return serviceHealth.getNode().getAddress();
}

ServerStore getUpstreamStore() {
return serverStore;
}
Expand Down
Loading