[Core] Make Preserving Proto Naming During Event JSON Conversion Configurable (#57705)

Recently, when we ran performance tests with task event generation
turned on. We saw some performance regression when the workloads ran on
very small CPU machines. With further investigation, the overhead mainly
comes from the name format convention when converting the proto message
to JSON format payload in the aggregator agent.

This PR adds an env var for the config to control the name conversion
behavior and update the corresponding tests.

Also note that, eventually we are planning to remove this config turn
off the field name conversion by default after migrated all the current
event usage.

---------

Signed-off-by: Mengjin Yan <mengjinyan3@gmail.com>
Signed-off-by: elliot-barn <elliot.barnwell@anyscale.com>

Author: MengjinYan

python/ray/dashboard/modules/aggregator/aggregator_agent.py

@@ -66,6 +66,12 @@
 PUBLISH_EVENTS_TO_EXTERNAL_HTTP_SERVICE = ray_constants.env_bool(
     f"{env_var_prefix}_PUBLISH_EVENTS_TO_EXTERNAL_HTTP_SERVICE", True
 )
+# flag to control whether preserve the proto field name when converting the events to
+# JSON. If True, the proto field name will be preserved. If False, the proto field name
+# will be converted to camel case.
+PRESERVE_PROTO_FIELD_NAME = ray_constants.env_bool(
+    f"{env_var_prefix}_PRESERVE_PROTO_FIELD_NAME", False
+)
 
 
 class AggregatorAgent(
@@ -124,6 +130,7 @@ def __init__(self, dashboard_agent) -> None:
                     endpoint=self._events_export_addr,
                     executor=self._executor,
                     events_filter_fn=self._can_expose_event,
+                    preserve_proto_field_name=PRESERVE_PROTO_FIELD_NAME,
                 ),
                 event_buffer=self._event_buffer,
                 common_metric_tags=self._common_tags,

python/ray/dashboard/modules/aggregator/publisher/async_publisher_client.py

@@ -66,12 +66,14 @@ def __init__(
         executor: ThreadPoolExecutor,
         events_filter_fn: Callable[[object], bool],
         timeout: float = PUBLISHER_TIMEOUT_SECONDS,
+        preserve_proto_field_name: bool = False,
     ) -> None:
         self._endpoint = endpoint
         self._executor = executor
         self._events_filter_fn = events_filter_fn
         self._timeout = aiohttp.ClientTimeout(total=timeout)
         self._session = None
+        self._preserve_proto_field_name = preserve_proto_field_name
 
     async def publish(self, batch: PublishBatch) -> PublishStats:
         events_batch: list[events_base_event_pb2.RayEvent] = batch.events
@@ -89,7 +91,11 @@ async def publish(self, batch: PublishBatch) -> PublishStats:
             self._executor,
             lambda: [
                 json.loads(
-                    message_to_json(e, always_print_fields_with_no_presence=True)
+                    message_to_json(
+                        e,
+                        always_print_fields_with_no_presence=True,
+                        preserving_proto_field_name=self._preserve_proto_field_name,
+                    )
                 )
                 for e in filtered
             ],