diff --git a/assets/js/source_lv_hooks.js b/assets/js/source_lv_hooks.js index 55c59d6e23..49a4baf459 100644 --- a/assets/js/source_lv_hooks.js +++ b/assets/js/source_lv_hooks.js @@ -26,7 +26,7 @@ hooks.SourceLogsSearchList = { const hook = this activateDelegatedTooltips(this.el, '[data-toggle="tooltip"]') - window.scrollTo(0, document.body.scrollHeight) + // window.scrollTo(0, document.body.scrollHeight) const observer = new IntersectionObserver((entries, observer) => { diff --git a/lib/logflare/logs/log_event.ex b/lib/logflare/logs/log_event.ex index e0bd3c796a..e6a001ef66 100644 --- a/lib/logflare/logs/log_event.ex +++ b/lib/logflare/logs/log_event.ex @@ -25,7 +25,6 @@ defmodule Logflare.LogEvent do field :body, :map, default: %{} field :valid, :boolean field :drop, :boolean, default: false - field :is_from_stale_query, :boolean field :timestamp_inferred, :boolean, default: false field :ingested_at, :utc_datetime_usec field :source_uuid, Ecto.UUID.Atom diff --git a/lib/logflare/logs/logs_search.ex b/lib/logflare/logs/logs_search.ex index f47d205fd4..4cfba12ccb 100644 --- a/lib/logflare/logs/logs_search.ex +++ b/lib/logflare/logs/logs_search.ex @@ -44,6 +44,7 @@ defmodule Logflare.Logs.Search do %{error: nil} = so <- apply_local_timestamp_correction(so), %{error: nil} = so <- apply_timestamp_filter_rules(so), %{error: nil} = so <- apply_filters(so), + %{error: nil} = so <- apply_cursor(so), %{error: nil} = so <- apply_select_rules(so), %{error: nil} = so <- do_query(so), %{error: nil} = so <- apply_warning_conditions(so), diff --git a/lib/logflare/logs/search_operation.ex b/lib/logflare/logs/search_operation.ex index 26acb1225c..f830f26d1d 100644 --- a/lib/logflare/logs/search_operation.ex +++ b/lib/logflare/logs/search_operation.ex @@ -35,8 +35,30 @@ defmodule Logflare.Logs.SearchOperation do field :chart_data_shape_id, atom(), default: nil, enforce: true field :type, :events | :aggregates field :status, {atom(), String.t() | [String.t()]} + field :event_page_request, event_page_request() + field :event_page_result, event_page_result() + field :has_more_events?, boolean(), default: false end + @type event_cursor :: %{timestamp: integer(), id: String.t()} + @type event_page_intent :: :within_range | :extend_previous | :extend_next + @type event_page_request :: %{ + intent: event_page_intent(), + boundary: integer() | DateTime.t() | NaiveDateTime.t() | nil, + cursor: event_cursor() | nil + } + @type event_page_result :: %{ + request: event_page_request(), + has_more?: boolean(), + cursor: event_cursor() | nil + } + + @spec event_page_direction(event_page_intent()) :: :previous | :next + def event_page_direction(:extend_next), do: :next + + def event_page_direction(intent) when intent in [:within_range, :extend_previous], + do: :previous + def new(params) do so = struct(__MODULE__, params) diff --git a/lib/logflare/logs/search_operations.ex b/lib/logflare/logs/search_operations.ex index 6f4c54387d..dacddfab9e 100644 --- a/lib/logflare/logs/search_operations.ex +++ b/lib/logflare/logs/search_operations.ex @@ -47,6 +47,32 @@ defmodule Logflare.Logs.SearchOperations do @spec max_chart_ticks :: integer() def max_chart_ticks, do: @default_max_n_chart_ticks + @spec default_limit :: pos_integer() + def default_limit, do: @default_limit + + @spec fetch_limit :: pos_integer() + def fetch_limit, do: @default_limit + 1 + + @spec event_page_params(SO.t(), SO.event_page_intent(), SO.event_cursor() | nil) :: + {:ok, %{event_page_request: SO.event_page_request(), tailing?: false}} | :error + def event_page_params(%SO{tailing?: false} = so, intent, cursor) + when (intent == :within_range and is_map(cursor)) or + (intent in [:extend_previous, :extend_next] and (is_map(cursor) or is_nil(cursor))) do + with {:ok, boundary} <- event_page_boundary(so, intent, cursor) do + {:ok, + %{ + event_page_request: %{ + intent: intent, + boundary: boundary, + cursor: cursor + }, + tailing?: false + }} + end + end + + def event_page_params(%SO{}, _intent, _cursor), do: :error + @spec do_query(SO.t()) :: SO.t() def do_query(%SO{} = so) do with {:ok, response} <- execute_backend_query(so) do @@ -122,15 +148,109 @@ defmodule Logflare.Logs.SearchOperations do @spec apply_query_defaults(SO.t()) :: SO.t() def apply_query_defaults(%SO{} = so) do + intent = if so.event_page_request, do: so.event_page_request.intent, else: :within_range + direction = SO.event_page_direction(intent) + query = from(table_name(so)) |> select(%{}) - |> order_by([t], desc: t.timestamp) - |> limit(@default_limit) + |> order_events(direction) + |> limit(^fetch_limit()) %{so | query: query} end + defp event_page_boundary(%SO{} = so, intent, cursor) + when intent in [:extend_previous, :extend_next] and + (is_map(cursor) or intent == :extend_next) do + case event_range_bounds(so) do + {:ok, bounds} -> {:ok, Map.fetch!(bounds, event_page_boundary_key(intent))} + :error -> {:ok, nil} + end + end + + defp event_page_boundary(%SO{} = so, intent, _cursor) do + with {:ok, bounds} <- event_range_bounds(so) do + boundary = + case intent do + :within_range -> nil + intent -> Map.fetch!(bounds, event_page_boundary_key(intent)) + end + + {:ok, boundary} + end + end + + defp event_page_boundary_key(:extend_previous), do: :min + defp event_page_boundary_key(:extend_next), do: :max + + defp event_range_bounds(%SO{} = so) do + if bounded_timestamp_filters?(so.lql_ts_filters) do + bounds = + SearchOperationHelpers.get_min_max_filter_timestamps( + so.lql_ts_filters, + chart_period(so) + ) + + {:ok, Map.take(bounds, [:min, :max])} + else + :error + end + end + + defp bounded_timestamp_filters?(filters) do + Enum.any?(filters, &(&1.operator == :range and length(&1.values || []) == 2)) or + (Enum.any?(filters, &(&1.operator in [:>, :>=])) and + Enum.any?(filters, &(&1.operator in [:<, :<=]))) + end + + defp order_events(query, :previous), + do: order_by(query, [t], desc: t.timestamp, desc: t.id) + + defp order_events(query, :next), do: order_by(query, [t], asc: t.timestamp, asc: t.id) + + @spec apply_cursor(SO.t()) :: SO.t() + def apply_cursor(%SO{event_page_request: %{cursor: nil}} = so), do: so + + def apply_cursor(%SO{event_page_request: %{intent: intent, cursor: cursor}} = so) do + %{timestamp: timestamp, id: id} = cursor + timestamp = normalize_event_timestamp(so.backend_type, timestamp) + direction = SO.event_page_direction(intent) + + %{so | query: where(so.query, ^cursor_condition(direction, timestamp, id))} + end + + def apply_cursor(%SO{} = so), do: so + + defp normalize_event_timestamp(:postgres, timestamp) when is_integer(timestamp), + do: DateTime.from_unix!(timestamp, :microsecond) + + defp normalize_event_timestamp(_backend_type, timestamp), do: timestamp + + defp cursor_condition(:previous, timestamp, id) when is_integer(timestamp) do + dynamic( + [t], + t.timestamp < fragment("TIMESTAMP_MICROS(?)", ^timestamp) or + (t.timestamp == fragment("TIMESTAMP_MICROS(?)", ^timestamp) and t.id < ^id) + ) + end + + defp cursor_condition(:next, timestamp, id) when is_integer(timestamp) do + dynamic( + [t], + t.timestamp > fragment("TIMESTAMP_MICROS(?)", ^timestamp) or + (t.timestamp == fragment("TIMESTAMP_MICROS(?)", ^timestamp) and t.id > ^id) + ) + end + + defp cursor_condition(:previous, timestamp, id) do + dynamic([t], t.timestamp < ^timestamp or (t.timestamp == ^timestamp and t.id < ^id)) + end + + defp cursor_condition(:next, timestamp, id) do + dynamic([t], t.timestamp > ^timestamp or (t.timestamp == ^timestamp and t.id > ^id)) + end + @spec apply_halt_conditions(SO.t()) :: SO.t() def apply_halt_conditions(%SO{} = so) do chart_period = chart_period(so) @@ -239,6 +359,11 @@ defmodule Logflare.Logs.SearchOperations do %{so | rows: rows} end + def apply_timestamp_filter_rules(%SO{type: :events, event_page_request: %{intent: intent}} = so) + when intent in [:extend_previous, :extend_next] do + apply_event_page_boundary(so) + end + def apply_timestamp_filter_rules(%SO{backend_type: :postgres, type: :events} = so) do %{so | query: apply_postgres_event_timestamp_filter_rules(so)} end @@ -342,6 +467,82 @@ defmodule Logflare.Logs.SearchOperations do %{so | query: q} end + defp apply_event_page_boundary( + %SO{ + event_page_request: %{intent: intent, boundary: nil, cursor: %{timestamp: timestamp}} + } = so + ) do + direction = SO.event_page_direction(intent) + timestamp = normalize_event_timestamp(so.backend_type, timestamp) + + %{so | query: apply_event_page_partition_filter(so.query, so, direction, timestamp)} + end + + defp apply_event_page_boundary(%SO{event_page_request: %{boundary: nil}} = so), do: so + + defp apply_event_page_boundary( + %SO{ + event_page_request: %{intent: intent, boundary: boundary, cursor: %{}}, + query: query + } = so + ) do + boundary = normalize_event_timestamp(so.backend_type, boundary) + direction = SO.event_page_direction(intent) + %{so | query: apply_event_page_partition_filter(query, so, direction, boundary)} + end + + defp apply_event_page_boundary( + %SO{event_page_request: %{intent: intent, boundary: boundary}} = so + ) do + boundary = normalize_event_timestamp(so.backend_type, boundary) + direction = SO.event_page_direction(intent) + query = where(so.query, ^boundary_condition(direction, boundary)) + + %{so | query: apply_event_page_partition_filter(query, so, direction, boundary)} + end + + defp boundary_condition(:previous, boundary) when is_integer(boundary), + do: dynamic([t], t.timestamp < fragment("TIMESTAMP_MICROS(?)", ^boundary)) + + defp boundary_condition(:next, boundary) when is_integer(boundary), + do: dynamic([t], t.timestamp > fragment("TIMESTAMP_MICROS(?)", ^boundary)) + + defp boundary_condition(:previous, boundary), do: dynamic([t], t.timestamp < ^boundary) + defp boundary_condition(:next, boundary), do: dynamic([t], t.timestamp > ^boundary) + + defp apply_event_page_partition_filter(query, %SO{backend_type: :postgres}, _, _), do: query + + defp apply_event_page_partition_filter( + query, + %SO{partition_by: :timestamp}, + direction, + boundary + ) do + date = boundary |> event_boundary_datetime() |> DateTime.to_date() + + case direction do + :previous -> where(query, [t], fragment("EXTRACT(DATE FROM ?)", t.timestamp) <= ^date) + :next -> where(query, [t], fragment("EXTRACT(DATE FROM ?)", t.timestamp) >= ^date) + end + end + + defp apply_event_page_partition_filter(query, %SO{partition_by: :pseudo}, direction, boundary) do + date = boundary |> event_boundary_datetime() |> DateTime.to_date() + + case direction do + :previous -> where(query, partition_date() <= ^date or in_streaming_buffer()) + :next -> where(query, partition_date() >= ^date or in_streaming_buffer()) + end + end + + defp event_boundary_datetime(timestamp) when is_integer(timestamp), + do: DateTime.from_unix!(timestamp, :microsecond) + + defp event_boundary_datetime(%DateTime{} = timestamp), do: timestamp + + defp event_boundary_datetime(%NaiveDateTime{} = timestamp), + do: DateTime.from_naive!(timestamp, "Etc/UTC") + defp apply_bq_aggregate_timestamp_filters(query, so, filters, chart_period) do period = to_bq_interval_token(chart_period) tick_count = SearchOperationHelpers.default_period_tick_count(chart_period) diff --git a/lib/logflare/logs/search_query_executor.ex b/lib/logflare/logs/search_query_executor.ex index 2c7ef8251c..b0d914c235 100644 --- a/lib/logflare/logs/search_query_executor.ex +++ b/lib/logflare/logs/search_query_executor.ex @@ -11,6 +11,7 @@ defmodule Logflare.Logs.SearchQueryExecutor do alias Logflare.LogEvent alias Logflare.Logs.Search alias Logflare.Logs.SearchOperation, as: SO + alias Logflare.Logs.SearchOperations alias Logflare.Utils.Tasks @query_timeout 30_000 @@ -37,11 +38,19 @@ defmodule Logflare.Logs.SearchQueryExecutor do end def query(pid, params) do - GenServer.call(pid, {:query, params}, @query_timeout) + GenServer.call(pid, {:query, query_params(params)}, @query_timeout) + end + + def query_page(pid, params, intent, cursor) do + GenServer.call( + pid, + {:query_page, query_params(params), intent, cursor}, + @query_timeout + ) end def query_agg(pid, params) do - GenServer.call(pid, {:query_agg, params}, @query_timeout) + GenServer.call(pid, {:query_agg, query_params(params)}, @query_timeout) end def cancel_agg(pid) do @@ -75,6 +84,25 @@ defmodule Logflare.Logs.SearchQueryExecutor do {:reply, :ok, %{state | event_task: {new_ref, new_params}}} end + def handle_call({:query_page, params, intent, cursor}, {lv_pid, _ref}, state) do + search_op = SO.new(params) + + case SearchOperations.event_page_params(search_op, intent, cursor) do + {:ok, page_params} -> + {ref, _params} = state.event_task + + if ref, do: Task.shutdown(ref, :brutal_kill) + + search_op = struct(search_op, page_params) + new_ref = start_search_task(lv_pid, search_op) + + {:reply, :ok, %{state | event_task: {new_ref, page_params}}} + + :error -> + {:reply, :error, state} + end + end + def handle_call({:query_agg, new_params}, {lv_pid, _ref}, state) do {ref, _params} = state.agg_task @@ -118,61 +146,40 @@ defmodule Logflare.Logs.SearchQueryExecutor do end @impl true - def handle_info({_ref, {:search_result, lv_pid, %{events: events_so}}}, state) do - Logger.debug( - "SearchQueryExecutor: Getting search events for #{pid_to_string(lv_pid)} / #{state.source_id} source..." - ) - - {_ref, params} = state.event_task - - rows = Enum.map(events_so.rows, &LogEvent.make_from_db(&1, %{source: params.source})) - - old_rows = if params.search_op_log_events, do: params.search_op_log_events.rows, else: [] - - # prevents removal of log events loaded - # during initial tailing query - log_events = - old_rows - |> Enum.reject(& &1.is_from_stale_query) - |> Enum.concat(rows) - |> Enum.uniq_by(&{&1.body, &1.id}) - |> Enum.sort_by(& &1.body["timestamp"], &>=/2) - |> Enum.take(100) - - send( - state.caller, - {:search_result, - %{ - events: %{events_so | rows: log_events} - }} - ) - - {:noreply, %{state | event_task: {nil, nil}}} + def handle_info({ref, {:search_result, lv_pid, %{events: events_so}}}, state) do + if active_task?(state.event_task, ref) do + handle_event_result(lv_pid, events_so, state) + else + {:noreply, state} + end end @impl true - def handle_info({_ref, {:search_result, lv_pid, %{aggregates: aggregates_so}}}, state) do - Logger.debug( - "SearchQueryExecutor: Getting search aggregates for #{pid_to_string(lv_pid)} / #{state.source_id} source..." - ) - - {_ref, _params} = state.agg_task - - send( - state.caller, - {:search_result, - %{ - aggregates: aggregates_so - }} - ) - - {:noreply, %{state | agg_task: {nil, nil}}} + def handle_info({ref, {:search_result, lv_pid, %{aggregates: aggregates_so}}}, state) do + if active_task?(state.agg_task, ref) do + handle_aggregate_result(lv_pid, aggregates_so, state) + else + {:noreply, state} + end end @impl true - def handle_info({_ref, {:search_error, _lv_pid, %SO{} = search_op}}, state) do - send(state.caller, {:search_error, search_op}) - {:noreply, state} + def handle_info({ref, {:search_error, _lv_pid, %SO{type: :aggregates} = search_op}}, state) do + if active_task?(state.agg_task, ref) do + send(state.caller, {:search_error, search_op}) + {:noreply, %{state | agg_task: {nil, nil}}} + else + {:noreply, state} + end + end + + def handle_info({ref, {:search_error, _lv_pid, %SO{} = search_op}}, state) do + if active_task?(state.event_task, ref) do + send(state.caller, {:search_error, search_op}) + {:noreply, %{state | event_task: {nil, nil}}} + else + {:noreply, state} + end end # handles task shutdown messages @@ -188,9 +195,53 @@ defmodule Logflare.Logs.SearchQueryExecutor do {:noreply, state} end + defp handle_event_result(lv_pid, events_so, state) do + Logger.debug( + "SearchQueryExecutor: Getting search events for #{pid_to_string(lv_pid)} / #{state.source_id} source..." + ) + + page_size = SearchOperations.default_limit() + raw_rows = events_so.rows + has_sentinel_row? = has_sentinel_row?(raw_rows) + + page_rows = + raw_rows + |> Enum.take(page_size) + |> Enum.map(&LogEvent.make_from_db(&1, %{source: events_so.source})) + |> uniq_sort_log_events() + + event_page_result = + event_page_result(events_so.event_page_request, page_rows, has_sentinel_row?) + + events_so = %{ + events_so + | rows: page_rows, + has_more_events?: has_sentinel_row?, + event_page_result: event_page_result + } + + send(state.caller, {:search_result, %{events: events_so}}) + + {:noreply, %{state | event_task: {nil, nil}}} + end + + defp handle_aggregate_result(lv_pid, aggregates_so, state) do + Logger.debug( + "SearchQueryExecutor: Getting search aggregates for #{pid_to_string(lv_pid)} / #{state.source_id} source..." + ) + + send(state.caller, {:search_result, %{aggregates: aggregates_so}}) + + {:noreply, %{state | agg_task: {nil, nil}}} + end + + def start_search_task(lv_pid, %SO{} = so), do: do_start_search_task(lv_pid, so) + def start_search_task(lv_pid, params) do - so = SO.new(params) + do_start_search_task(lv_pid, SO.new(params)) + end + defp do_start_search_task(lv_pid, so) do Tasks.async(fn -> so |> Search.search() @@ -219,4 +270,46 @@ defmodule Logflare.Logs.SearchQueryExecutor do end end) end + + defp uniq_sort_log_events(log_events) do + log_events + |> Enum.uniq_by(&{&1.body["timestamp"], event_id(&1)}) + |> Enum.sort_by(&{&1.body["timestamp"], event_id(&1)}, :desc) + end + + defp has_sentinel_row?(rows), do: length(rows) >= SearchOperations.fetch_limit() + + defp log_event_cursor(%LogEvent{} = event) do + %{timestamp: event.body["timestamp"], id: event_id(event)} + end + + defp event_page_result(nil, _rows, _has_more?), do: nil + + defp event_page_result(request, rows, has_more?) do + %{ + request: request, + has_more?: has_more?, + cursor: page_edge_cursor(rows, SO.event_page_direction(request.intent)) + } + end + + defp page_edge_cursor([], _direction), do: nil + defp page_edge_cursor(rows, :previous), do: rows |> List.last() |> log_event_cursor() + defp page_edge_cursor(rows, :next), do: rows |> List.first() |> log_event_cursor() + + defp active_task?({%Task{ref: ref}, _params}, ref), do: true + defp active_task?({ref, _params}, ref) when is_reference(ref), do: true + defp active_task?(_task, _ref), do: false + + defp event_id(%LogEvent{id: id, body: body}), do: id || body["id"] + + defp query_params(params) do + Map.drop(params, [ + :range_extension_patch, + :search_op, + :search_op_log_events, + :search_op_log_aggregates, + :streams + ]) + end end diff --git a/lib/logflare/lql/rules.ex b/lib/logflare/lql/rules.ex index 4d65938e9d..cf29d64fbf 100644 --- a/lib/logflare/lql/rules.ex +++ b/lib/logflare/lql/rules.ex @@ -16,6 +16,8 @@ defmodule Logflare.Lql.Rules do @type lql_rule :: ChartRule.t() | FilterRule.t() | FromRule.t() | SelectRule.t() @type lql_rules :: [lql_rule()] + @type timestamp_extension_direction :: :previous | :next + @type timestamp_value :: integer() | Date.t() | DateTime.t() | NaiveDateTime.t() # ============================================================================= # Rule Type Extraction @@ -276,6 +278,50 @@ defmodule Logflare.Lql.Rules do |> Enum.concat(new_timestamp_rules) end + @doc """ + Replaces timestamp filters with an absolute range extended to an event timestamp. + + The previous direction replaces the lower edge, while the next direction + replaces the upper edge. All other rules are preserved. + """ + @spec extend_timestamp_range( + lql_rules(), + timestamp_extension_direction(), + timestamp_value(), + Calendar.time_zone() + ) :: lql_rules() + def extend_timestamp_range(lql_rules, direction, event_timestamp, timezone \\ "Etc/UTC") + when is_list(lql_rules) and direction in [:previous, :next] and is_binary(timezone) do + timestamp_values = + lql_rules + |> get_timestamp_filters() + |> Enum.flat_map(×tamp_values/1) + |> Enum.map(&normalize_timestamp/1) + + case timestamp_values do + [] -> + lql_rules + + timestamp_values -> + event_timestamp = normalize_timestamp(event_timestamp, timezone) + + values = + case direction do + :previous -> [event_timestamp, Enum.max(timestamp_values)] + :next -> [Enum.min(timestamp_values), event_timestamp] + end + + timestamp_rule = + FilterRule.build( + path: "timestamp", + operator: :range, + values: values + ) + + update_timestamp_rules(lql_rules, [timestamp_rule]) + end + end + @doc """ Creates new timestamp filters by jumping forward or backward in time. @@ -299,6 +345,40 @@ defmodule Logflare.Lql.Rules do FilterRule.shorthand_timestamp?(filter_rule) end + defp timestamp_values(%FilterRule{value: value, values: values}) do + [value | List.wrap(values)] + |> Enum.reject(&is_nil/1) + end + + defp normalize_timestamp(timestamp) when is_integer(timestamp) do + timestamp + |> DateTime.from_unix!(:microsecond) + |> DateTime.to_naive() + end + + defp normalize_timestamp(%DateTime{} = timestamp) do + timestamp + |> DateTime.shift_zone!("Etc/UTC") + |> DateTime.to_naive() + end + + defp normalize_timestamp(%NaiveDateTime{} = timestamp), do: timestamp + defp normalize_timestamp(%Date{} = timestamp), do: NaiveDateTime.new!(timestamp, ~T[00:00:00]) + + defp normalize_timestamp(timestamp, timezone) when is_integer(timestamp) do + timestamp + |> DateTime.from_unix!(:microsecond) + |> normalize_timestamp(timezone) + end + + defp normalize_timestamp(%DateTime{} = timestamp, timezone) do + timestamp + |> DateTime.shift_zone!(timezone) + |> DateTime.to_naive() + end + + defp normalize_timestamp(timestamp, _timezone), do: normalize_timestamp(timestamp) + # ============================================================================= # LQL Parser Warnings # ============================================================================= diff --git a/lib/logflare_web/live/log_event_live.ex b/lib/logflare_web/live/log_event_live.ex index 3c2d22688f..e8fc8198a5 100644 --- a/lib/logflare_web/live/log_event_live.ex +++ b/lib/logflare_web/live/log_event_live.ex @@ -24,6 +24,8 @@ defmodule LogflareWeb.LogEventLive do lql = params["lql"] || "" is_tailing = params["tailing?"] == "true" + search_timezone = params["tz"] || preferred_timezone(socket.assigns) + opts = [ source: source, @@ -49,7 +51,7 @@ defmodule LogflareWeb.LogEventLive do |> assign(:log_event_id, params["uuid"]) |> assign(:lql, lql) |> assign(:tailing?, is_tailing) - |> assign(:tz, params["tz"]) + |> assign(:search_timezone, search_timezone) |> assign(:timestamp, timestamp) {:ok, socket} @@ -68,4 +70,15 @@ defmodule LogflareWeb.LogEventLive do defp maybe_put_timestamp(opts, timestamp), do: Keyword.put(opts, :timestamp, DateTime.truncate(timestamp, :second)) + + @spec preferred_timezone(map()) :: String.t() + defp preferred_timezone(%{team_user: %{preferences: %{timezone: timezone}}}) + when is_binary(timezone), + do: timezone + + defp preferred_timezone(%{user: %{preferences: %{timezone: timezone}}}) + when is_binary(timezone), + do: timezone + + defp preferred_timezone(_assigns), do: "Etc/UTC" end diff --git a/lib/logflare_web/live/log_event_live/search_log_event_viewer_component.ex b/lib/logflare_web/live/log_event_live/search_log_event_viewer_component.ex index 33d8a0c3ae..0ef4c9285d 100644 --- a/lib/logflare_web/live/log_event_live/search_log_event_viewer_component.ex +++ b/lib/logflare_web/live/log_event_live/search_log_event_viewer_component.ex @@ -37,7 +37,7 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do params = event_params(assigns) |> Map.merge(%{log_event_id: id, timestamp: d}) - |> Map.put(:lql, assigns.params["lql"] || "") + |> Map.put(:lql, assigns.lql) socket = socket @@ -94,11 +94,7 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do @impl true def render(%{source: source, log_event: %LE{body: body} = le} = assigns) do - tz = - if assigns.team_user, - do: Map.get(assigns.team_user.preferences || %{}, :timezone, "Etc/UTC"), - else: Map.get(assigns.user.preferences || %{}, :timezone, "Etc/UTC") - + tz = assigns.search_timezone timestamp = Timex.from_unix(body["timestamp"], :microsecond) local_timestamp = @@ -110,7 +106,7 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do LogView.render("log_event_body.html", source: source, source_schema_flat_map: assigns.source_schema_flat_map, - search_params: assigns.search_params, + search_params: %{"tz" => tz}, team: assigns.team, body: body, fmt_body: BqSchema.encode_metadata(body), @@ -119,8 +115,7 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do lql: assigns.lql, lql_schema: get_lql_schema(source), timestamp: timestamp, - local_timezone: tz, - search_timezone: assigns.search_params["tz"] || tz, + local_timezone: assigns.search_timezone, local_timestamp: local_timestamp ) end @@ -136,13 +131,12 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do team = socket.assigns[:team] || assigns[:team] source = socket.assigns[:source] || assigns[:source] timestamp = socket.assigns[:timestamp] || assigns[:timestamp] - lql = socket.assigns[:lql] || assigns[:lql] || assigns.params["lql"] || "" + lql = assigns[:lql] || socket.assigns[:lql] || "" source_schema_flat_map = socket.assigns[:source_schema_flat_map] || assigns[:source_schema_flat_map] - search_params = - socket.assigns[:search_params] || extract_search_params(assigns) + search_timezone = assigns[:search_timezone] || socket.assigns[:search_timezone] || "Etc/UTC" socket |> assign(:user, user) @@ -152,7 +146,7 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do |> assign(:timestamp, timestamp) |> assign(:lql, lql) |> assign(:source_schema_flat_map, source_schema_flat_map) - |> assign(:search_params, search_params) + |> assign(:search_timezone, search_timezone) |> assign(:error, nil) end @@ -172,9 +166,4 @@ defmodule LogflareWeb.Search.LogEventViewerComponent do _ -> SchemaBuilder.initial_table_schema() end end - - defp extract_search_params(%{params: params}) when is_map(params), - do: Map.take(params, ["tz"]) - - defp extract_search_params(_assigns), do: %{} end diff --git a/lib/logflare_web/live/search_live/event_context_component.ex b/lib/logflare_web/live/search_live/event_context_component.ex index f678c0e327..0c1a041180 100644 --- a/lib/logflare_web/live/search_live/event_context_component.ex +++ b/lib/logflare_web/live/search_live/event_context_component.ex @@ -14,18 +14,19 @@ defmodule LogflareWeb.SearchLive.EventContextComponent do %{ params: %{ "log-event-timestamp" => log_timestamp, - "log-event-id" => log_event_id, - "querystring" => query_string, - "source-id" => source_id, - "timezone" => timezone - } + "log-event-id" => log_event_id + }, + querystring: query_string, + search_timezone: timezone, + source: source } = assigns + source_id = source.id + event_timestamp = log_timestamp |> String.to_integer() |> Timex.from_unix(:microsecond) lql_rules = - Sources.get_source_for_lv_param(source_id) - |> prepare_lql_rules(query_string, event_timestamp) + prepare_lql_rules(source, query_string, event_timestamp) {:ok, socket @@ -33,7 +34,7 @@ defmodule LogflareWeb.SearchLive.EventContextComponent do |> assign(target_event_id: log_event_id, timezone: timezone) |> assign(is_truncated_before: false) |> assign(is_truncated_after: false) - |> assign(source: Sources.get_source_for_lv_param(source_id)) + |> assign(source: source) |> assign(:logs, AsyncResult.loading()) |> start_async(:logs, fn -> search_logs(log_event_id, event_timestamp, source_id, lql_rules) diff --git a/lib/logflare_web/live/search_live/form_components.ex b/lib/logflare_web/live/search_live/form_components.ex index dfeeb35893..d9d3cc5f23 100644 --- a/lib/logflare_web/live/search_live/form_components.ex +++ b/lib/logflare_web/live/search_live/form_components.ex @@ -135,7 +135,6 @@ defmodule LogflareWeb.SearchLive.FormComponents do attr :uri_params, :map, required: true attr :lql_rules, :list, required: true attr :user, Logflare.User, required: true - attr :search_op_log_events, :any, default: nil attr :search_op_log_aggregates, :any, default: nil attr :has_results?, :boolean attr :source, Logflare.Sources.Source, required: true diff --git a/lib/logflare_web/live/search_live/log_event_components.ex b/lib/logflare_web/live/search_live/log_event_components.ex index b6da8d20e3..21d5f8341b 100644 --- a/lib/logflare_web/live/search_live/log_event_components.ex +++ b/lib/logflare_web/live/search_live/log_event_components.ex @@ -21,22 +21,47 @@ defmodule LogflareWeb.SearchLive.LogEventComponents do @default_empty_event_message "(empty event message)" attr :search_op_log_events, :map, default: nil + attr :search_op_log_aggregates, :map, default: nil + attr :log_events, :any, default: [] attr :last_query_completed_at, :any, default: nil attr :loading, :boolean, required: true + attr :pagination_available?, :boolean, default: false + attr :unbounded_pagination?, :boolean, default: false + attr :event_page_loading, :atom, default: nil + attr :next_events_exhausted?, :boolean, default: false attr :search_timezone, :string, required: true - attr :tailing?, :boolean, required: true - attr :querystring, :string, required: true attr :empty_event_message_placeholder, :string, default: @default_empty_event_message attr :source_schema_flat_map, :map, default: %{} attr :search_op, Logflare.Logs.SearchOperation def results_list(assigns) do - assigns = assign(assigns, :select_fields, build_select_fields(assigns.search_op)) + assigns = + assigns + |> assign(:select_fields, build_select_fields(assigns.search_op)) + |> assign(:event_page_loading?, not is_nil(assigns.event_page_loading)) + |> assign( + :top_pagination_available?, + assigns.pagination_available? or + (assigns.unbounded_pagination? and + match?(%{has_more_events?: true}, assigns.search_op_log_events)) + ) ~H""" -
-