jerryshao commented on code in PR #13700: URL: https://github.com/apache/gravitino/pull/13700#discussion_r4267653305
########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java: ########## @@ -0,0 +1,165 @@ +/* + * 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.gravitino.job.k8s; + +import com.google.common.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Caches the resources listed in a namespace for a short time, so that querying the status of many + * jobs in the same namespace costs one LIST call instead of one GET call per job. + * + * <p>A resource missing from the cache isn't necessarily gone: it may have been created after the + * namespace was listed, or the namespace can't be listed at all. The caller must fall back to a GET + * call on a cache miss. + * + * @param <T> the resource type + */ +final class NamespacedResourceCache<T extends HasMetadata> { + + private static final class Snapshot<T> { + + private final long listedAtMs; + + private final Map<String, T> resources; + + private Snapshot(long listedAtMs, Map<String, T> resources) { + this.listedAtMs = listedAtMs; + this.resources = resources; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(NamespacedResourceCache.class); + + private final long ttlMs; + + private final Function<String, List<T>> lister; + + private final Predicate<RuntimeException> isFatal; + + private final LongSupplier clock; + + private final Map<String, Snapshot<T>> snapshots = new ConcurrentHashMap<>(); + + // Listing a namespace is a network call, so it is guarded by a lock of its own rather than by + // the snapshot map, which must never block. + private final Map<String, Object> listLocks = new ConcurrentHashMap<>(); + + // Bumped by every invalidation, so that a listing started before it isn't cached. + private final AtomicLong invalidations = new AtomicLong(); + Review Comment: Fixed. a8b4730d2 first made the invalidation per namespace, and 00998d7c5 removes the counters altogether: each namespace has one listing in the cache, being made or made, and an invalidation removes it, so a listing in flight is not cached and other namespaces are not affected. `testInvalidationOfAnotherNamespace` and `testInvalidatedWhileListing` cover both sides. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java: ########## @@ -0,0 +1,165 @@ +/* + * 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.gravitino.job.k8s; + +import com.google.common.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Caches the resources listed in a namespace for a short time, so that querying the status of many + * jobs in the same namespace costs one LIST call instead of one GET call per job. + * + * <p>A resource missing from the cache isn't necessarily gone: it may have been created after the + * namespace was listed, or the namespace can't be listed at all. The caller must fall back to a GET + * call on a cache miss. + * + * @param <T> the resource type + */ +final class NamespacedResourceCache<T extends HasMetadata> { + + private static final class Snapshot<T> { + + private final long listedAtMs; + + private final Map<String, T> resources; + + private Snapshot(long listedAtMs, Map<String, T> resources) { + this.listedAtMs = listedAtMs; + this.resources = resources; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(NamespacedResourceCache.class); + + private final long ttlMs; + + private final Function<String, List<T>> lister; + + private final Predicate<RuntimeException> isFatal; + + private final LongSupplier clock; + + private final Map<String, Snapshot<T>> snapshots = new ConcurrentHashMap<>(); + + // Listing a namespace is a network call, so it is guarded by a lock of its own rather than by + // the snapshot map, which must never block. + private final Map<String, Object> listLocks = new ConcurrentHashMap<>(); + + // Bumped by every invalidation, so that a listing started before it isn't cached. + private final AtomicLong invalidations = new AtomicLong(); + + /** + * Creates a cache. + * + * @param ttlMs how long a namespace listing is cached, caching is disabled if not positive + * @param lister lists the resources of a namespace + * @param isFatal tells the failures of the lister that getting a resource by itself would run + * into as well, for example an API server that can't be reached. They are thrown to the + * caller of {@link #find(String, String)} instead of being cached as an empty listing. + * @param clock the current time in milliseconds + */ + NamespacedResourceCache( + long ttlMs, + Function<String, List<T>> lister, + Predicate<RuntimeException> isFatal, + LongSupplier clock) { + this.ttlMs = ttlMs; + this.lister = lister; + this.isFatal = isFatal; + this.clock = clock; + } + + /** + * Finds a resource in the cached listing of its namespace, and lists the namespace again if the + * cached listing has expired. A namespace that can't be listed, for example because listing it is + * forbidden, isn't listed again until the cached failure expires. + * + * @param namespace the namespace of the resource + * @param name the name of the resource + * @return the resource, or empty if it isn't in the listing or the namespace can't be listed + * @throws RuntimeException the failure to list the namespace, if it is a fatal one + */ + Optional<T> find(String namespace, String name) { + if (ttlMs <= 0) { + return Optional.empty(); + } + + Snapshot<T> snapshot = snapshots.get(namespace); + if (isExpired(snapshot)) { + // Only one thread lists a namespace at a time, the others wait for its listing. + synchronized (listLocks.computeIfAbsent(namespace, ns -> new Object())) { + snapshot = snapshots.get(namespace); + if (isExpired(snapshot)) { + long invalidationsBefore = invalidations.get(); + snapshot = snapshotOf(namespace); + if (invalidations.get() == invalidationsBefore) { + snapshots.put(namespace, snapshot); + } Review Comment: Fixed. a8b4730d2 first made the invalidation per namespace, and 00998d7c5 removes the counters altogether: each namespace has one listing in the cache, being made or made, and an invalidation removes it, so a listing in flight is not cached and other namespaces are not affected. `testInvalidationOfAnotherNamespace` and `testInvalidatedWhileListing` cover both sides. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java: ########## @@ -0,0 +1,165 @@ +/* + * 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.gravitino.job.k8s; + +import com.google.common.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Caches the resources listed in a namespace for a short time, so that querying the status of many + * jobs in the same namespace costs one LIST call instead of one GET call per job. + * + * <p>A resource missing from the cache isn't necessarily gone: it may have been created after the + * namespace was listed, or the namespace can't be listed at all. The caller must fall back to a GET + * call on a cache miss. + * + * @param <T> the resource type + */ +final class NamespacedResourceCache<T extends HasMetadata> { + + private static final class Snapshot<T> { + + private final long listedAtMs; + + private final Map<String, T> resources; + + private Snapshot(long listedAtMs, Map<String, T> resources) { + this.listedAtMs = listedAtMs; + this.resources = resources; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(NamespacedResourceCache.class); + + private final long ttlMs; + + private final Function<String, List<T>> lister; + + private final Predicate<RuntimeException> isFatal; + + private final LongSupplier clock; + + private final Map<String, Snapshot<T>> snapshots = new ConcurrentHashMap<>(); + + // Listing a namespace is a network call, so it is guarded by a lock of its own rather than by + // the snapshot map, which must never block. + private final Map<String, Object> listLocks = new ConcurrentHashMap<>(); + + // Bumped by every invalidation, so that a listing started before it isn't cached. + private final AtomicLong invalidations = new AtomicLong(); + + /** + * Creates a cache. + * + * @param ttlMs how long a namespace listing is cached, caching is disabled if not positive + * @param lister lists the resources of a namespace + * @param isFatal tells the failures of the lister that getting a resource by itself would run + * into as well, for example an API server that can't be reached. They are thrown to the + * caller of {@link #find(String, String)} instead of being cached as an empty listing. + * @param clock the current time in milliseconds + */ + NamespacedResourceCache( + long ttlMs, + Function<String, List<T>> lister, + Predicate<RuntimeException> isFatal, + LongSupplier clock) { + this.ttlMs = ttlMs; + this.lister = lister; + this.isFatal = isFatal; + this.clock = clock; + } + + /** + * Finds a resource in the cached listing of its namespace, and lists the namespace again if the + * cached listing has expired. A namespace that can't be listed, for example because listing it is + * forbidden, isn't listed again until the cached failure expires. + * + * @param namespace the namespace of the resource + * @param name the name of the resource + * @return the resource, or empty if it isn't in the listing or the namespace can't be listed + * @throws RuntimeException the failure to list the namespace, if it is a fatal one + */ + Optional<T> find(String namespace, String name) { + if (ttlMs <= 0) { + return Optional.empty(); + } + + Snapshot<T> snapshot = snapshots.get(namespace); + if (isExpired(snapshot)) { + // Only one thread lists a namespace at a time, the others wait for its listing. + synchronized (listLocks.computeIfAbsent(namespace, ns -> new Object())) { + snapshot = snapshots.get(namespace); + if (isExpired(snapshot)) { + long invalidationsBefore = invalidations.get(); + snapshot = snapshotOf(namespace); + if (invalidations.get() == invalidationsBefore) { + snapshots.put(namespace, snapshot); + } + } + } + } + return Optional.ofNullable(snapshot.resources.get(name)); + } + + /** + * Drops the cached listing of a namespace, after a resource in it is changed. + * + * @param namespace the namespace + */ + void invalidate(String namespace) { + invalidations.incrementAndGet(); + snapshots.remove(namespace); + } Review Comment: Fixed. a8b4730d2 first made the invalidation per namespace, and 00998d7c5 removes the counters altogether: each namespace has one listing in the cache, being made or made, and an invalidation removes it, so a listing in flight is not cached and other namespaces are not affected. `testInvalidationOfAnotherNamespace` and `testInvalidatedWhileListing` cover both sides. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java: ########## @@ -0,0 +1,165 @@ +/* + * 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.gravitino.job.k8s; + +import com.google.common.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Caches the resources listed in a namespace for a short time, so that querying the status of many + * jobs in the same namespace costs one LIST call instead of one GET call per job. + * + * <p>A resource missing from the cache isn't necessarily gone: it may have been created after the + * namespace was listed, or the namespace can't be listed at all. The caller must fall back to a GET + * call on a cache miss. + * + * @param <T> the resource type + */ +final class NamespacedResourceCache<T extends HasMetadata> { + + private static final class Snapshot<T> { + + private final long listedAtMs; + + private final Map<String, T> resources; + + private Snapshot(long listedAtMs, Map<String, T> resources) { + this.listedAtMs = listedAtMs; + this.resources = resources; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(NamespacedResourceCache.class); + + private final long ttlMs; + + private final Function<String, List<T>> lister; + + private final Predicate<RuntimeException> isFatal; + + private final LongSupplier clock; + + private final Map<String, Snapshot<T>> snapshots = new ConcurrentHashMap<>(); + + // Listing a namespace is a network call, so it is guarded by a lock of its own rather than by + // the snapshot map, which must never block. + private final Map<String, Object> listLocks = new ConcurrentHashMap<>(); Review Comment: Fixed in 00998d7c5. The lock map is gone: the cache holds one `CompletableFuture` per namespace, so concurrent queries of a namespace wait for the same listing, and expired listings are dropped on the next listing. The map only holds the namespaces listed within the TTL or being listed. `testConcurrentFindsListOnce` checks that two concurrent queries of a namespace make one listing and that another namespace isn't held up by it. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/NamespacedResourceCache.java: ########## @@ -0,0 +1,165 @@ +/* + * 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.gravitino.job.k8s; + +import com.google.common.collect.ImmutableMap; +import io.fabric8.kubernetes.api.model.HasMetadata; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; +import java.util.function.LongSupplier; +import java.util.function.Predicate; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Caches the resources listed in a namespace for a short time, so that querying the status of many + * jobs in the same namespace costs one LIST call instead of one GET call per job. + * + * <p>A resource missing from the cache isn't necessarily gone: it may have been created after the + * namespace was listed, or the namespace can't be listed at all. The caller must fall back to a GET + * call on a cache miss. + * + * @param <T> the resource type + */ +final class NamespacedResourceCache<T extends HasMetadata> { + + private static final class Snapshot<T> { + + private final long listedAtMs; + + private final Map<String, T> resources; + + private Snapshot(long listedAtMs, Map<String, T> resources) { + this.listedAtMs = listedAtMs; + this.resources = resources; + } + } + + private static final Logger LOG = LoggerFactory.getLogger(NamespacedResourceCache.class); + + private final long ttlMs; + + private final Function<String, List<T>> lister; + + private final Predicate<RuntimeException> isFatal; + + private final LongSupplier clock; + + private final Map<String, Snapshot<T>> snapshots = new ConcurrentHashMap<>(); + + // Listing a namespace is a network call, so it is guarded by a lock of its own rather than by + // the snapshot map, which must never block. + private final Map<String, Object> listLocks = new ConcurrentHashMap<>(); + + // Bumped by every invalidation, so that a listing started before it isn't cached. + private final AtomicLong invalidations = new AtomicLong(); + + /** + * Creates a cache. + * + * @param ttlMs how long a namespace listing is cached, caching is disabled if not positive + * @param lister lists the resources of a namespace + * @param isFatal tells the failures of the lister that getting a resource by itself would run + * into as well, for example an API server that can't be reached. They are thrown to the + * caller of {@link #find(String, String)} instead of being cached as an empty listing. + * @param clock the current time in milliseconds + */ + NamespacedResourceCache( + long ttlMs, + Function<String, List<T>> lister, + Predicate<RuntimeException> isFatal, + LongSupplier clock) { + this.ttlMs = ttlMs; + this.lister = lister; + this.isFatal = isFatal; + this.clock = clock; + } + + /** + * Finds a resource in the cached listing of its namespace, and lists the namespace again if the + * cached listing has expired. A namespace that can't be listed, for example because listing it is + * forbidden, isn't listed again until the cached failure expires. + * + * @param namespace the namespace of the resource + * @param name the name of the resource + * @return the resource, or empty if it isn't in the listing or the namespace can't be listed + * @throws RuntimeException the failure to list the namespace, if it is a fatal one + */ + Optional<T> find(String namespace, String name) { + if (ttlMs <= 0) { + return Optional.empty(); + } + + Snapshot<T> snapshot = snapshots.get(namespace); + if (isExpired(snapshot)) { + // Only one thread lists a namespace at a time, the others wait for its listing. + synchronized (listLocks.computeIfAbsent(namespace, ns -> new Object())) { Review Comment: Fixed in 00998d7c5. The lock map is gone: the cache holds one `CompletableFuture` per namespace, so concurrent queries of a namespace wait for the same listing, and expired listings are dropped on the next listing. The map only holds the namespaces listed within the TTL or being listed. `testConcurrentFindsListOnce` checks that two concurrent queries of a namespace make one listing and that another namespace isn't held up by it. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ ########## plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/PodLogUtils.java: ########## @@ -0,0 +1,117 @@ +/* + * 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.gravitino.job.k8s; + +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.List; + +/** Reads the output of a job from the log of its pod. */ +public final class PodLogUtils { + + private static final int READ_BUFFER_SIZE = 8192; + + // The largest array size that is safe to allocate on every JVM. + private static final int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8; + + private PodLogUtils() {} + + /** + * Reads the last lines of a log stream, within its last bytes. Only the last bytes are kept in + * memory while reading. If the kept bytes start in the middle of a line, that partial line is + * dropped, unless it is the only line. + * + * @param log the log stream, which is read to its end + * @param maxLines the maximum number of lines to return + * @param maxBytes the maximum number of bytes to read from the tail of the log + * @return the last lines of the log + * @throws IOException if the log can't be read + */ + public static List<String> readLastLines(InputStream log, int maxLines, int maxBytes) + throws IOException { + Preconditions.checkArgument(maxLines > 0, "maxLines must be positive"); + Preconditions.checkArgument(maxBytes > 0, "maxBytes must be positive"); + + // Keep one more byte than the window, to tell whether the window starts at a line boundary. + // The window is capped so that the buffer, of twice its size, fits in an array. + int keep = (int) Math.min(maxBytes + 1L, MAX_BUFFER_SIZE / 2); + int window = keep - 1; + int maxCapacity = Math.max(2 * keep, READ_BUFFER_SIZE); + // The buffer grows with the log, as most logs are much shorter than the window. + byte[] tail = new byte[READ_BUFFER_SIZE]; + int length = 0; + byte[] chunk = new byte[READ_BUFFER_SIZE]; + int read; + while ((read = log.read(chunk)) >= 0) { Review Comment: No change needed: `InputStream.read(byte[])` returns 0 only for an empty array ("If the length of b is zero, then no bytes are read and 0 is returned; otherwise, there is an attempt to read at least one byte" and it blocks until one is available), and the buffer here is never empty. So the loop always progresses or ends at -1. `readLastLines` was also fuzzed with random short reads in an earlier review. _🤖 Addressed by [Claude Code](https://claude.com/claude-code)_ -- 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]
