[rollout] fix: mlflow consecutive slashes (verl-project#4446)

### What does this PR do?

MLFlow would not work with metrics that have // in its item name, it
will yield error like so:

```
  File "/usr/local/lib/python3.12/dist-packages/mlflow/tracking/client.py", line 2511, in log_batch
    return self._tracking_client.log_batch(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/dist-packages/mlflow/telemetry/track.py", line 30, in wrapper
    result = func(*args, **kwargs)
             ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/dist-packages/mlflow/tracking/_tracking_service/client.py", line 581, in log_batch
    self.store.log_batch(run_id=run_id, metrics=metrics_batch, params=[], tags=[])
  File "/usr/local/lib/python3.12/dist-packages/mlflow/store/tracking/rest_store.py", line 906, in log_batch
    self._call_endpoint(LogBatch, req_body)
  File "/usr/local/lib/python3.12/dist-packages/mlflow/store/tracking/rest_store.py", line 208, in _call_endpoint
    return call_endpoint(
           ^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/dist-packages/mlflow/utils/rest_utils.py", line 596, in call_endpoint
    response = verify_rest_response(response, endpoint)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/dist-packages/mlflow/utils/rest_utils.py", line 315, in verify_rest_response
    raise RestException(json.loads(response.text))
mlflow.exceptions.RestException: INVALID_PARAMETER_VALUE: Invalid value "val-aux//reward/mean_at_1" for parameter 'metrics[0].name' supplied: Names may be treated as files in certain cases, and must not resolve to other names when treated as such. This name would resolve to 'val-aux/reward/mean_at_1'
```

### Test

Added testing for this behavior into `TestMlflowLoggingAdapter`.

### Design & Code Changes

Used regular expression to parse and substituted multiple slashes
pattern

Author: BaiqingL

tests/utils/test_mlflow_key_sanitization.py

@@ -20,24 +20,44 @@
 
 class TestMlflowLoggingAdapter(unittest.TestCase):
     def test_sanitize_key_and_warning(self):
+        """Test key sanitization for invalid characters and consecutive slashes with warnings."""
         adapter = _MlflowLoggingAdapter()
-        data = {"valid_key": 1.0, "invalid@key!": 2.0, "another/valid-key": 3.0, "bad key#": 4.0}
+        data = {
+            "valid_key": 1.0,
+            "invalid@key!": 2.0,
+            "another/valid-key": 3.0,
+            "bad key#": 4.0,
+            "val-aux//reward/mean_at_1": 5.0,
+            "val-core///acc/best_at_5": 6.0,
+            "metric////with/many////slashes": 7.0,
+        }
         # Patch mlflow.log_metrics to capture the metrics actually sent
         with (
             patch("mlflow.log_metrics") as mock_log_metrics,
             patch.object(adapter, "logger") as mock_logger,
         ):
             adapter.log(data, step=5)
-            # Check that keys are sanitized
+            # Check that invalid characters are sanitized
             sent_metrics = mock_log_metrics.call_args[1]["metrics"]
             self.assertIn("invalid_at_key_", sent_metrics)  # @ becomes _at_, ! becomes _
             self.assertIn("bad key_", sent_metrics)  # # becomes _, space remains
             self.assertNotIn("invalid@key!", sent_metrics)
             self.assertNotIn("bad key#", sent_metrics)
-            # Check that a warning was logged for each sanitized key
+            # Check that consecutive slashes are collapsed to single slashes
+            self.assertIn("val-aux/reward/mean_at_1", sent_metrics)
+            self.assertIn("val-core/acc/best_at_5", sent_metrics)
+            self.assertIn("metric/with/many/slashes", sent_metrics)
+            self.assertNotIn("val-aux//reward/mean_at_1", sent_metrics)
+            self.assertNotIn("val-core///acc/best_at_5", sent_metrics)
+            # Check that warnings were logged for all sanitized keys
             warning_msgs = [str(call) for call in mock_logger.warning.call_args_list]
+            # Warnings for invalid characters
             self.assertTrue(any("invalid@key!" in msg and "invalid_at_key_" in msg for msg in warning_msgs))
             self.assertTrue(any("bad key#" in msg and "bad key_" in msg for msg in warning_msgs))
+            # Warnings for consecutive slashes
+            self.assertTrue(any("val-aux//reward/mean_at_1" in msg for msg in warning_msgs))
+            self.assertTrue(any("val-core///acc/best_at_5" in msg for msg in warning_msgs))
+            self.assertTrue(any("metric////with/many////slashes" in msg for msg in warning_msgs))
 
 
 if __name__ == "__main__":

verl/utils/tracking.py

@@ -278,13 +278,16 @@ def __init__(self):
         self._invalid_chars_pattern = re.compile(
             r"[^/\w.\- :]"
         )  # Allowed: slashes, alphanumerics, underscores, periods, dashes, colons, and spaces.
+        self._consecutive_slashes_pattern = re.compile(r"/+")
 
     def log(self, data, step):
         import mlflow
 
         def sanitize_key(key):
             # First replace @ with _at_ for backward compatibility
             sanitized = key.replace("@", "_at_")
+            # Replace consecutive slashes with a single slash (MLflow treats them as file paths)
+            sanitized = self._consecutive_slashes_pattern.sub("/", sanitized)
             # Then replace any other invalid characters with _
             sanitized = self._invalid_chars_pattern.sub("_", sanitized)
             if sanitized != key: