chore: fix flaky tests, adjusts exit trapping to only certain process…

…es in SourceSup (#2134)

* chore: fix flaky schema check

* Revert "fix: reduce exit trapping and error logging on supervision tree termination"

This reverts commit e2fb447.

* chore: resolve flaky tests, remove unnecessary exit trapping

lib/logflare/source/bigquery/schema.ex

@@ -41,6 +41,7 @@ defmodule Logflare.Source.BigQuery.Schema do
 
     # TODO: remove source_id from metadata to reduce confusion
     Logger.metadata(source_id: args[:source_token], source_token: args[:source_token])
+    Process.flag(:trap_exit, true)
 
     persist(0)
 
@@ -84,15 +85,6 @@ defmodule Logflare.Source.BigQuery.Schema do
     end
   end
 
-  def terminate(reason, state) do
-    # Do Shutdown Stuff
-    Logger.info("Going Down - #{inspect(reason)} - #{__MODULE__}", %{
-      source_id: state.source_token
-    })
-
-    reason
-  end
-
   # TODO: remove, external procs should not have access to internal state
   def get_state(source_token) when is_atom(source_token) do
     with {:ok, pid} <- Backends.lookup(__MODULE__, source_token) do

lib/logflare/source/recent_logs_server.ex

@@ -97,8 +97,6 @@ defmodule Logflare.Source.RecentLogsServer do
   def init(args) do
     source = Keyword.get(args, :source)
 
-    # keep for legacy notifications on whether source tree dies.
-    # TODO: use process monitoring outside of supervision tree.
     Process.flag(:trap_exit, true)
     Logger.metadata(source_id: source.token, source_token: source.token)
 

lib/logflare/source/supervisor.ex

@@ -253,7 +253,7 @@ defmodule Logflare.Source.Supervisor do
         maybe_restart_mismatched_source_pipelines(source_token)
 
       %{v1: {:error, _}, v2: {:error, _}} ->
-        Logger.debug("Source.Supervisor - SourceSup not found, starting...",
+        Logger.info("Source.Supervisor - SourceSup not found, starting...",
           source_id: source_token,
           source_token: source_token
         )

test/logflare/backends_test.exs

@@ -176,7 +176,7 @@ defmodule Logflare.BackendsTest do
       ChannelTopics.subscribe_source(source_token)
       le = build(:log_event, source: source)
       assert {:ok, 1} = Backends.ingest_logs([le], source)
-      :timer.sleep(500)
+      :timer.sleep(1000)
 
       TestUtils.retry_assert(fn ->
         assert_received %_{event: "rate", payload: %{rate: _}}
@@ -217,6 +217,7 @@ defmodule Logflare.BackendsTest do
       assert {:ok, 0} = Backends.ingest_logs([le], source)
       # only the init message in RLS
       assert [_] = Backends.list_recent_logs_local(source)
+      :timer.sleep(1000)
     end
 
     test "route to source with lql", %{user: user} do
@@ -225,7 +226,7 @@ defmodule Logflare.BackendsTest do
       source = Logflare.Repo.preload(source, :rules, force: true)
       start_supervised!({SourceSup, source}, id: :source)
       start_supervised!({SourceSup, target}, id: :target)
-      :timer.sleep(1000)
+      :timer.sleep(500)
 
       assert {:ok, 2} =
                Backends.ingest_logs(
@@ -236,7 +237,7 @@ defmodule Logflare.BackendsTest do
                  source
                )
 
-      :timer.sleep(500)
+      :timer.sleep(1000)
       # init message + 2 events
       assert Backends.list_recent_logs_local(source) |> length() == 3
       # init message + 1 events
@@ -252,10 +253,10 @@ defmodule Logflare.BackendsTest do
       start_supervised!({SourceSup, source}, id: :source)
       start_supervised!({SourceSup, target}, id: :target)
       start_supervised!({SourceSup, other_target}, id: :other_target)
-      :timer.sleep(1000)
+      :timer.sleep(500)
 
       assert {:ok, 1} = Backends.ingest_logs([%{"event_message" => "testing 123"}], source)
-      :timer.sleep(500)
+      :timer.sleep(1000)
       # init message + 1 events
       assert Backends.list_recent_logs_local(source) |> length() == 2
       # init message + 1 events
@@ -299,6 +300,7 @@ defmodule Logflare.BackendsTest do
                )
 
       assert_receive ^ref, 2_000
+      :timer.sleep(1000)
     end
   end
 
@@ -337,6 +339,8 @@ defmodule Logflare.BackendsTest do
       TestUtils.retry_assert(fn ->
         assert_received {^ref, %{"event_message" => "some event"}}
       end)
+
+      :timer.sleep(1000)
     end
   end
 

test/logflare/logs/logs_test.exs

@@ -112,7 +112,8 @@ defmodule Logflare.LogsTest do
       ]
 
       assert :ok = Logs.ingest_logs(batch, source)
-      assert_receive :ok, 1000
+      assert_receive :ok, 1_500
+      :timer.sleep(1_500)
     end
   end
 

test/logflare/partners_test.exs

@@ -5,6 +5,7 @@ defmodule Logflare.PartnerTest do
   alias Logflare.Partners.PartnerUser
   alias Logflare.Repo
   alias Logflare.User
+  alias Logflare.Google.CloudResourceManager
 
   describe "get/1" do
     test "returns the partner with given id" do
@@ -81,6 +82,9 @@ defmodule Logflare.PartnerTest do
 
   describe "delete_user/2" do
     test "deletes user and removes association with partner" do
+      CloudResourceManager
+      |> expect(:set_iam_policy, fn -> nil end)
+
       partner = insert(:partner)
 
       {:ok, %{id: id} = user} = Partners.create_user(partner, %{"email" => TestUtils.gen_email()})

test/logflare/sql_test.exs

@@ -1,9 +1,10 @@
 defmodule Logflare.SqlTest do
   @moduledoc false
-  use Logflare.DataCase, async: false
+  use Logflare.DataCase
   alias Logflare.SingleTenant
   alias Logflare.Sql
   alias Logflare.Backends.Adaptor.PostgresAdaptor
+  alias Logflare.Backends.AdaptorSupervisor
   @logflare_project_id "logflare-project-id"
   @user_project_id "user-project-id"
   @user_dataset_id "user-dataset-id"
@@ -440,7 +441,7 @@ defmodule Logflare.SqlTest do
       source = insert(:source, user: user, name: "c.d.e")
       backend = insert(:backend, type: :postgres, sources: [source], config: config)
 
-      pid = start_supervised!({PostgresAdaptor, {source, backend}})
+      pid = start_supervised!({AdaptorSupervisor, {source, backend}})
 
       log_event =
         Logflare.LogEvent.make(

test/logflare_web/controllers/endpoints_controller_test.exs

@@ -4,6 +4,7 @@ defmodule LogflareWeb.EndpointsControllerTest do
   alias Logflare.SingleTenant
   alias Logflare.Backends
   alias Logflare.Source
+  alias Logflare.Sources
   alias Logflare.SystemMetrics.AllLogsLogged
 
   setup do
@@ -178,6 +179,8 @@ defmodule LogflareWeb.EndpointsControllerTest do
 
     test "GET a basic sandboxed query with from table", %{conn: initial_conn, user: user} do
       for source <- Logflare.Repo.all(Source) do
+        source = Sources.preload_defaults(source)
+
         Backends.ingest_logs(
           [%{"event_message" => "some message", "project" => "default"}],
           source

test/logflare_web/controllers/user_controller_test.exs

@@ -3,6 +3,7 @@ defmodule LogflareWeb.UserControllerTest do
   import LogflareWeb.Router.Helpers
   use LogflareWeb.ConnCase
   alias Logflare.Users
+  alias Logflare.Google.CloudResourceManager
 
   describe "UserController update" do
     setup do
@@ -123,6 +124,9 @@ defmodule LogflareWeb.UserControllerTest do
     end
 
     test "succeeds", %{conn: conn} do
+      CloudResourceManager
+      |> expect(:set_iam_policy, fn -> nil end)
+
       conn = delete(conn, ~p"/account")
       assert redirected_to(conn, 302) =~ ~p"/auth/login?user_deleted=true"
     end
@@ -137,6 +141,9 @@ defmodule LogflareWeb.UserControllerTest do
     end
 
     test "bug: should be able to delete a partner-provisioned account", %{conn: conn} do
+      CloudResourceManager
+      |> expect(:set_iam_policy, fn -> nil end)
+
       conn = delete(conn, ~p"/account")
       assert redirected_to(conn, 302) =~ ~p"/auth/login"
     end

test/test_helper.exs

@@ -3,6 +3,7 @@ Mix.Task.run("app.start")
 ExUnit.start()
 
 # Mimick mocks setup
+Mimic.copy(Logflare.Google.CloudResourceManager)
 Mimic.copy(Logflare.Mailer)
 Mimic.copy(Logflare.Logs)
 Mimic.copy(Logflare.Logs.LogEvents)