Skip to content

Commit

Permalink
Merge pull request #1700 from hanbingleixue/xds-flowcontrol
Browse files Browse the repository at this point in the history
Add the common module with XDS circuit breaker and rate limiting function
  • Loading branch information
Sherlockhan authored Dec 25, 2024
2 parents b669b3b + 7a8bd82 commit ff41cc7
Show file tree
Hide file tree
Showing 34 changed files with 2,448 additions and 16 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ public class XdsAbort {
/**
* The percentage of requests/ operations/ connections that will be aborted with the error code
*/
private int percentage;
private FractionalPercent percentage;

public int getHttpStatus() {
return httpStatus;
Expand All @@ -41,11 +41,11 @@ public void setHttpStatus(int httpStatus) {
this.httpStatus = httpStatus;
}

public int getPercentage() {
public FractionalPercent getPercentage() {
return percentage;
}

public void setPercentage(int percentage) {
public void setPercentage(FractionalPercent percentage) {
this.percentage = percentage;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ public class XdsDelay {
/**
* The percentage of requests on which the delay will be injected
*/
private int percentage;
private FractionalPercent percentage;

public long getFixedDelay() {
return fixedDelay;
Expand All @@ -41,11 +41,11 @@ public void setFixedDelay(long fixedDelay) {
this.fixedDelay = fixedDelay;
}

public int getPercentage() {
public FractionalPercent getPercentage() {
return percentage;
}

public void setPercentage(int percentage) {
public void setPercentage(FractionalPercent percentage) {
this.percentage = percentage;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,10 @@ public class XdsAbortTest {
@Test
public void testXdsAbort() {
XdsAbort abort = new XdsAbort();
abort.setPercentage(100);
FractionalPercent percent = new FractionalPercent();
abort.setPercentage(percent);
abort.setHttpStatus(200);
Assert.assertEquals(200, abort.getHttpStatus());
Assert.assertEquals(100, abort.getPercentage(), 0);
Assert.assertEquals(percent, abort.getPercentage());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -29,9 +29,10 @@ public class XdsDelayTest {
@Test
public void testXdsDelay() {
XdsDelay delay = new XdsDelay();
delay.setPercentage(100);
FractionalPercent fractionalPercent = new FractionalPercent();
delay.setPercentage(fractionalPercent);
delay.setFixedDelay(200L);
Assert.assertEquals(200, delay.getFixedDelay());
Assert.assertEquals(100, delay.getPercentage(), 0);
Assert.assertEquals(fractionalPercent, delay.getPercentage());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -294,7 +294,12 @@ private static XdsHttpFault parseHttpFault(HTTPFault httpFault) {

private static XdsAbort parseAbort(FaultAbort faultAbort) {
XdsAbort xdsAbort = new XdsAbort();
xdsAbort.setPercentage(faultAbort.getPercentage().getNumerator());
io.sermant.core.service.xds.entity.FractionalPercent fractionalPercent =
new io.sermant.core.service.xds.entity.FractionalPercent();
fractionalPercent.setNumerator(faultAbort.getPercentage().getNumerator());
fractionalPercent.setDenominator(DenominatorType.getValueByName(faultAbort.getPercentage()
.getDenominator().name()));
xdsAbort.setPercentage(fractionalPercent);
xdsAbort.setHttpStatus(faultAbort.getHttpStatus());
return xdsAbort;
}
Expand All @@ -303,7 +308,12 @@ private static XdsDelay parseDelay(FaultDelay faultDelay) {
XdsDelay xdsDelay = new XdsDelay();
long fixedDelay = Duration.ofSeconds(faultDelay.getFixedDelay().getSeconds()).toMillis();
xdsDelay.setFixedDelay(fixedDelay);
xdsDelay.setPercentage(faultDelay.getPercentage().getNumerator());
io.sermant.core.service.xds.entity.FractionalPercent fractionalPercent =
new io.sermant.core.service.xds.entity.FractionalPercent();
fractionalPercent.setNumerator(faultDelay.getPercentage().getNumerator());
fractionalPercent.setDenominator(DenominatorType.getValueByName(faultDelay.getPercentage()
.getDenominator().name()));
xdsDelay.setPercentage(fractionalPercent);
return xdsDelay;
}

Expand Down
22 changes: 18 additions & 4 deletions sermant-plugins/sermant-flowcontrol/config/config.yaml
Original file line number Diff line number Diff line change
@@ -1,6 +1,20 @@
# FlowControl configuration
flow.control.plugin:
useCseRule: true # whether to configure cse rules
enable-start-monitor: false # whether to enable indicator monitoring
enable-system-adaptive: false # whether to enable system adaptive flow control
enable-system-rule: false # whether to enable system rule flow control
# whether to configure cse rules
useCseRule: true
# whether to enable indicator monitoring
enable-start-monitor: false
# whether to enable system adaptive flow control
enable-system-adaptive: false
# whether to enable system rule flow control
enable-system-rule: false
xds.flow.control.config:
# Whether to enable Xds flow control
enable: false
retry:
# The specified response status codes that need to be retried. Retry will be performed when the response's status
# code matches one of the specified codes.
x-sermant-retriable-status-codes:
# The specified response header names that need to be retried. Retry will be performed when the response contains
# the specified headers.
x-sermant-retriable-header-names:
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,21 @@ public class CommonConst {
*/
public static final String REQUEST_START_TIME = "requestStartTime";

/**
* the connect for request address
*/
public static final String CONNECT = ":";

/**
* point
*/
public static final String ESCAPED_POINT = "\\.";

/**
* Default response status code
*/
public static final int DEFAULT_RESPONSE_CODE = -1;

private CommonConst() {
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
/*
* Copyright (C) 2021-2022 Huawei Technologies Co., Ltd. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/

package io.sermant.flowcontrol.common.config;

import io.sermant.core.config.common.ConfigFieldKey;
import io.sermant.core.config.common.ConfigTypeKey;
import io.sermant.core.plugin.config.PluginConfig;

import java.util.List;

/**
* retry configuration class
*
* @author zhouss
* @since 2022-01-28
*/
@ConfigTypeKey("xds.flow.control.config")
public class XdsFlowControlConfig implements PluginConfig {
/**
* Specify the response code for retry, and retry will be executed when the response code is included
*/
@ConfigFieldKey("retry.x-sermant-retriable-status-codes")
private List<String> retryStatusCodes;

/**
* Specify the response code for retry, and retry will be executed when the response header is included
*/
@ConfigFieldKey("retry.x-sermant-retriable-header-names")
private List<String> retryHeaderNames;

/**
* xds flow control switch
*/
private boolean enable;

public List<String> getRetryStatusCodes() {
return retryStatusCodes;
}

public void setRetryStatusCodes(List<String> retryStatusCodes) {
this.retryStatusCodes = retryStatusCodes;
}

public List<String> getRetryHeaderNames() {
return retryHeaderNames;
}

public void setRetryHeaderNames(List<String> retryHeaderNames) {
this.retryHeaderNames = retryHeaderNames;
}

public boolean isEnable() {
return enable;
}

public void setEnable(boolean enable) {
this.enable = enable;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
/*
* Copyright (C) 2024-2024 Sermant Authors. All rights reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package io.sermant.flowcontrol.common.core.match;

import io.sermant.core.service.xds.entity.XdsHeaderMatcher;
import io.sermant.core.service.xds.entity.XdsPathMatcher;
import io.sermant.core.service.xds.entity.XdsRoute;
import io.sermant.core.service.xds.entity.XdsRouteAction;
import io.sermant.core.service.xds.entity.XdsRouteAction.XdsClusterWeight;
import io.sermant.core.service.xds.entity.XdsRouteAction.XdsWeightedClusters;
import io.sermant.core.service.xds.entity.XdsRouteMatch;
import io.sermant.core.utils.CollectionUtils;
import io.sermant.core.utils.StringUtils;
import io.sermant.flowcontrol.common.entity.FlowControlScenario;
import io.sermant.flowcontrol.common.entity.RequestEntity;
import io.sermant.flowcontrol.common.util.RandomUtil;
import io.sermant.flowcontrol.common.xds.handler.XdsHandler;

import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
* XdsRouteMatchManager, Get BusinessEntity based on xDS routing rules
*
* @author zhp
* @since 2024-12-20
**/
public enum XdsRouteMatchManager {
/**
* singleton
*/
INSTANCE;

/**
* get matched scenario information
*
* @param requestEntity request-information
* @param serviceName service name
* @return matched business information
*/
public FlowControlScenario getMatchedScenarioInfo(RequestEntity requestEntity, String serviceName) {
FlowControlScenario scenario = new FlowControlScenario();
scenario.setServiceName(serviceName);
Optional<XdsRoute> matchedRouteOptional = getMatchedRoute(requestEntity, serviceName);
if (!matchedRouteOptional.isPresent()) {
return scenario;
}
XdsRoute matchedRoute = matchedRouteOptional.get();
scenario.setRouteName(matchedRoute.getName());
scenario.setClusterName(selectClusterByRoute(matchedRoute));
return scenario;
}

private Optional<XdsRoute> getMatchedRoute(RequestEntity requestEntity, String serviceName) {
List<XdsRoute> routes =
XdsHandler.INSTANCE.getServiceRouteByServiceName(serviceName);
for (XdsRoute route : routes) {
XdsRouteMatch routeMatch = route.getRouteMatch();

// check path matching
if (!isPathMatched(routeMatch.getPathMatcher(), requestEntity.getApiPath())) {
continue;
}

// check head matching
if (!isHeadersMatched(routeMatch.getHeaderMatchers(), requestEntity.getHeaders())) {
continue;
}
return Optional.of(route);
}
return Optional.empty();
}

private boolean isPathMatched(XdsPathMatcher matcher, String path) {
return matcher.isMatch(path);
}

private boolean isHeadersMatched(List<XdsHeaderMatcher> matchers, Map<String, String> headers) {
return matchers.stream()
.allMatch(xdsHeaderMatcher -> xdsHeaderMatcher.isMatch(headers));
}

private String selectClusterByRoute(XdsRoute matchedRoute) {
XdsRouteAction routeAction = matchedRoute.getRouteAction();
String cluster = routeAction.getCluster();
if (!routeAction.isWeighted() || routeAction.getWeightedClusters() == null) {
return cluster;
}
XdsWeightedClusters weightedClusters = routeAction.getWeightedClusters();
List<XdsClusterWeight> clusters = weightedClusters.getClusters();
int totalWeight = weightedClusters.getTotalWeight();
if (CollectionUtils.isEmpty(clusters) || totalWeight == 0) {
return StringUtils.EMPTY;
}
int randomWeight = RandomUtil.randomInt(totalWeight);

int currentWeight = 0;
for (XdsClusterWeight clusterWeight : clusters) {
currentWeight += clusterWeight.getWeight();
if (randomWeight < currentWeight) {
return clusterWeight.getClusterName();
}
}
return StringUtils.EMPTY;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,10 @@

package io.sermant.flowcontrol.common.entity;

import java.util.Collections;
import java.util.List;
import java.util.Map;

/**
* flow control response
*
Expand All @@ -33,6 +37,11 @@ public class FlowControlResponse {
*/
private Object result;

/**
* response
*/
private Map<String, List<String>> headers;

/**
* Whether to replace the actual response result, if true, replace
*/
Expand All @@ -54,6 +63,21 @@ public FlowControlResponse(String msg, int code) {
this.code = code;
}

/**
* flow control response results
*
* @param msg prompt message
* @param code response code
* @param headers response headers
* @param result response result
*/
public FlowControlResponse(String msg, int code, Map<String, List<String>> headers, Object result) {
this.msg = msg;
this.code = code;
this.headers = headers;
this.result = result;
}

/**
* flow control response results
*
Expand Down Expand Up @@ -91,4 +115,8 @@ public Object getResult() {
public boolean isReplaceResult() {
return isReplaceResult;
}

public Map<String, List<String>> getHeaders() {
return headers == null ? Collections.emptyMap() : headers;
}
}
Loading

0 comments on commit ff41cc7

Please sign in to comment.