生产更新:同步已提交的海外仓分析及业务修复(2026-10-16) #27

Merged
admin merged 138 commits from test into main 2026-10-16 18:02:59 +08:00
5 changed files with 279 additions and 0 deletions
Showing only changes of commit 4c9878c338 - Show all commits
@@ -0,0 +1,55 @@
package com.baiyue.lastleg.selfoperated.analytics.query.diagnostics;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
/** Request-local, parameter-free client timings. No SQL or business identities are retained. */
final class QueryDiagnostics {
static final ThreadLocal<State> CURRENT = new ThreadLocal<>();
private static final ObjectMapper JSON = new ObjectMapper();
static final String PREFIX = "com.baiyue.lastleg.selfoperated.analytics.query.";
private static final Set<String> METHODS = Set.of("AnalyticsReadMapper.scopes", "AnalyticsReadMapper.publishedRun",
"AnalyticsReadMapper.applyBarriers", "AnalyticsReadMapper.count", "AnalyticsReadMapper.page",
"AnalyticsReadMapper.inventoryCount", "AnalyticsReadMapper.inventoryPage", "DashboardReadMapper.daily", "DashboardReadMapper.report");
static String method(String id) {
if (!id.startsWith(PREFIX)) return null;
String method = id.substring(PREFIX.length());
return METHODS.contains(method) ? method : null;
}
static String phase(String method) {
if (method.endsWith(".scopes") || method.endsWith(".publishedRun") || method.endsWith(".applyBarriers")) return "CONTEXT";
if (method.endsWith(".daily")) return "DAILY";
if (method.endsWith(".report")) return "REPORT";
return method.endsWith("Count") || method.endsWith(".count") ? "COUNT" : "PAGE";
}
static final class Timing {
public long calls, failures, nanos;
void add(long elapsed, boolean failed) { calls++; nanos += elapsed; if (failed) failures++; }
}
static final class MethodTiming {
public final String phase;
public final Timing executor = new Timing(), statement = new Timing();
MethodTiming(String method) { phase = QueryDiagnostics.phase(method); }
}
static final class State {
final long started = System.nanoTime();
final Map<String, MethodTiming> methods = new TreeMap<>();
final Deque<String> frames = new ArrayDeque<>();
int statementDepth;
MethodTiming timing(String method) { return methods.computeIfAbsent(method, MethodTiming::new); }
String json(String endpoint, String requestId, int status, boolean failed) {
long total = System.nanoTime() - started;
long executor = methods.values().stream().mapToLong(t -> t.executor.nanos).sum();
Map<String,Object> report = new LinkedHashMap<>();
report.put("event", "SELFOP_QUERY_DIAGNOSTICS"); report.put("endpoint", endpoint);
report.put("requestId", requestId != null && requestId.matches("[A-Za-z0-9._:-]{1,128}") ? requestId : null);
report.put("httpStatus", status); report.put("escapedException", failed);
report.put("requestNanos", total); report.put("unattributedNanos", Math.max(0, total-executor));
report.put("timingMeaning", "CLIENT_EXECUTOR_AND_STATEMENT_INCLUDING_MAPPING_NOT_SERVER_DB_TIME");
report.put("poolWaitAttribution", "NOT_MEASURED_PER_REQUEST"); report.put("methods", methods);
try { return JSON.writeValueAsString(report); }
catch (JsonProcessingException e) { return "{\"event\":\"SELFOP_QUERY_DIAGNOSTICS_SERIALIZATION_FAILED\"}"; }
}
}
}
@@ -0,0 +1,26 @@
package com.baiyue.lastleg.selfoperated.analytics.query.diagnostics;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.web.servlet.FilterRegistrationBean;
import org.springframework.context.annotation.*;
import org.springframework.core.Ordered;
/** Explicit TEST opt-in; no datasource decoration or changes to business queries. */
@Configuration(proxyBeanMethods=false)
@ConditionalOnProperty(name="DEPLOY_ENV", havingValue="test", matchIfMissing=false)
public class QueryDiagnosticsConfiguration {
@Bean
@ConditionalOnProperty(name="baiyue.self-operated-analytics.query-diagnostics.enabled", havingValue="true", matchIfMissing=false)
public QueryDiagnosticsInterceptor queryDiagnosticsInterceptor() { return new QueryDiagnosticsInterceptor(); }
@Bean
@ConditionalOnProperty(name="baiyue.self-operated-analytics.query-diagnostics.enabled", havingValue="true", matchIfMissing=false)
public FilterRegistrationBean<QueryDiagnosticsFilter> queryDiagnosticsFilter() {
var log = LoggerFactory.getLogger(QueryDiagnosticsFilter.class);
var registration = new FilterRegistrationBean<>(new QueryDiagnosticsFilter(log::info));
registration.setOrder(Ordered.HIGHEST_PRECEDENCE+1);
registration.setDispatcherTypes(jakarta.servlet.DispatcherType.REQUEST);
return registration;
}
}
@@ -0,0 +1,33 @@
package com.baiyue.lastleg.selfoperated.analytics.query.diagnostics;
import com.baiyue.common.web.CorrelationContext;
import jakarta.servlet.*;
import jakarta.servlet.http.*;
import org.slf4j.MDC;
import org.springframework.web.filter.OncePerRequestFilter;
import java.io.IOException;
import java.util.Map;
import java.util.function.Consumer;
public final class QueryDiagnosticsFilter extends OncePerRequestFilter {
private static final String BASE = "/api/platform/v1/last-leg/self-operated-analytics/";
private static final Map<String,String> ENDPOINTS = Map.of(BASE+"dashboard", "DASHBOARD",
BASE+"sku-sales-summary", "SUMMARY", BASE+"inventory-coverage", "INVENTORY");
private final Consumer<String> sink;
QueryDiagnosticsFilter(Consumer<String> sink) { this.sink = sink; }
@Override protected boolean shouldNotFilter(HttpServletRequest request) {
return !"GET".equals(request.getMethod()) || !ENDPOINTS.containsKey(request.getRequestURI());
}
@Override protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain chain)
throws ServletException, IOException {
var state = new QueryDiagnostics.State(); QueryDiagnostics.CURRENT.set(state);
String requestId = MDC.get(CorrelationContext.REQUEST_ID); boolean failed = true;
try { chain.doFilter(request, response); failed = false; }
finally {
QueryDiagnostics.CURRENT.remove();
// Diagnostic failures must never replace the original request result or exception.
try { sink.accept(state.json(ENDPOINTS.get(request.getRequestURI()), requestId, response.getStatus(), failed)); }
catch (RuntimeException ignored) { logger.warn("SELFOP_QUERY_DIAGNOSTICS_OUTPUT_FAILED"); }
}
}
}
@@ -0,0 +1,42 @@
package com.baiyue.lastleg.selfoperated.analytics.query.diagnostics;
import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.executor.Executor;
import org.apache.ibatis.executor.statement.StatementHandler;
import org.apache.ibatis.mapping.*;
import org.apache.ibatis.plugin.*;
import org.apache.ibatis.session.*;
import java.sql.Statement;
@Intercepts({
@Signature(type=Executor.class, method="query", args={MappedStatement.class,Object.class,RowBounds.class,ResultHandler.class}),
@Signature(type=Executor.class, method="query", args={MappedStatement.class,Object.class,RowBounds.class,ResultHandler.class,CacheKey.class,BoundSql.class}),
@Signature(type=StatementHandler.class, method="query", args={Statement.class,ResultHandler.class})
})
public final class QueryDiagnosticsInterceptor implements Interceptor {
@Override public Object intercept(Invocation invocation) throws Throwable {
var state = QueryDiagnostics.CURRENT.get();
if (state == null) return invocation.proceed();
boolean executor = invocation.getTarget() instanceof Executor;
String method;
boolean record;
if (executor) {
method = QueryDiagnostics.method(((MappedStatement) invocation.getArgs()[0]).getId());
record = method != null && state.frames.isEmpty();
state.frames.push(method == null ? "" : method);
} else {
method = state.frames.peek();
record = method != null && !method.isEmpty() && state.statementDepth == 0;
state.statementDepth++;
}
long start = System.nanoTime(); boolean failed = true;
try { Object result = invocation.proceed(); failed = false; return result; }
finally {
if (record) {
var timing = state.timing(method);
(executor ? timing.executor : timing.statement).add(System.nanoTime()-start, failed);
}
if (executor) state.frames.pop(); else state.statementDepth--;
}
}
}
@@ -0,0 +1,123 @@
package com.baiyue.lastleg.selfoperated.analytics.query.diagnostics;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.servlet.ServletException;
import org.apache.ibatis.cache.CacheKey;
import org.apache.ibatis.executor.Executor;
import org.apache.ibatis.executor.statement.StatementHandler;
import org.apache.ibatis.mapping.*;
import org.apache.ibatis.session.*;
import org.junit.jupiter.api.*;
import org.slf4j.MDC;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.boot.web.servlet.FilterRegistrationBean;
import org.springframework.mock.web.*;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.*;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;
import static org.mockito.ArgumentMatchers.*;
class QueryDiagnosticsTest {
private static final String BASE="/api/platform/v1/last-leg/self-operated-analytics/";
private static final String ID=QueryDiagnostics.PREFIX+"DashboardReadMapper.report";
@AfterEach void cleanup() { QueryDiagnostics.CURRENT.remove(); MDC.clear(); }
@Test void onlyExplicitTestAndEnabledCreatesBeans() {
var runner=new ApplicationContextRunner().withUserConfiguration(QueryDiagnosticsConfiguration.class);
runner.run(c->assertEquals(0,c.getBeansOfType(QueryDiagnosticsInterceptor.class).size()));
runner.withPropertyValues("DEPLOY_ENV=test").run(c->assertEquals(0,c.getBeansOfType(FilterRegistrationBean.class).size()));
runner.withPropertyValues("DEPLOY_ENV=production","baiyue.self-operated-analytics.query-diagnostics.enabled=true")
.run(c->assertEquals(0,c.getBeansOfType(QueryDiagnosticsInterceptor.class).size()));
runner.withPropertyValues("baiyue.self-operated-analytics.query-diagnostics.enabled=true")
.run(c->assertEquals(0,c.getBeansOfType(QueryDiagnosticsInterceptor.class).size()));
runner.withPropertyValues("DEPLOY_ENV=test","baiyue.self-operated-analytics.query-diagnostics.enabled=true")
.run(c->{assertEquals(1,c.getBeansOfType(QueryDiagnosticsInterceptor.class).size());assertEquals(1,c.getBeansOfType(FilterRegistrationBean.class).size());});
}
@Test void pathsAndMethodsAreExactAndExceptionsCleanRequestState() throws Exception {
List<String> lines=new ArrayList<>();var filter=new QueryDiagnosticsFilter(lines::add);
for(String path:List.of("dashboard/extra","sku-daily-series","collection-status"))
filter.doFilter(new MockHttpServletRequest("GET",BASE+path),new MockHttpServletResponse(),(a,b)->assertNull(QueryDiagnostics.CURRENT.get()));
filter.doFilter(new MockHttpServletRequest("POST",BASE+"dashboard"),new MockHttpServletResponse(),(a,b)->assertNull(QueryDiagnostics.CURRENT.get()));
assertTrue(lines.isEmpty());
var failure=new ServletException("PRIVATE_TOKEN");
var caught=assertThrows(ServletException.class,()->filter.doFilter(new MockHttpServletRequest("GET",BASE+"dashboard"),
new MockHttpServletResponse(),(a,b)->{assertNotNull(QueryDiagnostics.CURRENT.get());throw failure;}));
assertSame(failure,caught);assertNull(QueryDiagnostics.CURRENT.get());
assertTrue(new ObjectMapper().readTree(lines.get(0)).get("escapedException").asBoolean());
assertFalse(lines.get(0).contains("PRIVATE_TOKEN"));
var badSink=new QueryDiagnosticsFilter(s->{throw new IllegalStateException("PRIVATE");});
assertSame(failure,assertThrows(ServletException.class,()->badSink.doFilter(new MockHttpServletRequest("GET",BASE+"dashboard"),
new MockHttpServletResponse(),(a,b)->{throw failure;})));
assertNull(QueryDiagnostics.CURRENT.get());
}
private static MappedStatement statement(String id) {
var config=new Configuration();
return new MappedStatement.Builder(config,id,p->new BoundSql(config,"SELECT PRIVATE_SQL",List.of(),p),SqlCommandType.SELECT).build();
}
@Test void cacheHitCountsExecutorButNoStatementAndDoesNotExposeParameters() throws Exception {
var plugin=new QueryDiagnosticsInterceptor();var executor=(Executor)plugin.plugin(mock(Executor.class));
List<String> lines=new ArrayList<>();var filter=new QueryDiagnosticsFilter(lines::add);
MDC.put("requestId","synthetic-request-1");
var request=new MockHttpServletRequest("GET",BASE+"sku-sales-summary");request.setQueryString("sourceAccount=PRIVATE_ACCOUNT");
request.addHeader("Authorization","PRIVATE_TOKEN");
filter.doFilter(request,new MockHttpServletResponse(),(a,b)->{
try { executor.query(statement(ID),Map.of("sku","PRIVATE_SKU"),RowBounds.DEFAULT,null); }
catch(java.sql.SQLException e){throw new ServletException(e);}
});
var json=new ObjectMapper().readTree(lines.get(0));var timing=json.path("methods").path("DashboardReadMapper.report");
assertEquals(1,timing.path("executor").path("calls").asInt());assertEquals(0,timing.path("statement").path("calls").asInt());
assertEquals("synthetic-request-1",json.path("requestId").asText());assertEquals("SUMMARY",json.path("endpoint").asText());
assertFalse(lines.get(0).contains("PRIVATE"));assertFalse(lines.get(0).contains("\n"));assertNull(QueryDiagnostics.CURRENT.get());
}
@Test void nestedExecutorAndStatementAreNotDoubleCountedAndFailureIsPreserved() throws Exception {
var plugin=new QueryDiagnosticsInterceptor();var delegate=mock(Executor.class);
var executor=(Executor)plugin.plugin(delegate);var ms=statement(ID);var bound=ms.getBoundSql(null);
var statementDelegate=mock(StatementHandler.class);var handler=(StatementHandler)plugin.plugin(statementDelegate);
var state=new QueryDiagnostics.State();QueryDiagnostics.CURRENT.set(state);
when(delegate.query(eq(ms),any(),eq(RowBounds.DEFAULT),isNull())).thenAnswer(i->
executor.query(ms,i.getArgument(1),RowBounds.DEFAULT,null,new CacheKey(),bound));
when(delegate.query(eq(ms),any(),eq(RowBounds.DEFAULT),isNull(),any(CacheKey.class),eq(bound)))
.thenAnswer(i->handler.query(mock(Statement.class),null));
var innerDelegate=mock(StatementHandler.class);var innerHandler=(StatementHandler)plugin.plugin(innerDelegate);
when(innerDelegate.query(any(),isNull())).thenReturn(List.of());
when(statementDelegate.query(any(),isNull())).thenAnswer(i->innerHandler.query(i.getArgument(0),null));
executor.query(ms,Map.of("secret","PRIVATE"),RowBounds.DEFAULT,null);
var timing=state.methods.get("DashboardReadMapper.report");
assertEquals(1,timing.executor.calls);assertEquals(1,timing.statement.calls);assertTrue(state.frames.isEmpty());
var failure=new java.sql.SQLException("PRIVATE_SQL_FAILURE");
when(statementDelegate.query(any(),isNull())).thenThrow(failure);
assertSame(failure,assertThrows(java.sql.SQLException.class,()->executor.query(ms,Map.of(),RowBounds.DEFAULT,null)));
assertEquals(2,timing.executor.calls);assertEquals(2,timing.statement.calls);
assertEquals(1,timing.executor.failures);assertEquals(1,timing.statement.failures);
assertTrue(state.frames.isEmpty());assertEquals(0,state.statementDepth);
executor.query(statement("other.Mapper.privateMethod"),Map.of(),RowBounds.DEFAULT,null);
assertEquals(1,state.methods.size());
}
@Test void concurrentRequestsHaveIndependentStateAndCleanup() throws Exception {
var ready=new CyclicBarrier(2);var pool=Executors.newFixedThreadPool(2);
try {
List<Future<String>> futures=new ArrayList<>();
for(int index=0;index<2;index++){final int n=index;futures.add(pool.submit(()->{
List<String> lines=new ArrayList<>();MDC.put("requestId","request-"+n);
try {
new QueryDiagnosticsFilter(lines::add).doFilter(new MockHttpServletRequest("GET",BASE+"inventory-coverage"),new MockHttpServletResponse(),(a,b)->{
var state=QueryDiagnostics.CURRENT.get();state.timing("AnalyticsReadMapper.inventoryPage").executor.add(n+1,false);
try{ready.await(5,TimeUnit.SECONDS);}catch(Exception e){throw new ServletException(e);}
assertSame(state,QueryDiagnostics.CURRENT.get());
});
assertNull(QueryDiagnostics.CURRENT.get());return lines.get(0);
} finally {MDC.clear();}
}));}
for(int i=0;i<2;i++){var json=new ObjectMapper().readTree(futures.get(i).get(10,TimeUnit.SECONDS));
assertEquals("request-"+i,json.path("requestId").asText());
assertEquals(i+1,json.path("methods").path("AnalyticsReadMapper.inventoryPage").path("executor").path("nanos").asLong());}
} finally {pool.shutdownNow();}
}
}