rkhachatryan commented on code in PR #27989: URL: https://github.com/apache/flink/pull/27989#discussion_r3143943308
########## 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: Can we avoid `HashMap.equals` to save CPU cycles here? For example, in `applyLimits` we could count the number of changes to the map; and if it's zero then return the original hashmap (and then here we could use `==`). Or just return `tuple(map, is_modified)`. ########## 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)) { + // No-op truncation can't produce new collisions. Return the input to avoid retaining + // the freshly-allocated transformed map. + return rawVariables; + } + if (slotStates == null) { + return transformed; + } + + final long slotHash = slotHash(metricName, transformed); + final long rawHash = attributesHash(rawVariables); + final SlotState state = slotStates.get(slotHash); + if (state == null) { + slotStates.put(slotHash, new SlotState(rawHash)); + return transformed; + } + if (state.warned || state.firstRawHash == rawHash) { + return transformed; + } + LOG.warn( + "Truncation collision at metric '{}': this registration truncated {} and " + + "collapsed onto transformed attributes {} already seen for a " + + "different raw input. Exported series at this slot will be " + + "ambiguous at the downstream backend.", + metricName, + truncatedEntries(rawVariables, transformed), + transformed); + state.warned = true; + collisionCount++; + return transformed; + } + + /** + * Returns the subset of {@code raw} entries whose value was actually truncated. Dropped + * attributes don't contribute to the collision signature for this warning. + */ + private static Map<String, String> truncatedEntries( + final Map<String, String> raw, final Map<String, String> transformed) { + final Map<String, String> diff = new HashMap<>(); + for (Map.Entry<String, String> e : raw.entrySet()) { + final String newValue = transformed.get(e.getKey()); + if (newValue != null && !newValue.equals(e.getValue())) { + diff.put(e.getKey(), e.getValue()); + } + } + return diff; + } + + /** + * @return the cumulative number of collisions observed since construction. + */ + @VisibleForTesting + long getCollisionCount() { + return collisionCount; + } + + @VisibleForTesting + int getTrackedSlotCount() { + return slotStates == null ? 0 : slotStates.size(); + } + + @VisibleForTesting + Map<String, Integer> getAttributeValueLimits() { + return attributeValueLimits; + } + + // ----------------------------------------------------------------------- + + private Map<String, String> applyLimits(final Map<String, String> variables) { + final int globalLimit = + attributeValueLimits.getOrDefault(GLOBAL_LIMIT_KEY, Integer.MAX_VALUE); + final Map<String, String> result = new HashMap<>(); + for (Map.Entry<String, String> entry : variables.entrySet()) { + final String key = entry.getKey(); + final String value = entry.getValue(); + final int configuredLimit = attributeValueLimits.getOrDefault(key, globalLimit); + if (configuredLimit == 0) { + // Drop the attribute entirely (user-requested via 0). + continue; + } + if (configuredLimit > 0 && value.length() > configuredLimit) { Review Comment: Is value guaranteed to be non-null? ########## flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryReporterOptions.java: ########## @@ -132,6 +132,71 @@ private OpenTelemetryReporterOptions() {} "Timeout in milliseconds for waiting on async export completion.") .build()); + /** Prefix key used to identify attribute value length limit configuration entries. */ + public static final String ATTRIBUTE_VALUE_LENGTH_LIMITS_PREFIX_KEY = + "transform.attribute-value-length-limits."; + + /** + * Config option for attribute value length limits. Only used for documentation purposes — the + * actual option is prefix-based and parsed by {@link MetricAttributeTransformer}. + * + * <p>For example, to limit the {@code task_name} attribute to 60 characters set {@code + * metrics.reporter.otel.transform.attribute-value-length-limits.task_name: 60} in the Flink + * configuration. The special key {@code *} defines a global limit for all attributes not + * explicitly listed. {@code 0} drops an attribute; negative values disable the limit for that + * attribute (useful to override a global cap). + */ + @PublicEvolving + public static final ConfigOption<Integer> ATTRIBUTE_VALUE_LENGTH_LIMITS = + ConfigOptions.key(ATTRIBUTE_VALUE_LENGTH_LIMITS_PREFIX_KEY + "<attribute-name>") + .intType() + .noDefaultValue() + .withDescription( + Description.builder() + .text( + "Limits of the exported attribute values length. Only applies to the metric " + + "reporter; ignored by the trace and event reporters. " + + "Configuration is prefix based, " + + "for example to limit `task_name` attribute set " + + "`metrics.reporter.otel.transform.attribute-value-length-limits.task_name: 60` " + + "in the config for OTel reporter. " + + "A special key '*' can be used to define a global limit for all attributes not " + + "explicitly listed. " + + "For example `metrics.reporter.otel.transform.attribute-value-length-limits.*: 1024` " + + "will limit all attributes to attributeValue.substring(0, 1024). " + + "Global limit defaults to Integer.MAX_VALUE if not set. Individual attribute " + + "limits always override the global limit and are verified by exact match on the " + + "attribute name. " + + "0 can be used to drop an attribute. Negative values are interpreted as no limit " + + "for the attribute (can be used for global limit overrides).") + .build()); + + /** Config key for the collision tracker's maximum size. */ + public static final String COLLISION_TRACKING_MAX_SLOTS_KEY = + "transform.collision-tracking-max-slots"; Review Comment: Are these options prefixed anyhow? I guess not - and it might be not very clear from the option name what is it about. How about adding `exporter.` prefix or something more relevant? ########## flink-metrics/flink-metrics-otel/src/main/java/org/apache/flink/metrics/otel/OpenTelemetryMetricReporter.java: ########## @@ -156,6 +166,7 @@ protected MetricExporter createExporter(final MetricConfig metricConfig) { @Override public synchronized void close() { + transformer = null; Review Comment: It's reasonable assume that `notifyOfAddedMetric` is not called after `close`; However, it'd be safer to not nullify it here, maybe just reset cache. WDYT? ########## 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)) { + // No-op truncation can't produce new collisions. Return the input to avoid retaining + // the freshly-allocated transformed map. + return rawVariables; + } + if (slotStates == null) { + return transformed; + } + + final long slotHash = slotHash(metricName, transformed); + final long rawHash = attributesHash(rawVariables); + final SlotState state = slotStates.get(slotHash); + if (state == null) { + slotStates.put(slotHash, new SlotState(rawHash)); + return transformed; + } + if (state.warned || state.firstRawHash == rawHash) { + return transformed; + } + LOG.warn( + "Truncation collision at metric '{}': this registration truncated {} and " + + "collapsed onto transformed attributes {} already seen for a " + + "different raw input. Exported series at this slot will be " + + "ambiguous at the downstream backend.", + metricName, + truncatedEntries(rawVariables, transformed), + transformed); Review Comment: Since hash collisions are possible, I'd phrase it as **_possible_** metric collision. -- 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]
