Izeren commented on code in PR #27989: URL: https://github.com/apache/flink/pull/27989#discussion_r3147154707
########## flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/MetricAttributeTransformer.java: ########## @@ -0,0 +1,333 @@ +/* + * 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.flink.metrics.otel; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.metrics.MetricConfig; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; +import javax.annotation.concurrent.NotThreadSafe; + +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.apache.flink.metrics.otel.OpenTelemetryReporterOptions.ATTRIBUTE_VALUE_LENGTH_LIMITS_PREFIX_KEY; +import static org.apache.flink.metrics.otel.OpenTelemetryReporterOptions.COLLISION_TRACKING_MAX_SLOTS; + +/** + * Applies per-attribute and global length limits to metric attribute values before they are + * exported via the OpenTelemetry reporter, and best-effort detects cases where truncation caused + * two distinct raw attribute maps to collapse to the same transformed output for a given metric + * name (which results in ambiguous exported series at the downstream backend). + * + * <p>Configuration is prefix-based under {@value + * OpenTelemetryReporterOptions#ATTRIBUTE_VALUE_LENGTH_LIMITS_PREFIX_KEY}. The special key {@code *} + * sets a global limit for all attributes not explicitly listed. + * + * <p>Semantics of individual limit values: + * + * <ul> + * <li>Positive: truncate the attribute value to that many characters. + * <li>Zero: drop the attribute entirely. + * <li>Negative: keep the full value (overrides a global cap). + * </ul> + * + * <p><b>Collision detection</b> fires at most once per (name, transformed-attributes) slot when two + * distinct raw attribute maps have been observed for that slot. The tracker stores only 64-bit + * hashes, in a bounded LRU so that memory stays contained when attribute cardinality is high (e.g. + * a per-attempt UUID). The cap is configured via {@link + * OpenTelemetryReporterOptions#COLLISION_TRACKING_MAX_SLOTS}; {@code 0} disables tracking entirely. + * On overflow the least-recently-touched slot is evicted, and a previously warned slot that later + * gets evicted may fire its warning again on re-entry. + * + * <p>This class is <b>not thread-safe</b>. Callers are expected to serialize calls to {@link + * #transform}. + * + * @see OpenTelemetryReporterOptions#ATTRIBUTE_VALUE_LENGTH_LIMITS + */ +@NotThreadSafe +final class MetricAttributeTransformer { + + private static final Logger LOG = LoggerFactory.getLogger(MetricAttributeTransformer.class); + + static final String GLOBAL_LIMIT_KEY = "*"; + + /** Load factor for the tracking map; matches {@link java.util.HashMap}'s default. */ + private static final float TRACKING_MAP_LOAD_FACTOR = 0.75f; + + /** Upper bound on the tracking map's initial capacity. */ + private static final int TRACKING_MAP_INITIAL_CAPACITY = 1024; + + /** Access-ordered iteration so {@code removeEldestEntry} evicts the true LRU entry. */ + private static final boolean LRU_ACCESS_ORDER = true; + + /** Parsed attribute-to-limit mapping, including {@code *} for the global limit. */ + private final Map<String, Integer> attributeValueLimits; + + /** + * Bounded LRU of slot-hash → tracking state, or {@code null} when tracking is disabled. The + * slot-hash encodes (metricName, transformedVariables); {@link SlotState} records the first + * raw-hash observed at that slot and whether the slot has already warned. + */ + @Nullable private final Map<Long, SlotState> slotStates; + + private long collisionCount = 0; + + /** + * Per-slot tracking state. Mutating {@code warned} in place avoids a per-warning allocation; + * the immutable slot-hash key drives LRU ordering regardless. + */ + private static final class SlotState { + final long firstRawHash; + boolean warned; + + SlotState(final long firstRawHash) { + this.firstRawHash = firstRawHash; + } + } + + /** + * Constructs a {@link MetricAttributeTransformer} by reading length-limit entries from the + * supplied {@link MetricConfig}. + */ + MetricAttributeTransformer(final MetricConfig metricConfig) { + this.attributeValueLimits = parseAttributeLimits(metricConfig); + if (!attributeValueLimits.isEmpty()) { + LOG.info( + "Metric attribute transformer is configured with value length limits: {}", + attributeValueLimits); + } + this.slotStates = buildSlotStateMap(metricConfig); + } + + @Nullable + private static Map<Long, SlotState> buildSlotStateMap(final MetricConfig metricConfig) { + final int maxSlots = readCollisionTrackingMaxSlots(metricConfig); + if (maxSlots == 0) { + return null; + } + return new LinkedHashMap<>( + TRACKING_MAP_INITIAL_CAPACITY, TRACKING_MAP_LOAD_FACTOR, LRU_ACCESS_ORDER) { + @Override + protected boolean removeEldestEntry(final Map.Entry<Long, SlotState> eldest) { + return size() > maxSlots; + } + }; + } + + private static int readCollisionTrackingMaxSlots(final MetricConfig metricConfig) { + final int defaultValue = COLLISION_TRACKING_MAX_SLOTS.defaultValue(); + final int parsed; + try { + parsed = metricConfig.getInteger(COLLISION_TRACKING_MAX_SLOTS.key(), defaultValue); + } catch (final NumberFormatException e) { + LOG.warn( + "Skipping invalid format for {} with value: {}; falling back to default {}", + COLLISION_TRACKING_MAX_SLOTS.key(), + metricConfig.get(COLLISION_TRACKING_MAX_SLOTS.key()), + defaultValue, + e); + return defaultValue; + } + if (parsed < 0) { + LOG.warn( + "Ignoring negative value {} for {}; falling back to default {}", + parsed, + COLLISION_TRACKING_MAX_SLOTS.key(), + defaultValue); + return defaultValue; + } + return parsed; + } + + /** + * Applies configured length limits to {@code rawVariables} and best-effort tracks whether the + * resulting (name, transformed-attributes) slot has been seen before with a different raw + * input. Emits at most one WARN per slot. + * + * <p>If no limits are configured, or if truncation would produce a map equal to the input, + * {@code rawVariables} is returned unchanged (reference-equal) and no tracking state is + * touched. + * + * @param metricName the fully-qualified metric name + * @param rawVariables the raw metric attribute map + * @return the transformed attribute map to store alongside the metric + */ + Map<String, String> transform(final String metricName, final Map<String, String> rawVariables) { + if (attributeValueLimits.isEmpty()) { + return rawVariables; + } + final Map<String, String> transformed = applyLimits(rawVariables); + if (transformed.equals(rawVariables)) { Review Comment: I have swapped method to `getTransformedOrNull`. It doesn't stop creation of the copy of original map though, but I don't see a nice way to optimize it (splitting into detect + apply is still similarly expensive). -- 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]
