Merge branch 'OBS-1994' into 'enterprise'

feat(events): add case-insensitive message search to events page

See merge request dkinternal/observability/dataops-observability!63

Author: ci bot

common/entity_services/event_service.py

@@ -2,7 +2,7 @@
 
 from collections import defaultdict
 
-from peewee import JOIN
+from peewee import JOIN, fn
 
 from common.entities import (
     Component,
@@ -15,6 +15,7 @@
     RunTask,
     Task,
 )
+from common.entities.event import ApiEventType
 
 from .helpers import ListRules, Page, ProjectEventFilters
 
@@ -57,6 +58,20 @@ def get_events_with_rules(*, rules: ListRules, filters: ProjectEventFilters) ->
             filter_list.append(EventEntity.timestamp >= filters.date_range_start)
         if filters.date_range_end:
             filter_list.append(EventEntity.timestamp < filters.date_range_end)
+        if filters.search:
+            lowered_payload = fn.LOWER(EventEntity.v2_payload.cast("CHAR"))
+            search_term = f"%{filters.search.lower()}%"
+            json_search = lambda path: fn.JSON_SEARCH(lowered_payload, "one", search_term, None, path).is_null(False)
+            filter_list.append(
+                (
+                    (EventEntity.type == ApiEventType.MESSAGE_LOG)
+                    & (json_search("$.log_entries[*].message") | json_search("$.log_entries[*].message_details"))
+                )
+                | (
+                    (EventEntity.type == ApiEventType.METRIC_LOG)
+                    & (json_search("$.metric_entries[*].key") | json_search("$.metric_entries[*].value"))
+                )
+            )
 
         query = query.where(*filter_list)
         page = Page[EventEntity].get_paginated_results(query, EventEntity.timestamp, rules)

common/entity_services/helpers/filter_rules.py

@@ -40,6 +40,7 @@
 KEY_QUERY_NAME: str = "key"
 
 # List events endpoints
+SEARCH_QUERY_NAME: str = "search"
 EVENT_TYPE_QUERY_NAME: str = "event_type"
 EVENT_ID_QUERY_NAME: str = "event_id"
 JOURNEY_ID_QUERY_NAME: str = "journey_id"
@@ -111,6 +112,7 @@ def _normalize_tools(params: MultiDict, field_name: str) -> list[str]:
     PROJECT_ID_QUERY_NAME: ParamConfig("project_ids", MultiDict.getlist),
     RUN_ID_QUERY_NAME: ParamConfig("run_ids", MultiDict.getlist),
     RUN_KEY_QUERY_NAME: ParamConfig("run_keys", MultiDict.getlist),
+    SEARCH_QUERY_NAME: ParamConfig("search", MultiDict.get),
     START_RANGE_QUERY_NAME: ParamConfig("start_range", _date_or_none),
     START_RANGE_BEGIN_QUERY_NAME: ParamConfig("start_range_begin", _date_or_none),
     START_RANGE_END_QUERY_NAME: ParamConfig("start_range_end", _date_or_none),
@@ -149,6 +151,7 @@ class Filters:
     project_ids: list[str] = field(default_factory=list)
     run_ids: list[str] = field(default_factory=list)
     run_keys: list[str] = field(default_factory=list)
+    search: Optional[str] = None
     start_range: Optional[datetime] = None
     start_range_begin: Optional[datetime] = None
     start_range_end: Optional[datetime] = None
@@ -331,6 +334,7 @@ def from_params(cls, params: MultiDict, project_ids: list[UUID] | None = None) -
                 TASK_ID_QUERY_NAME,
                 DATE_RANGE_START_QUERY_NAME,
                 DATE_RANGE_END_QUERY_NAME,
+                SEARCH_QUERY_NAME,
             ],
         )
 

common/events/converters.py

@@ -231,6 +231,7 @@ def handle_message_log_v2(self, event: v2.MessageLogUserEvent) -> bool:
         self.v1_event = v1.MessageLogEvent(
             log_level=message.level.name,
             message=message.message,
+            message_details=message.message_details,
             event_type=v1.MessageLogEvent.__name__,
             **self._extract_common_attributes(event),
             **self._extract_component_data(
@@ -509,7 +510,11 @@ def handle_run_status(self, event: v1.RunStatusEvent) -> bool:
     def handle_message_log(self, event: v1.MessageLogEvent) -> bool:
         self.v2_event = v2.MessageLogUserEvent(
             event_payload=v2.MessageLog(
-                log_entries=[v2.LogEntry(level=v2.LogLevel[event.log_level], message=event.message)],
+                log_entries=[
+                    v2.LogEntry(
+                        level=v2.LogLevel[event.log_level], message=event.message, message_details=event.message_details
+                    )
+                ],
                 component=self._extract_component_data(event),
                 **self._extract_common_payload_attributes(event),
             ),

common/events/v1/message_log_event.py

@@ -30,6 +30,10 @@ class MessageLogEventBaseSchema(Schema):
         validate=not_empty(),
         metadata={"description": "Required. The body of the message to log.", "example": "The job has completed."},
     )
+    message_details = Str(
+        load_default=None,
+        metadata={"description": "Optional. Additional details about the message."},
+    )
 
 
 class MessageLogEventSchema(MessageLogEventBaseSchema, EventSchema):
@@ -46,6 +50,7 @@ class MessageLogEvent(Event):
 
     log_level: str
     message: str
+    message_details: str | None = None
 
     __schema__ = MessageLogEventSchema
     __api_schema__ = MessageLogEventApiSchema

common/events/v2/message_log.py

@@ -31,6 +31,7 @@ class LogLevel(std_Enum):
 class LogEntry:
     level: LogLevel
     message: str
+    message_details: str | None = None
 
 
 @dataclass
@@ -50,6 +51,10 @@ class LogEntrySchema(Schema):
         validate=not_empty(),
         metadata={"description": "Required. The body of the message to log.", "example": "The job has completed."},
     )
+    message_details = Str(
+        load_default=None,
+        metadata={"description": "Optional. Additional details about the message."},
+    )
 
     @post_load
     def to_dataclass(self, data: dict, **_: Any) -> LogEntry:

common/tests/unit/entity_services/helpers/test_filter_rules.py

@@ -167,6 +167,18 @@ def test_filters_invalid_instance_status():
         Filters.validate_instance_status(["RANDOM"])
 
 
+@pytest.mark.unit
+def test_project_event_filters_search():
+    filters = ProjectEventFilters.from_params(MultiDict([("search", "some query")]))
+    assert filters.search == "some query"
+
+
+@pytest.mark.unit
+def test_project_event_filters_search_empty():
+    filters = ProjectEventFilters.from_params(MultiDict([]))
+    assert filters.search is None
+
+
 @pytest.mark.unit
 def test_upcoming_instances_filters_valid():
     params = MultiDict(

common/tests/unit/events/test_converters.py

@@ -123,6 +123,47 @@ def test_convert_message_log_events(base_fields, id_fields, component_fields):
             assert event == to_v1(to_v2(event))
 
 
+@pytest.mark.unit
+def test_convert_message_log_with_message_details(base_fields, id_fields, component_fields):
+    """Test that message_details round-trips through v1 -> v2 -> v1 conversion."""
+    data = {
+        **{k: v[0] for k, v in base_fields.items()},
+        **{k: v[0] for k, v in id_fields.items()},
+        **{k: v[0] for k, v in component_fields["batch"].items()},
+        "log_level": MessageEventLogLevel.ERROR.name,
+        "message": "test message",
+        "message_details": "some extra details about the error",
+        "event_type": MessageLogEvent.__name__,
+    }
+    event = instantiate_event_from_data(data)
+    assert event.message_details == "some extra details about the error"
+    v2_event = to_v2(event)
+    assert v2_event.event_payload.log_entries[0].message_details == "some extra details about the error"
+    roundtrip = to_v1(v2_event)
+    assert roundtrip == event
+    assert roundtrip.message_details == "some extra details about the error"
+
+
+@pytest.mark.unit
+def test_convert_message_log_without_message_details(base_fields, id_fields, component_fields):
+    """Test that None message_details round-trips correctly."""
+    data = {
+        **{k: v[0] for k, v in base_fields.items()},
+        **{k: v[0] for k, v in id_fields.items()},
+        **{k: v[0] for k, v in component_fields["batch"].items()},
+        "log_level": MessageEventLogLevel.INFO.name,
+        "message": "test message",
+        "event_type": MessageLogEvent.__name__,
+    }
+    event = instantiate_event_from_data(data)
+    assert event.message_details is None
+    v2_event = to_v2(event)
+    assert v2_event.event_payload.log_entries[0].message_details is None
+    roundtrip = to_v1(v2_event)
+    assert roundtrip == event
+    assert roundtrip.message_details is None
+
+
 @pytest.mark.unit
 @pytest.mark.converters_slow
 def test_convert_metric_log_events(base_fields, id_fields, component_fields):

common/tests/unit/events/v1/test_message_log.py

@@ -80,6 +80,24 @@ def test_message_log_min_length_message(message_log_event_data):
         MessageLogEvent.from_dict(message_log_event_data)
 
 
+@pytest.mark.unit
+def test_message_log_with_message_details(unidentified_message_log_data):
+    unidentified_message_log_data["message_details"] = "Some extra details"
+    event = MessageLogEvent.as_event_from_request(unidentified_message_log_data)
+    assert event.message_details == "Some extra details"
+    data = event.as_dict()
+    assert data["message_details"] == "Some extra details"
+    restored = MessageLogEvent.from_dict(data)
+    assert restored.message_details == "Some extra details"
+
+
+@pytest.mark.unit
+def test_message_log_without_message_details(unidentified_message_log_data):
+    assert "message_details" not in unidentified_message_log_data
+    event = MessageLogEvent.as_event_from_request(unidentified_message_log_data)
+    assert event.message_details is None
+
+
 @pytest.mark.unit
 def test_message_log_validate_default_values(message_log_event_data):
     message_log_event_data.pop("task_key", None)

common/tests/unit/events/v2/test_message_log.py

@@ -48,6 +48,20 @@ def test_log_entries(default_base_payload_dict, message_log_data, log_entry, def
     assert MessageLogSchema().load(message_log_data) == expected_result
 
 
+@pytest.mark.unit
+def test_log_entry_with_message_details():
+    data = {"level": "WARNING", "message": "Something happened", "message_details": "Extra detail here"}
+    entry = LogEntrySchema().load(data)
+    assert entry == LogEntry(level=LogLevel.WARNING, message="Something happened", message_details="Extra detail here")
+
+
+@pytest.mark.unit
+def test_log_entry_without_message_details():
+    data = {"level": "INFO", "message": "No details"}
+    entry = LogEntrySchema().load(data)
+    assert entry.message_details is None
+
+
 @pytest.mark.unit
 def test_message_log_empty(message_log_data):
     del message_log_data["log_entries"]

observability_api/endpoints/v1/projects.py

@@ -248,6 +248,13 @@ def get(self, project_id: UUID) -> Response:
               type: string
               format: date
             required: false
+          - in: query
+            name: search
+            schema:
+              type: string
+            required: false
+            description: Optional. A case-insensitive search query. Matches MESSAGE_LOG events by message and
+                         message_details, and METRIC_LOG events by metric key and value.
           - in: query
             name: event_type
             description: Optional. If specified, the results will be limited to events with one of the specified event