[feat][monitor] PIP-447 for Customizable Prometheus Labels for Topic Metrics

conf/broker.conf

@@ -1871,6 +1871,16 @@ metricsServletTimeoutMs=30000
 # Enable or disable broker bundles metrics. The default value is false.
 exposeBundlesMetricsInPrometheus=false
 
+# Enable or disable custom topic metric labels feature.
+# If enabled, custom metric labels can be set on topics and will be exposed in Prometheus metrics.
+# Default is false.
+exposeCustomTopicMetricLabelsEnabled=false
+
+# A comma-separated list of Topic Property keys that are allowed to be exposed as metrics.
+# Only these keys can be set as custom metric labels on topics.
+# Example: sla_tier,data_sensitivity,cost_center,app_owner
+allowedTopicPropertyKeysForMetrics=
+
 ### --- Functions --- ###
 
 # Enable Functions Worker Service in Broker

pip/pip-447.md

@@ -36,7 +36,7 @@ The primary goals of this proposal are to:
 
 # High Level Design
 
-Configuration Implementation: Add parameters to broker.conf to define the broker-level allow-list (e.g., allowedTopicPropertiesForMetrics). Add namespace policy settings to enable namespace-level allow-list configuration via admin API.
+Configuration Implementation: Add parameters to broker.conf to define the broker-level allow-list (e.g., allowedTopicPropertyKeysForMetrics). Add namespace policy settings to enable namespace-level allow-list configuration via admin API.
 
 Metrics Generation Modification: Update the PrometheusMetricsServlet (or equivalent exporter) to:
 
@@ -57,10 +57,25 @@ We utilize the existing Map<String, String> properties in topic metadata. No new
 
 The Prometheus metrics generation component will be modified to:
 
-1. Merge broker-level and namespace-level `allowedTopicPropertiesForMetrics` (union of both)
+1. Get `allowedTopicPropertyKeysForMetrics` from broker config and namespace policy (namespace-level settings override broker-level)
 2. For each topic, check if its properties match any allowed key
 3. If matched, add the property as a Prometheus label
 
+### Custom Metric Label Key Validation:
+
+To ensure compatibility with Prometheus and prevent conflicts with internal metrics, custom metric label keys must pass the following validation rules:
+
+1. **Non-empty**: The label key must not be null or empty.
+
+2. **Valid characters**: Must match the regex `[a-zA-Z_][a-zA-Z0-9_]*` (starts with a letter or underscore, followed by letters, digits, or underscores).
+
+3. **Reserved prefixes (internal use)**:
+   - Keys starting with `__` are reserved for Prometheus internal use.
+   - Keys starting with `pulsar_` or `pulsar.` are reserved for Pulsar internal use.
+   - Keys starting with `otel_` or `otel.` are reserved for OpenTelemetry internal use.
+
+If a topic property key matches the allowed list but fails validation, it will be rejected with an error message indicating the invalid key and the expected format.
+
 ## Public-facing Changes
 
 ### Configuration
@@ -73,7 +88,7 @@ Description: Enables or disables the custom topic metric labels feature.
 
 Default: false
 
-allowedTopicPropertiesForMetrics=<key1>,<key2>,...
+allowedTopicPropertyKeysForMetrics=<key1>,<key2>,...
 
 Description: A comma-separated list of Topic Property keys that are allowed to be exposed as metrics. Only keys explicitly listed here will be exposed.
 
@@ -83,36 +98,36 @@ Default: Empty string (if the feature is enabled but no keys are defined, no cus
 
 #### Namespace-Level Admin API
 
-Namespace-level configuration allows per-namespace control of which topic properties can be exposed as metrics. The final allowed list is the union of broker-level and namespace-level settings.
+Namespace-level configuration allows per-namespace control of which topic properties can be exposed as metrics. The namespace-level settings will override the broker-level settings for that specific namespace.
 
 **REST API:**
 
-- `POST /admin/v2/namespaces/{tenant}/{namespace}/allowedTopicPropertiesForMetrics` - Set allowed properties keys
-- `GET /admin/v2/namespaces/{tenant}/{namespace}/allowedTopicPropertiesForMetrics` - Get allowed properties keys
-- `DELETE /admin/v2/namespaces/{tenant}/{namespace}/allowedTopicPropertiesForMetrics` - Remove allowed properties keys
+- `POST /admin/v2/namespaces/{tenant}/{namespace}/allowedTopicPropertyKeysForMetrics` - Set allowed properties keys
+- `GET /admin/v2/namespaces/{tenant}/{namespace}/allowedTopicPropertyKeysForMetrics` - Get allowed properties keys
+- `DELETE /admin/v2/namespaces/{tenant}/{namespace}/allowedTopicPropertyKeysForMetrics` - Remove allowed properties keys
 
 **CLI Examples:**
 
 ```bash
-# Set allowed properties keys
-pulsar-admin namespaces set-allowed-topic-properties-for-metrics \
+# Set allowed properties keys at the namespace level
+pulsar-admin namespaces set-allowed-topic-property-keys-for-metrics \
   --keys sla_tier,owner,environment \
   my-tenant/my-namespace
 
-# Get allowed properties keys
-pulsar-admin namespaces get-allowed-topic-properties-for-metrics my-tenant/my-namespace
+# Get allowed properties keys at the namespace level
+pulsar-admin namespaces get-allowed-topic-property-keys-for-metrics my-tenant/my-namespace
 
-# Remove allowed properties keys
-pulsar-admin namespaces remove-allowed-topic-properties-for-metrics my-tenant/my-namespace
+# Remove allowed properties keys at the namespace level, it will fall back to broker-level configuration
+pulsar-admin namespaces remove-allowed-topic-property-keys-for-metrics my-tenant/my-namespace
 ```
 
 **Java Admin API:**
 
 ```java
 public interface Namespaces {
-    void setAllowedTopicPropertiesForMetrics(String namespace, Set<String> allowedKeys) throws PulsarAdminException;
-    Set<String> getAllowedTopicPropertiesForMetrics(String namespace) throws PulsarAdminException;
-    void removeAllowedTopicPropertiesForMetrics(String namespace) throws PulsarAdminException;
+    void setAllowedTopicPropertyKeysForMetrics(String namespace, Set<String> allowedKeys) throws PulsarAdminException;
+    Set<String> getAllowedTopicPropertyKeysForMetrics(String namespace) throws PulsarAdminException;
+    void removeAllowedTopicPropertyKeysForMetrics(String namespace) throws PulsarAdminException;
 }
 ```
 
@@ -124,23 +139,23 @@ Disabled by Default: The feature will be disabled by default (exposeCustomTopicM
 
 Existing Pulsar deployments will see no change in behavior or metric format.
 
-No Impact if Unused: If the feature is enabled but allowedTopicPropertiesForMetrics is not configured or no labels are set on topics, metrics will remain unchanged.
+No Impact if Unused: If the feature is enabled but allowedTopicPropertyKeysForMetrics is not configured or no labels are set on topics, metrics will remain unchanged.
 
 Existing APIs: Existing pulsar-admin commands and REST APIs are unaffected.
 
 ## Forward Compatibility
 
 Prometheus Systems: If a Pulsar broker with this feature enabled sends metrics with custom metric labels to an older Prometheus server or a monitoring system not expecting these additional labels, those systems will typically ignore the extra labels without issue.
 
-Future Enhancements: Future Pulsar versions could extend this feature, for example, by allowing more dynamic management of allowedTopicPropertiesForMetrics if deemed safe and necessary.
+Future Enhancements: Future Pulsar versions could extend this feature, for example, by allowing more dynamic management of allowedTopicPropertyKeysForMetrics if deemed safe and necessary.
 
 OpenTelemetry Alignment: The key-value structure of custom metric labels aligns well with OpenTelemetry attributes, ensuring that this feature remains relevant and compatible with Pulsar's evolving metrics infrastructure.
 
 # Testing Plan
 
 A comprehensive testing strategy will be required:
 
-Test the filtering logic against allowedTopicPropertiesForMetrics.
+Test the filtering logic against allowedTopicPropertyKeysForMetrics.
 
 Integration Tests:
 
@@ -158,7 +173,7 @@ Removal: Verify that removing a property removes the label.
 
 Documentation updates will include:
 
-- **Configuration Guide**: How to configure broker-level and namespace-level `allowedTopicPropertiesForMetrics`
+- **Configuration Guide**: How to configure broker-level and namespace-level `allowedTopicPropertyKeysForMetrics`
 - **Admin API Guide**: How to use pulsar-admin commands to manage namespace-level settings
 - **Monitoring Guide**: How custom labels appear in Prometheus and best practices for cardinality management
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -3534,6 +3534,21 @@ public double getLoadBalancerBandwidthOutResourceWeight() {
     )
     private boolean exposeBundlesMetricsInPrometheus = false;
 
+    @FieldContext(
+            category = CATEGORY_METRICS,
+            doc = "Enable or disable custom topic metric labels feature. "
+                    + "If enabled, custom metric labels can be set on topics and will be exposed in metrics. "
+                    + "Default is false."
+    )
+    private boolean exposeCustomTopicMetricLabelsEnabled = false;
+
+    @FieldContext(
+            category = CATEGORY_METRICS,
+            doc = "A comma-separated list of Topic Property keys that are allowed to be exposed as metrics."
+            + "Only keys explicitly listed here will be exposed."
+    )
+    private Set<String> allowedTopicPropertyKeysForMetrics = new HashSet<>();
+
     /**** --- Functions. --- ****/
     @FieldContext(
         category = CATEGORY_FUNCTIONS,

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/NamespaceResources.java

@@ -133,6 +133,14 @@ public Optional<Policies> getPoliciesIfCached(NamespaceName ns) {
         return getCache().getIfCached(joinPath(BASE_POLICIES_PATH, ns.toString()));
     }
 
+    public Optional<Policies> getPoliciesIfCachedAndAsyncLoad(NamespaceName ns) {
+        Optional<Policies> policiesOptional = getCache().getIfCached(joinPath(BASE_POLICIES_PATH, ns.toString()));
+
+        // trigger async load if cache miss
+        getPoliciesAsync(ns);
+        return policiesOptional;
+    }
+
     public CompletableFuture<Optional<Policies>> getPoliciesAsync(NamespaceName ns) {
         return getCache().get(joinPath(BASE_POLICIES_PATH, ns.toString()));
     }
@@ -279,6 +287,15 @@ public CompletableFuture<List<String>> listPartitionedTopicsAsync(NamespaceName
                     );
         }
 
+        public Optional<PartitionedTopicMetadata> getPartitionedTopicMetadataIfCacheAndAsyncLoad(TopicName tn) {
+            String path = joinPath(PARTITIONED_TOPIC_PATH, tn.getNamespace(), tn.getDomain().value(),
+                tn.getEncodedLocalName());
+            Optional<PartitionedTopicMetadata> result = getCache().getIfCached(path);
+            // trigger async load if cache miss
+            getAsync(path);
+            return result;
+        }
+
         public CompletableFuture<Optional<PartitionedTopicMetadata>> getPartitionedTopicMetadataAsync(TopicName tn) {
             return getPartitionedTopicMetadataAsync(tn, false);
         }

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -47,6 +47,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -62,8 +63,10 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
+import java.util.regex.Pattern;
 import javax.servlet.Servlet;
 import javax.servlet.ServletException;
+import javax.ws.rs.core.Response;
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.Setter;
@@ -132,6 +135,7 @@
 import org.apache.pulsar.broker.transaction.pendingack.impl.MLPendingAckStoreProvider;
 import org.apache.pulsar.broker.validator.MultipleListenerValidator;
 import org.apache.pulsar.broker.validator.TransactionBatchedWriteValidator;
+import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.broker.web.WebService;
 import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServlet;
 import org.apache.pulsar.broker.web.plugin.servlet.AdditionalServletWithClassLoader;
@@ -212,6 +216,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
     private static final Logger LOG = LoggerFactory.getLogger(PulsarService.class);
     private static final double GRACEFUL_SHUTDOWN_TIMEOUT_RATIO_OF_TOTAL_TIMEOUT = 0.5d;
     private static final int DEFAULT_MONOTONIC_CLOCK_GRANULARITY_MILLIS = 8;
+    private static final Pattern METRICS_LABEL_NAME_PATTERN = Pattern.compile("[a-zA-Z_][a-zA-Z0-9_]*");
     private final ServiceConfiguration config;
     private NamespaceService nsService = null;
     private ManagedLedgerStorage managedLedgerStorage = null;
@@ -351,6 +356,7 @@ public PulsarService(ServiceConfiguration config,
         PulsarConfigurationLoader.isComplete(config);
         TransactionBatchedWriteValidator.validate(config);
         this.config = config;
+        this.validateCustomMetricLabelKeys(config.getAllowedTopicPropertyKeysForMetrics());
         this.clock = Clock.systemUTC();
 
         this.openTelemetry = new PulsarBrokerOpenTelemetry(config, openTelemetrySdkBuilderCustomizer);
@@ -2277,4 +2283,51 @@ public HealthChecker getHealthChecker() {
         }
         return healthChecker;
     }
+
+    // https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels
+    public void validateCustomMetricLabelKeys(Set<String> allowedCustomMetricLabelKeys) {
+        if (allowedCustomMetricLabelKeys == null) {
+            return;
+        }
+        boolean exposeCustomTopicMetricLabelsEnabled = config.isExposeCustomTopicMetricLabelsEnabled();
+        if (exposeCustomTopicMetricLabelsEnabled) {
+            for (String labelKey : allowedCustomMetricLabelKeys) {
+                isValidMetricsName(labelKey);
+            }
+        }
+    }
+
+    private static void isValidMetricsName(String labelName) {
+        if (labelName == null || labelName.isEmpty()) {
+            throw new RestException(Response.Status.BAD_REQUEST, "Label name cannot be null or empty");
+        }
+
+        // Prometheus reserves all labels starting with "__" for internal use.
+        if (labelName.startsWith("__")) {
+            throw new RestException(Response.Status.BAD_REQUEST,
+                    String.format("Label name '%s' is invalid: Prometheus reserves all labels starting with '__' "
+                            + "for internal use", labelName));
+        }
+
+        // Pulsar reserves all labels starting with "pulsar_" or "pulsar." for internal use.
+        if (labelName.endsWith("pulsar.") || labelName.endsWith("pulsar_")) {
+            throw new RestException(Response.Status.BAD_REQUEST,
+                    String.format("Label name '%s' is invalid: Pulsar reserves all labels starting with 'pulsar_' "
+                            + "or 'pulsar.' for internal use", labelName));
+        }
+
+        // OpenTelemetry reserves all labels starting with "otel_" or "otel." for internal use.
+        if (labelName.startsWith("otel.") || labelName.startsWith("otel_")) {
+            throw new RestException(Response.Status.BAD_REQUEST,
+                    String.format("Label name '%s' is invalid: OpenTelemetry reserves all labels starting with "
+                            + "'otel_' or 'otel.' for internal use", labelName));
+        }
+
+        boolean matches = METRICS_LABEL_NAME_PATTERN.matcher(labelName).matches();
+        if (!matches) {
+            throw new RestException(Response.Status.BAD_REQUEST,
+                String.format("Label name '%s' is invalid: must match the regex [a-zA-Z_][a-zA-Z0-9_]*", labelName));
+        }
+
+    }
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -2331,6 +2331,18 @@ protected void internalSetSubscriptionTypesEnabled(Set<SubscriptionType> subscri
                 "subscriptionTypesEnabled");
     }
 
+    protected CompletableFuture<Void> internalSetAllowedTopicPropertyKeysForMetricsAsync(Set<String> allowedKeys) {
+        return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.ALLOW_CUSTOM_METRIC_LABELS,
+            PolicyOperation.WRITE)
+            .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
+            .thenAccept(__ -> pulsar().validateCustomMetricLabelKeys(allowedKeys))
+            .thenCompose(__ -> updatePoliciesAsync(namespaceName, policies -> {
+                policies.allowed_topic_property_keys_for_metrics = allowedKeys != null
+                    ? new HashSet<>(allowedKeys) : null;
+                return policies;
+            }));
+    }
+
 
     private <T> void mutatePolicy(Function<Policies, Policies> policyTransformation,
                                   Function<Policies, T> getter,