Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import java.time.temporal.ChronoUnit;
import java.util.HexFormat;
import java.util.Objects;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -48,10 +49,11 @@
* can be overridden, see {@link
* io.javaoperatorsdk.operator.api.config.ConfigurationService#clusterScopedEventNamespace()}.
*
* <p>Events are named deterministically, after the object they are about plus a hash of everything
* that identifies the event, so that recording the same event again resolves to the event already
* recorded for it. Repeat occurrences are then counted on that event rather than recorded as copies
* of it, see {@link DefaultEventSink}.
* <p>By default, events are named deterministically, after the object they are about plus a hash of
* everything that identifies the event, so that recording the same event again resolves to the
* event already recorded for it. Repeat occurrences are then counted on that event rather than
* recorded as copies of it, see {@link DefaultEventSink}. How events aggregate, how they are named
* and whether they carry an owner reference can be configured, see {@link #builder(EventSink)}.
*/
public class DefaultEventRecorder implements EventRecorder {

Expand All @@ -73,10 +75,79 @@ public class DefaultEventRecorder implements EventRecorder {

private static final int IDENTITY_HASH_LENGTH = 32;

/** What the API server accepts as an object name, see RFC 1123 on DNS subdomains. */
private static final Pattern RFC_1123_SUBDOMAIN =
Pattern.compile("[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*");

private final EventSink sink;
private final EventNamingStrategy namingStrategy;
private final EventKeyStrategy keyStrategy;
private final boolean ownerReference;

public DefaultEventRecorder(EventSink sink) {
this(sink, EventNamingStrategy.none(), EventKeyStrategy.none(), false);
}

private DefaultEventRecorder(
EventSink sink,
EventNamingStrategy namingStrategy,
EventKeyStrategy keyStrategy,
boolean ownerReference) {
this.sink = sink;
this.namingStrategy = namingStrategy;
this.keyStrategy = keyStrategy;
this.ownerReference = ownerReference;
}

public static Builder builder(EventSink sink) {
return new Builder(sink);
}

/** Builder for {@link DefaultEventRecorder}. */
public static final class Builder {

private final EventSink sink;
private EventNamingStrategy namingStrategy = EventNamingStrategy.none();
private EventKeyStrategy keyStrategy = EventKeyStrategy.none();
private boolean ownerReference = false;

private Builder(EventSink sink) {
this.sink = Objects.requireNonNull(sink, "sink must not be null");
}

/** The strategy naming recorded events, see {@link EventNamingStrategy}. */
public Builder namingStrategy(EventNamingStrategy namingStrategy) {
this.namingStrategy =
Objects.requireNonNull(namingStrategy, "namingStrategy must not be null");
return this;
}

/**
* The strategy deriving the default aggregation key of records that do not set one, see {@link
* EventKeyStrategy}. The key is ignored for events whose name is set by the record or resolved
* by the naming strategy, see {@link EventNamingStrategy}.
*/
public Builder keyStrategy(EventKeyStrategy keyStrategy) {
this.keyStrategy = Objects.requireNonNull(keyStrategy, "keyStrategy must not be null");
return this;
}

/**
* When set, recorded events carry an {@code ownerReference} to the object they are about. The
* reference expresses ownership for tooling that reads it; note that the Kubernetes garbage
* collector ignores events, so it does not cause cascade deletion, events expire through the
* event TTL either way. Records can override this per event via {@link
* EventRecord.Builder#ownedByRegarding(boolean)}. The reference is only set when the object
* already has a uid.
*/
public Builder ownerReference(boolean ownerReference) {
this.ownerReference = ownerReference;
return this;
}

public DefaultEventRecorder build() {
return new DefaultEventRecorder(sink, namingStrategy, keyStrategy, ownerReference);
}
}

/**
Expand Down Expand Up @@ -111,14 +182,17 @@ private static String resolve() {
public void record(EventRecord event, Context<?> context) {
Objects.requireNonNull(context, "the context of the reconciliation must not be null");
Objects.requireNonNull(event, "event must not be null");
Event assembled = null;
try {
sink.emit(toEvent(context, event), context);
assembled = toEvent(context, event);
sink.emit(assembled, context);
} catch (Exception e) {
// recording an event must never break the caller: a controller that fails to reconcile
// because it could not write an event is strictly worse than one that records nothing
log.warn(
"Could not record {} event with reason {} for resource {} in namespace {}",
"Could not record {} event named {} with reason {} for resource {} in namespace {}",
event.type(),
assembled != null ? assembled.getMetadata().getName() : "unknown",
event.reason(),
context.getPrimaryResource().getMetadata().getName(),
context.getPrimaryResource().getMetadata().getNamespace(),
Expand Down Expand Up @@ -164,6 +238,23 @@ protected Event toEvent(Context<?> context, EventRecord record) {
.withNewSource()
.withComponent(record.reportingComponent().orElse(controllerName))
.endSource();
boolean ownedByRegarding = record.ownedByRegarding().orElse(ownerReference);
if (ownedByRegarding && regarding.getMetadata().getUid() == null) {
log.debug(
"Not setting the owner reference on the event about {}: the object has no uid yet",
regarding.getMetadata().getName());
}
if (ownedByRegarding && regarding.getMetadata().getUid() != null) {
builder
.editMetadata()
.addNewOwnerReference()
.withApiVersion(regarding.getApiVersion())
.withKind(regarding.getKind())
.withName(regarding.getMetadata().getName())
.withUid(regarding.getMetadata().getUid())
.endOwnerReference()
.endMetadata();
}
record.action().ifPresent(builder::withAction);
return builder.build();
}
Expand All @@ -181,17 +272,56 @@ private String eventNamespace(HasMetadata regarding, Context<?> context) {
CLUSTER_SCOPED_EVENT_NAMESPACE);
}

private String eventName(HasMetadata regarding, EventRecord record, String reportingController) {
return record
.name()
.filter(name -> !name.isBlank())
.or(() -> namingStrategy.nameFor(regarding, record).filter(name -> !name.isBlank()))
.map(DefaultEventRecorder::truncateToMaxNameLength)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
.filter(DefaultEventRecorder::isValidEventName)
.orElseGet(() -> identityHashName(regarding, record, reportingController));
}

private static boolean isValidEventName(String name) {
if (RFC_1123_SUBDOMAIN.matcher(name).matches()) {
return true;
}
log.warn(
"Falling back to the default event name: {} is not a valid RFC 1123 DNS subdomain", name);
return false;
}

private static String truncateToMaxNameLength(String name) {
if (name.length() <= MAX_NAME_LENGTH) {
return name;
}
var truncated = name.substring(0, MAX_NAME_LENGTH);
while (truncated.endsWith("-") || truncated.endsWith(".")) {
truncated = truncated.substring(0, truncated.length() - 1);
}
log.warn(
"Truncated the name of event {} to {} to stay within the Kubernetes name limit",
name,
truncated);
return truncated;
}

/**
* Names events {@code <object name>.<hash>}, following the convention of the Go client, hashing
* everything that makes two events the same event: the object, the type, the reason, the
* reporting component and, unless the record sets a {@link EventRecord#key()}, the message. The
* name is therefore stable across occurrences, which is what lets the sink recognise a repeat,
* and stays so across operator restarts and between replicas, unlike a name remembered in memory.
* reporting component and, unless the record sets a {@link EventRecord#key()} or the recorder is
* built with a default {@link EventKeyStrategy}, the message. The name is therefore stable across
* occurrences, which is what lets the sink recognise a repeat, and stays so across operator
* restarts and between replicas, unlike a name remembered in memory.
*
* <p>The object is identified by its uid, with the kind as a fallback for objects that do not
* have one yet, such as a dependent resource that has only been built so far.
*
* <p>This is the fallback when the record does not set a name and the naming strategy resolves to
* nothing, see {@link EventNamingStrategy}.
*/
private String eventName(HasMetadata regarding, EventRecord record, String reportingController) {
private String identityHashName(
HasMetadata regarding, EventRecord record, String reportingController) {
var metadata = regarding.getMetadata();
var identity =
String.join(
Expand All @@ -201,7 +331,10 @@ private String eventName(HasMetadata regarding, EventRecord record, String repor
record.type().value(),
record.reason(),
record.reportingComponent().orElse(reportingController),
record.key().orElseGet(() -> requireNonNullElse(record.message(), "")));
record
.key()
.or(() -> keyStrategy.keyFor(regarding, record))
.orElseGet(() -> requireNonNullElse(record.message(), "")));

var suffix = "." + identityDigest(identity);
var prefix = metadata.getName();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
/*
* Copyright Java Operator SDK Authors
*
* Licensed 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 io.javaoperatorsdk.operator.api.event;

import java.util.Optional;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.Experimental;

import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;

/**
* Derives the default aggregation key of an event, used when the {@link EventRecord} does not set
* one explicitly. The key identifies an event among the events about the same object, so that
* repeated occurrences resolve to the same event rather than to one event each, see {@link
* EventRecord#key()}.
*
* <p>An empty result leaves the record without a default key, which keeps the message part of the
* event identity.
*
* <p>Implementations are called from concurrent reconciliations and must be thread safe.
*/
@Experimental(API_MIGHT_CHANGE)
@FunctionalInterface
public interface EventKeyStrategy {

Optional<String> keyFor(HasMetadata regarding, EventRecord record);

/** No default key: the message stays part of the event identity. */
static EventKeyStrategy none() {
return (regarding, record) -> Optional.empty();
}

/**
* Aggregates by event type and reason: all occurrences of a reason resolve to one event whose
* count grows and whose message is replaced with the latest one. The right choice for events that
* report a current state rather than individual occurrences.
*/
static EventKeyStrategy byReason() {
return (regarding, record) -> Optional.of(record.type().value() + "/" + record.reason());
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
/*
* Copyright Java Operator SDK Authors
*
* Licensed 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 io.javaoperatorsdk.operator.api.event;

import java.util.Optional;

import io.fabric8.kubernetes.api.model.HasMetadata;
import io.javaoperatorsdk.operator.api.reconciler.Experimental;

import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE;

/**
* Names the event recorded about an object. The name is what the sink looks a recorded event up by,
* so it is also the aggregation identity: two records resolving to the same name are counted as
* occurrences of one event. A name must therefore be unique among the events it should not
* aggregate with, and stable across operator restarts and replicas.
*
* <p>A name must be a valid RFC 1123 DNS subdomain: at most 253 lowercase alphanumeric characters,
* {@code -} or {@code .}, starting and ending with an alphanumeric character. Names longer than the
* limit are truncated. A name derived from the object and a fixed lowercase suffix (such as {@code
* <object>-status-report}) satisfies all of this by construction.
*
* <p>An empty result or an invalid name falls back to the default {@code <object>.<identity hash>}
* name, rather than the event being lost to the API server rejecting the name.
*
* <p>Implementations are called from concurrent reconciliations and must be thread safe.
*/
@Experimental(API_MIGHT_CHANGE)
@FunctionalInterface
public interface EventNamingStrategy {

Optional<String> nameFor(HasMetadata regarding, EventRecord record);
Comment thread
afalhambra-hivemq marked this conversation as resolved.

/** No custom naming: every event gets the default {@code <object>.<identity hash>} name. */
static EventNamingStrategy none() {
return (regarding, record) -> Optional.empty();
}
}
Loading
Loading