feat(scheduler): 增加CRM重点表的独占调度

2-6点CRM重点表独占a4集群

service-common/src/main/java/com/lanyuanxiaoyao/service/common/Constants.java

@@ -224,6 +224,7 @@ public interface Constants {
 
     String SCHEDULE_RECOMMEND = "schedule_recommend";
     String SCHEDULE_FORCE = "schedule_force";
+    String SCHEDULE_ESCAPE = "schedule_escape";
 
     BiFunction<TableMeta, String, String> FIELD_COVERT = (tableMeta, field) -> {
         if (TableMeta.SourceType.TELEPG.equals(tableMeta.getSourceType())) {

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/configuration/ScheduleStrategyProvider.java

@@ -32,7 +32,7 @@ public class ScheduleStrategyProvider {
                 // ODS重点表调度
                 ScheduleStrategyImpl.simple("ods_focus_schedule", "ODS 重点表调度", OdsFocusScheduleJob.class, "0 30 23 * * ?"),
                 // CRM重点表调度
-                ScheduleStrategyImpl.simple("crm_focus_schedule", "CRM 重点表补充调度", CrmFocusScheduleJob.class, "0 15 3,4 * * ?"),
+                ScheduleStrategyImpl.simple("crm_focus_schedule", "CRM 重点表补充调度", CrmFocusScheduleJob.class, "0 0/10 2,3,4,5,6 * * ?"),
                 // 忙时调度
                 ScheduleStrategyImpl.simple("focus_schedule", "重点表跨天调度", FocusUnVersionUpdateScheduleJob.class, "0 0,10,20,40 0,1 * * ?"),
                 ScheduleStrategyImpl.simple(false, "remove_scheduled", "跨天调度时及时删除已完成跨天的任务", RemoveScheduledJob.class, "0 * 0,1 * * ?")

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/compaction/CrmFocusScheduleJob.java

@@ -1,5 +1,6 @@
 package com.lanyuanxiaoyao.service.scheduler.quartz.compaction;
 
+import cn.hutool.core.util.StrUtil;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.lanyuanxiaoyao.service.common.Constants;
 import com.lanyuanxiaoyao.service.common.utils.TableMetaHelper;
@@ -8,6 +9,8 @@ import com.lanyuanxiaoyao.service.forest.service.InfoService;
 import com.lanyuanxiaoyao.service.scheduler.utils.ScheduleHelper;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import org.eclipse.collections.api.factory.Maps;
+import org.eclipse.collections.api.list.ImmutableList;
 import org.quartz.DisallowConcurrentExecution;
 import org.quartz.JobExecutionContext;
 import org.slf4j.Logger;
@@ -16,7 +19,7 @@ import org.springframework.cloud.client.discovery.DiscoveryClient;
 import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder;
 
 /**
- * 日常普通调度
+ * Crm重点表调度
  *
  * @author ZhangJiacheng
  * @date 2023-05-11
@@ -40,13 +43,16 @@ public class CrmFocusScheduleJob extends BaseScheduleJob {
 
     public static void schedule(DiscoveryClient discoveryClient, InfoService infoService, HudiService hudiService, ObjectMapper mapper, String comment) {
         logger.info("Crm focus schedule");
+        ImmutableList<String> unUpdateVersionTableIds = infoService.nonUpdatedVersionTables();
         ScheduleHelper.schedule(
                 discoveryClient,
                 infoService,
                 hudiService,
                 mapper,
-                meta -> TableMetaHelper.existsTag(meta.getTags(), Constants.TAGS_CRM_FOCUS),
+                meta -> TableMetaHelper.existsTag(meta.getTags(), Constants.TAGS_CRM_FOCUS)
-                comment
+                        && unUpdateVersionTableIds.contains(StrUtil.format("{}-{}", meta.getFlinkJobId(), meta.getAlias())),
+                comment,
+                Maps.immutable.of(Constants.SCHEDULE_FORCE, Constants.CLUSTER_A4)
         );
     }
 

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/compaction/DistributeScheduleJob.java

@@ -67,14 +67,18 @@ public class DistributeScheduleJob extends BaseScheduleJob {
         if (metadata.containsKey(Constants.SCHEDULE_RECOMMEND)) {
             String recommendCluster = metadata.get(Constants.SCHEDULE_RECOMMEND);
             Optional<Cluster> cluster = clusters.select(s -> StrUtil.equals(recommendCluster, s.cluster())).getFirstOptional();
-            if (cluster.isPresent() && cluster.get().available()) {
+            if (cluster.isPresent() && cluster.get().available(metadata)) {
                 return cluster.get().queue();
             } else {
                 logger.warn(StrUtil.format("{} cluster not found or busy"));
             }
         }
         for (Cluster cluster : clusters) {
-            if (cluster.available()) {
+            if (cluster.available(metadata)) {
+                String escapeCluster = metadata.getOrDefault(Constants.SCHEDULE_ESCAPE, null);
+                if (StrUtil.isNotBlank(escapeCluster) && StrUtil.equals(escapeCluster, cluster.cluster())) {
+                    continue;
+                }
                 return cluster.queue();
             }
         }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/cluster/A4Cluster.java

@@ -3,6 +3,7 @@ package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.cluster;
 import com.lanyuanxiaoyao.service.common.Constants;
 import com.lanyuanxiaoyao.service.forest.service.YarnService;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.AvailableStrategy;
+import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.DatetimeLimit;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.QueueSizeLimit;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.YarnQueueUsedLimit;
 import org.springframework.cloud.client.discovery.DiscoveryClient;
@@ -23,7 +24,8 @@ public class A4Cluster extends Cluster {
                 Constants.COMPACTION_QUEUE_A4,
                 AvailableStrategy.and(
                         new QueueSizeLimit(client, Constants.COMPACTION_QUEUE_A4, 10),
-                        new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_A4, "ten_iap.datalake", 0.8)
+                        new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_A4, "ten_iap.datalake", 0.8),
+                        new DatetimeLimit(false, "* * 2,3,4,5,6 * * ? *")
                 )
         );
     }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/cluster/B5Cluster.java

@@ -22,7 +22,7 @@ public class B5Cluster extends Cluster {
                         new QueueSizeLimit(client, Constants.COMPACTION_QUEUE_B5, 10),
                         new YarnQueueUsedLimit(yarnService, Constants.CLUSTER_B5, "ten_iap.datalake", 0.9)
                 ) */
-                () -> false
+                metadata -> false
         );
     }
 }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/cluster/Cluster.java

@@ -3,6 +3,7 @@ package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.cluster;
 import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.core.util.StrUtil;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.AvailableStrategy;
+import org.eclipse.collections.api.map.ImmutableMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,12 +38,12 @@ public class Cluster {
         return queue;
     }
 
-    public boolean available() {
+    public boolean available(ImmutableMap<String, String> metadata) {
         try {
             if (ObjectUtil.isNull(availableStrategy)) {
                 return true;
             }
-            return availableStrategy.available();
+            return availableStrategy.available(metadata);
         } catch (Throwable throwable) {
             logger.error(StrUtil.format("Check cluster {} available fail", this.cluster), throwable);
         }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/strategy/AvailableStrategy.java

@@ -2,6 +2,7 @@ package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy;
 
 import org.eclipse.collections.api.factory.Lists;
 import org.eclipse.collections.api.list.ImmutableList;
+import org.eclipse.collections.api.map.ImmutableMap;
 
 /**
  * 判断集群是否有可用资源的策略
@@ -22,7 +23,7 @@ public interface AvailableStrategy {
         return new WhetherAvailableStrategy(condition, trueOption, falseOption);
     }
 
-    boolean available();
+    boolean available(ImmutableMap<String, String> metadata);
 
     abstract class BaseAvailableStrategy implements AvailableStrategy {
         protected final ImmutableList<AvailableStrategy> strategies;
@@ -38,8 +39,8 @@ public interface AvailableStrategy {
         }
 
         @Override
-        public boolean available() {
+        public boolean available(ImmutableMap<String, String> metadata) {
-            return strategies.allSatisfy(AvailableStrategy::available);
+            return strategies.allSatisfy(strategy -> strategy.available(metadata));
         }
     }
 
@@ -49,8 +50,8 @@ public interface AvailableStrategy {
         }
 
         @Override
-        public boolean available() {
+        public boolean available(ImmutableMap<String, String> metadata) {
-            return strategies.anySatisfy(AvailableStrategy::available);
+            return strategies.anySatisfy(strategy -> strategy.available(metadata));
         }
     }
 
@@ -66,8 +67,8 @@ public interface AvailableStrategy {
         }
 
         @Override
-        public boolean available() {
+        public boolean available(ImmutableMap<String, String> metadata) {
-            return condition.available() ? trueOption.available() : falseOption.available();
+            return condition.available(metadata) ? trueOption.available(metadata) : falseOption.available(metadata);
         }
     }
 }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/strategy/DatetimeLimit.java

@@ -6,6 +6,7 @@ import com.cronutils.model.time.ExecutionTime;
 import com.cronutils.parser.CronParser;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
+import org.eclipse.collections.api.map.ImmutableMap;
 
 /**
  * 时间限制
@@ -25,7 +26,7 @@ public class DatetimeLimit implements AvailableStrategy {
     }
 
     @Override
-    public boolean available() {
+    public boolean available(ImmutableMap<String, String> metadata) {
         return runInPeriod == execution.isMatch(ZonedDateTime.now().withNano(0));
     }
 }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/strategy/MetadataLimit.java

@@ -0,0 +1,22 @@
+package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy;
+
+import java.util.function.Predicate;
+import org.eclipse.collections.api.map.ImmutableMap;
+
+/**
+ * 通过Metadata信息限制调度
+ *
+ * @author lanyuanxiaoyao
+ */
+public class MetadataLimit implements AvailableStrategy {
+    private final Predicate<ImmutableMap<String, String>> predicate;
+
+    public MetadataLimit(Predicate<ImmutableMap<String, String>> predicate) {
+        this.predicate = predicate;
+    }
+
+    @Override
+    public boolean available(ImmutableMap<String, String> metadata) {
+        return predicate.test(metadata);
+    }
+}

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/strategy/QueueSizeLimit.java

@@ -1,6 +1,7 @@
 package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy;
 
 import com.lanyuanxiaoyao.service.configuration.utils.QueueUtil;
+import org.eclipse.collections.api.map.ImmutableMap;
 import org.springframework.cloud.client.discovery.DiscoveryClient;
 
 /**
@@ -21,7 +22,7 @@ public class QueueSizeLimit implements AvailableStrategy {
     }
 
     @Override
-    public boolean available() {
+    public boolean available(ImmutableMap<String, String> metadata) {
         return QueueUtil.size(discoveryClient, queue) < limit;
     }
 }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/quartz/distribute/strategy/YarnQueueUsedLimit.java

@@ -2,6 +2,7 @@ package com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy;
 
 import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnQueue;
 import com.lanyuanxiaoyao.service.forest.service.YarnService;
+import org.eclipse.collections.api.map.ImmutableMap;
 
 /**
  * Yarn 队列剩余资源限制
@@ -28,7 +29,7 @@ public class YarnQueueUsedLimit implements AvailableStrategy {
     }
 
     @Override
-    public boolean available() {
+    public boolean available(ImmutableMap<String, String> metadata) {
         return queueUsed(cluster, queue) < limit;
     }
 }

service-scheduler/src/main/java/com/lanyuanxiaoyao/service/scheduler/utils/ScheduleHelper.java

@@ -108,7 +108,10 @@ public class ScheduleHelper {
                                 .thenComparing(TableMetaWrapper::getCompactionDuration, Comparator.reverseOrder()))
                 .collect(meta -> new QueueItem<>(
                         StrUtil.format("{}-{}", meta.getFlinkJobId(), meta.getAlias()),
-                        metadata.toMap(),
+                        metadata
+                                .toMap()
+                                .withKeyValue(Constants.METRICS_LABEL_FLINK_JOB_ID, meta.getFlinkJobId().toString())
+                                .withKeyValue(Constants.METRICS_LABEL_ALIAS, meta.getAlias()),
                         meta.getPriority(),
                         ScheduleJob.builder()
                                 .id(IdUtil.nanoId(10))

service-scheduler/src/test/java/com/lanyuanxiaoyao/service/scheduler/TestCache.java

@@ -0,0 +1,36 @@
+package com.lanyuanxiaoyao.service.scheduler;
+
+import com.github.benmanes.caffeine.cache.CacheLoader;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.LoadingCache;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * @author lanyuanxiaoyao
+ */
+public class TestCache {
+    public static void main(String[] args) {
+        CacheLoaderImpl loader = new CacheLoaderImpl();
+        LoadingCache<String, Boolean> cache = Caffeine.newBuilder()
+                .build(loader);
+        loader.setBindCache(cache);
+        System.out.println(cache.asMap());
+        System.out.println(cache.get("one"));
+        System.out.println(cache.asMap());
+    }
+
+    public static class CacheLoaderImpl implements CacheLoader<String, Boolean> {
+        private LoadingCache<String, Boolean> bindCache;
+
+        public void setBindCache(LoadingCache<String, Boolean> bindCache) {
+            this.bindCache = bindCache;
+        }
+
+        @Override
+        public @Nullable Boolean load(@NonNull String key) throws Exception {
+            bindCache.put("two", false);
+            return true;
+        }
+    }
+}

service-scheduler/src/test/java/com/lanyuanxiaoyao/service/scheduler/TestStrategy.java

@@ -2,6 +2,7 @@ package com.lanyuanxiaoyao.service.scheduler;
 
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.AvailableStrategy;
 import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.DatetimeLimit;
+import org.eclipse.collections.api.factory.Maps;
 
 /**
  * @author lanyuanxiaoyao
@@ -10,8 +11,8 @@ import com.lanyuanxiaoyao.service.scheduler.quartz.distribute.strategy.DatetimeL
 public class TestStrategy {
     public static void main(String[] args) {
         AvailableStrategy strategy = AvailableStrategy.and(
-                new DatetimeLimit(false, "* * 7-19 28-30 4 ? 2024")
+                new DatetimeLimit(false, "* * 2,3,4,5,6 * * ? *")
         );
-        System.out.println(strategy.available());
+        System.out.println(strategy.available(Maps.immutable.empty()));
     }
 }