Plural Cloud CRUD

Basic CRUD mutations to drive a Plural cloud experience, allowing for api-based provisioning of the Plural Console.

At the moment, only supporting aws in us-east-1, can expand as needed.  Persistence is managed by a set of cloud-specific cockroach clusters.

.tool-versions

@@ -1,2 +1,2 @@
-erlang 24.3.4.14
-elixir 1.12.3
+erlang 24.3.4.17
+elixir 1.13.4

Dockerfile

@@ -1,4 +1,4 @@
-FROM bitwalker/alpine-elixir:1.12.3 AS builder
+FROM bitwalker/alpine-elixir:1.13.4 AS builder
 
 # The following are build arguments used to change variable parts of the image.
 # The name of your application/release (required)

Makefile

@@ -49,10 +49,10 @@ else
 endif
 
 testup: ## sets up dependent services for test
-	docker-compose up -d
+	docker compose up -d
 
 testdown: ## tear down test dependencies
-	docker-compose down
+	docker compose down
 
 connectdb: ## proxies the db in kubernetes via kubectl
 	@echo "run psql -U forge -h 127.0.0.1 forge to connect"

apps/core/lib/core/application.ex

@@ -25,7 +25,7 @@ defmodule Core.Application do
 
   def broker() do
     case conf(:start_broker) do
-      true -> [{Core.Conduit.Broker, []}]
+      true -> [{Core.Conduit.Broker, []}, Core.Services.Coud.Poller]
       _ -> []
     end
   end

apps/core/lib/core/clients/console.ex

@@ -0,0 +1,96 @@
+defmodule Core.Clients.Console do
+  require Logger
+
+  @clusters_q """
+  query {
+    clusters(first: 100) {
+      edges { node { name id distro metadata } }
+    }
+  }
+  """
+
+  @create_svc_q """
+  mutation Create($clusterId: ID!, $attributes: ServiceDeploymentAttributes!) {
+    createServiceDeployment(clusterId: $clusterId, attributes: $attributes) {
+      id
+    }
+  }
+  """
+
+  @delete_svc_q """
+  mutation Delete($id: ID!) {
+    deleteServiceDeployment(id: $id) {
+      id
+    }
+  }
+  """
+
+  @update_svc_q """
+  mutation Update($id: ID!, $attributes: ServiceUpdateAttributes!) {
+    updateServiceDeployment(id: $id) {
+      id
+    }
+  }
+  """
+
+  @repo_q """
+  query Repo($url: String!) {
+    gitRepository(url: $url) {
+      id
+    }
+  }
+  """
+
+  def new(url, token) do
+    Req.new(base_url: url, auth: "Token #{token}")
+    |> AbsintheClient.attach()
+  end
+
+  def clusters(client) do
+    Req.post(client, graphql: @clusters_q)
+    |> case do
+      {:ok, %Req.Response{body: %{"clusters" => %{"edges" => edges}}}} -> {:ok, Enum.map(edges, & &1["node"])}
+      res ->
+        Logger.warn "Failed to fetch clusters: #{inspect(res)}"
+        {:error, "could not fetch clusters"}
+    end
+  end
+
+  def repo(client, url) do
+    Req.post(client, graphql: {@repo_q, %{url: url}})
+    |> case do
+      {:ok, %Req.Response{body: %{"gitRepository" => %{"id" => id}}}} -> {:ok, id}
+      res ->
+        Logger.warn "Failed to fetch clusters: #{inspect(res)}"
+        {:error, "could not fetch repo"}
+    end
+  end
+
+  def create_service(client, cluster_id, attrs) do
+    Req.post(client, graphql: {@create_svc_q, %{clusterId: cluster_id, attributes: attrs}})
+    |> service_resp("createServiceDeployment")
+  end
+
+  def update_service(client, id, attrs) do
+    Req.post(client, graphql: {@update_svc_q, %{id: id, attributes: attrs}})
+    |> service_resp("updateServiceDeployment")
+  end
+
+  def delete_service(client, id) do
+    Req.post(client, graphql: {@delete_svc_q, %{id: id}})
+    |> service_resp("deleteServiceDeployment")
+  end
+
+  defp service_resp({:ok, %Req.Response{status: 200, body: body}}, field) do
+    case body[field] do
+      %{"id" => id} -> {:ok, id}
+      err ->
+        Logger.warn "invalid console gql response: #{inspect(err)}"
+    end
+  end
+
+  defp service_resp(resp, _) do
+    Logger.error "failed to fetch from console: #{inspect(resp)}"
+    {:error, "console error"}
+  end
+end

apps/core/lib/core/pubsub/events.ex

@@ -90,3 +90,7 @@ defmodule Core.PubSub.ClusterDependencyCreated, do: use Piazza.PubSub.Event
 defmodule Core.PubSub.DeferredUpdateCreated, do: use Piazza.PubSub.Event
 
 defmodule Core.PubSub.UpgradesPromoted, do: use Piazza.PubSub.Event
+
+defmodule Core.PubSub.ConsoleInstanceCreated, do: use Piazza.PubSub.Event
+defmodule Core.PubSub.ConsoleInstanceUpdated, do: use Piazza.PubSub.Event
+defmodule Core.PubSub.ConsoleInstanceDeleted, do: use Piazza.PubSub.Event

apps/core/lib/core/pubsub/protocols/fanout.ex

@@ -235,3 +235,21 @@ defimpl Core.PubSub.Fanout, for: [Core.PubSub.RoleCreated, Core.PubSub.RoleUpdat
     |> Enum.count()
   end
 end
+
+defimpl Core.PubSub.Fanout, for: Core.PubSub.ConsoleInstanceCreated do
+  alias Core.Services.Cloud.Workflow
+
+  def fanout(%{item: instance}), do: Workflow.provision(instance)
+end
+
+defimpl Core.PubSub.Fanout, for: Core.PubSub.ConsoleInstanceUpdated do
+  alias Core.Services.Cloud.Workflow
+
+  def fanout(%{item: instance}), do: Workflow.sync(instance)
+end
+
+defimpl Core.PubSub.Fanout, for: Core.PubSub.ConsoleInstanceDeleted do
+  alias Core.Services.Cloud.Workflow
+
+  def fanout(%{item: instance}), do: Workflow.deprovision(instance)
+end

apps/core/lib/core/schema/cloud_cluster.ex

@@ -0,0 +1,42 @@
+defmodule Core.Schema.CloudCluster do
+  use Piazza.Ecto.Schema
+
+  defenum Cloud, aws: 0
+
+  @saturation 1000
+
+  @region_map %{
+    aws: ~w(us-east-1)
+  }
+
+  schema "cloud_clusters" do
+    field :name,        :string
+    field :external_id, :binary_id
+    field :cloud,       Cloud
+    field :region,      :string
+    field :count,       :integer
+
+    timestamps()
+  end
+
+  def for_cloud(query \\ __MODULE__, cloud) do
+    from(c in query, where: c.cloud == ^cloud)
+  end
+
+  def unsaturated(query \\ __MODULE__) do
+    from(c in query, where: c.count < @saturation)
+  end
+
+  def for_region(query \\ __MODULE__, region) do
+    from(c in query, where: c.region == ^region)
+  end
+
+  def region_information(), do: @region_map
+
+  def changeset(model, attrs \\ %{}) do
+    model
+    |> cast(attrs, ~w(name external_id cloud region)a)
+    |> unique_constraint(:name)
+    |> validate_required(~w(name external_id cloud region)a)
+  end
+end

apps/core/lib/core/schema/cockroach_cluster.ex

@@ -0,0 +1,34 @@
+defmodule Core.Schema.CockroachCluster do
+  use Piazza.Ecto.Schema
+  alias Piazza.Ecto.EncryptedString
+  alias Core.Schema.CloudCluster
+
+  @saturation 1000
+
+  schema "cockroach_clusters" do
+    field :name,        :string
+    field :cloud,       CloudCluster.Cloud
+    field :region,      :string
+    field :url,         EncryptedString
+    field :certificate, :string
+    field :endpoints,   :map
+    field :count,       :integer, default: 0
+
+    timestamps()
+  end
+
+  def for_cloud(query \\ __MODULE__ , cloud) do
+    from(c in query, where: c.cloud == ^cloud)
+  end
+
+  def unsaturated(query \\ __MODULE__) do
+    from(c in query, where: c.count < @saturation)
+  end
+
+  def changeset(model, attrs \\ %{}) do
+    model
+    |> cast(attrs, ~w(name cloud region url certificate endpoints)a)
+    |> unique_constraint(:name)
+    |> validate_required(~w(name cloud region url certificate endpoints)a)
+  end
+end

apps/core/lib/core/schema/console_instance.ex

@@ -0,0 +1,117 @@
+defmodule Core.Schema.ConsoleInstance do
+  use Piazza.Ecto.Schema
+  alias Piazza.Ecto.EncryptedString
+  alias Core.Schema.{CockroachCluster, CloudCluster, User}
+
+  defenum Size, small: 0, medium: 1, large: 2
+  defenum Status,
+    pending: 0,
+    database_created: 1,
+    deployment_created: 2,
+    provisioned: 3,
+    deployment_deleted: 4,
+    database_deleted: 5
+
+  @region_map %{
+    aws: ~w(us-east-1)
+  }
+
+  schema "console_instances" do
+    field :name,          :string
+    field :status,        Status
+    field :subdomain,     :string
+    field :url,           :string
+    field :external_id,   :string
+    field :cloud,         CloudCluster.Cloud
+    field :size,          Size
+    field :region,        :string
+
+    field :deleted_at,    :utc_datetime_usec
+
+    embeds_one :instance_status, InstanceStatus, on_replace: :update do
+      field :db,  :boolean, default: false
+      field :svc, :boolean, default: false
+    end
+
+    embeds_one :configuration, Configuration, on_replace: :update do
+      field :database,       :string
+      field :dbuser,         :string
+      field :dbpassword,     EncryptedString
+      field :subdomain,      :string
+      field :jwt_secret,     EncryptedString
+      field :owner_name,     :string
+      field :owner_email,    :string
+      field :admin_password, EncryptedString
+      field :aes_key,        EncryptedString
+      field :encryption_key, EncryptedString
+      field :client_id,      :string
+      field :client_secret,  EncryptedString
+      field :plural_token,   EncryptedString
+      field :kas_api,        EncryptedString
+      field :kas_private,    EncryptedString
+      field :kas_redis,      EncryptedString
+    end
+
+    belongs_to :cockroach, CockroachCluster
+    belongs_to :cluster,   CloudCluster
+    belongs_to :owner,     User
+
+    timestamps()
+  end
+
+  def for_account(query \\ __MODULE__, account_id) do
+    from(c in query,
+      join: u in assoc(c, :owner),
+      where: u.account_id == ^account_id
+    )
+  end
+
+  def ordered(query \\ __MODULE__, order \\ [asc: :name]) do
+    from(c in query, order_by: ^order)
+  end
+
+  def regions(), do: @region_map
+
+  @valid ~w(name cloud size region status subdomain url external_id cockroach_id cluster_id owner_id)a
+
+  def changeset(model, attrs \\ %{}) do
+    model
+    |> cast(attrs, @valid)
+    |> cast_embed(:configuration, with: &configuration_changeset/2)
+    |> cast_embed(:instance_status, with: &status_changeset/2)
+    |> validate_required(@valid -- [:external_id])
+    |> unique_constraint(:subdomain)
+    |> unique_constraint(:name)
+    |> validate_format(:name, ~r/[a-z][a-z0-9]{5,10}/, message: "must be an alphanumeric string between 5 and 10 characters")
+    |> validate_region()
+  end
+
+  defp validate_region(cs) do
+    cloud = get_field(cs, :cloud)
+    regions = @region_map[cloud]
+    validate_change(cs, :region, fn :region, reg ->
+      case reg in regions do
+        true -> []
+        _ -> [region: "Invalid region #{reg} for cloud #{cloud}"]
+      end
+    end)
+  end
+
+  @conf_valid ~w(
+    database dbuser dbpassword
+    subdomain jwt_secret owner_name owner_email admin_password aes_key
+    encryption_key client_id client_secret plural_token
+    kas_api kas_private kas_redis
+  )a
+
+  defp configuration_changeset(model, attrs) do
+    model
+    |> cast(attrs, @conf_valid)
+    |> validate_required(@conf_valid)
+  end
+
+  defp status_changeset(model, attrs) do
+    model
+    |> cast(attrs, ~w(db svc)a)
+  end
+end