diff --git a/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py b/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py
index 12db5b9a68..8d0009938c 100644
--- a/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-flask/src/opentelemetry/instrumentation/flask/__init__.py
@@ -95,6 +95,38 @@ def hello():
     if __name__ == "__main__":
         app.run(debug=True)
 
+Custom Metrics Attributes using Labeler
+***************************************
+The Flask instrumentation reads from a Labeler utility that supports adding custom attributes 
+to the HTTP metrics recorded by the instrumentation.
+
+.. code-block:: python
+
+    from flask import Flask
+    from opentelemetry.instrumentation.flask import FlaskInstrumentor
+    from opentelemetry.instrumentation._labeler import get_labeler
+
+    app = Flask(__name__)
+    FlaskInstrumentor().instrument_app(app)
+
+    @app.route("/user/<user_id>")
+    def user_profile(user_id):
+        # Get the labeler for the current request
+        labeler = get_labeler()
+        
+        # Add custom attributes to Flask instrumentation metrics
+        labeler.add("user_id", user_id)
+        labeler.add("user_type", "registered")
+        
+        # Or, add multiple attributes at once
+        labeler.add_attributes({
+            "feature_flag": "new_ui",
+            "experiment_group": "control"
+        })
+        
+        return f"User profile for {user_id}"
+
+
 Configuration
 -------------
 
@@ -268,6 +300,7 @@ def response_hook(span: Span, status: str, response_headers: List):
 from opentelemetry.instrumentation.flask.package import _instruments
 from opentelemetry.instrumentation.flask.version import __version__
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+from opentelemetry.instrumentation._labeler import enhance_metric_attributes
 from opentelemetry.instrumentation.propagators import (
     get_global_response_propagator,
 )
@@ -408,6 +441,11 @@ def _start_response(status, response_headers, *args, **kwargs):
                     request_route
                 )
 
+            # Enhance attributes with any custom labeler attributes
+            duration_attrs_old = enhance_metric_attributes(
+                duration_attrs_old
+            )
+
             duration_histogram_old.record(
                 max(round(duration_s * 1000), 0), duration_attrs_old
             )
@@ -419,6 +457,11 @@ def _start_response(status, response_headers, *args, **kwargs):
             if request_route:
                 duration_attrs_new[HTTP_ROUTE] = str(request_route)
 
+            # Enhance attributes with any custom labeler attributes
+            duration_attrs_new = enhance_metric_attributes(
+                duration_attrs_new
+            )
+
             duration_histogram_new.record(
                 max(duration_s, 0), duration_attrs_new
             )
@@ -446,6 +489,10 @@ def _before_request():
             flask_request_environ,
             sem_conv_opt_in_mode=sem_conv_opt_in_mode,
         )
+        # Enhance attributes with custom labeler attributes
+        attributes = enhance_metric_attributes(
+            attributes
+        )
         if flask.request.url_rule:
             # For 404 that result from no route found, etc, we
             # don't have a url_rule.
diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/__init__.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/__init__.py
new file mode 100644
index 0000000000..50031765a8
--- /dev/null
+++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/__init__.py
@@ -0,0 +1,39 @@
+# Copyright The OpenTelemetry 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.
+
+"""
+Labeler that supports per-request custom attribute addition to web framework
+instrumentor-originating OpenTelemetry metrics.
+
+This was inspired by OpenTelemetry Go's net/http instrumentation Labeler
+https://github.com/open-telemetry/opentelemetry-go-contrib/pull/306
+"""
+
+from opentelemetry.instrumentation._labeler._internal import (
+    Labeler,
+    get_labeler,
+    set_labeler,
+    clear_labeler,
+    get_labeler_attributes,
+    enhance_metric_attributes,
+)
+
+__all__ = [
+    "Labeler",
+    "get_labeler",
+    "set_labeler", 
+    "clear_labeler",
+    "get_labeler_attributes",
+    "enhance_metric_attributes",
+]
diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/_internal/__init__.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/_internal/__init__.py
new file mode 100644
index 0000000000..01be55a991
--- /dev/null
+++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/_internal/__init__.py
@@ -0,0 +1,179 @@
+# Copyright The OpenTelemetry 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.
+
+import threading
+from typing import Dict, Union, Optional, Any
+import contextvars
+
+# Context variable to store the current labeler
+_labeler_context: contextvars.ContextVar[Optional["Labeler"]] = contextvars.ContextVar(
+    "otel_labeler", default=None
+)
+
+
+class Labeler:
+    """
+    Labeler can be used by instrumented code or distro to add custom attributes
+    to the metrics recorded by those OpenTelemetry instrumentations reading it.
+    
+    Labeler accumulates custom attributes for OpenTelemetry metrics for the
+    current request in context.
+
+    This feature is experimental and unstable.
+    """
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._attributes: Dict[str, Union[str, int, float, bool]] = {}
+
+    def add(self, key: str, value: Union[str, int, float, bool]) -> None:
+        """
+        Add a single attribute to the labeler.
+        
+        Args:
+            key: The attribute key
+            value: The attribute value (must be a primitive type)
+        """
+        if not isinstance(value, (str, int, float, bool)):
+            raise ValueError(f"Attribute value must be str, int, float, or bool, got {type(value)}")
+        
+        with self._lock:
+            self._attributes[key] = value
+
+    def add_attributes(self, attributes: Dict[str, Union[str, int, float, bool]]) -> None:
+        """
+        Add multiple attributes to the labeler.
+        
+        Args:
+            attributes: Dictionary of attributes to add
+        """
+        for key, value in attributes.items():
+            if not isinstance(value, (str, int, float, bool)):
+                raise ValueError(f"Attribute value for '{key}' must be str, int, float, or bool, got {type(value)}")
+        
+        with self._lock:
+            self._attributes.update(attributes)
+
+    def get_attributes(self) -> Dict[str, Union[str, int, float, bool]]:
+        """
+        Returns a copy of all attributes added to the labeler.
+        """
+        with self._lock:
+            return self._attributes.copy()
+
+    def clear(self) -> None:
+        with self._lock:
+            self._attributes.clear()
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._attributes)
+
+
+def get_labeler() -> Labeler:
+    """
+    Get the Labeler instance for the current request context.
+    
+    If no Labeler exists in the current context, a new one is created
+    and stored in the context.
+    
+    Returns:
+        Labeler instance for the current request, or a new empty Labeler
+        if not in a request context
+    """
+    labeler = _labeler_context.get()
+    if labeler is None:
+        labeler = Labeler()
+        _labeler_context.set(labeler)
+    return labeler
+
+
+def set_labeler(labeler: Labeler) -> None:
+    """
+    Set the Labeler instance for the current request context.
+    
+    Args:
+        labeler: The Labeler instance to set
+    """
+    _labeler_context.set(labeler)
+
+
+def clear_labeler() -> None:
+    """
+    Clear the Labeler instance from the current request context.
+    """
+    _labeler_context.set(None)
+
+
+def get_labeler_attributes() -> Dict[str, Union[str, int, float, bool]]:
+    """
+    Get attributes from the current labeler, if any.
+    
+    Returns:
+        Dictionary of custom attributes, or empty dict if no labeler exists
+    """
+    labeler = _labeler_context.get()
+    if labeler is None:
+        return {}
+    return labeler.get_attributes()
+
+
+def enhance_metric_attributes(
+    base_attributes: Dict[str, Any], 
+    include_custom: bool = True,
+    max_custom_attrs: int = 20,
+    max_attr_value_length: int = 100
+) -> Dict[str, Any]:
+    """
+    This function combines base_attributes with custom attributes
+    from the current labeler.
+    
+    Custom attributes are skipped if they would override base_attributes,
+    exceed max_custom_attrs number, or are not simple types (str, int, float,
+    bool). If custom attributes have string values exceeding the
+    max_attr_value_length, then they are truncated.
+    
+    Args:
+        base_attributes: The base attributes for the metric
+        include_custom: Whether to include custom labeler attributes
+        max_custom_attrs: Maximum number of custom attributes to include
+        max_attr_value_length: Maximum length for string attribute values
+        
+    Returns:
+        Dictionary combining base and custom attributes
+    """
+    if not include_custom:
+        return base_attributes.copy()
+    
+    custom_attributes = get_labeler_attributes()
+    if not custom_attributes:
+        return base_attributes.copy()
+    
+    enhanced_attributes = base_attributes.copy()
+    
+    added_count = 0
+    for key, value in custom_attributes.items():
+        if added_count >= max_custom_attrs:
+            break
+        if key in base_attributes:
+            continue
+
+        if isinstance(value, str) and len(value) > max_attr_value_length:
+            value = value[:max_attr_value_length]
+
+        if isinstance(value, (str, int, float, bool)):
+            enhanced_attributes[key] = value
+            added_count += 1
+    
+    return enhanced_attributes
diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/example.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/example.py
new file mode 100644
index 0000000000..f9d049c942
--- /dev/null
+++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/_labeler/example.py
@@ -0,0 +1,52 @@
+# Copyright The OpenTelemetry 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.
+"""
+Example Flask application demonstrating how to use Labeler for adding
+custom attributes to metrics generated by the Flask instrumentor.
+"""
+
+from flask import Flask
+from opentelemetry.instrumentation.flask import FlaskInstrumentor
+from opentelemetry.instrumentation._labeler import get_labeler
+
+app = Flask(__name__)
+FlaskInstrumentor().instrument_app(app)
+
+
+@app.route("/healthcheck")
+def healthcheck():
+    # Get the labeler for the current request
+    labeler = get_labeler()
+
+    labeler.add_attributes({
+        "endpoint_type": "healthcheck",
+        "internal_request": True,
+    })
+    return "OK"
+
+
+@app.route("/user/<user_id>")
+def user_profile(user_id):
+    labeler = get_labeler()
+    
+    # Can add individual attributes or multiple at once
+    labeler.add("user_id", user_id)
+    labeler.add_attributes({
+        "has_premium": user_id in ["123", "456"],
+        "experiment_group": "control",
+        "feature_enabled": True,
+        "user_segment": "active"
+    })
+    
+    return f"Got user profile for {user_id}"
diff --git a/opentelemetry-instrumentation/tests/test_labeler.py b/opentelemetry-instrumentation/tests/test_labeler.py
new file mode 100644
index 0000000000..e60b4e58f8
--- /dev/null
+++ b/opentelemetry-instrumentation/tests/test_labeler.py
@@ -0,0 +1,166 @@
+"""
+Test cases for the common Labeler functionality in opentelemetry-instrumentation.
+"""
+
+import unittest
+import threading
+import contextvars
+
+from opentelemetry.instrumentation._labeler import (
+    Labeler,
+    get_labeler,
+    set_labeler,
+    clear_labeler,
+    get_labeler_attributes,
+)
+
+
+class TestLabeler(unittest.TestCase):
+    def setUp(self):
+        clear_labeler()
+
+    def test_labeler_init(self):
+        labeler = Labeler()
+        self.assertEqual(labeler.get_attributes(), {})
+        self.assertEqual(len(labeler), 0)
+
+    def test_add_single_attribute(self):
+        labeler = Labeler()
+        labeler.add("test_key", "test_value")
+        attributes = labeler.get_attributes()
+        self.assertEqual(attributes, {"test_key": "test_value"})
+        self.assertEqual(len(labeler), 1)
+
+    def test_add_multiple_attributes(self):
+        labeler = Labeler()
+        labeler.add("key1", "value1")
+        labeler.add("key2", 42)
+        labeler.add("key3", True)
+        labeler.add("key4", 3.14)
+        attributes = labeler.get_attributes()
+        expected = {"key1": "value1", "key2": 42, "key3": True, "key4": 3.14}
+        self.assertEqual(attributes, expected)
+        self.assertEqual(len(labeler), 4)
+
+    def test_add_attributes_dict(self):
+        labeler = Labeler()
+        attrs = {"key1": "value1", "key2": 42, "key3": False}
+        labeler.add_attributes(attrs)
+        attributes = labeler.get_attributes()
+        self.assertEqual(attributes, attrs)
+
+    def test_invalid_attribute_types(self):
+        labeler = Labeler()
+        
+        with self.assertRaises(ValueError):
+            labeler.add("key", [1, 2, 3])
+        
+        with self.assertRaises(ValueError):
+            labeler.add("key", {"nested": "dict"})
+        
+        with self.assertRaises(ValueError):
+            labeler.add_attributes({"key": None})
+
+    def test_overwrite_attribute(self):
+        labeler = Labeler()
+        labeler.add("key1", "original")
+        labeler.add("key1", "updated")
+        attributes = labeler.get_attributes()
+        self.assertEqual(attributes, {"key1": "updated"})
+
+    def test_clear_attributes(self):
+        labeler = Labeler()
+        labeler.add("key1", "value1")
+        labeler.add("key2", "value2")
+        labeler.clear()
+        self.assertEqual(labeler.get_attributes(), {})
+        self.assertEqual(len(labeler), 0)
+
+    def test_thread_safety(self):
+        labeler = Labeler()
+        num_threads = 10
+        num_operations = 100
+        
+        def worker(thread_id):
+            for i in range(num_operations):
+                labeler.add(f"thread_{thread_id}_key_{i}", f"value_{i}")
+        
+        # Start multiple threads
+        threads = []
+        for thread_id in range(num_threads):
+            thread = threading.Thread(target=worker, args=(thread_id,))
+            threads.append(thread)
+            thread.start()
+        
+        # Wait for all threads to complete
+        for thread in threads:
+            thread.join()
+        
+        # Check that all attributes were added
+        attributes = labeler.get_attributes()
+        expected_count = num_threads * num_operations
+        self.assertEqual(len(attributes), expected_count)
+
+
+class TestLabelerContext(unittest.TestCase):
+    def setUp(self):
+        clear_labeler()
+
+    def test_get_labeler_creates_new(self):
+        """Test that get_labeler creates a new labeler if none exists."""
+        labeler = get_labeler()
+        self.assertIsInstance(labeler, Labeler)
+        self.assertEqual(labeler.get_attributes(), {})
+
+    def test_get_labeler_returns_same_instance(self):
+        """Test that get_labeler returns the same instance within context."""
+        labeler1 = get_labeler()
+        labeler1.add("test", "value")
+        labeler2 = get_labeler()
+        self.assertIs(labeler1, labeler2)
+        self.assertEqual(labeler2.get_attributes(), {"test": "value"})
+
+    def test_set_labeler(self):
+        custom_labeler = Labeler()
+        custom_labeler.add("custom", "value")
+        set_labeler(custom_labeler)
+        retrieved_labeler = get_labeler()
+        self.assertIs(retrieved_labeler, custom_labeler)
+        self.assertEqual(retrieved_labeler.get_attributes(), {"custom": "value"})
+
+    def test_clear_labeler(self):
+        labeler = get_labeler()
+        labeler.add("test", "value")
+        clear_labeler()
+        # Should get a new labeler after clearing
+        new_labeler = get_labeler()
+        self.assertIsNot(new_labeler, labeler)
+        self.assertEqual(new_labeler.get_attributes(), {})
+
+    def test_get_labeler_attributes(self):
+        clear_labeler()
+        attrs = get_labeler_attributes()
+        self.assertEqual(attrs, {})
+        labeler = get_labeler()
+        labeler.add("test", "value")
+        attrs = get_labeler_attributes()
+        self.assertEqual(attrs, {"test": "value"})
+
+    def test_context_isolation(self):
+        def context_worker(context_id, results):
+            labeler = get_labeler()
+            labeler.add("context_id", context_id)
+            labeler.add("value", f"context_{context_id}")
+            results[context_id] = labeler.get_attributes()
+        
+        results = {}
+        
+        # Run in different contextvars contexts
+        for i in range(3):
+            ctx = contextvars.copy_context()
+            ctx.run(context_worker, i, results)
+        
+        # Each context should have its own labeler with its own values
+        for i in range(3):
+            expected = {"context_id": i, "value": f"context_{i}"}
+            self.assertEqual(results[i], expected)