|
|
@@ -7,13 +7,21 @@ import com.adx.tencent.conversionsync.ConversionSyncRunner;
|
|
|
import com.adx.tencent.conversionsync.ConversionSyncService;
|
|
|
import com.adx.tencent.conversionsync.ManualTencentCallbackService;
|
|
|
import com.adx.tencent.conversionsync.RetryService;
|
|
|
+import org.springframework.data.redis.core.Cursor;
|
|
|
+import org.springframework.data.redis.core.ScanOptions;
|
|
|
+import org.springframework.data.redis.core.StringRedisTemplate;
|
|
|
import org.springframework.http.HttpStatus;
|
|
|
import org.springframework.lang.Nullable;
|
|
|
import org.springframework.web.bind.annotation.*;
|
|
|
import org.springframework.web.server.ResponseStatusException;
|
|
|
|
|
|
+import java.time.Duration;
|
|
|
+import java.util.ArrayList;
|
|
|
import java.util.List;
|
|
|
import java.util.Map;
|
|
|
+import java.util.UUID;
|
|
|
+import java.util.concurrent.CompletableFuture;
|
|
|
+import java.util.concurrent.TimeUnit;
|
|
|
|
|
|
/**
|
|
|
* 管理接口:手动触发转化同步和回调重试。
|
|
|
@@ -28,19 +36,22 @@ public class AdminController {
|
|
|
private final ConversionSyncService conversionSyncer;
|
|
|
private final RetryService retryService;
|
|
|
private final ManualTencentCallbackService manualTencentCallbackService;
|
|
|
+ private final StringRedisTemplate redisTemplate;
|
|
|
|
|
|
public AdminController(ConversionClient conversionClient,
|
|
|
@Nullable ConversionSyncRunner conversionSyncRunner,
|
|
|
@Nullable ConversionBackfillJobService backfillJobService,
|
|
|
@Nullable ConversionSyncService conversionSyncer,
|
|
|
@Nullable RetryService retryService,
|
|
|
- @Nullable ManualTencentCallbackService manualTencentCallbackService) {
|
|
|
+ @Nullable ManualTencentCallbackService manualTencentCallbackService,
|
|
|
+ @Nullable StringRedisTemplate redisTemplate) {
|
|
|
this.conversionClient = conversionClient;
|
|
|
this.conversionSyncRunner = conversionSyncRunner;
|
|
|
this.backfillJobService = backfillJobService;
|
|
|
this.conversionSyncer = conversionSyncer;
|
|
|
this.retryService = retryService;
|
|
|
this.manualTencentCallbackService = manualTencentCallbackService;
|
|
|
+ this.redisTemplate = redisTemplate;
|
|
|
}
|
|
|
|
|
|
@PostMapping("/conversions/query")
|
|
|
@@ -103,6 +114,193 @@ public class AdminController {
|
|
|
return manualTencentCallbackService.replayDeductedCallback(id);
|
|
|
}
|
|
|
|
|
|
+ @PostMapping("/redis/tighten-bid-media-ttl")
|
|
|
+ public Object tightenBidMediaTtl(@RequestBody(required = false) Map<String, Object> body) {
|
|
|
+ if (redisTemplate == null) {
|
|
|
+ throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
|
|
|
+ "redis template is not configured");
|
|
|
+ }
|
|
|
+ long ttlSeconds = toLong(body == null ? null : body.get("ttlSeconds"));
|
|
|
+ if (ttlSeconds <= 0) {
|
|
|
+ ttlSeconds = Duration.ofDays(3).getSeconds();
|
|
|
+ }
|
|
|
+ long scanCount = toLong(body == null ? null : body.get("scanCount"));
|
|
|
+ if (scanCount <= 0) {
|
|
|
+ scanCount = 2000;
|
|
|
+ }
|
|
|
+ long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
|
|
|
+ if (maxKeys <= 0) {
|
|
|
+ maxKeys = 200000;
|
|
|
+ }
|
|
|
+ List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
|
|
|
+ if (patterns.isEmpty()) {
|
|
|
+ patterns = defaultBidMediaPatterns();
|
|
|
+ }
|
|
|
+
|
|
|
+ List<Map<String, Object>> results = new ArrayList<>();
|
|
|
+ long totalScanned = 0;
|
|
|
+ long totalUpdated = 0;
|
|
|
+ for (String pattern : patterns) {
|
|
|
+ TtlTightenResult result = tightenTtl(pattern, ttlSeconds, scanCount, Math.max(0, maxKeys - totalScanned));
|
|
|
+ results.add(Map.of(
|
|
|
+ "pattern", pattern,
|
|
|
+ "scanned", result.scanned(),
|
|
|
+ "updated", result.updated(),
|
|
|
+ "stoppedByLimit", result.stoppedByLimit()
|
|
|
+ ));
|
|
|
+ totalScanned += result.scanned();
|
|
|
+ totalUpdated += result.updated();
|
|
|
+ if (totalScanned >= maxKeys) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return Map.of(
|
|
|
+ "ttlSeconds", ttlSeconds,
|
|
|
+ "ttlHuman", Duration.ofSeconds(ttlSeconds).toString(),
|
|
|
+ "scanCount", scanCount,
|
|
|
+ "maxKeys", maxKeys,
|
|
|
+ "scanned", totalScanned,
|
|
|
+ "updated", totalUpdated,
|
|
|
+ "patterns", results
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
+ @PostMapping("/redis/delete-bid-media-expiring-before")
|
|
|
+ public Object deleteBidMediaExpiringBefore(@RequestBody(required = false) Map<String, Object> body) {
|
|
|
+ if (redisTemplate == null) {
|
|
|
+ throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
|
|
|
+ "redis template is not configured");
|
|
|
+ }
|
|
|
+ long ttlLessThanSeconds = toLong(body == null ? null : body.get("ttlLessThanSeconds"));
|
|
|
+ if (ttlLessThanSeconds <= 0) {
|
|
|
+ ttlLessThanSeconds = Duration.ofDays(2).getSeconds();
|
|
|
+ }
|
|
|
+ long scanCount = toLong(body == null ? null : body.get("scanCount"));
|
|
|
+ if (scanCount <= 0) {
|
|
|
+ scanCount = 1000;
|
|
|
+ }
|
|
|
+ long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
|
|
|
+ if (maxKeys <= 0) {
|
|
|
+ maxKeys = 50000;
|
|
|
+ }
|
|
|
+ List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
|
|
|
+ if (patterns.isEmpty()) {
|
|
|
+ patterns = defaultBidMediaPatterns();
|
|
|
+ }
|
|
|
+
|
|
|
+ String jobId = UUID.randomUUID().toString();
|
|
|
+ RedisMaintenanceJob job = new RedisMaintenanceJob(jobId, "RUNNING", ttlLessThanSeconds, scanCount, maxKeys,
|
|
|
+ 0, 0, null, List.of());
|
|
|
+ saveRedisMaintenanceJob(job);
|
|
|
+ List<String> taskPatterns = List.copyOf(patterns);
|
|
|
+ long taskTtlLessThanSeconds = ttlLessThanSeconds;
|
|
|
+ long taskScanCount = scanCount;
|
|
|
+ long taskMaxKeys = maxKeys;
|
|
|
+ CompletableFuture.runAsync(() -> runDeleteExpiringJob(jobId, taskPatterns, taskTtlLessThanSeconds,
|
|
|
+ taskScanCount, taskMaxKeys));
|
|
|
+ return Map.of(
|
|
|
+ "jobId", jobId,
|
|
|
+ "status", "RUNNING",
|
|
|
+ "ttlLessThanSeconds", ttlLessThanSeconds,
|
|
|
+ "ttlLessThanHuman", Duration.ofSeconds(ttlLessThanSeconds).toString(),
|
|
|
+ "scanCount", scanCount,
|
|
|
+ "maxKeys", maxKeys,
|
|
|
+ "patterns", taskPatterns
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
+ @GetMapping("/redis/delete-bid-media-expiring-before/{jobId}")
|
|
|
+ public Object getDeleteBidMediaExpiringBeforeJob(@PathVariable("jobId") String jobId) {
|
|
|
+ Map<Object, Object> job = readRedisMaintenanceJob(jobId);
|
|
|
+ if (job == null) {
|
|
|
+ throw new ResponseStatusException(HttpStatus.NOT_FOUND, "job not found");
|
|
|
+ }
|
|
|
+ return job;
|
|
|
+ }
|
|
|
+
|
|
|
+ private void runDeleteExpiringJob(String jobId, List<String> patterns, long ttlLessThanSeconds,
|
|
|
+ long scanCount, long maxKeys) {
|
|
|
+ List<Map<String, Object>> results = new ArrayList<>();
|
|
|
+ long totalScanned = 0;
|
|
|
+ long totalDeleted = 0;
|
|
|
+ try {
|
|
|
+ for (String pattern : patterns) {
|
|
|
+ DeleteExpiringResult result = deleteExpiringKeys(pattern, ttlLessThanSeconds, scanCount,
|
|
|
+ Math.max(0, maxKeys - totalScanned));
|
|
|
+ results.add(Map.of(
|
|
|
+ "pattern", pattern,
|
|
|
+ "scanned", result.scanned(),
|
|
|
+ "deleted", result.deleted(),
|
|
|
+ "stoppedByLimit", result.stoppedByLimit()
|
|
|
+ ));
|
|
|
+ totalScanned += result.scanned();
|
|
|
+ totalDeleted += result.deleted();
|
|
|
+ saveRedisMaintenanceJob(new RedisMaintenanceJob(jobId, "RUNNING", ttlLessThanSeconds,
|
|
|
+ scanCount, maxKeys, totalScanned, totalDeleted, null, List.copyOf(results)));
|
|
|
+ if (totalScanned >= maxKeys) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ saveRedisMaintenanceJob(new RedisMaintenanceJob(jobId, "COMPLETED", ttlLessThanSeconds,
|
|
|
+ scanCount, maxKeys, totalScanned, totalDeleted, null, List.copyOf(results)));
|
|
|
+ } catch (Exception e) {
|
|
|
+ saveRedisMaintenanceJob(new RedisMaintenanceJob(jobId, "FAILED", ttlLessThanSeconds,
|
|
|
+ scanCount, maxKeys, totalScanned, totalDeleted, rootMessage(e), List.copyOf(results)));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ @PostMapping("/redis/delete-bid-media-expiring-before-sync")
|
|
|
+ public Object deleteBidMediaExpiringBeforeSync(@RequestBody(required = false) Map<String, Object> body) {
|
|
|
+ if (redisTemplate == null) {
|
|
|
+ throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE,
|
|
|
+ "redis template is not configured");
|
|
|
+ }
|
|
|
+ long ttlLessThanSeconds = toLong(body == null ? null : body.get("ttlLessThanSeconds"));
|
|
|
+ if (ttlLessThanSeconds <= 0) {
|
|
|
+ ttlLessThanSeconds = Duration.ofDays(2).getSeconds();
|
|
|
+ }
|
|
|
+ long scanCount = toLong(body == null ? null : body.get("scanCount"));
|
|
|
+ if (scanCount <= 0) {
|
|
|
+ scanCount = 1000;
|
|
|
+ }
|
|
|
+ long maxKeys = toLong(body == null ? null : body.get("maxKeys"));
|
|
|
+ if (maxKeys <= 0) {
|
|
|
+ maxKeys = 50000;
|
|
|
+ }
|
|
|
+ List<String> patterns = parseStringList(body == null ? null : body.get("patterns"));
|
|
|
+ if (patterns.isEmpty()) {
|
|
|
+ patterns = defaultBidMediaPatterns();
|
|
|
+ }
|
|
|
+
|
|
|
+ List<Map<String, Object>> results = new ArrayList<>();
|
|
|
+ long totalScanned = 0;
|
|
|
+ long totalDeleted = 0;
|
|
|
+ for (String pattern : patterns) {
|
|
|
+ DeleteExpiringResult result = deleteExpiringKeys(pattern, ttlLessThanSeconds, scanCount,
|
|
|
+ Math.max(0, maxKeys - totalScanned));
|
|
|
+ results.add(Map.of(
|
|
|
+ "pattern", pattern,
|
|
|
+ "scanned", result.scanned(),
|
|
|
+ "deleted", result.deleted(),
|
|
|
+ "stoppedByLimit", result.stoppedByLimit()
|
|
|
+ ));
|
|
|
+ totalScanned += result.scanned();
|
|
|
+ totalDeleted += result.deleted();
|
|
|
+ if (totalScanned >= maxKeys) {
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return Map.of(
|
|
|
+ "ttlLessThanSeconds", ttlLessThanSeconds,
|
|
|
+ "ttlLessThanHuman", Duration.ofSeconds(ttlLessThanSeconds).toString(),
|
|
|
+ "scanCount", scanCount,
|
|
|
+ "maxKeys", maxKeys,
|
|
|
+ "scanned", totalScanned,
|
|
|
+ "deleted", totalDeleted,
|
|
|
+ "patterns", results
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
private static int toInt(Object v) {
|
|
|
if (v == null) return 0;
|
|
|
if (v instanceof Number n) return n.intValue();
|
|
|
@@ -115,6 +313,113 @@ public class AdminController {
|
|
|
try { return Long.parseLong(v.toString()); } catch (NumberFormatException e) { return 0L; }
|
|
|
}
|
|
|
|
|
|
+ private TtlTightenResult tightenTtl(String pattern, long ttlSeconds, long scanCount, long maxKeys) {
|
|
|
+ if (maxKeys <= 0 || pattern == null || pattern.isBlank()) {
|
|
|
+ return new TtlTightenResult(0, 0, true);
|
|
|
+ }
|
|
|
+ long scanned = 0;
|
|
|
+ long updated = 0;
|
|
|
+ ScanOptions options = ScanOptions.scanOptions().match(pattern).count(scanCount).build();
|
|
|
+ try (Cursor<String> cursor = redisTemplate.scan(options)) {
|
|
|
+ while (cursor.hasNext()) {
|
|
|
+ String key = cursor.next();
|
|
|
+ scanned++;
|
|
|
+ Long currentTtl = redisTemplate.getExpire(key, TimeUnit.SECONDS);
|
|
|
+ if (currentTtl == null || currentTtl < 0 || currentTtl > ttlSeconds) {
|
|
|
+ Boolean ok = redisTemplate.expire(key, Duration.ofSeconds(ttlSeconds));
|
|
|
+ if (Boolean.TRUE.equals(ok)) {
|
|
|
+ updated++;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (scanned >= maxKeys) {
|
|
|
+ return new TtlTightenResult(scanned, updated, true);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return new TtlTightenResult(scanned, updated, false);
|
|
|
+ }
|
|
|
+
|
|
|
+ private DeleteExpiringResult deleteExpiringKeys(String pattern, long ttlLessThanSeconds, long scanCount, long maxKeys) {
|
|
|
+ if (maxKeys <= 0 || pattern == null || pattern.isBlank()) {
|
|
|
+ return new DeleteExpiringResult(0, 0, true);
|
|
|
+ }
|
|
|
+ long scanned = 0;
|
|
|
+ long deleted = 0;
|
|
|
+ ScanOptions options = ScanOptions.scanOptions().match(pattern).count(scanCount).build();
|
|
|
+ try (Cursor<String> cursor = redisTemplate.scan(options)) {
|
|
|
+ while (cursor.hasNext()) {
|
|
|
+ String key = cursor.next();
|
|
|
+ scanned++;
|
|
|
+ Long currentTtl = redisTemplate.getExpire(key, TimeUnit.SECONDS);
|
|
|
+ if (currentTtl != null && currentTtl > 0 && currentTtl < ttlLessThanSeconds) {
|
|
|
+ Boolean ok = redisTemplate.delete(key);
|
|
|
+ if (Boolean.TRUE.equals(ok)) {
|
|
|
+ deleted++;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (scanned >= maxKeys) {
|
|
|
+ return new DeleteExpiringResult(scanned, deleted, true);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return new DeleteExpiringResult(scanned, deleted, false);
|
|
|
+ }
|
|
|
+
|
|
|
+ private void saveRedisMaintenanceJob(RedisMaintenanceJob job) {
|
|
|
+ try {
|
|
|
+ String key = redisJobKey(job.jobId());
|
|
|
+ Map<String, String> values = Map.of(
|
|
|
+ "jobId", job.jobId(),
|
|
|
+ "status", job.status(),
|
|
|
+ "ttlLessThanSeconds", String.valueOf(job.ttlLessThanSeconds()),
|
|
|
+ "scanCount", String.valueOf(job.scanCount()),
|
|
|
+ "maxKeys", String.valueOf(job.maxKeys()),
|
|
|
+ "scanned", String.valueOf(job.scanned()),
|
|
|
+ "deleted", String.valueOf(job.deleted()),
|
|
|
+ "error", job.error() == null ? "" : job.error(),
|
|
|
+ "patterns", job.patterns().toString()
|
|
|
+ );
|
|
|
+ redisTemplate.opsForHash().putAll(key, values);
|
|
|
+ redisTemplate.expire(key, Duration.ofHours(2));
|
|
|
+ } catch (Exception ignored) {
|
|
|
+ // Job 状态只用于管理查询,写状态失败不影响清理任务继续执行。
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private Map<Object, Object> readRedisMaintenanceJob(String jobId) {
|
|
|
+ Map<Object, Object> payload = redisTemplate.opsForHash().entries(redisJobKey(jobId));
|
|
|
+ if (payload == null || payload.isEmpty()) {
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ return payload;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<String> parseStringList(Object raw) {
|
|
|
+ if (!(raw instanceof Iterable<?> iterable)) {
|
|
|
+ return List.of();
|
|
|
+ }
|
|
|
+ List<String> values = new ArrayList<>();
|
|
|
+ for (Object item : iterable) {
|
|
|
+ if (item != null && !item.toString().isBlank()) {
|
|
|
+ values.add(item.toString());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return values;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static List<String> defaultBidMediaPatterns() {
|
|
|
+ return List.of(
|
|
|
+ "adx:bid:*",
|
|
|
+ "adx:media:*",
|
|
|
+ "adx:honor:bid:*",
|
|
|
+ "adx:honor:media:*",
|
|
|
+ "adx:kuaishou:bid:*",
|
|
|
+ "adx:kuaishou:media:*",
|
|
|
+ "adx:vivo:bid:*",
|
|
|
+ "adx:vivo:media:*"
|
|
|
+ );
|
|
|
+ }
|
|
|
+
|
|
|
private static List<Integer> parseOffsets(Map<String, Object> body) {
|
|
|
if (body == null || body.isEmpty()) {
|
|
|
return defaultBackfillOffsets();
|
|
|
@@ -133,4 +438,24 @@ public class AdminController {
|
|
|
private static List<Integer> defaultBackfillOffsets() {
|
|
|
return List.of(-1, -2, -3, -4, -5, -6, -7);
|
|
|
}
|
|
|
+
|
|
|
+ private static String redisJobKey(String jobId) {
|
|
|
+ return "adx:admin:redis-maintenance-job:" + jobId;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static String rootMessage(Throwable e) {
|
|
|
+ Throwable cur = e;
|
|
|
+ while (cur.getCause() != null) {
|
|
|
+ cur = cur.getCause();
|
|
|
+ }
|
|
|
+ return cur.getMessage() == null ? cur.getClass().getName() : cur.getMessage();
|
|
|
+ }
|
|
|
+
|
|
|
+ private record TtlTightenResult(long scanned, long updated, boolean stoppedByLimit) {}
|
|
|
+
|
|
|
+ private record DeleteExpiringResult(long scanned, long deleted, boolean stoppedByLimit) {}
|
|
|
+
|
|
|
+ private record RedisMaintenanceJob(String jobId, String status, long ttlLessThanSeconds, long scanCount,
|
|
|
+ long maxKeys, long scanned, long deleted, String error,
|
|
|
+ List<Map<String, Object>> patterns) {}
|
|
|
}
|