From 9fb103e979c2e53519b5b184371f353ded015fe4 Mon Sep 17 00:00:00 2001 From: htyoung Date: Tue, 13 Aug 2024 15:42:15 +0800 Subject: [PATCH] [opt](fe) Optimize calculate load job num metric in FE (#39267) cherry pick from https://github.com/apache/doris/pull/31952 https://github.com/apache/doris/pull/34020 Co-authored-by: Lei Zhang <27994433+SWJTU-ZhangLei@users.noreply.github.com> --- .../apache/doris/load/loadv2/LoadManager.java | 24 +++++++---- .../org/apache/doris/metric/MetricRepo.java | 40 +++++++++++++------ 2 files changed, 43 insertions(+), 21 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java index 122182c65c..46964268a6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java @@ -60,6 +60,7 @@ import com.google.common.collect.Maps; import com.google.common.collect.Sets; import org.apache.commons.collections.CollectionUtils; import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.time.StopWatch; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -343,6 +344,11 @@ public class LoadManager implements Writable { job.unprotectReadEndOperation(operation); LOG.info(new LogBuilder(LogKey.LOAD_JOB, operation.getId()).add("operation", operation) .add("msg", "replay end load job").build()); + + // When idToLoadJob size increase 10000 roughly, we run removeOldLoadJob to reduce mem used + if ((idToLoadJob.size() > 0) && (idToLoadJob.size() % 10000 == 0)) { + removeOldLoadJob(); + } } /** @@ -406,14 +412,10 @@ public class LoadManager implements Writable { /** * Get load job num, used by metric. **/ - public long getLoadJobNum(JobState jobState, EtlJobType jobType) { - readLock(); - try { - return idToLoadJob.values().stream().filter(j -> j.getState() == jobState && j.getJobType() == jobType) - .count(); - } finally { - readUnlock(); - } + public Map, Long> getLoadJobNum() { + return idToLoadJob.values().stream().collect(Collectors.groupingBy( + loadJob -> Pair.of(loadJob.getJobType(), loadJob.getState()), + Collectors.counting())); } /** @@ -475,6 +477,8 @@ public class LoadManager implements Writable { } private void removeLoadJobIf(Predicate pred) { + long removeJobNum = 0; + StopWatch stopWatch = StopWatch.createStarted(); writeLock(); try { Iterator> iter = idToLoadJob.entrySet().iterator(); @@ -483,10 +487,14 @@ public class LoadManager implements Writable { if (pred.test(job)) { iter.remove(); jobRemovedTrigger(job); + removeJobNum++; } } } finally { writeUnlock(); + stopWatch.stop(); + LOG.info("end to removeOldLoadJob, removeJobNum:{} cost:{} ms", + removeJobNum, stopWatch.getTime()); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java index 0a771fa7ad..a18b57ac78 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/metric/MetricRepo.java @@ -23,6 +23,7 @@ import org.apache.doris.alter.AlterJobV2.JobType; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.TabletInvertedIndex; import org.apache.doris.common.Config; +import org.apache.doris.common.Pair; import org.apache.doris.common.ThreadPoolManager; import org.apache.doris.common.util.NetUtils; import org.apache.doris.load.EtlJobType; @@ -42,6 +43,7 @@ import org.apache.doris.transaction.TransactionStatus; import com.codahale.metrics.Histogram; import com.codahale.metrics.MetricRegistry; +import com.google.common.collect.Maps; import com.google.common.collect.Sets; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -130,19 +132,19 @@ public final class MetricRepo { public static GaugeMetricImpl GAUGE_REQUEST_PER_SECOND; public static GaugeMetricImpl GAUGE_QUERY_ERR_RATE; public static GaugeMetricImpl GAUGE_MAX_TABLET_COMPACTION_SCORE; + private static Map, Long> loadJobNum = Maps.newHashMap(); private static ScheduledThreadPoolExecutor metricTimer = ThreadPoolManager.newDaemonScheduledThreadPool(1, "metric-timer-pool", true); private static MetricCalculator metricCalculator = new MetricCalculator(); - // init() should only be called after catalog is contructed. + // init() should only be called after catalog is constructed. public static synchronized void init() { if (isInit) { return; } // load jobs - LoadManager loadManger = Env.getCurrentEnv().getLoadManager(); for (EtlJobType jobType : EtlJobType.values()) { if (jobType == EtlJobType.UNKNOWN) { continue; @@ -154,7 +156,7 @@ public final class MetricRepo { if (!Env.getCurrentEnv().isMaster()) { return 0L; } - return loadManger.getLoadJobNum(state, jobType); + return MetricRepo.getLoadJobNum(jobType, state); } }; gauge.addLabel(new MetricLabel("job", "load")).addLabel(new MetricLabel("type", jobType.name())) @@ -175,7 +177,7 @@ public final class MetricRepo { Set states = Sets.newHashSet(); states.add(jobState); List jobs = routineLoadManager.getRoutineLoadJobByState(states); - return Long.valueOf(jobs.size()); + return (long) jobs.size(); } }; gauge.addLabel(new MetricLabel("job", "load")).addLabel(new MetricLabel("type", "ROUTINE_LOAD")) @@ -508,7 +510,7 @@ public final class MetricRepo { private static void initSystemMetrics() { // TCP retransSegs - GaugeMetric tcpRetransSegs = (GaugeMetric) new GaugeMetric( + GaugeMetric tcpRetransSegs = new GaugeMetric( "snmp", MetricUnit.NOUNIT, "All TCP packets retransmitted") { @Override public Long getValue() { @@ -519,7 +521,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(tcpRetransSegs); // TCP inErrs - GaugeMetric tpcInErrs = (GaugeMetric) new GaugeMetric( + GaugeMetric tpcInErrs = new GaugeMetric( "snmp", MetricUnit.NOUNIT, "The number of all problematic TCP packets received") { @Override public Long getValue() { @@ -530,7 +532,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(tpcInErrs); // TCP inSegs - GaugeMetric tpcInSegs = (GaugeMetric) new GaugeMetric( + GaugeMetric tpcInSegs = new GaugeMetric( "snmp", MetricUnit.NOUNIT, "The number of all TCP packets received") { @Override public Long getValue() { @@ -541,7 +543,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(tpcInSegs); // TCP outSegs - GaugeMetric tpcOutSegs = (GaugeMetric) new GaugeMetric( + GaugeMetric tpcOutSegs = new GaugeMetric( "snmp", MetricUnit.NOUNIT, "The number of all TCP packets send with RST") { @Override public Long getValue() { @@ -552,7 +554,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(tpcOutSegs); // Memory Total - GaugeMetric memTotal = (GaugeMetric) new GaugeMetric( + GaugeMetric memTotal = new GaugeMetric( "meminfo", MetricUnit.BYTES, "Total usable memory") { @Override public Long getValue() { @@ -563,7 +565,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(memTotal); // Memory Free - GaugeMetric memFree = (GaugeMetric) new GaugeMetric( + GaugeMetric memFree = new GaugeMetric( "meminfo", MetricUnit.BYTES, "The amount of physical memory not used by the system") { @Override public Long getValue() { @@ -574,7 +576,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(memFree); // Memory Total - GaugeMetric memAvailable = (GaugeMetric) new GaugeMetric("meminfo", MetricUnit.BYTES, + GaugeMetric memAvailable = new GaugeMetric("meminfo", MetricUnit.BYTES, "An estimate of how much memory is available for starting new applications, without swapping") { @Override public Long getValue() { @@ -585,7 +587,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(memAvailable); // Buffers - GaugeMetric buffers = (GaugeMetric) new GaugeMetric("meminfo", MetricUnit.BYTES, + GaugeMetric buffers = new GaugeMetric("meminfo", MetricUnit.BYTES, "Memory in buffer cache, so relatively temporary storage for raw disk blocks") { @Override public Long getValue() { @@ -596,7 +598,7 @@ public final class MetricRepo { DORIS_METRIC_REGISTER.addSystemMetrics(buffers); // Cached - GaugeMetric cached = (GaugeMetric) new GaugeMetric( + GaugeMetric cached = new GaugeMetric( "meminfo", MetricUnit.BYTES, "Memory in the pagecache (Diskcache and Shared Memory)") { @Override public Long getValue() { @@ -664,6 +666,9 @@ public final class MetricRepo { // update the metrics first updateMetrics(); + // update load job metrics + updateLoadJobMetrics(); + // jvm JvmService jvmService = new JvmService(); JvmStats jvmStats = jvmService.stats(); @@ -701,4 +706,13 @@ public final class MetricRepo { public static synchronized List getMetricsByName(String name) { return DORIS_METRIC_REGISTER.getMetricsByName(name); } + + private static void updateLoadJobMetrics() { + LoadManager loadManager = Env.getCurrentEnv().getLoadManager(); + MetricRepo.loadJobNum = loadManager.getLoadJobNum(); + } + + private static long getLoadJobNum(EtlJobType jobType, JobState jobState) { + return MetricRepo.loadJobNum.getOrDefault(Pair.of(jobType, jobState), 0L); + } }