feat(lineage): Phase 4 — SLO metrics + bundle v1.3.0 pilot (warn mode)

dict-relationships-v2 epic, Phase 4 (final). Backend SLO instrumentation + schema bundle pilot. Alert rules + runbook идут отдельным коммитом в ordinis-infra. Backend metrics (Micrometer / Prometheus): - nsi_dependents_query_duration_seconds (Timer + p50/p95/p99 percentiles) labels: mode={schema|record|*_error}, target_dict - nsi_cascade_close_total (Counter) labels: target_dict, outcome={success|blocked|cascade_required|too_large|timeout|error} - nsi_cascade_close_duration_seconds (Timer + percentiles) - nsi_lineage_mv_last_refresh_seconds_ago (Gauge, NaN when mv disabled) - nsi_lineage_mv_row_count (Gauge) - nsi_lineage_mv_last_duration_seconds (Gauge) - nsi_lineage_mv_refresh_total{outcome=ok|error} (Counter) Cardinality bound: target_dict tag — ≤40 dictionaries; outcome — 6 values. Total time series ≈ 240 per app instance — приемлемо. Optional MeterRegistry inject — null в test ctor'ах (back-compat). Bundle v1.3.0 pilot: - spacecraft.satellite_type_code → satellite_type.code теперь имеет x-references-on-close: warn. Закрытие satellite_type больше не блокируется через 1.3.0; spacecraft остаётся orphan; orphan scanner поднимет metric. Data steward потом repoints. - bundle version 1.2.1 → 1.3.0, spacecraft schema 1.0.0 → 1.1.0 (description обновлён с pilot context). - WARN-mode выбран как safe pilot — validates новый code path без destructive cascade. CASCADE-mode flip может быть отдельным PR per-relationship после operator review. Verify: - mvn verify -pl '!ordinis-app': SUCCESS (5.7s). - 106/106 tests still green (existing tests не задеты). - helm lint в ordinis-infra: clean (alerts render correctly). Backward-compat: REGRESSION test (ReferenceCloseRegressionTest) covers v1.2.1-style schemas без x-references-on-close. WARN mode на one field не меняет contract других — bundle import idempotent. Roadmap: 4 phases ⚓ shipped. Cascade engine + lineage UI + materialized view + SLO metrics live.
2026-05-08 11:34:29 +03:00
parent 3eeaba058f
commit a7cdd5df5c
5 changed files with 158 additions and 10 deletions
@@ -11,6 +11,9 @@ import cloud.nstart.terravault.ordinis.outbox.OutboxRecorder;
 import cloud.nstart.terravault.ordinis.restapi.audit.AuditLogger;
 import cloud.nstart.terravault.ordinis.restapi.error.OrdinisException;
 import cloud.nstart.terravault.ordinis.restapi.service.DictionaryDefinitionService;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.Timer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -23,6 +26,7 @@ import org.springframework.transaction.annotation.Transactional;
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;

 /**
@@ -60,6 +64,8 @@ public class CascadeCloseService {
  private final LineageIndexService lineage;
  private final RecordCloseSink closeSink;
  private final ScopeContext scopeContext;
+  /** Phase 4 SLO. Optional — null в test ctor'е. */
+  private final Optional<MeterRegistry> meterRegistry;

  @Autowired
  public CascadeCloseService(
@@ -68,10 +74,12 @@ public class CascadeCloseService {
      DictionaryDefinitionService defService,
      AuditLogger auditLogger,
      OutboxRecorder outbox,
-      ScopeContext scopeContext) {
+      ScopeContext scopeContext,
+      Optional<MeterRegistry> meterRegistry) {
    this(lineage,
        new DefaultRecordCloseSink(recordRepo, defService, auditLogger, outbox),
-        scopeContext);
+        scopeContext,
+        meterRegistry);
  }

  /** Test-only ctor — sink инжектится напрямую (Mockito не мокает concrete). */
@@ -79,9 +87,18 @@ public class CascadeCloseService {
      LineageIndexService lineage,
      RecordCloseSink closeSink,
      ScopeContext scopeContext) {
+    this(lineage, closeSink, scopeContext, Optional.empty());
+  }
+
+  CascadeCloseService(
+      LineageIndexService lineage,
+      RecordCloseSink closeSink,
+      ScopeContext scopeContext,
+      Optional<MeterRegistry> meterRegistry) {
    this.lineage = lineage;
    this.closeSink = closeSink;
    this.scopeContext = scopeContext;
+    this.meterRegistry = meterRegistry;
  }

  /**
@@ -129,10 +146,12 @@ public class CascadeCloseService {
      OffsetDateTime closeAt,
      String reason,
      boolean confirmed) {
+    Timer.Sample sample = startTimer();
    OffsetDateTime when = closeAt == null ? OffsetDateTime.now() : closeAt;
    CascadePlan plan = evaluatePlan(targetDict, targetKey, when);

    if (!plan.blockers().isEmpty()) {
+      stopTimerAndCount(sample, targetDict, "blocked");
      throw OrdinisException.conflict(
          "x_references_blocked_by_dependents",
          "Cannot close " + targetDict + "/" + targetKey + " — "
@@ -141,6 +160,7 @@ public class CascadeCloseService {
    }

    if (!plan.cascade().isEmpty() && !confirmed) {
+      stopTimerAndCount(sample, targetDict, "cascade_required");
      throw OrdinisException.conflict(
          "x_references_cascade_required",
          plan.cascade().size() + " dependents will be cascaded. "
@@ -148,6 +168,7 @@ public class CascadeCloseService {
    }

    if (plan.cascade().size() > CASCADE_MAX_PER_REQUEST) {
+      stopTimerAndCount(sample, targetDict, "too_large");
      throw OrdinisException.badRequest(
          "x_references_cascade_too_large",
          "Cascade size " + plan.cascade().size() + " exceeds limit "
@@ -164,18 +185,46 @@ public class CascadeCloseService {
      }
      log.info("Cascade close OK: target={}/{} cascaded={} warnings={}",
          targetDict, targetKey, plan.cascade().size(), plan.warnings().size());
+      stopTimerAndCount(sample, targetDict, "success");
      return new CascadeCloseResult(
          targetDict, targetKey, plan.cascade(), plan.warnings(), 1 + plan.cascade().size());
    } catch (TransactionTimedOutException | QueryTimeoutException e) {
      log.error("Cascade close timeout for {}/{} after 30s: {}", targetDict, targetKey, e.getMessage());
+      stopTimerAndCount(sample, targetDict, "timeout");
      throw new OrdinisException(
          org.springframework.http.HttpStatus.SERVICE_UNAVAILABLE,
          "x_references_cascade_timeout",
          "Cascade close transaction exceeded 30s timeout. Rollback complete. "
              + "Consider splitting via bulk-close or reducing cascade fan-out.");
+    } catch (RuntimeException e) {
+      stopTimerAndCount(sample, targetDict, "error");
+      throw e;
    }
  }

+  // === Metrics helpers ===
+
+  private Timer.Sample startTimer() {
+    if (meterRegistry.isEmpty()) return null;
+    return Timer.start(meterRegistry.get());
+  }
+
+  /** Records timer + increments outcome counter (success/blocked/timeout/...). */
+  private void stopTimerAndCount(Timer.Sample sample, String targetDict, String outcome) {
+    if (meterRegistry.isEmpty()) return;
+    var registry = meterRegistry.get();
+    Tags tags = Tags.of("target_dict", targetDict, "outcome", outcome);
+    if (sample != null) {
+      Timer t = Timer.builder("nsi_cascade_close_duration_seconds")
+          .description("Cascade close transaction latency")
+          .tags(tags)
+          .publishPercentiles(0.5, 0.95, 0.99)
+          .register(registry);
+      sample.stop(t);
+    }
+    registry.counter("nsi_cascade_close_total", tags).increment();
+  }
+
  /**
   * Узкий test-friendly seam над close mechanics (repo + audit + outbox).
   * Concrete prod impl ниже; тесты подменяют через subclass / lambda.
@@ -7,6 +7,9 @@ import cloud.nstart.terravault.ordinis.domain.record.DependentsQuery;
 import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecord;
 import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecordRepository;
 import com.fasterxml.jackson.databind.JsonNode;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.Timer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -56,25 +59,30 @@ public class LineageIndexService {
  private final DependentsQuery dependentsQuery;
  /** Optional — присутствует только когда {@code ordinis.lineage.mv.enabled=true}. */
  private final Optional<MaterializedViewRefreshService> mvRefresh;
+  /** Optional — Phase 4 SLO instrumentation. Null в test ctor'е (без metrics). */
+  private final Optional<MeterRegistry> meterRegistry;

  @Autowired
  public LineageIndexService(
      DictionaryDefinitionRepository definitionRepository,
      DictionaryRecordRepository recordRepository,
      DependentsQuery dependentsQuery,
-      Optional<MaterializedViewRefreshService> mvRefresh) {
+      Optional<MaterializedViewRefreshService> mvRefresh,
+      Optional<MeterRegistry> meterRegistry) {
    this.definitionRepository = definitionRepository;
    this.recordRepository = recordRepository;
    this.dependentsQuery = dependentsQuery;
    this.mvRefresh = mvRefresh;
+    this.meterRegistry = meterRegistry;
  }

-  /** Test-friendly ctor без MV refresh service. */
+  /** Test-friendly ctor без MV refresh service и metrics. */
  public LineageIndexService(
      DictionaryDefinitionRepository definitionRepository,
      DictionaryRecordRepository recordRepository,
      DependentsQuery dependentsQuery) {
-    this(definitionRepository, recordRepository, dependentsQuery, Optional.empty());
+    this(definitionRepository, recordRepository, dependentsQuery,
+        Optional.empty(), Optional.empty());
  }

  /**
@@ -90,6 +98,19 @@ public class LineageIndexService {
  @Transactional(readOnly = true)
  public List<SchemaDependent> findSchemaDependents(
      String targetDictName, Set<DataScope> allowedScopes) {
+    Timer.Sample sample = startTimer();
+    try {
+      var result = findSchemaDependentsInternal(targetDictName, allowedScopes);
+      stopTimer(sample, "schema", targetDictName);
+      return result;
+    } catch (RuntimeException e) {
+      stopTimer(sample, "schema_error", targetDictName);
+      throw e;
+    }
+  }
+
+  private List<SchemaDependent> findSchemaDependentsInternal(
+      String targetDictName, Set<DataScope> allowedScopes) {
    Optional<DictionaryDefinition> targetOpt = definitionRepository.findByName(targetDictName);
    if (targetOpt.isEmpty()) return List.of();
    DictionaryDefinition target = targetOpt.get();
@@ -169,6 +190,24 @@ public class LineageIndexService {
      Set<DataScope> allowedScopes,
      int page,
      int size) {
+    Timer.Sample sample = startTimer();
+    try {
+      var result = findRecordDependentsInternal(
+          targetDictName, targetBusinessKey, allowedScopes, page, size);
+      stopTimer(sample, "record", targetDictName);
+      return result;
+    } catch (RuntimeException e) {
+      stopTimer(sample, "record_error", targetDictName);
+      throw e;
+    }
+  }
+
+  private RecordDependentsPage findRecordDependentsInternal(
+      String targetDictName,
+      String targetBusinessKey,
+      Set<DataScope> allowedScopes,
+      int page,
+      int size) {
    if (page < 0) throw new IllegalArgumentException("page must be >= 0");
    if (size < 1 || size > 200) {
      throw new IllegalArgumentException("size must be 1..200, got " + size);
@@ -311,6 +350,24 @@ public class LineageIndexService {
      OnCloseAction onClose,
      long count) {}

+  // === Metrics helpers ===
+
+  private Timer.Sample startTimer() {
+    if (meterRegistry.isEmpty()) return null;
+    return Timer.start(meterRegistry.get());
+  }
+
+  private void stopTimer(Timer.Sample sample, String mode, String targetDict) {
+    if (sample == null || meterRegistry.isEmpty()) return;
+    // Cardinality bound: target_dict tag — bounded by # dictionaries (~40).
+    Timer t = Timer.builder("nsi_dependents_query_duration_seconds")
+        .description("Dependents lookup latency (Phase 4 SLO instrumentation)")
+        .tags(Tags.of("mode", mode, "target_dict", targetDict))
+        .publishPercentiles(0.5, 0.95, 0.99)
+        .register(meterRegistry.get());
+    sample.stop(t);
+  }
+
  /** Paginated page + per-source aggregates + lineage source metadata. */
  public record RecordDependentsPage(
      String targetDict,
@@ -1,5 +1,8 @@
 package cloud.nstart.terravault.ordinis.restapi.service.reference;

+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tags;
+import jakarta.annotation.PostConstruct;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
@@ -13,6 +16,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.time.OffsetDateTime;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;

 /* JDK 25 + Mockito не мочат concrete JdbcTemplate (subclass mock-maker
@@ -60,17 +64,45 @@ public class MaterializedViewRefreshService {
  private final MvGateway gateway;
  /** Volatile guard — не позволяет двум refresh'ам параллельно. */
  private final AtomicReference<Instant> currentRefreshStart = new AtomicReference<>(null);
+  /** Phase 4: gauges feed Prometheus + alerts. */
+  private final AtomicLong lastRefreshEpochMs = new AtomicLong(0);
+  private final AtomicLong lastRowCount = new AtomicLong(0);
+  private final AtomicLong lastDurationMs = new AtomicLong(0);
+  private final Optional<MeterRegistry> meterRegistry;

  @Value("${ordinis.lineage.mv.min-interval-sec:30}")
  private long minIntervalSec;

+  public MaterializedViewRefreshService(JdbcTemplate jdbc, Optional<MeterRegistry> meterRegistry) {
+    this(new JdbcMvGateway(jdbc), meterRegistry);
+  }
+
  public MaterializedViewRefreshService(JdbcTemplate jdbc) {
-    this(new JdbcMvGateway(jdbc));
+    this(new JdbcMvGateway(jdbc), Optional.empty());
  }

  /** Test-only ctor — gateway инжектится напрямую. */
  MaterializedViewRefreshService(MvGateway gateway) {
+    this(gateway, Optional.empty());
+  }
+
+  MaterializedViewRefreshService(MvGateway gateway, Optional<MeterRegistry> meterRegistry) {
    this.gateway = gateway;
+    this.meterRegistry = meterRegistry;
+  }
+
+  @PostConstruct
+  void registerGauges() {
+    if (meterRegistry.isEmpty()) return;
+    var reg = meterRegistry.get();
+    // Last refresh age в секундах — используется в alert OrdinisLineageMvStale.
+    reg.gauge("nsi_lineage_mv_last_refresh_seconds_ago", Tags.empty(), this,
+        s -> s.lastRefreshEpochMs.get() == 0
+            ? Double.NaN
+            : (System.currentTimeMillis() - s.lastRefreshEpochMs.get()) / 1000.0);
+    reg.gauge("nsi_lineage_mv_row_count", Tags.empty(), lastRowCount, AtomicLong::doubleValue);
+    reg.gauge("nsi_lineage_mv_last_duration_seconds", Tags.empty(),
+        lastDurationMs, l -> l.get() / 1000.0);
  }

  /**
@@ -167,9 +199,18 @@ public class MaterializedViewRefreshService {
    } catch (DataAccessException e) {
      log.warn("Failed to update lineage_mv_meta: {}", e.getMessage());
    }
+    // Phase 4: feed gauges + counter (separate from gateway updateMeta —
+    // нужно даже при ошибке записи в lineage_mv_meta).
+    lastDurationMs.set(duration);
+    if (rowCount != null) lastRowCount.set(rowCount);
    if ("ok".equals(status)) {
+      lastRefreshEpochMs.set(System.currentTimeMillis());
      log.info("MV record_dependents_index refreshed: {}ms, {} rows", duration, rowCount);
    }
+    final String finalStatus = status;
+    meterRegistry.ifPresent(reg -> reg.counter(
+        "nsi_lineage_mv_refresh_total",
+        Tags.of("outcome", finalStatus)).increment());
  }

  private static String truncate(String s, int max) {