Akash3121 commented on code in PR #10812: URL: https://github.com/apache/paimon/pull/10812#discussion_r4437856792
########## paimon-filesystems/paimon-jindo/src/main/java/org/apache/paimon/jindo/JindoIOStats.java: ########## @@ -0,0 +1,671 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.jindo; + +import org.apache.paimon.annotation.VisibleForTesting; +import org.apache.paimon.utils.JsonSerdeUtil; +import org.apache.paimon.utils.Pair; + +import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode; + +import com.aliyun.jindodata.common.JindoConstant; +import com.aliyun.jindodata.common.JindoHadoopSystem; +import com.aliyun.jindodata.store.JindoMetricsStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.lang.management.BufferPoolMXBean; +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryPoolMXBean; +import java.lang.management.MemoryUsage; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +/** + * Counters of the Jindo streams in the class loader of paimon-jindo, which is the whole JVM unless + * jobs bundle their own copy, logged with native, process and container memory and a WARN near OOM. + */ +public final class JindoIOStats { + + private static final Logger LOG = LoggerFactory.getLogger(JindoIOStats.class); + + static final StreamCounter INPUT = new StreamCounter(); + static final StreamCounter OUTPUT = new StreamCounter(); + + static final AtomicLong INTERVAL_MILLIS = new AtomicLong(); + private static final AtomicBoolean RUNNING = new AtomicBoolean(false); + private static volatile Pair<JindoHadoopSystem, org.apache.hadoop.fs.Path> nativeSource; + private static volatile long startRetryMillis; + + @VisibleForTesting + static volatile ThreadFactory reporterThreads = r -> new Thread(r, "paimon-jindo-io-stats"); + + private JindoIOStats() {} + + /** + * Counts a stream just opened by a FileIO with this interval and returns its counter, or null + * when the interval is null, zero or negative. The shortest positive interval is used. + */ + @Nullable + static StreamCounter track(StreamCounter counter, @Nullable Duration interval) { + if (interval == null || interval.isZero() || interval.isNegative()) { + return null; + } + long millis = Math.max(1, interval.toMillis()); + long current = INTERVAL_MILLIS.get(); + if (current == 0 || millis < current) { + INTERVAL_MILLIS.accumulateAndGet(millis, (c, m) -> c == 0 ? m : Math.min(c, m)); + } + counter.opened(); + startReporter(); + return counter; + } + + private static void startReporter() { + if (RUNNING.get() + || System.currentTimeMillis() < startRetryMillis + || !RUNNING.compareAndSet(false, true)) { + return; + } + try { + Reporter reporter = + new Reporter( + INTERVAL_MILLIS, + Cgroup.detect( + Paths.get("/proc/self/cgroup"), Paths.get("/sys/fs/cgroup")), + Paths.get("/proc/self"), + RUNNING); + Thread thread = reporterThreads.newThread(reporter::run); + thread.setDaemon(true); + // the inherited loader may be a job's, which must stay collectable + thread.setContextClassLoader(JindoIOStats.class.getClassLoader()); + thread.start(); + } catch (Throwable t) { + // e.g. no native thread near the memory limit; the stream must still reach its caller + startRetryMillis = System.currentTimeMillis() + INTERVAL_MILLIS.get(); + RUNNING.set(false); + LOG.warn( + "Failed to start the Jindo IO stats reporter, retrying with a later stream.", + t); + } + } + + /** Registers a file system that is never closed, used to dump the Jindo native metrics. */ + static void registerNativeMetrics(JindoHadoopSystem fs, org.apache.hadoop.fs.Path path) { + if (nativeSource == null) { + nativeSource = Pair.of(fs, path); + } + } + + /** Open, close and byte counters of one stream direction. */ + static final class StreamCounter { + + private final AtomicLong opened = new AtomicLong(); + private final AtomicLong closed = new AtomicLong(); + private final AtomicLong active = new AtomicLong(); + private final AtomicLong peak = new AtomicLong(); + private final LongAdder bytes = new LongAdder(); + + void opened() { + opened.incrementAndGet(); + peak.accumulateAndGet(active.incrementAndGet(), Math::max); + } + + void closed() { + closed.incrementAndGet(); + active.decrementAndGet(); + } + + void bytes(long n) { + if (n > 0) { + bytes.add(n); + } + } + + long open() { + return active.get(); + } + + long openedTotal() { + return opened.get(); + } + + long closedTotal() { + return closed.get(); + } + + long bytesTotal() { + return bytes.sum(); + } + + /** + * Returns the peak since the previous call, at least {@code open}, and restarts it from the + * current count without losing a concurrent update. + */ + long takePeak(long open) { + while (true) { + long previous = peak.get(); + if (peak.compareAndSet(previous, active.get())) { + return Math.max(previous, open); + } + } + } + } + + /** Samples memory every second and logs the statistics until the streams go idle. */ + static final class Reporter { + + private static final long SAMPLE_MILLIS = 1000; + private static final double[] ALARM_RATIOS = {0.8, 0.9, 0.95}; + private static final double ALARM_RESET_RATIO = 0.7; + + private final AtomicLong intervalMillis; + private final List<Cgroup> cgroups; + private final Path procSelf; + private final AtomicBoolean running; + private final int[] alarmLevels; + + @Nullable private JindoMetricsStore nativeStore; + private boolean nativeDisabled; + @Nullable private String jemalloc; + private long lastEvents; + + Reporter( + AtomicLong intervalMillis, + List<Cgroup> cgroups, + Path procSelf, + AtomicBoolean running) { + this.intervalMillis = intervalMillis; + this.cgroups = cgroups; + this.procSelf = procSelf; + this.running = running; + this.alarmLevels = new int[cgroups.size()]; + } + + /** Runs until an interval passes without stream activity; the next open starts another. */ + void run() { + long lastReport = System.currentTimeMillis(); + while (true) { + try { + Thread.sleep(Math.min(SAMPLE_MILLIS, intervalMillis.get())); + } catch (InterruptedException e) { + running.set(false); + return; + } + try { + String alarm = checkAlarm(); + if (alarm != null) { + LOG.warn(alarm); + } + long now = System.currentTimeMillis(); + // read every time, so that a shorter interval set by a later FileIO applies + if (now - lastReport >= intervalMillis.get()) { + lastReport = now; + if (hasActivity()) { + LOG.info(snapshot()); + } else if (stop()) { + return; + } + } + } catch (Throwable t) { + LOG.debug("Failed to collect Jindo IO stats.", t); + } + } + } + + /** Returns a WARN message when the working set of a cgroup level enters its next alarm. */ + @Nullable + String checkAlarm() { + Cgroup alarmed = null; + double alarmedRatio = 0; + // each level keeps its own alarm, so a parent's WARN does not mute a child's + for (int i = 0; i < cgroups.size(); i++) { + Cgroup cgroup = cgroups.get(i); + long limit = cgroup.limit(); + long workingSet = cgroup.workingSet(); + if (limit <= 0 || workingSet < 0) { + continue; + } + double ratio = (double) workingSet / limit; + if (ratio < ALARM_RESET_RATIO) { + alarmLevels[i] = 0; + } + int level = 0; + while (level < ALARM_RATIOS.length && ratio >= ALARM_RATIOS[level]) { + level++; + } + if (level > alarmLevels[i]) { + alarmLevels[i] = level; Review Comment: If multiple cgroup levels cross a threshold in the same sample, this loop advances `alarmLevels` for all of them but logs only the level with the highest ratio. For example, if a parent reaches 90% while its child reaches 85%, only the parent is logged; if the parent later drops and the child remains at 85%, the child is already marked as warned and produces no WARN. That contradicts the intended guarantee that one level’s warning does not mute another. Could we advance only the level actually emitted (so the other is emitted on the next sample), or emit every newly crossed level, and add a simultaneous-crossing regression test? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
