Add Prometheus Query Operations

These are basic fetch functions to query data from a prometheus endpoint.

3 files changed, 104 insertions, 0 deletions

diff --git a/lib/ex_prometheus_io.ex b/lib/ex_prometheus_io.ex
index e1c9427..cac39a5 100644
--- a/lib/ex_prometheus_io.ex
+++ b/lib/ex_prometheus_io.ex
@@ -1,2 +1,33 @@
 defmodule ExPrometheusIo do
+  use Application
+
+  def start(_, _) do
+    ExPrometheusIo.Supervisor.start_link
+  end
+
+  def query(query, _opts \\ []) do
+    query_opts = [query]
+    spawn_query(:fetch_query, query_opts)
+  end
+
+  def range(query, start_ts, end_ts, step, _opts \\ []) do
+    query_opts = [query, start_ts, end_ts, step]
+    spawn_query(:fetch_range, query_opts)
+  end
+
+  def series(matches, _opts \\ []) do
+    spawn_query(:fetch_series, [matches])
+  end
+
+  defp spawn_query(fetch, query_opts, _opts \\ []) do
+    query_ref = make_ref()
+    query_opts = query_opts ++ [query_ref, self()]
+    {:ok, pid} = Task.Supervisor.start_child(
+      ExPrometheusIo.QuerySupervisor,
+      ExPrometheusIo.Query,
+      fetch,
+      query_opts)
+    {pid, query_ref}
+  end
+
 end
diff --git a/lib/ex_prometheus_io/query.ex b/lib/ex_prometheus_io/query.ex
new file mode 100644
index 0000000..2bba7bd
--- /dev/null
+++ b/lib/ex_prometheus_io/query.ex
@@ -0,0 +1,56 @@
+defmodule ExPrometheusIo.Query do
+
+  def fetch_query(query_str, query_ref, owner) do
+    "query=#{query_str}" <> "&time=#{:os.system_time(:seconds)}"
+    |> fetch_json("query")
+    |> Poison.decode
+    |> send_results(query_ref, owner)
+  end
+
+  def fetch_range(query_str, start_ts, end_ts, step, query_ref, owner) do
+    "query=#{query_str}"
+    <> "&start=#{start_ts}"
+    <> "&end=#{end_ts}"
+    <> "&step=#{step}"
+    |> fetch_json("query_range")
+    |> Poison.decode
+    |> send_results(query_ref, owner)
+  end
+
+  def fetch_series(matches, query_ref, owner) do
+    matches
+    |> Enum.map(fn(match) -> "match[]=#{match}" end)
+    |> Enum.join("&")
+    |> fetch_json("series")
+    |> Poison.decode
+    |> send_results(query_ref, owner)
+  end
+
+  defp fetch_json(query_str, query_type) do
+    {:ok, {_, _, body}} = :httpc.request(
+      "http://"
+      <> prometheus_host
+      <> "/api/v1/#{query_type}?"
+      <> query_str
+      |> String.to_char_list())
+    body
+  end
+
+  defp send_results({:error, :invalid} = results, query_ref, owner) do
+    send(owner, {:prometheus_results, query_ref, results})
+  end
+
+  defp send_results(
+      {:ok, %{"status" => "success", "data" => results}}, query_ref, owner) do
+    send(owner, {:prometheus_results, query_ref, results})
+  end
+
+  defp send_results({:ok, %{"error" => message}}, query_ref, owner) do
+    send(owner, {:prometheus_results, query_ref, {:error, message}})
+  end
+
+  defp prometheus_host do
+    Application.fetch_env!(:ex_prometheus_io, :hostname)
+  end
+
+end
diff --git a/lib/ex_prometheus_io/supervisor.ex b/lib/ex_prometheus_io/supervisor.ex
new file mode 100644
index 0000000..1cb8c66
--- /dev/null
+++ b/lib/ex_prometheus_io/supervisor.ex
@@ -0,0 +1,17 @@
+defmodule ExPrometheusIo.Supervisor do
+  use Supervisor
+
+  def start_link() do
+    Supervisor.start_link(__MODULE__, [], name: __MODULE__)
+  end
+
+  def init(_) do
+    children = [
+      supervisor(Task.Supervisor, [[name: ExPrometheusIo.QuerySupervisor]])
+    ]
+
+    opts = [strategy: :one_for_one]
+    supervise(children, opts)
+  end
+
+end