feat: use feature provider invalidator worker (#72)

Update applications to refresh cached feature flags / machines when they
are updated. This was already the case in zebra and guard, and now we
add it to all the other applications consuming feature flags too.

Ref: renderedtext/project-tasks#2202

Author: lucaspin

ee/audit/lib/audit/application.ex

@@ -25,6 +25,7 @@ defmodule Audit.Application do
       {{Audit.Streamer.Scheduler, []}, enabled?("START_STREAMER")},
       {{Cachex, [name: Audit.Cache]}, true},
       {provider, System.get_env("FEATURE_YAML_PATH") != nil},
+      {{Audit.FeatureProviderInvalidatorWorker, []}, true},
       {Supervisor.child_spec({Cachex, :feature_provider_cache}, id: :feature_provider_cache),
        true}
     ]

ee/audit/lib/audit/feature_provider_invalidator_worker.ex

@@ -0,0 +1,44 @@
+defmodule Audit.FeatureProviderInvalidatorWorker do
+  require Logger
+
+  @doc """
+  This module consumes RabbitMQ feature and machine state change events
+  and invalidates features and machines caches.
+  """
+
+  use Tackle.Multiconsumer,
+    url: Application.get_env(:audit, :amqp_url),
+    service: "audit",
+    routes: [
+      {"feature_exchange", "features_changed", :features_changed},
+      {"feature_exchange", "organization_features_changed", :organization_features_changed}
+    ],
+    # This queue is used to consume events from the feature exchange.
+    # It is declared as non-durable, auto-delete and exclusive.
+    # This means that the queue will be deleted when the consumer disconnects.
+    # This is the desired behavior, because these events are used to invalidate pod-level caches.
+    queue: :dynamic,
+    queue_opts: [
+      durable: false,
+      auto_delete: true,
+      exclusive: true
+    ],
+    connection_id: Audit.FeatureProviderInvalidatorWorker
+
+  def features_changed(_message) do
+    log("invalidating features")
+    {:ok, _} = FeatureProvider.list_features(reload: true)
+    :ok
+  end
+
+  def organization_features_changed(message) do
+    event = InternalApi.Feature.OrganizationFeaturesChanged.decode(message)
+    log("invalidating features for org #{event.org_id}")
+    {:ok, _} = FeatureProvider.list_features(reload: true, param: event.org_id)
+    :ok
+  end
+
+  defp log(message) do
+    Logger.info("[FEATURE PROVIDER INVALIDATOR WORKER] #{message}")
+  end
+end

ee/audit/lib/internal_api/feature.pb.ex

@@ -221,6 +221,48 @@ defmodule InternalApi.Feature.Availability do
   field(:quantity, 2, type: :uint32)
 end
 
+defmodule InternalApi.Feature.MachinesChanged do
+  @moduledoc false
+  use Protobuf, syntax: :proto3
+  @type t :: %__MODULE__{}
+
+  defstruct []
+end
+
+defmodule InternalApi.Feature.OrganizationMachinesChanged do
+  @moduledoc false
+  use Protobuf, syntax: :proto3
+
+  @type t :: %__MODULE__{
+          org_id: String.t()
+        }
+
+  defstruct [:org_id]
+
+  field(:org_id, 1, type: :string)
+end
+
+defmodule InternalApi.Feature.FeaturesChanged do
+  @moduledoc false
+  use Protobuf, syntax: :proto3
+  @type t :: %__MODULE__{}
+
+  defstruct []
+end
+
+defmodule InternalApi.Feature.OrganizationFeaturesChanged do
+  @moduledoc false
+  use Protobuf, syntax: :proto3
+
+  @type t :: %__MODULE__{
+          org_id: String.t()
+        }
+
+  defstruct [:org_id]
+
+  field(:org_id, 1, type: :string)
+end
+
 defmodule InternalApi.Feature.FeatureService.Service do
   @moduledoc false
   use GRPC.Service, name: "InternalApi.Feature.FeatureService"

front/config/config.exs

@@ -29,8 +29,6 @@ config :front,
 
 config :front, default_user_name: "Semaphore User"
 
-config :front, :cache_settings, organization_features_ttl: :timer.minutes(15)
-
 config :feature_provider, provider: {Front.FeatureHubProvider, []}
 config :front, :superjerry_client, {Support.FakeClients.Superjerry, []}
 config :front, :scouter_client, {Front.Clients.Scouter, []}

front/config/test.exs

@@ -64,7 +64,6 @@ config :wallaby, max_wait_time: 10_000
 
 config :joken, current_time_adapter: Support.TimeMock
 
-config :front, :cache_settings, organization_features_ttl: 0
 config :front, guard_grpc_timeout: 1_000
 config :front, permission_patrol_timeout: 1_000
 

front/lib/front/application.ex

@@ -21,7 +21,8 @@ defmodule Front.Application do
         {Phoenix.PubSub, [name: Front.PubSub, adapter: Phoenix.PubSub.PG2]},
         FrontWeb.Endpoint,
         {Task.Supervisor, [name: Front.TaskSupervisor]},
-        Front.Tracing.Store
+        Front.Tracing.Store,
+        Front.FeatureProviderInvalidatorWorker
       ] ++ reactor() ++ cache() ++ telemetry() ++ feature_provider(provider)
 
     opts = [strategy: :one_for_one, name: Front.Supervisor]

front/lib/front/feature_hub_provider.ex

@@ -11,75 +11,21 @@ defmodule Front.FeatureHubProvider do
 
   import Front.Utils
 
-  @cache_version :crypto.hash(:md5, File.read(__ENV__.file) |> elem(1)) |> Base.encode64()
-
-  defp cache_key(org_id, operation),
-    do: "feature_hub/#{@cache_version}/#{org_id}/#{operation}"
-
-  @impl FeatureProvider.Provider
-  def provide_features(org_id, opts \\ []) do
-    use_cache? = Keyword.get(opts, :use_cache, true)
-
-    if use_cache? do
-      cache_fetch(org_id, "list_organization_features", fn ->
-        do_list_features(org_id, update_cache: true)
-      end)
-    else
-      do_list_features(org_id)
-    end
-  end
-
   @impl FeatureProvider.Provider
-  def provide_machines(org_id, opts \\ []) do
-    use_cache? = Keyword.get(opts, :use_cache, true)
-
-    if use_cache? do
-      cache_fetch(org_id, "list_organization_machines", fn ->
-        do_list_machines(org_id, update_cache: true)
-      end)
-    else
-      do_list_machines(org_id)
-    end
-  end
-
-  defp cache_fetch(org_id, operation, callback) do
-    cache_key(org_id, operation)
-    |> Front.Cache.get()
-    |> case do
-      {:ok, results} ->
-        Watchman.increment({"feature_hub.#{operation}.cache_hit", [org_id]})
-        ok(Front.Cache.decode(results))
-
-      {:not_cached, _} ->
-        Watchman.increment({"feature_hub.#{operation}.cache_miss", [org_id]})
-        callback.()
-    end
-  end
-
-  defp do_list_features(org_id, opts \\ []) do
-    update_cache = Keyword.get(opts, :update_cache, false)
-
+  def provide_features(org_id, _opts \\ []) do
     FeatureClient.list_organization_features(%{org_id: org_id})
     |> unwrap(fn response ->
       features =
         response.organization_features
         |> Enum.map(&feature_from_grpc/1)
         |> Enum.filter(&FeatureProvider.Feature.visible?/1)
 
-      if update_cache do
-        Front.Async.run(fn ->
-          cache_key(org_id, "list_organization_features")
-          |> Front.Cache.set(Front.Cache.encode(features), cache_ttl())
-        end)
-      end
-
       ok(features)
     end)
   end
 
-  defp do_list_machines(org_id, opts \\ []) do
-    update_cache = Keyword.get(opts, :update_cache, false)
-
+  @impl FeatureProvider.Provider
+  def provide_machines(org_id, _opts \\ []) do
     FeatureClient.list_organization_machines(%{org_id: org_id})
     |> unwrap(fn response ->
       machines =
@@ -88,13 +34,6 @@ defmodule Front.FeatureHubProvider do
         |> Enum.filter(&FeatureProvider.Machine.enabled?/1)
         |> order_by_machine_type()
 
-      if update_cache do
-        Front.Async.run(fn ->
-          cache_key(org_id, "list_organization_machines")
-          |> Front.Cache.set(Front.Cache.encode(machines), cache_ttl())
-        end)
-      end
-
       ok(machines)
     end)
   end
@@ -145,8 +84,4 @@ defmodule Front.FeatureHubProvider do
       |> String.split("-", parts: 2)
     end)
   end
-
-  defp cache_ttl do
-    Application.get_env(:front, :cache_settings)[:features_ttl]
-  end
 end

front/lib/front/feature_provider_invalidator_worker.ex

@@ -0,0 +1,59 @@
+defmodule Front.FeatureProviderInvalidatorWorker do
+  require Logger
+
+  @doc """
+  This module consumes RabbitMQ feature and machine state change events
+  and invalidates features and machines caches.
+  """
+
+  use Tackle.Multiconsumer,
+    url: Application.get_env(:front, :amqp_url),
+    service: "front",
+    routes: [
+      {"feature_exchange", "machines_changed", :machines_changed},
+      {"feature_exchange", "organization_machines_changed", :organization_machines_changed},
+      {"feature_exchange", "features_changed", :features_changed},
+      {"feature_exchange", "organization_features_changed", :organization_features_changed}
+    ],
+    # This queue is used to consume events from the feature exchange.
+    # It is declared as non-durable, auto-delete and exclusive.
+    # This means that the queue will be deleted when the consumer disconnects.
+    # This is the desired behavior, because these events are used to invalidate pod-level caches.
+    queue: :dynamic,
+    queue_opts: [
+      durable: false,
+      auto_delete: true,
+      exclusive: true
+    ],
+    connection_id: Front.FeatureProviderInvalidatorWorker
+
+  def machines_changed(_message) do
+    log("invalidating machines")
+    {:ok, _} = FeatureProvider.list_machines(reload: true)
+    :ok
+  end
+
+  def organization_machines_changed(message) do
+    event = InternalApi.Feature.OrganizationMachinesChanged.decode(message)
+    log("invalidating machines for org #{event.org_id}")
+    {:ok, _} = FeatureProvider.list_machines(reload: true, param: event.org_id)
+    :ok
+  end
+
+  def features_changed(_message) do
+    log("invalidating features")
+    {:ok, _} = FeatureProvider.list_features(reload: true)
+    :ok
+  end
+
+  def organization_features_changed(message) do
+    event = InternalApi.Feature.OrganizationFeaturesChanged.decode(message)
+    log("invalidating features for org #{event.org_id}")
+    {:ok, _} = FeatureProvider.list_features(reload: true, param: event.org_id)
+    :ok
+  end
+
+  defp log(message) do
+    Logger.info("[FEATURE PROVIDER INVALIDATOR WORKER] #{message}")
+  end
+end

github_notifier/lib/github_notifier/application.ex

@@ -22,6 +22,7 @@ defmodule GithubNotifier.Application do
         {{Services.PipelineStartedNotifier, []}, enabled?("START_CONSUMERS")},
         {{Services.PipelineFinishedNotifier, []}, enabled?("START_CONSUMERS")},
         {{Services.PipelineSummaryAvailableNotifier, []}, enabled?("START_CONSUMERS")},
+        {{GithubNotifier.FeatureProviderInvalidatorWorker, []}, true},
         feature_provider()
       ])
 

github_notifier/lib/github_notifier/feature_provider_invalidator_worker.ex

@@ -0,0 +1,44 @@
+defmodule GithubNotifier.FeatureProviderInvalidatorWorker do
+  require Logger
+
+  @doc """
+  This module consumes RabbitMQ feature and machine state change events
+  and invalidates features and machines caches.
+  """
+
+  use Tackle.Multiconsumer,
+    url: Application.get_env(:github_notifier, :amqp_url),
+    service: "github_notifier",
+    routes: [
+      {"feature_exchange", "features_changed", :features_changed},
+      {"feature_exchange", "organization_features_changed", :organization_features_changed}
+    ],
+    # This queue is used to consume events from the feature exchange.
+    # It is declared as non-durable, auto-delete and exclusive.
+    # This means that the queue will be deleted when the consumer disconnects.
+    # This is the desired behavior, because these events are used to invalidate pod-level caches.
+    queue: :dynamic,
+    queue_opts: [
+      durable: false,
+      auto_delete: true,
+      exclusive: true
+    ],
+    connection_id: GithubNotifier.FeatureProviderInvalidatorWorker
+
+  def features_changed(_message) do
+    log("invalidating features")
+    {:ok, _} = FeatureProvider.list_features(reload: true)
+    :ok
+  end
+
+  def organization_features_changed(message) do
+    event = InternalApi.Feature.OrganizationFeaturesChanged.decode(message)
+    log("invalidating features for org #{event.org_id}")
+    {:ok, _} = FeatureProvider.list_features(reload: true, param: event.org_id)
+    :ok
+  end
+
+  defp log(message) do
+    Logger.info("[FEATURE PROVIDER INVALIDATOR WORKER] #{message}")
+  end
+end