Skip to content

<fix>[management-network]: fix callback endpoint selection #4653

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

Open
MatheMatrix wants to merge 1 commit into
base: feature-5.5.28-management-dual-stack
Choose a base branch
from
Open

<fix>[management-network]: fix callback endpoint selection #4653

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
75 changes: 75 additions & 0 deletions core/src/main/java/org/zstack/core/Platform.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@
import static org.zstack.utils.CollectionDSL.e;
import static org.zstack.utils.CollectionDSL.map;
import static org.zstack.utils.StringDSL.ln;
import static org.zstack.utils.clouderrorcode.CloudOperationsErrorCode.ORG_ZSTACK_CORE_PLATFORM_10003;
import static org.zstack.utils.clouderrorcode.CloudOperationsErrorCode.ORG_ZSTACK_CORE_PLATFORM_10004;
import static org.zstack.utils.clouderrorcode.CloudOperationsErrorCode.ORG_ZSTACK_CORE_PLATFORM_10005;

public class Platform {
private static final CLogger logger = CLoggerImpl.getLogger(Platform.class);
Expand Down Expand Up @@ -1058,6 +1061,78 @@ public static ErrorableValue<String> getManagementServerIp(String targetIp) {
return getManagementNodeEndpointData().selectForTarget(ManagementEndpointData.EndpointType.NODE, targetIp);
}

public static ErrorableValue<List<RemoteEndpoint>> resolveRemoteEndpoints(String endpoint) {
String normalizedEndpoint = IPv6NetworkUtils.stripHostUrlBrackets(endpoint == null ? null : endpoint.trim());
if (StringUtils.isBlank(normalizedEndpoint)) {
return ErrorableValue.ofErrorCode(argerr(ORG_ZSTACK_CORE_PLATFORM_10003,
"cannot resolve an empty remote endpoint"));
}

if (NetworkUtils.isIpAddress(normalizedEndpoint)) {
return resolveRemoteEndpoints(normalizedEndpoint, Collections.singletonList(normalizedEndpoint));
}

try {
List<String> addresses = Arrays.stream(InetAddress.getAllByName(normalizedEndpoint))
.map(InetAddress::getHostAddress)
.collect(Collectors.toList());
return resolveRemoteEndpoints(normalizedEndpoint, addresses);
} catch (UnknownHostException e) {
return ErrorableValue.ofErrorCode(operr(ORG_ZSTACK_CORE_PLATFORM_10004,
"cannot resolve remote endpoint[%s]: %s", normalizedEndpoint, e.getMessage()));
}
}

public static ErrorableValue<List<RemoteEndpoint>> resolveRemoteEndpoints(String endpoint, Collection<String> addresses) {
List<RemoteEndpoint> endpoints = new ArrayList<>();
ErrorCode lastError = null;
Set<String> uniqueAddresses = new LinkedHashSet<>();
if (addresses != null) {
for (String address : addresses) {
String connectIp = normalizeManagementIp(IPv6NetworkUtils.stripHostUrlBrackets(address));
if (!NetworkUtils.isIpAddress(connectIp) || !uniqueAddresses.add(connectIp)) {
continue;
}

ErrorableValue<String> callbackIp = getManagementServerIp(connectIp);
if (callbackIp.isSuccess()) {
endpoints.add(new RemoteEndpoint(connectIp, callbackIp.result));
} else {
lastError = callbackIp.error;
}
}
}

if (!endpoints.isEmpty()) {
return ErrorableValue.of(endpoints);
}

if (lastError != null) {
return ErrorableValue.ofErrorCode(lastError);
}

return ErrorableValue.ofErrorCode(argerr(ORG_ZSTACK_CORE_PLATFORM_10005,
"remote endpoint[%s] has no valid IPv4 or IPv6 address", endpoint));
}

public static class RemoteEndpoint {
private final String connectIp;
private final String callbackIp;

public RemoteEndpoint(String connectIp, String callbackIp) {
this.connectIp = connectIp;
this.callbackIp = callbackIp;
}

public String getConnectIp() {
return connectIp;
}

public String getCallbackIp() {
return callbackIp;
}
}

public static int getManagementNodeServicePort() {
return Integer.parseInt(System.getProperty("RESTFacade.port", "8080"));
}
Expand Down
23 changes: 10 additions & 13 deletions core/src/main/java/org/zstack/core/ansible/AnsibleRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,22 +390,19 @@ private void cleanup() {

public void run(ReturnValueCompletion<Boolean> completion) {
try {
ErrorableValue<List<Platform.RemoteEndpoint>> resolvedEndpoints = Platform.resolveRemoteEndpoints(targetIp);
if (!resolvedEndpoints.isSuccess()) {
completion.fail(resolvedEndpoints.error);
return;
}
Platform.RemoteEndpoint resolvedEndpoint = resolvedEndpoints.result.get(0);
targetIp = resolvedEndpoint.getConnectIp();

String selectedManagementNodeIp = managementNodeIp;
if (selectedManagementNodeIp == null) {
if (NetworkUtils.isIpAddress(targetIp)) {
ErrorableValue<String> managementNodeEndpoint = Platform.getManagementServerIp(targetIp);
if (!managementNodeEndpoint.isSuccess()) {
completion.fail(managementNodeEndpoint.error);
return;
}
selectedManagementNodeIp = managementNodeEndpoint.result;
} else {
selectedManagementNodeIp = restf.getHostName();
}
}
if (NetworkUtils.isIpAddress(targetIp)) {
updateCheckersManagementNodeIp(selectedManagementNodeIp);
selectedManagementNodeIp = resolvedEndpoint.getCallbackIp();
}
updateCheckersManagementNodeIp(selectedManagementNodeIp);

if (!forceRun && !isNeedRun()) {
completion.success(false);
Expand Down
72 changes: 55 additions & 17 deletions core/src/main/java/org/zstack/core/rest/RESTFacadeImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,11 +243,7 @@ public static String buildCallbackUrl(String hostName, int port, String path) {
}

public static ErrorableValue<String> selectCallbackUrl(String requestUrl, Map<String, String> headers, String defaultCallbackUrl, int port, String path) {
if (headers != null && headers.keySet().stream().anyMatch(RESTConstant.CALLBACK_URL::equalsIgnoreCase)) {
return ErrorableValue.of(defaultCallbackUrl);
}

if (CoreGlobalProperty.UNIT_TEST_ON) {
if (hasCallbackUrlHeader(headers) || CoreGlobalProperty.UNIT_TEST_ON) {
return ErrorableValue.of(defaultCallbackUrl);
}

Expand All @@ -256,15 +252,41 @@ public static ErrorableValue<String> selectCallbackUrl(String requestUrl, Map<St
return ErrorableValue.of(defaultCallbackUrl);
}

if (!NetworkUtils.isIpv4Address(host) && !IPv6NetworkUtils.isIpv6Address(host)) {
return ErrorableValue.of(defaultCallbackUrl);
ErrorableValue<Platform.RemoteEndpoint> endpoint = resolveRequestEndpoint(host);
if (!endpoint.isSuccess()) {
return ErrorableValue.ofErrorCode(endpoint.error);
}
return ErrorableValue.of(buildCallbackUrl(endpoint.result.getCallbackIp(), port, path));
}

private static boolean hasCallbackUrlHeader(Map<String, String> headers) {
return headers != null && headers.keySet().stream().anyMatch(RESTConstant.CALLBACK_URL::equalsIgnoreCase);
}

private static ErrorableValue<Platform.RemoteEndpoint> resolveRequestEndpoint(String host) {
ErrorableValue<List<Platform.RemoteEndpoint>> endpoints = Platform.resolveRemoteEndpoints(host);
if (!endpoints.isSuccess()) {
return ErrorableValue.ofErrorCode(endpoints.error);
}
return ErrorableValue.of(endpoints.result.get(0));
}

private static boolean isHttpsHostname(String requestUrl, String host) {
try {
return "https".equalsIgnoreCase(new URI(requestUrl).getScheme()) && !NetworkUtils.isIpAddress(host);
} catch (URISyntaxException e) {
return false;
}
}

ErrorableValue<String> callbackIp = Platform.getManagementServerIp(host);
if (!callbackIp.isSuccess()) {
return ErrorableValue.ofErrorCode(callbackIp.error);
private static String replaceRequestHost(String requestUrl, String connectIp) {
try {
URI uri = new URI(requestUrl);
return new URI(uri.getScheme(), uri.getUserInfo(), connectIp, uri.getPort(),
uri.getPath(), uri.getQuery(), uri.getFragment()).toString();
} catch (URISyntaxException e) {
throw new CloudRuntimeException(String.format("cannot replace request host in url[%s]", requestUrl), e);
}
return ErrorableValue.of(buildCallbackUrl(callbackIp.result, port, path));
}

private static String extractRequestHost(String requestUrl) {
Expand Down Expand Up @@ -434,11 +456,27 @@ public void asyncJsonGet(final String url, final String body, Map<String, String

@Override
public void asyncJson(final String url, final String body, Map<String, String> headers, HttpMethod method, final AsyncRESTCallback callback, final TimeUnit unit, final long timeout) {
ErrorableValue<String> selectedCallbackUrl = selectCallbackUrl(url, headers, callbackUrl, port, path);
if (!selectedCallbackUrl.isSuccess()) {
callback.fail(selectedCallbackUrl.error);
return;
String selectedCallbackUrl = callbackUrl;
String targetUrl = url;
if (!hasCallbackUrlHeader(headers) && !CoreGlobalProperty.UNIT_TEST_ON) {
String host = extractRequestHost(url);
if (host != null) {
if (isHttpsHostname(url, host)) {
callback.fail(operr(ORG_ZSTACK_CORE_REST_10016,
"cannot use hostname[%s] for asynchronous HTTPS request[%s] because its resolved address cannot be pinned without changing TLS server identity",
host, url));
return;
}
ErrorableValue<Platform.RemoteEndpoint> endpoint = resolveRequestEndpoint(host);
if (!endpoint.isSuccess()) {
callback.fail(endpoint.error);
return;
}
selectedCallbackUrl = buildCallbackUrl(endpoint.result.getCallbackIp(), port, path);
targetUrl = replaceRequestHost(url, endpoint.result.getConnectIp());
}
}
final String actualTargetUrl = targetUrl;

synchronized (interceptors) {
for (BeforeAsyncJsonPostInterceptor ic : interceptors) {
Expand Down Expand Up @@ -469,7 +507,7 @@ public void asyncJson(final String url, final String body, Map<String, String> h
HttpHeaders requestHeaders = new HttpHeaders();
requestHeaders.setContentLength(body.length());
requestHeaders.set(RESTConstant.TASK_UUID, taskUuid);
requestHeaders.set(RESTConstant.CALLBACK_URL, selectedCallbackUrl.result);
requestHeaders.set(RESTConstant.CALLBACK_URL, selectedCallbackUrl);
MediaType JSON = MediaType.parseMediaType("application/json; charset=utf-8");
requestHeaders.setContentType(JSON);
if (headers != null) {
Expand Down Expand Up @@ -667,7 +705,7 @@ public long getSentAtMillis() {
logger.trace(String.format("json %s [%s], %s", method.toString(), url, req));
}

ListenableFuture<ResponseEntity<String>> f = asyncRestTemplate.exchange(url, method, req, String.class);
ListenableFuture<ResponseEntity<String>> f = asyncRestTemplate.exchange(actualTargetUrl, method, req, String.class);
f.addCallback(rsp -> {}, e -> wrapper.fail(err(ORG_ZSTACK_CORE_REST_10003, SysErrors.HTTP_ERROR, e.getLocalizedMessage())));
} catch (RestClientException e) {
logger.warn(String.format("Unable to %s to %s: %s", method.toString(), url, e.getMessage()));
Expand Down
Loading