peerage_ec2

defmodule Peerage.Via.Ec2 do
  @moduledoc """
  A Peerage provider for easy clustering on AWS EC2 and Elastic Beanstalk.
  """
  @behaviour Peerage.Provider

  alias Peerage.Via.Ec2.Xml
  require Logger

  @doc """
  Periodically polls the metadata and EC2 API's for other nodes in the same "cluster."
  """
  def poll() do
    fetch_instance_id()
    |> fetch_cluster_name()
    |> fetch_running_services()
    |> format_services_list()
  end

  defp fetch_instance_id() do
    # EC2 provides an instance metadata API endpoint. We'll perform
    # a request to determine the ID of the running instance.
    #
    # AWS Documentation: http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html
    metadata_api = 'http://169.254.169.254/latest/meta-data/instance-id'

    case request(metadata_api) do
      {:ok, {{_, 200, _}, _headers, body}} ->
        to_string(body)

      error ->
        Logger.error("Peerage.Via.Ec2 hit error in fetch_instance_id: #{inspect(error)}")
        :error
    end
  end

  defp fetch_cluster_name(:error), do: :error

  defp fetch_cluster_name(instance_id) do
    # Having retrieved the instance_id of the current EC2 instance,
    # we'll peform a signed/authenticated request to Amazon's EC2
    # DescribeInstances API to retrieve the name of the `cluster`
    # of instances we've tagged.
    query = ExAws.EC2.describe_instances(filters: ["instance-id": [instance_id]])
    response = ExAws.request(query)

    case response do
      {:ok, %{status_code: 200, body: body}} ->
        body
        |> to_charlist
        |> Xml.parse()
        |> Xml.first("//tagSet/item[key='#{tag_name(:cluster)}']/value")
        |> Xml.text()

      error ->
        Logger.error("Peerage.Via.Ec2 hit error in fetch_cluster_name: #{inspect(error)}")
        :error
    end
  end

  defp fetch_running_services(:error), do: :error

  defp fetch_running_services(cluster_name) do
    # Having retrieved the cluster_name, we'll peform a
    # signed/authenticated request to Amazon's EC2
    # DescribeInstances API to retrieve all of the
    # running services in that cluster.
    #
    # Note that, an InstanceState code of 16 represents
    # a running EC2 service.
    #
    # AWS Documentation: http://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_InstanceState.html
    query =
      ExAws.EC2.describe_instances(
        filters: ["instance-state-code": ["16"], "tag:#{tag_name(:cluster)}": [cluster_name]]
      )

    response = ExAws.request(query)

    case response do
      {:ok, %{status_code: 200, body: body}} ->
        instances = body |> to_charlist |> Xml.parse()

        Enum.map(Xml.all(instances, "//instancesSet/item"), fn node ->
          host = Xml.first(node, "//privateIpAddress") |> Xml.text()
          service = Xml.first(node, "//tagSet/item[key='service']/value") |> Xml.text()
          %{host: host, name: service}
        end)

      error ->
        Logger.error("Peerage.Via.Ec2 hit error in fetch_running_services: #{inspect(error)}")
        :error
    end
  end

  defp format_services_list(:error), do: []

  defp format_services_list(services) do
    Enum.map(services, fn service ->
      String.to_atom("#{service.name}@" <> to_string(service.host))
    end)
  end

  defp describe_endpoint(filters) do
    query_string =
      filters
      |> Map.put("Action", "DescribeInstances")
      |> Map.put("Version", "2016-11-15")
      |> URI.encode_query()

    "https://ec2.amazonaws.com/?" <> query_string
  end

  defp request(uri), do: :httpc.request(:get, {uri, []}, [timeout: timeout()], [])
  defp tag_name(key), do: Application.fetch_env!(:peerage_ec2, :tags)[key]
  defp timeout(), do: Application.get_env(:peerage_ec2, :timeout, 1000)
end