peter-toth commented on code in PR #861:
URL:
https://github.com/apache/spark-kubernetes-operator/pull/861#discussion_r4069330734
##########
docs/configuration.md:
##########
@@ -105,6 +107,45 @@ Since `,` is the separator, an expression cannot contain
it, e.g. `{1,3}`. An in
is logged and matches only the `reason` identical to it. Note that `*` is not
a wildcard but an
invalid expression, so use `.*` to exclude all reasons.
+### Event frequency
+
+Independently of that, the operator publishes an event with the same `reason`
**and the same
+`message`** on the same resource at most once every
+`spark.kubernetes.operator.events.minIntervalSeconds` (5 minutes by default,
dynamically
+overridable). The two options differ in kind: `excludedReasons` blocks a
`reason` permanently,
+while this one only limits how often an unchanged event is repeated. Set it to
`0` to publish
+every event.
+
+```properties
+spark.kubernetes.operator.events.minIntervalSeconds=300
+```
+
+The first event of a `reason` on a resource is always published right away, so
a rare event such
+as a failure is never delayed. Only a repeat that says exactly what the last
one said is dropped:
+such a repeat merely bumps the `count` of the one `Event` object, yet still
costs a read and a
+write on the API server. A repeat whose `message` differs is always published,
because the
+operator rewrites the `message` of the existing `Event`, so the current cause
of a failure and the
+`SuspendHeld` message that retracts a stale `KueueAdmissionPending` keep
reaching the user. The
+resource is identified by its `metadata.uid`, so a resource that reuses the
name of a deleted one
+starts over.
+
+An event is only published when a reconciliation emits it, so the effective
period is this
+interval **rounded up to the next repeat**, not the interval itself. With the
defaults,
+`KueueAdmissionPending` is emitted every 120 seconds
+(`spark.kubernetes.operator.reconciler.intervalSeconds`), so a 300 second
interval publishes it
+every 360 seconds, and `SuspendHeld` is emitted every 1800 seconds
+(`spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds`), so
it is unaffected.
+
+Keep the interval **below the interval at which the repeats themselves are
emitted**, i.e. below
Review Comment:
**Finding 1.** "below both of the two options named above" means below
`spark.kubernetes.operator.reconciler.intervalSeconds`, which is 120. The
default of this option is 300, so the rule declares the shipped default a
misconfiguration, and five lines earlier the same section says that default is
fine. A user who follows the rule sets the interval under 120, at which point
every repeat is published and the pacing stops applying to
`KueueAdmissionPending` — the only event the change affects at defaults, per
the PR description.
The example given is sound, but it only supports the rule for `SuspendHeld`.
I measured all four cases on the timed recorder, recording a repeat every `R`
seconds for one hour and listing the publishes:
```
I=300 R=120 [0, 360, 720, 1080, 1440, 1800, 2160, 2520, 2880, 3240, 3600]
I=2400 R=1800 [0, 3600]
I=1800 R=1800 [0, 1800, 3600]
I=1801 R=1800 [0, 3600]
```
So the hazard is not `I` versus `R`, it is the effective period
`ceil(I/R)*R` reaching the TTL. For `SuspendHeld` at `R=1800` that does bind at
`I <= 1800`, exactly 15x looser than "below 120", and the last two rows show
the boundary is precisely there. For `KueueAdmissionPending` at `R=120` it
binds at `I <= 3480`.
`ceil(I/R)*R < I + R`, so one rule covers both without the reader doing
arithmetic:
```
Keep `minIntervalSeconds` plus the interval at which the repeats are emitted
within the
`--event-ttl` of the API server (one hour by default), i.e. below
`3600 -
spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds` for
`SuspendHeld`
and below `3600 - spark.kubernetes.operator.reconciler.intervalSeconds` for
`KueueAdmissionPending`. With the defaults that is 1800 seconds, so the
default 300 leaves plenty
of room. At `minIntervalSeconds=2400` the `SuspendHeld` repeat at 1800
seconds is dropped and the
next one lands at 3600 seconds, exactly the TTL, so the event expires in
between and the hold
stops being visible.
```
The same rule is stated in two more places and needs the same correction:
`SparkOperatorConf.java:373-374` in the option description, and
`docs/config_properties.md` which is generated from it. The reciprocal clause
added to `SUSPEND_HOLD_REQUEUE_INTERVAL_SECONDS` ("keep `minIntervalSeconds`
below this interval") is fine as it stands, since `I <= R` is what binds for
that reason.
##########
tests/e2e/helm/events-config-values.yaml:
##########
@@ -19,3 +19,6 @@ operatorConfiguration:
# The suspend-events suite asserts that a held resource republishes its
SuspendHeld event.
# The default is 1800 seconds, which no assert can wait for.
spark.kubernetes.operator.reconciler.suspendHoldRequeueIntervalSeconds=10
+ # The repeat above carries the same message, so the default 300 second
minimum interval would
+ # drop every one of them and the count would never rise within the assert
timeout.
+ spark.kubernetes.operator.events.minIntervalSeconds=0
Review Comment:
**Finding 2.** `0` takes the `minInterval <= 0` early return, so
`suspend-events` — the only suite that installs the operator with events
enabled — exercises none of the new code. The map, the `compute`, the sweep and
the interval comparison are all skipped, and the unit tests are the sole
coverage.
A small positive value gets the coverage for free, because the pacing is a
no-op whenever the interval is below the emit interval. With
`suspendHoldRequeueIntervalSeconds=10` already set two lines up, I ran
`minIntervalSeconds=5` for 120 seconds of repeats:
```
PROBE I=5 R=10 publishes at >>> [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100,
110, 120]
PROBE tracked entries >>> 1
```
Every repeat still publishes, so the `count` rises exactly as it does at `0`
and the existing assert is unaffected, while the positive-interval path now
runs on a real cluster with a real `uid`. That last part is the bit unit tests
cannot cover: `uidOf` reads
`context.getPrimaryResource().getMetadata().getUid()`, and every unit test
hands it a hand-built `SparkApplication` rather than one that came back from
the API server.
```suggestion
# A small positive value keeps every repeat published, since the pacing
is a no-op below the
# requeue interval above, while still exercising the minimum-interval
path on a real resource.
spark.kubernetes.operator.events.minIntervalSeconds=5
```
##########
spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/ConfigurableEventRecorder.java:
##########
@@ -115,6 +166,107 @@ private static boolean isExcluded(String reason) {
.anyMatch(regex -> matches(regex, reason));
}
+ /**
+ * Returns whether the given event was already recorded on the given
resource less than {@link
+ *
org.apache.spark.k8s.operator.config.SparkOperatorConf#KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS}
+ * ago. The first record of a {@link RecordedEvent} always returns false, so
a rare event such as
+ * a failure is never delayed. The uid rather than the name identifies the
resource, since a name
+ * can be reused by a later resource.
+ *
+ * <p>The message is part of the identity because {@link DefaultEventSink}
rewrites the message
+ * and the timestamp of the existing Event on every repeat, so only a repeat
that says exactly
+ * what the last one said is free of new information. A reason whose message
varies between
+ * repeats, such as one embedding the cause of a failure, keeps reaching the
user.
+ *
+ * @param event The event to record.
+ * @param uid The uid of the resource the event is about, null when it
cannot be determined.
+ * @return Whether the event is a repeat that should be dropped.
+ */
+ private boolean withinMinInterval(EventRecord event, String uid) {
+ long minInterval = minIntervalNanos();
+ if (minInterval <= 0L) {
+ // Limiting is off, so the state is dropped rather than left behind with
nothing to sweep it.
+ discardIntervalState();
+ return false;
+ }
+ if (uid == null) {
+ // The resource cannot be identified, so the event is published rather
than attributed to
+ // the wrong resource. Every event the operator publishes is about a
primary resource.
+ return false;
+ }
+ long now = nanoTime.getAsLong();
+ sweep(now, minInterval);
+ AtomicBoolean admitted = new AtomicBoolean();
+ // compute is atomic per key, so concurrent reconcile threads cannot both
admit the same event.
+ lastRecorded.compute(
+ new RecordedEvent(uid, event.reason(), event.message()),
+ (key, previous) -> {
+ if (previous == null || now - previous >= minInterval) {
+ admitted.set(true);
+ return now;
+ }
+ return previous;
+ });
+ return !admitted.get();
+ }
+
+ /**
+ * Drops the interval state. Guarded on {@link Map#isEmpty()} because this
runs on every event
+ * while limiting is off, while {@link ConcurrentHashMap#clear()} walks the
whole table, which the
+ * map never shrinks: clearing an already empty map that once held many
entries is not free.
+ */
+ private void discardIntervalState() {
+ if (!lastRecorded.isEmpty()) {
+ lastRecorded.clear();
+ }
+ }
+
+ /**
+ * Drops the entries older than the minimum interval, at most once per
interval. An entry that old
+ * admits the next event anyway, so removing it changes no decision, which
is why the map can be
+ * kept to the events recorded within the last interval instead of growing
with every resource the
+ * operator has ever seen. Sweeping inline avoids a background thread for a
map that is only ever
+ * touched while an event is published.
+ *
+ * <p>The due time is the elapsed time since the last sweep rather than a
deadline computed from
+ * the interval in force when that sweep ran. A deadline would keep a
lowered interval waiting
+ * out the old one, and would overflow for an interval large enough to
saturate {@link
+ * TimeUnit#toNanos}, leaving the map unswept for the life of the process.
+ *
+ * @param now The current time, in nanoseconds.
+ * @param minInterval The minimum interval between two events, in
nanoseconds.
+ */
+ private void sweep(long now, long minInterval) {
+ long previous = lastSweep.get();
+ // Only the thread that wins the compareAndSet sweeps, the others carry on
recording.
+ if (now - previous >= minInterval && lastSweep.compareAndSet(previous,
now)) {
+ lastRecorded.entrySet().removeIf(entry -> now - entry.getValue() >=
minInterval);
+ }
+ }
+
+ /**
+ * Returns how many events are currently tracked for the minimum interval.
The sweep has no
+ * effect on which events are published, so only this makes it observable to
a test.
+ *
+ * @return The number of tracked events.
+ */
+ int trackedEventCount() {
+ return lastRecorded.size();
+ }
+
+ private static String uidOf(Context<?> context) {
+ HasMetadata resource = context.getPrimaryResource();
+ ObjectMeta metadata = resource == null ? null : resource.getMetadata();
+ return metadata == null ? null : metadata.getUid();
+ }
+
+ private static long minIntervalNanos() {
+ Long seconds = KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS.getValue();
+ // An unparseable override falls back to the default, but an override of
the literal 'null'
+ // resolves to null, which is treated as no limit rather than throwing out
of a reconciliation.
+ return seconds == null ? 0L : TimeUnit.SECONDS.toNanos(seconds);
Review Comment:
**Finding 3.** The comment is exactly right, which is why the branch
deserves the same test `KUBERNETES_EVENTS_ENABLED` got. I confirmed both halves
of it:
```
PROBE-NULL-RESOLVES-TO >>> null
PROBE-GARBAGE-RESOLVES-TO >>> 300
PROBE-NULL-PUBLISHES-BOTH >>> confirmed
```
`ConfigOption.resolveValue` sends a non-primitive type through
`objectMapper.readValue`, so `"null"` deserialises to `null` while
`"not-a-number"` throws `JsonProcessingException` and falls back to the default.
What makes it worth pinning is that the two options fail in *opposite*
directions on the same input: `eventsEnabled()` uses `Boolean.TRUE.equals`, so
a `"null"` override stops every event, while here it removes the limit
entirely. Both are the right choice for their option, and only one of them is
currently a test:
```java
@Test
void treatsAMalformedMinIntervalOverrideAsNoLimit() {
// The option is a boxed Long, so a malformed override can resolve to
null. Unlike the enabled
// flag, which then drops every event, a limit that cannot be read must
not drop any.
SparkOperatorConfManager.INSTANCE.refresh(
Map.of(
KUBERNETES_EVENTS_ENABLED.getKey(), "true",
KUBERNETES_EVENTS_MIN_INTERVAL_SECONDS.getKey(), "null"));
Context<?> context = contextOf("uid-1");
EventRecord first = EventRecord.normal("KueueAdmissionPending",
"queued");
EventRecord second = EventRecord.normal("KueueAdmissionPending",
"queued");
timedRecorder.record(first, context);
timedRecorder.record(second, context);
verify(delegate).record(first, context);
verify(delegate).record(second, context);
}
```
--
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]
