dongjoon-hyun commented on code in PR #826: URL: https://github.com/apache/spark-kubernetes-operator/pull/826#discussion_r4012440728
########## spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.spark.k8s.operator.utils; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_EVENTS_ENABLED; + +import java.util.function.Supplier; + +import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder; +import lombok.extern.slf4j.Slf4j; + +/** Utility class for publishing Kubernetes events about Spark resources. */ +@Slf4j +public final class EventUtils { + + /** Reason for an event describing an unhandled error thrown out of a reconciliation. */ + public static final String REASON_RECONCILE_ERROR = "ReconcileError"; + + /** Reason for an event describing a failure to persist the resource status. */ + public static final String REASON_STATUS_UPDATE_FAILED = "StatusUpdateFailed"; + + /** Reason for an event describing an unhandled error thrown out of a cleanup. */ + public static final String REASON_CLEANUP_ERROR = "CleanupError"; + + /** Maximum number of characters kept from an event message. */ + static final int MAX_MESSAGE_LENGTH = 1024; + + /** Maximum number of links followed when looking for the innermost cause of a failure. */ + private static final int MAX_CAUSE_DEPTH = 10; + + private EventUtils() {} + + /** + * Publishes a warning event about a Spark resource, if event publishing is enabled. + * + * <p>Publishing is best effort. Callers are never failed by an observability concern. + * + * @param recorderSupplier Supplies the event recorder bound to the resource. + * @param reason A short CamelCase reason, as expected by Kubernetes. + * @param message The message, truncated if it exceeds {@value #MAX_MESSAGE_LENGTH} characters. + */ + public static void warn( + Supplier<ResourceEventRecorder> recorderSupplier, String reason, String message) { + if (!KUBERNETES_EVENTS_ENABLED.getValue()) { + return; + } + try { + recorderSupplier.get().warn(reason, truncate(message)); Review Comment: `ResourceEventRecorder.warn(reason, message)` builds an `EventRecord` with no `key`, and `DefaultEventRecorder.eventName` digests `key().orElseGet(record::message)` into the Event's `metadata.name`. So the full message is the dedup identity: any per-attempt variation (fabric8 `KubernetesClientException` messages embed the request URL and `Status.toString()` including `retryAfterSeconds`) creates a brand-new Event object per attempt instead of bumping `count` on one. Suggest building the record explicitly with a stable key so repeats collapse into count increments: ```java recorderSupplier.get().record( EventRecord.builder() .type(EventType.WARNING) .reason(reason) .message(truncate(message)) .key(reason) .build()); ``` ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconciler.java: ########## @@ -135,6 +136,10 @@ public ErrorStatusUpdateControl<SparkApplication> updateErrorStatus( retryInfo.isLastAttempt()); } }); + EventUtils.warn( Review Comment: JOSDK calls `updateErrorStatus` on **every** retry attempt (`API_RETRY_MAX_ATTEMPTS` defaults to 15, 5s x1.5 uncapped backoff, so a ~48+ min episode), and `DefaultEventSink.emit` is a synchronous GET + create/patch on the reconciler thread. This emits 15 times per failing resource, against an API server that is often the cause of the failure. `retryInfo.isLastAttempt()` / `getAttemptCount()` is already in hand three lines above. Could we gate this inside the existing `ifPresent` block (e.g. first or last attempt only)? Same applies to `SparkClusterReconciler.updateErrorStatus`. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.spark.k8s.operator.utils; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_EVENTS_ENABLED; + +import java.util.function.Supplier; + +import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder; +import lombok.extern.slf4j.Slf4j; + +/** Utility class for publishing Kubernetes events about Spark resources. */ +@Slf4j +public final class EventUtils { + + /** Reason for an event describing an unhandled error thrown out of a reconciliation. */ + public static final String REASON_RECONCILE_ERROR = "ReconcileError"; + + /** Reason for an event describing a failure to persist the resource status. */ + public static final String REASON_STATUS_UPDATE_FAILED = "StatusUpdateFailed"; + + /** Reason for an event describing an unhandled error thrown out of a cleanup. */ + public static final String REASON_CLEANUP_ERROR = "CleanupError"; + + /** Maximum number of characters kept from an event message. */ + static final int MAX_MESSAGE_LENGTH = 1024; + + /** Maximum number of links followed when looking for the innermost cause of a failure. */ + private static final int MAX_CAUSE_DEPTH = 10; + + private EventUtils() {} + + /** + * Publishes a warning event about a Spark resource, if event publishing is enabled. + * + * <p>Publishing is best effort. Callers are never failed by an observability concern. + * + * @param recorderSupplier Supplies the event recorder bound to the resource. + * @param reason A short CamelCase reason, as expected by Kubernetes. + * @param message The message, truncated if it exceeds {@value #MAX_MESSAGE_LENGTH} characters. + */ + public static void warn( + Supplier<ResourceEventRecorder> recorderSupplier, String reason, String message) { + if (!KUBERNETES_EVENTS_ENABLED.getValue()) { Review Comment: Two things sit outside the best-effort guard: 1. Callers pass `"... " + EventUtils.describe(e)`, so the cause walk and string build run before this flag check on every failure, even with the default `enabled=false`. The `Supplier` defers the cheap part (the recorder) but not the expensive part (the message). 2. `KUBERNETES_EVENTS_ENABLED` is `Boolean.class` (not primitive), so `ConfigOption.resolveValue` goes through Jackson; `readValue("null", Boolean.class)` returns `null`, which is not caught, and `!getValue()` then NPEs. Because this line is outside the `try`, in the cleanup `catch (RuntimeException e) { warn(...); throw e; }` that NPE would *replace* the original cleanup failure, and in `persistStatus` it escapes the `return false` contract. Low probability, but easy to close: take the message as `Supplier<String>` (or check the flag at the call sites) and move the flag read inside the `try`. ########## spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/EventUtilsTest.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.spark.k8s.operator.utils; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_EVENTS_ENABLED; +import static org.apache.spark.k8s.operator.utils.TestUtils.setConfigKey; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; + +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class EventUtilsTest { + + private final ResourceEventRecorder recorder = mock(ResourceEventRecorder.class); + + private final AtomicInteger supplierCalls = new AtomicInteger(); + + private final Supplier<ResourceEventRecorder> recorderSupplier = + () -> { + supplierCalls.incrementAndGet(); + return recorder; + }; + + @AfterEach + void restoreEventsDisabled() { + // The option defaults to false. setConfigKey mutates the shared ConfigOption, so every test + // that enables events has to put it back or it leaks into the rest of the JVM. + setConfigKey(KUBERNETES_EVENTS_ENABLED, false); + } + + @Test + void warnDoesNothingWhenDisabled() { + // Left at the default of false, so nothing should be published. + EventUtils.warn(recorderSupplier, EventUtils.REASON_RECONCILE_ERROR, "driver pod rejected"); + + verifyNoInteractions(recorder); + // The recorder is resolved lazily, so a disabled operator never even asks for one. + assertThat(supplierCalls).hasValue(0); + } + + @Test + void warnSwallowsFailureFromRecorder() { Review Comment: Coverage gaps worth closing: - No test here verifies a publish on the success path; this one only asserts "no exception", so it passes even if the `recorder.warn(...)` call is removed. A `verify(recorder).warn(eq(REASON_RECONCILE_ERROR), eq("boom"))` would pin it. - `truncate()` has no test (and note it returns 1027 chars, `substring(0, 1024) + "..."`, while the javadoc says 1024). - `SparkClusterReconciler` received the same four-site change with no new tests, and `StatusRecorder.persistStatus`'s new site is never exercised (`StatusRecorderTest` uses `mock(BaseContext.class)`). - The new `SparkAppReconcilerTest` assertions use `contains("Reconciliation failed.")` / `contains("cannot finish deleting")`, which both the App and Cluster strings satisfy, so a kind copy-paste swap between the two reconcilers would go unnoticed. Asserting the full `"Spark App ..."` prefix would catch it. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/StatusRecorder.java: ########## @@ -150,6 +150,11 @@ public boolean persistStatus(BaseContext<CR> context, STATUS newStatus) { return true; } catch (KubernetesClientException e) { log.error("Error while persisting status to {}", newStatus, e); + EventUtils.warn( Review Comment: This catch is reached only after `API_STATUS_PATCH_MAX_ATTEMPTS` status patches already failed, i.e. the API server is rejecting or unreachable. The new call then issues two more blocking requests (GET + create/patch) to the same server on the reconciler thread, and `AppReconcileStep.attemptStatusUpdate` immediately requeues on `false` (bounded only by the per-resource rate limiter, ~5 loops / 15s). Cluster-side callers discard the boolean entirely. Worth skipping the emit when the failure is transport-level (`e.getCode() == 0` / `ReconcilerUtils.isTransientError`) or otherwise bounding it, so the observability path doesn't add load exactly when the control plane is degraded. ########## spark-operator/src/test/java/org/apache/spark/k8s/operator/reconciler/SparkAppReconcilerTest.java: ########## @@ -97,6 +108,11 @@ void beforeEach() { .appendNewStateAndPersist(any(SparkAppContext.class), any(ApplicationState.class)); } + @AfterEach Review Comment: nit: the PR flips the same flag through two different layers: `SparkOperatorConfManager.INSTANCE.refresh(Map.of("spark.kubernetes.operator.events.enabled", ...))` here vs. `TestUtils.setConfigKey` (reflection on `defaultValue`) in `EventUtilsTest`. Since `ConfigOption.resolveValue` consults the override layer first, `setConfigKey` is shadowed whenever an override is live, and `EventUtilsTest`'s `@AfterEach` can't restore it. Suggest one idiom in both classes (the `refresh(Map.of())` reset is the prevailing one in this repo), referencing the key via `KUBERNETES_EVENTS_ENABLED.getKey()` rather than a string literal. `updateErrorStatusPublishesNoEventWhenDisabled` also duplicates `EventUtilsTest.warnDoesNothingWhenDisabled` and could be dropped. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/EventUtils.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.spark.k8s.operator.utils; + +import static org.apache.spark.k8s.operator.config.SparkOperatorConf.KUBERNETES_EVENTS_ENABLED; + +import java.util.function.Supplier; + +import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder; +import lombok.extern.slf4j.Slf4j; + +/** Utility class for publishing Kubernetes events about Spark resources. */ +@Slf4j +public final class EventUtils { Review Comment: nit: this whole file is 4-space indented; the codebase (and `CLAUDE.md`, "Java with 2-space indentation") uses 2 spaces. Same for the new `@AfterEach` and three test methods in `SparkAppReconcilerTest`, and the `EventUtils.warn(` call sites use a 14-space hanging indent unlike the surrounding `+4` continuations. Checkstyle has no `Indentation` module and Spotless runs no formatter, so CI won't flag it. Please reformat the added code (google-java-format style). Related: the `KUBERNETES_EVENTS_ENABLED` description in `SparkOperatorConf` mixes trailing `" +` and leading `+ "` fragments that split phrases (`"into the" + " namespace "`); the neighboring options use one `+ "..."` fragment per line. ########## spark-operator/src/main/java/org/apache/spark/k8s/operator/context/BaseContext.java: ########## @@ -42,4 +43,12 @@ public abstract class BaseContext<CR extends BaseResource<?, ?, ?, ?, ?>> { * @return The Kubernetes client. */ public abstract KubernetesClient getClient(); + + /** + * Returns the event recorder bound to the resource associated with this context. Recording an + * event is best effort: the recorder logs and swallows write failures. + * + * @return The event recorder. + */ + public abstract ResourceEventRecorder getEventRecorder(); Review Comment: This abstract method plus the two identical overrides serve exactly one call site (`StatusRecorder.persistStatus`), while all four reconciler sites bypass it with `context::eventRecorder` on the JOSDK `Context`. That leaves two call conventions for the same value, glued together by the `Supplier` in `EventUtils`. If we keep the util approach, two overloads `warn(Context<?>, ...)` / `warn(BaseContext<?>, ...)` that check the flag and then call `eventRecorder()` inline would drop the `Supplier` and this abstract method (or make it concrete over a shared `josdkContext` field instead of duplicating in both subclasses). Also in `EventUtils`: `DefaultEventRecorder.record` already catches and logs emit failures, so the `catch (RuntimeException)` in `warn` only guards the supplier call; `rootCauseOf`'s `next.equals(rootCause)` is unreachable because `Throwable.getCause()` returns `null` when `cause == this`, and the depth bound alone already guarantees termination; and `truncate`'s null branch is unreachable since every caller concatenates a literal. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
