Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
import org.apache.hugegraph.define.WorkLoad;
import org.apache.hugegraph.util.Bytes;
import org.apache.hugegraph.util.E;
import org.apache.hugegraph.util.Log;
import org.slf4j.Logger;

import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.RateLimiter;
Expand All @@ -43,6 +45,8 @@
@PreMatching
public class LoadDetectFilter implements ContainerRequestFilter {

private static final Logger LOG = Log.logger(LoadDetectFilter.class);

private static final Set<String> WHITE_API_LIST = ImmutableSet.of(
"",
"apis",
Expand All @@ -54,11 +58,30 @@ public class LoadDetectFilter implements ContainerRequestFilter {
private static final RateLimiter GC_RATE_LIMITER =
RateLimiter.create(1.0 / 30);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ 两条拒绝路径共享同一个 REJECT_LOG_RATE_LIMITER

高负载拒绝和低内存拒绝共享同一个 RateLimiter(每秒 1 个 permit)。这意味着:

  • 如果高负载拒绝消耗了 permit,紧接着的低内存拒绝就不会被记录(反之亦然)
  • 运维人员在同时出现高负载和低内存时,可能只看到一种告警,遗漏另一种

建议为两种拒绝原因使用独立的 RateLimiter:

private static final RateLimiter BUSY_LOG_LIMITER = RateLimiter.create(1.0);
private static final RateLimiter MEMORY_LOG_LIMITER = RateLimiter.create(1.0);

// Log at most 1 request per second to avoid too many logs when server is under heavy load
private static final RateLimiter REJECT_LOG_RATE_LIMITER = RateLimiter.create(1.0);

@Context
private jakarta.inject.Provider<HugeConfig> configProvider;
@Context
private jakarta.inject.Provider<WorkLoad> loadProvider;

public static boolean isWhiteAPI(ContainerRequestContext context) {
List<PathSegment> segments = context.getUriInfo().getPathSegments();
E.checkArgument(!segments.isEmpty(), "Invalid request uri '%s'",
context.getUriInfo().getPath());
String rootPath = segments.get(0).getPath();
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ 方法可见性变更:gcIfNeededprivate static 改为 protected 实例方法

原方法是 private static void gcIfNeeded(),现改为 protected boolean gcIfNeeded()。变更影响:

  1. 扩大可见性(private → protected),破坏了封装性
  2. 从静态方法变为实例方法
  3. 主要目的似乎是为了测试时 override

建议保持 private 可见性,如果需要测试可测试性,可以:

  • 使用 package-private 可见性 + @VisibleForTesting 注解
  • 或者通过注入策略接口实现

同样的建议也适用于 allowRejectLog() 方法。

return WHITE_API_LIST.contains(rootPath);
}

private static boolean gcIfNeeded() {
if (GC_RATE_LIMITER.tryAcquire(1)) {
System.gc();
return true;
}
return false;
}

@Override
public void filter(ContainerRequestContext context) {
if (LoadDetectFilter.isWhiteAPI(context)) {
Expand All @@ -70,7 +93,14 @@ public void filter(ContainerRequestContext context) {
int maxWorkerThreads = config.get(ServerOptions.MAX_WORKER_THREADS);
WorkLoad load = this.loadProvider.get();
// There will be a thread doesn't work, dedicated to statistics
if (load.incrementAndGet() >= maxWorkerThreads) {
int currentLoad = load.incrementAndGet();
if (currentLoad >= maxWorkerThreads) {
if (REJECT_LOG_RATE_LIMITER.tryAcquire()) {
LOG.warn("Rejected request due to high worker load, method={}, path={}, " +
"currentLoad={}, maxWorkerThreads={}",
context.getMethod(), context.getUriInfo().getPath(),
Comment thread
contrueCT marked this conversation as resolved.
currentLoad, maxWorkerThreads);
}
throw new ServiceUnavailableException(String.format(
"The server is too busy to process the request, " +
"you can config %s to adjust it or try again later",
Expand All @@ -83,7 +113,17 @@ public void filter(ContainerRequestContext context) {
long presumableFreeMem = (Runtime.getRuntime().maxMemory() -
allocatedMem) / Bytes.MB;
if (presumableFreeMem < minFreeMemory) {
gcIfNeeded();
boolean gcTriggered = gcIfNeeded();
long allocatedMemAfterCheck = Runtime.getRuntime().totalMemory() -
Runtime.getRuntime().freeMemory();
long recheckedFreeMem = (Runtime.getRuntime().maxMemory() -
allocatedMemAfterCheck) / Bytes.MB;
LOG.warn("Rejected request due to low free memory, method={}, path={}, " +
Comment thread
contrueCT marked this conversation as resolved.
Outdated
Comment thread
contrueCT marked this conversation as resolved.
Outdated
"presumableFreeMemMB={}, recheckedFreeMemMB={}, gcTriggered={}, " +
"minFreeMemoryMB={}",
context.getMethod(), context.getUriInfo().getPath(),
presumableFreeMem, recheckedFreeMem, gcTriggered,
minFreeMemory);
Comment thread
contrueCT marked this conversation as resolved.
Outdated
throw new ServiceUnavailableException(String.format(
"The server available memory %s(MB) is below than " +
"threshold %s(MB) and can't process the request, " +
Expand All @@ -92,18 +132,4 @@ public void filter(ContainerRequestContext context) {
ServerOptions.MIN_FREE_MEMORY.name()));
}
}

public static boolean isWhiteAPI(ContainerRequestContext context) {
List<PathSegment> segments = context.getUriInfo().getPathSegments();
E.checkArgument(!segments.isEmpty(), "Invalid request uri '%s'",
context.getUriInfo().getPath());
String rootPath = segments.get(0).getPath();
return WHITE_API_LIST.contains(rootPath);
}

private static void gcIfNeeded() {
if (GC_RATE_LIMITER.tryAcquire(1)) {
System.gc();
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.hugegraph.unit;

import org.apache.hugegraph.core.RoleElectionStateMachineTest;
import org.apache.hugegraph.unit.api.filter.LoadDetectFilterTest;
import org.apache.hugegraph.unit.api.filter.PathFilterTest;
import org.apache.hugegraph.unit.cache.CacheManagerTest;
import org.apache.hugegraph.unit.cache.CacheTest;
Expand Down Expand Up @@ -78,6 +79,7 @@
@RunWith(Suite.class)
@Suite.SuiteClasses({
/* api filter */
LoadDetectFilterTest.class,
PathFilterTest.class,

/* cache */
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.hugegraph.unit.api.filter;

import java.util.List;
import java.util.stream.Collectors;

import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.hugegraph.api.filter.LoadDetectFilter;
import org.apache.hugegraph.config.HugeConfig;
import org.apache.hugegraph.config.ServerOptions;
import org.apache.hugegraph.define.WorkLoad;
import org.apache.hugegraph.testutil.Assert;
import org.apache.hugegraph.testutil.Whitebox;
import org.apache.hugegraph.unit.BaseUnitTest;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;

import jakarta.inject.Provider;
import jakarta.ws.rs.ServiceUnavailableException;
import jakarta.ws.rs.container.ContainerRequestContext;
import jakarta.ws.rs.core.PathSegment;
import jakarta.ws.rs.core.UriInfo;

public class LoadDetectFilterTest extends BaseUnitTest {

private LoadDetectFilter loadDetectFilter;
private ContainerRequestContext requestContext;
private UriInfo uriInfo;
private WorkLoad workLoad;

@Before
public void setup() {
this.requestContext = Mockito.mock(ContainerRequestContext.class);
this.uriInfo = Mockito.mock(UriInfo.class);
this.workLoad = new WorkLoad();

Mockito.when(this.requestContext.getUriInfo()).thenReturn(this.uriInfo);
Mockito.when(this.requestContext.getMethod()).thenReturn("GET");

this.loadDetectFilter = new LoadDetectFilter();
this.setLoadProvider(this.workLoad);
this.setConfigProvider(createConfig(8, 0));
}

@Test
public void testFilter_WhiteListPathIgnored() {
setupPath("", List.of(""));
this.workLoad.incrementAndGet();

this.loadDetectFilter.filter(this.requestContext);

Assert.assertEquals(1, this.workLoad.get().get());
}

@Test
public void testFilter_RejectsWhenWorkerLoadIsTooHigh() {
setupPath("graphs/hugegraph/vertices",
List.of("graphs", "hugegraph", "vertices"));
this.setConfigProvider(createConfig(2, 0));
this.workLoad.incrementAndGet();

ServiceUnavailableException exception = (ServiceUnavailableException) Assert.assertThrows(
ServiceUnavailableException.class,
() -> this.loadDetectFilter.filter(this.requestContext));

Assert.assertContains("The server is too busy to process the request",
exception.getMessage());
Assert.assertContains(ServerOptions.MAX_WORKER_THREADS.name(),
exception.getMessage());
}
Comment thread
contrueCT marked this conversation as resolved.

@Test
public void testFilter_RejectsWhenFreeMemoryIsTooLow() {
setupPath("graphs/hugegraph/vertices",
List.of("graphs", "hugegraph", "vertices"));
this.setConfigProvider(createConfig(8, Integer.MAX_VALUE));

ServiceUnavailableException exception = (ServiceUnavailableException) Assert.assertThrows(
Comment thread
contrueCT marked this conversation as resolved.
ServiceUnavailableException.class,
() -> this.loadDetectFilter.filter(this.requestContext));

Comment thread
contrueCT marked this conversation as resolved.
Assert.assertContains("The server available memory",
exception.getMessage());
Assert.assertContains(ServerOptions.MIN_FREE_MEMORY.name(),
exception.getMessage());
}

@Test
public void testFilter_AllowsRequestWhenLoadAndMemoryAreHealthy() {
setupPath("graphs/hugegraph/vertices",
List.of("graphs", "hugegraph", "vertices"));
this.setConfigProvider(createConfig(8, 0));

this.loadDetectFilter.filter(this.requestContext);

Assert.assertEquals(1, this.workLoad.get().get());
}

private HugeConfig createConfig(int maxWorkerThreads, int minFreeMemory) {
Configuration conf = new PropertiesConfiguration();
conf.setProperty(ServerOptions.MAX_WORKER_THREADS.name(), maxWorkerThreads);
conf.setProperty(ServerOptions.MIN_FREE_MEMORY.name(), minFreeMemory);
return new HugeConfig(conf);
}

private void setupPath(String path, List<String> segments) {
List<PathSegment> pathSegments = segments.stream()
.map(this::createPathSegment)
.collect(Collectors.toList());
Mockito.when(this.uriInfo.getPath()).thenReturn(path);
Mockito.when(this.uriInfo.getPathSegments()).thenReturn(pathSegments);
}

private PathSegment createPathSegment(String path) {
PathSegment segment = Mockito.mock(PathSegment.class);
Mockito.when(segment.getPath()).thenReturn(path);
return segment;
}

private void setLoadProvider(WorkLoad workLoad) {
Whitebox.setInternalState(this.loadDetectFilter, "loadProvider",
(Provider<WorkLoad>) () -> workLoad);
}

private void setConfigProvider(HugeConfig config) {
Whitebox.setInternalState(this.loadDetectFilter, "configProvider",
(Provider<HugeConfig>) () -> config);
}
}
Loading