Persistent storage (#937)

Start moving filesystems as an initial implementation.

README.md

@@ -157,10 +157,6 @@ The following environment variables configure Livebook:
     "attached:NODE:COOKIE" (Attached node) or "embedded" (Embedded).
     Defaults to "standalone".
 
-  * LIVEBOOK_FILE_SYSTEM_1, LIVEBOOK_FILE_SYSTEM_2, ... - configures additional
-    file systems. Each variable should hold a configuration string, which must
-    be of the form: "s3 BUCKET_URL ACCESS_KEY_ID SECRET_ACCESS_KEY".
-
   * LIVEBOOK_IP - sets the ip address to start the web application on.
     Must be a valid IPv4 or IPv6 address.
 

config/config.exs

@@ -19,6 +19,9 @@ config :mime, :types, %{
   "text/plain" => ["livemd"]
 }
 
+# Sets the default storage backend
+config :livebook, :storage, Livebook.Storage.Ets
+
 # Sets the default authentication mode to token
 config :livebook, :authentication_mode, :token
 

lib/livebook.ex

@@ -57,9 +57,8 @@ defmodule Livebook do
       |> Livebook.FileSystem.Utils.ensure_dir_path()
 
     local_file_system = Livebook.FileSystem.Local.new(default_path: root_path)
-    configured_file_systems = Livebook.Config.file_systems!("LIVEBOOK_FILE_SYSTEM_")
 
-    config :livebook, :file_systems, [local_file_system | configured_file_systems]
+    config :livebook, :default_file_systems, [local_file_system]
 
     autosave_path =
       if config_env() == :test do

lib/livebook/application.ex

@@ -16,6 +16,8 @@ defmodule Livebook.Application do
         LivebookWeb.Telemetry,
         # Start the PubSub system
         {Phoenix.PubSub, name: Livebook.PubSub},
+        # Start the storage module
+        Livebook.Storage.current(),
         # Periodid measurement of system resources
         Livebook.SystemResources,
         # Start the tracker server on this node
@@ -31,7 +33,8 @@ defmodule Livebook.Application do
         # Start the Endpoint (http/https)
         # We skip the access url as we do our own logging below
         {LivebookWeb.Endpoint, log_access_url: false}
-      ] ++ app_specs()
+      ] ++
+        app_specs()
 
     opts = [strategy: :one_for_one, name: Livebook.Supervisor]
 

lib/livebook/config.ex

@@ -39,17 +39,23 @@ defmodule Livebook.Config do
   """
   @spec file_systems() :: list(FileSystem.t())
   def file_systems() do
-    Application.fetch_env!(:livebook, :file_systems)
+    Application.fetch_env!(:livebook, :default_file_systems) ++
+      Enum.map(storage().all(:filesystem), &storage_to_fs/1)
   end
 
   @doc """
   Appends a new file system to the configured ones.
   """
   @spec append_file_system(FileSystem.t()) :: list(FileSystem.t())
-  def append_file_system(file_system) do
+  def append_file_system(%FileSystem.S3{} = file_system) do
-    file_systems = Enum.uniq(file_systems() ++ [file_system])
+    attributes =
-    Application.put_env(:livebook, :file_systems, file_systems, persistent: true)
+      file_system
-    file_systems
+      |> FileSystem.S3.to_config()
+      |> Map.to_list()
+
+    storage().insert(:filesystem, generate_filesystem_id(), [{:type, "s3"} | attributes])
+
+    file_systems()
   end
 
   @doc """
@@ -57,9 +63,13 @@ defmodule Livebook.Config do
   """
   @spec remove_file_system(FileSystem.t()) :: list(FileSystem.t())
   def remove_file_system(file_system) do
-    file_systems = List.delete(file_systems(), file_system)
+    storage().all(:filesystem)
-    Application.put_env(:livebook, :file_systems, file_systems, persistent: true)
+    |> Enum.find(&(storage_to_fs(&1) == file_system))
-    file_systems
+    |> case do
+      %{id: id} -> storage().delete(:filesystem, id)
+    end
+
+    file_systems()
   end
 
   @doc """
@@ -316,59 +326,24 @@ defmodule Livebook.Config do
     }
   end
 
-  @doc """
+  defp storage() do
-  Parses file systems list.
+    Livebook.Storage.current()
-
-  Appends subsequent numbers to the given env prefix (starting from 1)
-  and parses the env variables until `nil` is encountered.
-  """
-  def file_systems!(env_prefix) do
-    Stream.iterate(1, &(&1 + 1))
-    |> Stream.map(fn n ->
-      env = env_prefix <> Integer.to_string(n)
-      System.get_env(env)
-    end)
-    |> Stream.take_while(& &1)
-    |> Enum.map(&parse_file_system!/1)
   end
 
-  defp parse_file_system!(string) do
+  defp storage_to_fs(%{type: "s3"} = config) do
-    case string do
+    case FileSystem.S3.from_config(config) do
-      "s3 " <> config ->
+      {:ok, fs} ->
-        FileSystem.S3.from_config_string(config)
+        fs
 
-      _ ->
+      {:error, message} ->
         abort!(
-          ~s{unrecognised file system, expected "s3 BUCKET_URL ACCESS_KEY_ID SECRET_ACCESS_KEY", got: #{inspect(string)}}
+          ~s{unrecognised file system, expected "s3 BUCKET_URL ACCESS_KEY_ID SECRET_ACCESS_KEY", got: #{inspect(message)}}
         )
     end
-    |> case do
-      {:ok, file_system} -> file_system
-      {:error, message} -> abort!(message)
-    end
   end
 
-  @doc """
+  defp generate_filesystem_id() do
-  Returns environment variables configuration corresponding
+    :crypto.strong_rand_bytes(6) |> Base.url_encode64()
-  to the given file systems.
-
-  The first (default) file system is ignored.
-  """
-  def file_systems_as_env(file_systems)
-
-  def file_systems_as_env([_ | additional_file_systems]) do
-    additional_file_systems
-    |> Enum.with_index(1)
-    |> Enum.map(fn {file_system, n} ->
-      config = file_system_to_config_string(file_system)
-      ["LIVEBOOK_FILE_SYSTEM_", Integer.to_string(n), "=", ?", config, ?"]
-    end)
-    |> Enum.intersperse(" ")
-    |> IO.iodata_to_binary()
-  end
-
-  defp file_system_to_config_string(%FileSystem.S3{} = file_system) do
-    ["s3 ", FileSystem.S3.to_config_string(file_system)]
   end
 
   @doc """

lib/livebook/file_system/s3.ex

@@ -26,31 +26,27 @@ defmodule Livebook.FileSystem.S3 do
   end
 
   @doc """
-  Parses file system from a configuration string.
+  Parses file system from a configuration map.
-
-  The expected format is `"BUCKET_URL ACCESS_KEY_ID SECRET_ACCESS_KEY"`.
-
-  ## Examples
-
-      Livebook.FileSystem.S3.from_config_string("https://s3.eu-central-1.amazonaws.com/mybucket myaccesskeyid mysecret")
   """
-  @spec from_config_string(String.t()) :: {:ok, t()} | {:error, String.t()}
+  @spec from_config(map()) :: {:ok, t()} | {:error, String.t()}
-  def from_config_string(string) do
+  def from_config(config) do
-    case String.split(string) do
+    case config do
-      [bucket_url, access_key_id, secret_access_key] ->
+      %{
+        bucket_url: bucket_url,
+        access_key_id: access_key_id,
+        secret_access_key: secret_access_key
+      } ->
         {:ok, new(bucket_url, access_key_id, secret_access_key)}
 
-      args ->
+      _config ->
-        {:error, "S3 filesystem configuration expects 3 arguments, but got #{length(args)}"}
+        {:error,
+         "S3 filesystem config is expected to have 3 arguments: 'bucket_url', 'access_key_id' and 'secret_access_key', but got #{inspect(config)}"}
     end
   end
 
-  @doc """
+  @spec to_config(t()) :: map()
-  Formats the given file system into an equivalent configuration string.
+  def to_config(%__MODULE__{} = s3) do
-  """
+    Map.take(s3, [:bucket_url, :access_key_id, :secret_access_key])
-  @spec to_config_string(t()) :: String.t()
-  def to_config_string(file_system) do
-    "#{file_system.bucket_url} #{file_system.access_key_id} #{file_system.secret_access_key}"
   end
 end
 

lib/livebook/storage.ex

@@ -0,0 +1,45 @@
+defmodule Livebook.Storage do
+  @moduledoc """
+  Behaviour defining an interface for storing arbitrary data in
+  [Entity-Attribute-Value](https://en.wikipedia.org/wiki/Entity%E2%80%93attribute%E2%80%93value_model) fashion.
+  """
+
+  @type namespace :: atom()
+  @type entity_id :: binary()
+  @type attribute :: atom()
+  @type value :: binary()
+  @type timestamp :: non_neg_integer()
+
+  @type entity :: %{required(:id) => entity_id(), optional(attribute()) => value()}
+
+  @doc """
+  Returns a map identified by `entity_id` in `namespace`.
+
+      fetch(:filesystem, "rand-id")
+      #=> {:ok, %{id: "rand-id", type: "s3", bucket_url: "/...", secret: "abc", access_key: "xyz"}}
+
+  """
+  @callback fetch(namespace(), entity_id()) :: {:ok, entity()} | :error
+
+  @doc """
+  Returns all values in namespace.
+
+      all(:filesystem)
+      [%{id: "rand-id", type: "s3", bucket_url: "/...", secret: "abc", access_key: "xyz"}]
+
+  """
+  @callback all(namespace()) :: [entity()]
+
+  @doc """
+  Inserts given list of attribute-value paris to a entity belonging to specified namespace.
+  """
+  @callback insert(namespace(), entity_id(), [{attribute(), value()}]) :: :ok
+
+  @doc """
+  Deletes an entity of given id from given namespace.
+  """
+  @callback delete(namespace(), entity_id()) :: :ok
+
+  @spec current() :: module()
+  def current(), do: Application.fetch_env!(:livebook, :storage)
+end

lib/livebook/storage/ets.ex

@@ -0,0 +1,100 @@
+defmodule Livebook.Storage.Ets do
+  @moduledoc """
+  Ets implementation of `Livebook.Storage` behaviour.
+
+  The module is supposed to be started just once as it
+  is responsible for managing a named ets table.
+
+  `insert` and `delete` operations are supposed to be called using a GenServer
+  while all the lookups can be performed by directly accessing the named table.
+  """
+  @behaviour Livebook.Storage
+
+  @table_name __MODULE__
+
+  use GenServer
+
+  @impl Livebook.Storage
+  def fetch(namespace, entity_id) do
+    @table_name
+    |> :ets.lookup({namespace, entity_id})
+    |> case do
+      [] ->
+        :error
+
+      entries ->
+        entries
+        |> Enum.map(fn {_key, attr, val, _timestamp} -> {attr, val} end)
+        |> Map.new()
+        |> Map.put(:id, entity_id)
+        |> then(&{:ok, &1})
+    end
+  end
+
+  @impl Livebook.Storage
+  def all(namespace) do
+    @table_name
+    |> :ets.match({{namespace, :"$1"}, :"$2", :"$3", :_})
+    |> Enum.group_by(
+      fn [entity_id, _attr, _val] -> entity_id end,
+      fn [_id, attr, val] -> {attr, val} end
+    )
+    |> Enum.map(fn {entity_id, attributes} ->
+      attributes
+      |> Map.new()
+      |> Map.put(:id, entity_id)
+    end)
+  end
+
+  @impl Livebook.Storage
+  def insert(namespace, entity_id, attributes) do
+    GenServer.call(__MODULE__, {:insert, namespace, entity_id, attributes})
+  end
+
+  @impl Livebook.Storage
+  def delete(namespace, entity_id) do
+    GenServer.call(__MODULE__, {:delete, namespace, entity_id})
+  end
+
+  @spec start_link(keyword()) :: GenServer.on_start()
+  def start_link(opts) do
+    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  end
+
+  @impl GenServer
+  def init(_opts) do
+    table = :ets.new(@table_name, [:named_table, :protected, :duplicate_bag])
+
+    {:ok, %{table: table}}
+  end
+
+  @impl GenServer
+  def handle_call({:insert, namespace, entity_id, attributes}, _from, %{table: table} = state) do
+    match_head = {{namespace, entity_id}, :"$1", :_, :_}
+
+    guards =
+      Enum.map(attributes, fn {key, _val} ->
+        {:==, :"$1", key}
+      end)
+
+    :ets.select_delete(table, [{match_head, guards, [true]}])
+
+    timestamp = System.os_time(:millisecond)
+
+    attributes =
+      Enum.map(attributes, fn {attr, val} ->
+        {{namespace, entity_id}, attr, val, timestamp}
+      end)
+
+    :ets.insert(table, attributes)
+
+    {:reply, :ok, state}
+  end
+
+  @impl GenServer
+  def handle_call({:delete, namespace, entity_id}, _from, %{table: table} = state) do
+    :ets.delete(table, {namespace, entity_id})
+
+    {:reply, :ok, state}
+  end
+end

lib/livebook_cli/server.ex

@@ -220,7 +220,7 @@ defmodule LivebookCLI.Server do
       |> Livebook.FileSystem.Utils.ensure_dir_path()
 
     local_file_system = Livebook.FileSystem.Local.new(default_path: root_path)
-    opts_to_config(opts, [{:livebook, :file_systems, [local_file_system]} | config])
+    opts_to_config(opts, [{:livebook, :default_file_systems, [local_file_system]} | config])
   end
 
   defp opts_to_config([{:sname, sname} | opts], config) do

lib/livebook_web/live/settings_live.ex

@@ -8,14 +8,12 @@ defmodule LivebookWeb.SettingsLive do
   @impl true
   def mount(_params, _session, socket) do
     file_systems = Livebook.Config.file_systems()
-    file_systems_env = Livebook.Config.file_systems_as_env(file_systems)
 
     {:ok,
      socket
      |> SidebarHelpers.shared_home_handlers()
      |> assign(
        file_systems: file_systems,
-       file_systems_env: file_systems_env,
        page_title: "Livebook - Settings"
      )}
   end
@@ -68,15 +66,6 @@ defmodule LivebookWeb.SettingsLive do
               <h2 class="text-xl text-gray-800 font-semibold">
                 File systems
               </h2>
-              <span class="tooltip top" data-tooltip="Copy as environment variables">
-                <button class="icon-button"
-                  aria-label="copy as environment variables"
-                  phx-click={JS.dispatch("lb:clipcopy", to: "#file-systems-env-source")}
-                  disabled={@file_systems_env == ""}>
-                  <.remix_icon icon="clipboard-line" class="text-lg" />
-                </button>
-                <span class="hidden" id="file-systems-env-source"><%= @file_systems_env %></span>
-              </span>
             </div>
               <LivebookWeb.SettingsLive.FileSystemsComponent.render
                 file_systems={@file_systems}
@@ -162,8 +151,7 @@ defmodule LivebookWeb.SettingsLive do
 
   @impl true
   def handle_info({:file_systems_updated, file_systems}, socket) do
-    file_systems_env = Livebook.Config.file_systems_as_env(file_systems)
+    {:noreply, assign(socket, file_systems: file_systems)}
-    {:noreply, assign(socket, file_systems: file_systems, file_systems_env: file_systems_env)}
   end
 
   def handle_info(_message, socket), do: {:noreply, socket}

test/livebook/storage/ets_test.exs

@@ -0,0 +1,76 @@
+defmodule Livebook.Storage.EtsTest do
+  use ExUnit.Case, async: false
+
+  alias Livebook.Storage.Ets
+
+  describe "insert/3 and fetch/2" do
+    test "properly inserts a new key-value attribute" do
+      assert :ok = Ets.insert(:insert, "test", key1: "val1", key2: "val2")
+
+      assert {:ok,
+              %{
+                id: "test",
+                key1: "val1",
+                key2: "val2"
+              }} = Ets.fetch(:insert, "test")
+    end
+
+    test "replaces already existing attributes with new values" do
+      assert :ok = Ets.insert(:insert, "replace", key1: "val1", key2: "val2")
+
+      assert {:ok,
+              %{
+                key1: "val1",
+                key2: "val2"
+              }} = Ets.fetch(:insert, "replace")
+
+      assert :ok =
+               Ets.insert(:insert, "replace", key1: "updated_val1", key2: "val2", key3: "val3")
+
+      assert {:ok,
+              %{
+                key1: "updated_val1",
+                key2: "val2",
+                key3: "val3"
+              }} = Ets.fetch(:insert, "replace")
+    end
+  end
+
+  test "fetch/2" do
+    :ok = Ets.insert(:fetch, "test", key1: "val1")
+
+    assert {:ok,
+            %{
+              id: "test",
+              key1: "val1"
+            }} = Ets.fetch(:fetch, "test")
+
+    assert :error = Ets.fetch(:fetch, "unknown")
+  end
+
+  test "delete/2" do
+    :ok = Ets.insert(:delete, "test", key1: "val1")
+
+    assert {:ok, _entity} = Ets.fetch(:delete, "test")
+
+    assert :ok = Ets.delete(:delete, "test")
+
+    assert :error = Ets.fetch(:delete, "test")
+  end
+
+  describe "all/1" do
+    test "returns all inserted entities for given namespace" do
+      :ok = Ets.insert(:all, "test1", key1: "val1")
+      :ok = Ets.insert(:all, "test2", key1: "val1")
+
+      {:ok, entity1} = Ets.fetch(:all, "test1")
+      {:ok, entity2} = Ets.fetch(:all, "test2")
+
+      assert [^entity1, ^entity2] = Ets.all(:all)
+    end
+
+    test "returns an empty list if no entities exist for given namespace" do
+      assert [] = Ets.all(:unknown_namespace)
+    end
+  end
+end