Copilot commented on code in PR #13700: URL: https://github.com/apache/gravitino/pull/13700#discussion_r4267389484
########## 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: `listLocks` grows monotonically because entries are never removed, so a system that sees many distinct namespaces over time can accumulate an unbounded number of lock objects. Consider replacing this with a bounded lock strategy (e.g., Guava `Striped<Lock>` / `Striped.lazyWeakLock(...)`) or removing the per-namespace lock entry after the listing completes (being careful to avoid races), so memory use doesn't scale with the number of namespaces ever observed. ########## 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: The `invalidations` counter is global, so invalidating *any* namespace can prevent caching of an in-flight listing for an unrelated namespace. Under concurrent load this can significantly reduce cache effectiveness and increase LIST traffic. A more targeted approach is to track invalidation generations per namespace (e.g., `ConcurrentHashMap<String, AtomicLong>`), or store a per-namespace generation alongside the snapshot and compare only that namespace’s generation before/after listing. ########## 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: `listLocks` grows monotonically because entries are never removed, so a system that sees many distinct namespaces over time can accumulate an unbounded number of lock objects. Consider replacing this with a bounded lock strategy (e.g., Guava `Striped<Lock>` / `Striped.lazyWeakLock(...)`) or removing the per-namespace lock entry after the listing completes (being careful to avoid races), so memory use doesn't scale with the number of namespaces ever observed. ########## 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: The `invalidations` counter is global, so invalidating *any* namespace can prevent caching of an in-flight listing for an unrelated namespace. Under concurrent load this can significantly reduce cache effectiveness and increase LIST traffic. A more targeted approach is to track invalidation generations per namespace (e.g., `ConcurrentHashMap<String, AtomicLong>`), or store a per-namespace generation alongside the snapshot and compare only that namespace’s generation before/after listing. ########## 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: The `invalidations` counter is global, so invalidating *any* namespace can prevent caching of an in-flight listing for an unrelated namespace. Under concurrent load this can significantly reduce cache effectiveness and increase LIST traffic. A more targeted approach is to track invalidation generations per namespace (e.g., `ConcurrentHashMap<String, AtomicLong>`), or store a per-namespace generation alongside the snapshot and compare only that namespace’s generation before/after listing. ########## 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: This loop can spin indefinitely if `log.read(byte[])` returns `0` (which is permitted by the `InputStream` contract for some implementations), because neither `length` nor the stream position changes. To avoid a potential busy loop, use a `> 0` condition and handle `0` reads explicitly (e.g., retry, or break depending on expected stream behavior). -- 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]
