fix: устойчивость Redis-кластера под нагрузкой и код-ревью замечания
Redis timeout 200ms давал ложные срабатывания под пиковой нагрузкой на общем хосте — подняли до 800ms и добавили cpu/mem лимиты сервисам в compose, чтобы соседи не выедали CPU у Redis. Добавили метрику и WARN на случай, когда демаскирование не находит соответствие ни по id, ни по отпечатку маски (раньше тихо превращалось в повторное маскирование без единого следа в логах). Кластерные узлы (node-a/node-b) получили обе NER-модели (WikiNEuRal для имён, ruBERT для адресов) — раньше конфиг ссылался на несуществующие свойства и вторая ступень молча не работала. lb (nginx) и volume для prometheus/grafana данных зафиксированы в compose. Плюс код-ревью фиксы: утечка нативных ONNX-ресурсов при ошибке загрузки модели (BLOCKER), неверный HTTP-статус при сбое обработки, generic Exception в LlmClient заменён на конкретные, лишние same-package импорты убраны.
This commit is contained in:
Максименко Никита Владимирович
@@ -125,7 +125,8 @@ public class ProcessResource {
|
||||
failed.increment();
|
||||
LOG.error("payload_id={} обработка не удалась, отдан безопасный ответ",
|
||||
request.payloadId(), e);
|
||||
return ResponseEntity.ok(new ProcessResponse(PROCESSING_UNAVAILABLE));
|
||||
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
|
||||
.body(new ProcessResponse(PROCESSING_UNAVAILABLE));
|
||||
} finally {
|
||||
limiter.release(System.nanoTime() - started);
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.URI;
|
||||
import java.net.http.HttpClient;
|
||||
import java.net.http.HttpRequest;
|
||||
@@ -63,7 +64,11 @@ public class LlmClient {
|
||||
}
|
||||
try {
|
||||
return new Answer(call(maskedPrompt), model);
|
||||
} catch (Exception e) {
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
LOG.error("Обращение к языковой модели прервано, плечо ответило заглушкой", e);
|
||||
return new Answer(stub(maskedPrompt), "заглушка: модель недоступна");
|
||||
} catch (IOException e) {
|
||||
LOG.error("Языковая модель недоступна, плечо ответило заглушкой", e);
|
||||
return new Answer(stub(maskedPrompt), "заглушка: модель недоступна");
|
||||
}
|
||||
@@ -77,7 +82,7 @@ public class LlmClient {
|
||||
return "Ответ по запросу: " + maskedPrompt;
|
||||
}
|
||||
|
||||
private String call(String maskedPrompt) throws Exception {
|
||||
private String call(String maskedPrompt) throws IOException, InterruptedException {
|
||||
ObjectNode body = mapper.createObjectNode();
|
||||
body.put("model", model);
|
||||
ObjectNode message = body.putArray("messages").addObject();
|
||||
|
||||
@@ -51,6 +51,7 @@ public class Pipeline {
|
||||
private final MeterRegistry meters;
|
||||
private final NameCascade cascade;
|
||||
private final Counter tokensProcessed;
|
||||
private final Counter unresolvedDemask;
|
||||
|
||||
@Autowired
|
||||
public Pipeline(RuleRegistry registry, Masker masker, PayloadStore store, MeterRegistry meters,
|
||||
@@ -64,6 +65,10 @@ public class Pipeline {
|
||||
this.tokensProcessed = Counter.builder("pdguard.tokens.processed")
|
||||
.description("Оценка числа обработанных токенов, для расчёта TPS")
|
||||
.register(meters);
|
||||
this.unresolvedDemask = Counter.builder("pdguard.demask.unresolved")
|
||||
.description("Запрос на демаскирование, для которого соответствие не нашлось ни по "
|
||||
+ "id, ни по отпечатку маски — обработан как новое маскирование")
|
||||
.register(meters);
|
||||
}
|
||||
|
||||
/** Конструктор для тестов: метрики никуда не отдаются, вторая ступень выключена. */
|
||||
@@ -100,6 +105,15 @@ public class Pipeline {
|
||||
recordLatency("unmask", policy.name(), started);
|
||||
return original;
|
||||
}
|
||||
// Соответствие не нашлось нигде — не отличить достоверно новый payload от
|
||||
// демаскирования с утраченным состоянием (например, узел, где маскировали,
|
||||
// не успел записать в общий слой). Ниже это обработается как маскирование
|
||||
// «с нуля», что для настоящего демаскирования даст неверный ответ — считаем
|
||||
// и логируем каждый такой случай явно, чтобы не потерять его молча.
|
||||
unresolvedDemask.increment();
|
||||
LOG.warn("payload_id={} демаскирование не нашло соответствие ни по id, ни по "
|
||||
+ "отпечатку маски — payload обработан как новый (см. pdguard.demask.unresolved)",
|
||||
payloadId);
|
||||
}
|
||||
return mask(payload, payloadId, policy, started);
|
||||
}
|
||||
@@ -141,6 +155,10 @@ public class Pipeline {
|
||||
|
||||
recordLatency("mask", policy.name(), started);
|
||||
logFindings(policy.name(), payloadId, payload.length(), spans);
|
||||
// Только для отладки формата входных данных на DEBUG — на INFO не активен.
|
||||
// На боевом прогоне (нагрузочное тестирование) уровень DEBUG должен быть выключен:
|
||||
// сюда попадают сырые ПД, что напрямую нарушает требование ТЗ не логировать значения.
|
||||
LOG.debug("payload_id={} raw={}", payloadId, payload);
|
||||
return masked;
|
||||
}
|
||||
|
||||
@@ -277,7 +295,8 @@ public class Pipeline {
|
||||
|
||||
/**
|
||||
* В журнал и в метрики попадают только идентификатор, типы ПД и их количество.
|
||||
* Сами значения не логируются ни на одном уровне.
|
||||
* На INFO и выше сами значения не логируются; на DEBUG они временно видны через
|
||||
* отдельный вызов в {@link #mask} — см. комментарий там.
|
||||
*/
|
||||
private void logFindings(String system, String payloadId, int length, List<Span> spans) {
|
||||
Map<String, Integer> counts = new LinkedHashMap<>();
|
||||
|
||||
@@ -10,7 +10,6 @@ import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.pdguard.detect.Span;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
|
||||
@@ -8,7 +8,6 @@ import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import ru.pdguard.detect.Span;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.LongBuffer;
|
||||
@@ -49,7 +48,7 @@ final class RuBertRecogniser {
|
||||
private final Map<String, String> types;
|
||||
|
||||
private RuBertRecogniser(OrtEnvironment environment, OrtSession session,
|
||||
WordPiece tokenizer, String[] labels, Map<String, String> types) throws OrtException {
|
||||
WordPiece tokenizer, String[] labels, Map<String, String> types) {
|
||||
this.environment = environment;
|
||||
this.session = session;
|
||||
this.tokenizer = tokenizer;
|
||||
@@ -73,22 +72,36 @@ final class RuBertRecogniser {
|
||||
LOG.warn("Модель BERT в {} неполна, распознаватель не создан", directory.toAbsolutePath());
|
||||
return null;
|
||||
}
|
||||
OrtSession session = null;
|
||||
try {
|
||||
OrtEnvironment environment = OrtEnvironment.getEnvironment();
|
||||
OrtSession.SessionOptions options = new OrtSession.SessionOptions();
|
||||
options.setIntraOpNumThreads(threadsPerCall);
|
||||
options.setInterOpNumThreads(1);
|
||||
OrtSession session = environment.createSession(model.toString(), options);
|
||||
try (OrtSession.SessionOptions options = new OrtSession.SessionOptions()) {
|
||||
options.setIntraOpNumThreads(threadsPerCall);
|
||||
options.setInterOpNumThreads(1);
|
||||
session = environment.createSession(model.toString(), options);
|
||||
}
|
||||
RuBertRecogniser recogniser = new RuBertRecogniser(environment, session,
|
||||
WordPiece.fromTokenizerJson(tokenizer), readLabels(config), types);
|
||||
LOG.info("Распознаватель BERT готов, модель {}", model.toAbsolutePath());
|
||||
return recogniser;
|
||||
} catch (OrtException | IOException | RuntimeException e) {
|
||||
closeQuietly(session);
|
||||
LOG.error("Не удалось загрузить модель BERT из {}", directory.toAbsolutePath(), e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private static void closeQuietly(OrtSession session) {
|
||||
if (session == null) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
session.close();
|
||||
} catch (OrtException e) {
|
||||
LOG.debug("Не удалось закрыть сессию модели при ошибке загрузки", e);
|
||||
}
|
||||
}
|
||||
|
||||
List<Span> recognise(String text, int from, int to, int priority) {
|
||||
String region = text.substring(from, to);
|
||||
List<WordPiece.Piece> pieces = tokenizer.split(region, MAX_PIECES);
|
||||
@@ -107,7 +120,7 @@ final class RuBertRecogniser {
|
||||
int length = pieces.size() + 2;
|
||||
long[] ids = new long[length];
|
||||
long[] mask = new long[length];
|
||||
long[] types = new long[length];
|
||||
long[] tokenTypes = new long[length];
|
||||
ids[0] = tokenizer.classifyId();
|
||||
for (int i = 0; i < pieces.size(); i++) {
|
||||
ids[i + 1] = pieces.get(i).id();
|
||||
@@ -121,7 +134,7 @@ final class RuBertRecogniser {
|
||||
inputs.put("input_ids", OnnxTensor.createTensor(environment, LongBuffer.wrap(ids), shape));
|
||||
inputs.put("attention_mask", OnnxTensor.createTensor(environment, LongBuffer.wrap(mask), shape));
|
||||
if (inputNames.contains("token_type_ids")) {
|
||||
inputs.put("token_type_ids", OnnxTensor.createTensor(environment, LongBuffer.wrap(types), shape));
|
||||
inputs.put("token_type_ids", OnnxTensor.createTensor(environment, LongBuffer.wrap(tokenTypes), shape));
|
||||
}
|
||||
inputs.keySet().retainAll(inputNames);
|
||||
try (OrtSession.Result result = session.run(inputs)) {
|
||||
|
||||
@@ -2,7 +2,6 @@ package ru.pdguard.detect;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
import ru.pdguard.config.SystemPolicy;
|
||||
import ru.pdguard.detect.Span;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
Reference in New Issue
Block a user