codope commented on code in PR #20151: URL: https://github.com/apache/hudi/pull/20151#discussion_r4163867874
########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/ArchivalCadenceCheck.java: ########## @@ -0,0 +1,127 @@ +/* + * 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.hudi.utilities.health; + +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieArchivalConfig; + +/** + * Checks whether archival is keeping the active timeline bounded. + * + * <p>Archival moves old instants out of the active timeline into the archived timeline. When it + * stops, the active timeline grows without limit, and because it is read and parsed on essentially + * every table operation, everything touching the table gets progressively slower. Unlike a failed + * write this produces no error -- the table simply degrades. + * + * <p>Two independent signals are reported. The first compares the active timeline against the + * configured retention, with slack: archival runs periodically rather than on every commit, so a + * timeline slightly over the configured maximum is normal. The second is an absolute ceiling, + * because past a few thousand instants timeline loading is a problem regardless of what the + * configuration permits -- a table configured to retain a very large number of commits is + * misconfigured rather than healthy. + * + * <p>This check deliberately does not attribute the cause. A blocked archival is most often a + * savepoint, which {@link SavepointArchivalBlockCheck} diagnoses specifically; the two are meant to + * be read together. + */ +public class ArchivalCadenceCheck implements HealthCheck { + + static final String NAME = "archival"; + + /** + * How far past the configured retention the active timeline may grow before being called + * unhealthy. Archival is periodic, so the timeline routinely sits somewhat above the maximum + * between runs. + */ + static final double RETENTION_SLACK_FACTOR = 1.1; + + /** + * Active timeline size past which the table is called unhealthy regardless of configuration. + * Beyond roughly this many instants, loading and parsing the timeline is itself a performance + * problem on every operation against the table. + */ + static final int MAX_REASONABLE_ACTIVE_INSTANTS = 5000; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether archival is keeping the active timeline bounded"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) { + Option<String> missing = context.firstMissingConfig(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + int maxCommitsToKeep = Integer.parseInt(context.getProps().getString( + HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), + HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.defaultValue())); + + HoodieActiveTimeline activeTimeline = context.getMetaClient().getActiveTimeline(); + int completedInstants = activeTimeline.filterCompletedInstants().countInstants(); + int writeInstants = activeTimeline.getWriteTimeline().countInstants(); + int retentionThreshold = (int) Math.ceil(RETENTION_SLACK_FACTOR * maxCommitsToKeep); + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), maxCommitsToKeep) + .config("effective.unhealthy.threshold.write.instants", retentionThreshold) + .config("effective.absolute.ceiling.completed.instants", MAX_REASONABLE_ACTIVE_INSTANTS) + .config("observed.write.instants", writeInstants) + .config("observed.completed.instants", completedInstants); + + boolean overAbsoluteCeiling = completedInstants >= MAX_REASONABLE_ACTIVE_INSTANTS; + boolean overConfiguredRetention = writeInstants >= retentionThreshold; Review Comment: Looks like `archival` ignores the cleaner-driven retention the archiver actually enforces. `ArchivalUtils.getMinAndMaxInstantsToKeep` raises min and max whenever the cleaner retains more than keep.min.commits. The archiver also holds back for: - the earliest pending write instant (TimelineArchiverV2) - MOR compaction retention - the MDT's latest compaction So a supported setup like clean.commits.retained=100 with keep.max.commits=30 can be flagged UNHEALTHY. Should we check, and model, the cleaner keys the same way `getMinAndMaxInstantsToKeep` does, and name the actual blocker (oldest pending instant, MDT compaction) in the finding? Also, `getWriteTimeline()` includes requested and inflight compaction/clustering, while the archiver counts completed commits only. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/SavepointArchivalBlockCheck.java: ########## @@ -0,0 +1,150 @@ +/* + * 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.hudi.utilities.health; + +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.table.timeline.TimelineUtils; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieArchivalConfig; + +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.Collections; +import java.util.List; +import java.util.stream.Collectors; + +/** + * Checks for savepoints that are holding back archival. + * + * <p>Archival stops at the earliest savepoint and will not advance past it. A savepoint taken for a + * one-off restore and then forgotten therefore pins the active timeline indefinitely. The timeline + * grows without bound, every timeline load gets slower, and because archival is blocked the + * cleaner's retention interacts with it in ways that keep data files alive too. This is a quiet + * failure: the table keeps accepting writes and nothing errors. + * + * <p>Two signals are reported. A savepoint sitting further back than the configured archival + * retention is actively blocking archival now. A savepoint older than {@link #STALE_SAVEPOINT_DAYS} + * is flagged separately as likely forgotten, since savepoints are normally short-lived. + * + * <p>The remedy this check points at is always to release the savepoint once the operation it was + * taken for is done. It deliberately does not suggest loosening archival to step over savepoints. + */ +public class SavepointArchivalBlockCheck implements HealthCheck { + + static final String NAME = "savepoint"; + + /** + * Age past which a savepoint is called out as probably forgotten. Savepoints are normally taken + * around a specific operation and released soon after; one lingering for weeks is usually an + * oversight rather than an intent. + */ + static final int STALE_SAVEPOINT_DAYS = 7; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether a savepoint is blocking archival or has been left behind"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) { + Option<String> missing = context.firstMissingConfig(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + int maxCommitsToKeep = Integer.parseInt(context.getProps().getString( + HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), + HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.defaultValue())); + HoodieTimeline writeTimeline = context.getMetaClient().getActiveTimeline().getWriteTimeline(); + List<HoodieInstant> savepoints = context.getMetaClient().getActiveTimeline() + .getSavePointTimeline().filterCompletedInstants().getInstants(); + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(), maxCommitsToKeep) + .config("observed.savepoint.count", savepoints.size()); + + if (savepoints.isEmpty()) { + return result.status(HealthStatus.HEALTHY).summary("No savepoints on the table").build(); + } + + List<String> savepointTimes = savepoints.stream() + .map(HoodieInstant::requestedTime).collect(Collectors.toList()); + result.config("observed.savepoints", String.join(",", savepointTimes)); + + List<String> blocking = savepointsBeyondRetention(writeTimeline, savepointTimes, maxCommitsToKeep); + List<String> stale = savepointTimes.stream() + .filter(SavepointArchivalBlockCheck::isOlderThanStaleThreshold) + .collect(Collectors.toList()); + + if (!stale.isEmpty()) { + result.finding(String.format( + "Savepoint(s) %s are more than %d days old. If the operation they were taken for is " + + "finished, delete them so archival can advance.", + String.join(", ", stale), STALE_SAVEPOINT_DAYS)); + } + + if (blocking.isEmpty()) { + return result.status(HealthStatus.HEALTHY) + .summary(String.format("%d savepoint(s) present, none older than the archival retention of %d commits", + savepoints.size(), maxCommitsToKeep)) + .build(); + } + + return result.status(HealthStatus.UNHEALTHY) + .summary(String.format( + "%d savepoint(s) older than the archival retention are blocking archival: %s", + blocking.size(), String.join(", ", blocking))) + .finding(String.format( + "Archival stops at the earliest savepoint. The active timeline cannot shrink past %s " + + "until that savepoint is released.", blocking.get(0))) + .finding("Delete the savepoint once the operation it was taken for is complete, using the " Review Comment: looks like savepoint can ignore hoodie.archive.beyond.savepoint. When the key is true, the archiver skips savepointed commits and does not stop at them (TimelineArchiverV2), yet the check still reports "blocking archival". `remedyIsAlwaysToReleaseTheSavepoint` asserts the key is never consulted. The skill's guardrail against recommending it is fine, but the check has to honour it when it is already on. Also, on a table with no savepoints, the check gates on keep.max.commits first and reports SKIPPED. It should report HEALTHY without needing any configuration. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieTableHealthChecker.java: ########## @@ -0,0 +1,327 @@ +/* + * 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.hudi.utilities; + +import org.apache.hudi.common.config.TypedProperties; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.util.StringUtils; +import org.apache.hudi.hadoop.fs.HadoopFSUtils; +import org.apache.hudi.utilities.health.ArchivalCadenceCheck; +import org.apache.hudi.utilities.health.CleanerCadenceCheck; +import org.apache.hudi.utilities.health.CompactionCadenceCheck; +import org.apache.hudi.utilities.health.HealthCheck; +import org.apache.hudi.utilities.health.HealthCheckContext; +import org.apache.hudi.utilities.health.HealthCheckResult; +import org.apache.hudi.utilities.health.HealthStatus; +import org.apache.hudi.utilities.health.SavepointArchivalBlockCheck; + +import com.beust.jcommander.JCommander; +import com.beust.jcommander.Parameter; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializationFeature; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.extern.slf4j.Slf4j; +import org.apache.hadoop.conf.Configuration; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Reports whether a Hudi table's table services are keeping up with ingestion. + * + * <p>Intended to be run on a schedule -- daily is typical -- against tables an operator is + * responsible for. Each check answers one question about whether a service is progressing as + * expected, and the process exit code makes the overall verdict usable from a cron job or CI + * pipeline without parsing the output. + * + * <p>This tool is strictly read-only. It inspects timeline state and reports; it never schedules or + * runs a table service, and never writes to the table. + * + * <h3>Writer properties are required</h3> + * + * <p>Nearly every question this tool answers is relative to how the table is written. Whether + * fifty delta commits since the last compaction is fine or alarming depends on the configured + * trigger; whether a thousand instants in the active timeline is fine depends on the configured + * retention. Hudi has a default for each of these, so the tool could always produce an answer -- + * but for a table written with non-default settings that answer would be wrong, and wrong in the + * direction that erodes trust, reporting healthy tables as unhealthy. + * + * <p>So by default a check whose governing property was not supplied reports {@code SKIPPED} and + * names the property, rather than guessing. Supply properties with {@code --props} or + * {@code --hoodie-conf}, or pass {@code --apply-all-defaults} to explicitly accept Hudi defaults + * for anything unset. + * + * <h3>Exit codes</h3> + * <ul> + * <li>{@code 0} -- every check that ran reported healthy</li> + * <li>{@code 1} -- at least one check reported unhealthy</li> + * <li>{@code 2} -- bad arguments, or the table could not be read</li> + * </ul> + * + * <h3>Example</h3> + * <pre> + * spark-submit --master local \ + * --class org.apache.hudi.utilities.HoodieTableHealthChecker \ + * $HUDI_DIR/packaging/hudi-utilities-bundle/target/hudi-utilities-bundle_2.12-1.3.0-SNAPSHOT.jar \ + * --base-path s3://bucket/path/to/table \ + * --props s3://bucket/path/to/writer.properties \ + * --output JSON + * </pre> + */ +@Slf4j +public class HoodieTableHealthChecker implements Serializable { + + private static final long serialVersionUID = 1L; + + static final int EXIT_HEALTHY = 0; + static final int EXIT_UNHEALTHY = 1; + static final int EXIT_ERROR = 2; + + private static final ObjectMapper JSON_MAPPER = + new ObjectMapper().enable(SerializationFeature.INDENT_OUTPUT); + + private final Config cfg; + private final TypedProperties props; + private final HoodieTableMetaClient metaClient; + + public HoodieTableHealthChecker(Configuration hadoopConf, Config cfg) { + this.cfg = cfg; + this.props = UtilHelpers.buildProperties(hadoopConf, cfg.propsFilePath, cfg.configs); + this.metaClient = HoodieTableMetaClient.builder() + .setConf(HadoopFSUtils.getStorageConfWithCopy(hadoopConf)) + .setBasePath(cfg.basePath) + .setLoadActiveTimelineOnLoad(true) + .build(); + } + + /** All checks this tool knows about, in report order. */ + static List<HealthCheck> allChecks() { + return Arrays.asList( + new CompactionCadenceCheck(), + new CleanerCadenceCheck(), + new SavepointArchivalBlockCheck(), + new ArchivalCadenceCheck()); + } + + /** + * Runs the selected checks and renders the report. + * + * @return the exit code the process should terminate with. + */ + public int run() { + List<HealthCheck> selected = selectChecks(cfg.checks); + HealthCheckContext context = new HealthCheckContext(metaClient, props, cfg.applyAllDefaults); + + List<HealthCheckResult> results = new ArrayList<>(); + List<String> erroredChecks = new ArrayList<>(); + + for (HealthCheck check : selected) { + try { + results.add(check.check(context)); + } catch (Exception e) { + // One check failing to reach a verdict must not hide the others. Record it and continue. + log.error("Health check '{}' failed to run for table {}", check.getName(), cfg.basePath, e); + erroredChecks.add(check.getName() + ": " + e.getMessage()); + } + } + + if (OutputFormat.JSON == cfg.output) { + System.out.println(renderJson(results, erroredChecks)); + } else { + System.out.println(renderTable(results, erroredChecks)); + } + + if (!erroredChecks.isEmpty()) { + return EXIT_ERROR; + } + return results.stream().anyMatch(r -> r.getStatus().isUnhealthy()) ? EXIT_UNHEALTHY : EXIT_HEALTHY; Review Comment: so can an errored check outrank UNHEALTHY? i think we should keep the error, but let UNHEALTHY win the exit code, or give "unhealthy + errors" its own code. ########## hudi-utilities/src/main/java/org/apache/hudi/utilities/health/CleanerCadenceCheck.java: ########## @@ -0,0 +1,151 @@ +/* + * 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.hudi.utilities.health; + +import org.apache.hudi.avro.model.HoodieCleanMetadata; +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.util.CleanerUtils; +import org.apache.hudi.common.util.Option; +import org.apache.hudi.config.HoodieCleanConfig; + +/** + * Checks whether the cleaner is keeping pace with ingestion. + * + * <p>Cleaning is what releases storage: every commit leaves behind older file versions that stay on + * disk until a clean removes them. When the cleaner stops, the table keeps working and nothing + * errors -- storage simply grows, and with it the cost of every listing. Like the other cadence + * checks this one asks not "has a clean run recently" but whether write commits have accumulated + * well past the point where the configured trigger should have scheduled one. + * + * <p>Two allowances are made before calling the table unhealthy. The configured trigger is + * multiplied by a slack factor, since async cleaning trails ingestion by design. And a floor is + * applied on top, because the trigger defaults to one commit: without a floor, a cleaner that is + * two commits behind -- routine when cleaning is asynchronous or briefly loses a lock to a + * concurrent writer -- would be reported as a failure. + */ +public class CleanerCadenceCheck implements HealthCheck { + + static final String NAME = "cleaner"; + + /** Multiplier on the configured trigger, absorbing async cleaning running on its own cadence. */ + static final double TRIGGER_SLACK_FACTOR = 2.0; + + /** + * Minimum number of commits since the last clean before the table can be called unhealthy, + * regardless of how low the trigger is set. Keeps a trigger of one commit from turning ordinary + * async lag into an alert. + */ + static final int MIN_UNHEALTHY_COMMITS = 5; + + @Override + public String getName() { + return NAME; + } + + @Override + public String getDescription() { + return "Whether the cleaner is keeping pace with write commits"; + } + + @Override + public HealthCheckResult check(HealthCheckContext context) throws Exception { + Option<String> missing = context.firstMissingConfig(HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key()); + if (missing.isPresent()) { + return context.skipForMissingConfig(NAME, missing.get()); + } + + int triggerCommits = Integer.parseInt(context.getProps().getString( + HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key(), + HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.defaultValue())); + int threshold = Math.max((int) Math.ceil(TRIGGER_SLACK_FACTOR * triggerCommits), MIN_UNHEALTHY_COMMITS); + + HoodieActiveTimeline activeTimeline = context.getMetaClient().getActiveTimeline(); + HoodieTimeline completedWrites = activeTimeline.getCommitsTimeline().filterCompletedInstants(); + Option<HoodieInstant> lastClean = activeTimeline.getCleanerTimeline().filterCompletedInstants().lastInstant(); + + HealthCheckResult.Builder result = HealthCheckResult.builder(NAME) + .config(HoodieCleanConfig.CLEAN_TRIGGER_MAX_COMMITS.key(), triggerCommits) + .config("effective.unhealthy.threshold.commits", threshold) + .config("observed.completed.write.commits", completedWrites.countInstants()); + echoIfSupplied(context, result, HoodieCleanConfig.CLEANER_POLICY.key()); + echoIfSupplied(context, result, HoodieCleanConfig.CLEANER_COMMITS_RETAINED.key()); + + int commitsSinceClean; + if (lastClean.isPresent()) { Review Comment: i think bot is flagging correctly.. i wrote something similar false positive for archival here https://github.com/apache/hudi/pull/20151/changes#r4163867874 i think cleaner can give a false UNHEALTHY on new tables and on insert-only or bulk_insert tables. -- 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]
