WIP Time picker

.gitignore

@@ -24,3 +24,4 @@ erl_crash.dump
 /config/prod.secret.exs
 /rel
 /tarballs
+.DS_Store

lib/kafkamon/message.ex

@@ -0,0 +1,38 @@
+defmodule Kafkamon.Message do
+  @type t :: %{
+    key: String.t,
+    offset: integer,
+    partition: integer,
+    topic: String.t,
+    value: map(),
+  }
+
+  defstruct [
+    :key,
+    :offset,
+    :partition,
+    :topic,
+    :value,
+  ]
+
+  def new(offset: offset, partition: partition, topic: topic, value: value) do
+    %__MODULE__{
+      key: "#{topic}/#{partition}##{offset}",
+      offset: offset,
+      partition: partition,
+      topic: topic,
+      value: value,
+    }
+  end
+
+  def stringify_keys(message = %__MODULE__{}) do
+    %{
+      "value"     => message.value,
+      "key"       => message.key,
+      "partition" => message.partition,
+      "offset"    => message.offset,
+      "topic"     => message.topic,
+    }
+  end
+
+end

lib/kafkamon/topics_subscriber.ex

@@ -1,7 +1,7 @@
 defmodule Kafkamon.TopicsSubscriber do
   use GenServer
 
-  alias Reader.EventQueue.Consumer.Message
+  alias Kafkamon.Message
 
   @stream_wait_time_ms Application.fetch_env!(:kafkamon, :consumer_wait_ms)
 
@@ -57,15 +57,7 @@ defmodule Kafkamon.TopicsSubscriber do
   defp broadcast(messages) do
     messages
     |> Enum.reverse
-    |> Enum.map(fn message ->
-      %{
-        "value" => message.value,
-        "key" => "#{message.topic}/#{message.partition}##{message.offset}",
-        "partition" => message.partition,
-        "offset" => message.offset,
-        "topic" => message.topic,
-      }
-    end)
+    |> Enum.map(&Message.stringify_keys/1)
     |> Enum.group_by(& Map.get(&1, "topic"))
     |> Enum.each(fn {topic, channel_messages} ->
       Kafkamon.Endpoint.broadcast("topic:#{topic}",

lib/mix/tasks/kafkamon.test_producer.ex

@@ -1,32 +1,37 @@
 defmodule Mix.Tasks.Kafkamon.TestProducer do
   use Mix.Task
 
+  @event_type 'AvvoEvent.AvvoProAdded'
 
   @shortdoc "Write as many events as fast as we can to Kafka"
 
   def run(_args) do
-    KafkaEx.start nil, nil
+    {:ok, brokers} = KafkaImpl.Util.kafka_brokers()
+    {:ok, worker} = KafkaImpl.create_no_name_worker(brokers, :no_consumer_group)
 
     schema_path = "test/data/AvvoProAdded.avsc"
     {:ok, schema_json} = File.read(schema_path)
-    type = 'AvvoEvent.AvvoProAdded'
-    v = %{event: %{app_id: "a", name: "n", timestamp: 0}, lawyer_id: 0}
-    message = Avrolixr.Codec.encode!(v, schema_json, type)
 
     ProgressBar.render_spinner [
-      text: "Producing to 'test'",
+      text: "Producing to 'users'",
       spinner_color: IO.ANSI.magenta,
       interval: 100,
       frames: :braille
     ], fn ->
-      produce("test", message, 50_000)
+      produce("users", 50, worker, schema_json)
     end
   end
 
-  def produce(_topic, _message, 0), do: nil
+  def produce(_topic, 0, _worker, _), do: nil
 
-  def produce(topic, message, iterations_left) do
-    KafkaEx.produce topic, 0, message
-    produce(topic, message, iterations_left - 1)
+
+  def produce(topic, iterations_left, worker, schema_json) do
+    ts = DateTime.utc_now |> DateTime.to_unix()
+    v = %{event: %{app_id: "a", name: "n", timestamp: ts}, lawyer_id: 100000 + iterations_left}
+    message = Avrolixr.Codec.encode!(v, schema_json, @event_type)
+    partition = rem(iterations_left, 12)
+    KafkaEx.produce topic, partition, message, worker_name: worker
+    :timer.sleep 100
+    produce(topic, iterations_left - 1, worker, schema_json)
   end
 end

lib/reader/event_queue/broadcast.ex

@@ -1,6 +1,6 @@
 defmodule Reader.EventQueue.Broadcast do
   alias Phoenix.PubSub
-  alias Reader.EventQueue.Consumer.Message
+  alias Kafkamon.Message
 
   def subscribe(topic), do: PubSub.subscribe KafkamonInternal, pubsub_topic(topic)
 

lib/reader/event_queue/consumer.ex

@@ -2,6 +2,8 @@ defmodule Reader.EventQueue.Consumer do
   require Logger
   use GenServer
 
+  alias Kafkamon.Message
+
   defmodule State do
     @type t :: %{
       topic_name: String.t,
@@ -13,16 +15,6 @@ defmodule Reader.EventQueue.Consumer do
     defstruct [:topic_name, :partition_number, :brokers, :worker_pid, :offset]
   end
 
-  defmodule Message do
-    @type t :: %{
-      value: map(),
-      topic: String.t,
-      offset: integer,
-      partition: integer,
-    }
-    defstruct [:value, :topic, :offset, :partition]
-  end
-
   @stream_wait_time_ms Application.fetch_env!(:kafkamon, :consumer_wait_ms)
 
   @spec start_link(State.t, [name: String.t]) :: GenServer.on_start
@@ -45,34 +37,46 @@ defmodule Reader.EventQueue.Consumer do
     end
   end
 
-  def handle_info(:fetch, state) do
-    [response] = KafkaImpl.fetch(
-      state.topic_name,
-      state.partition_number,
-      [
-        offset: state.offset,
-        worker_name: state.worker_pid,
-        auto_commit: false,
-      ]
-    )
+  @spec fetch_at(pid|atom, DateTime.t) :: {:ok, list} | {:error, any}
+  def fetch_at(consumer, datetime = %DateTime{}) do
+    GenServer.call(consumer, {:fetch_at, datetime})
+  end
+
+  def handle_call({:fetch_at, datetime}, _from, state) do
+    with {:ok, offset} <- datetime |> Timex.to_datetime(:utc) |> Timex.to_erl |> offset_at(state),
+         {:ok, _last_offset, messages} <- Reader.KafkaFetch.fetch_messages(%{state | offset: offset}),
+         wrapped <- messages |> Enum.map(&parse_message(&1, state)) do
+
+      {:reply, {:ok, wrapped}, state}
+    else
+      :no_offset ->
+        Logger.warn "Topic #{state.topic_name}##{state.partition_number} had no offset"
+        {:reply, {:ok, []}, state}
+      error -> 
+      Logger.warn "Error fetching messages at #{inspect datetime} for #{state.topic_name}##{state.partition_number}: #{inspect error}"
+      {:reply, {:error, error}, state}
+    end
+  end
 
-    %{partitions: [%{last_offset: last_offset, message_set: messages}]} = response
+  def handle_info(:stream, state) do
+    {:ok, last_offset, messages} = Reader.KafkaFetch.fetch_messages(state)
 
     messages
-    |> Enum.each(fn %KafkaEx.Protocol.Fetch.Message{offset: offset, value: value} ->
-      broadcast_message(state.topic_name, state.partition_number, offset, value)
-    end)
+    |> Enum.map(&parse_message(&1, state))
+    |> Enum.each(&broadcast/1)
 
     next_offset = case last_offset do
       nil -> state.offset
       last_offset -> last_offset + 1
     end
 
-    Process.send_after self, :fetch, @stream_wait_time_ms
+    Process.send_after self, :stream, @stream_wait_time_ms
 
     {:noreply, %{state | offset: next_offset}}
   end
 
+  def handle_info(_msg, state), do: {:noreply, state}
+
   defp latest_offset(%{offset: offset}) when is_integer(offset), do: offset
   defp latest_offset(%{topic_name: topic, partition_number: partition, worker_pid: pid}) do
     case KafkaImpl.latest_offset(topic, partition, pid) |> KafkaImpl.Util.extract_offset do
@@ -83,25 +87,14 @@ defmodule Reader.EventQueue.Consumer do
     end
   end
 
-  defp broadcast_message(topic, partition, offset, value) do
-    with {:ok, decoded} <- decode(value) do
-      %Message{
+  def parse_message(message, state) do
+    with {:ok, decoded} <- Reader.KafkaFetch.decode(message.value) do
+      Message.new(
+        offset: message.offset,
+        partition: state.partition_number,
+        topic: state.topic_name,
         value: decoded,
-        topic: topic,
-        offset: offset,
-        partition: partition,
-      }
-      |> broadcast()
-    end
-  end
-
-  defp decode(value) do
-    try do
-      value |> Avrolixr.Codec.decode
-    rescue
-      error ->
-        Logger.error "Could not decode message: #{inspect error}. Base64 encoded: #{Base.encode64(value)}"
-        :error
+      )
     end
   end
 
@@ -114,7 +107,7 @@ defmodule Reader.EventQueue.Consumer do
   end
 
   defp begin_streaming() do
-    send self, :fetch
+    send self, :stream
     :ok
   end
 
@@ -126,4 +119,13 @@ defmodule Reader.EventQueue.Consumer do
         {:error, "Could not start kafka worker: #{inspect error}"}
     end
   end
+
+
+  defp offset_at(erltime, state) do
+    case KafkaImpl.offset(state.topic_name, state.partition_number, erltime, state.worker_pid) |> KafkaImpl.Util.extract_offset do
+      :no_offset -> :no_offset
+      {:ok, offset} -> {:ok, offset}
+      x -> x # {:error, message}
+    end |> IO.inspect
+  end
 end

lib/reader/event_queue/foreman.ex

@@ -1,6 +1,18 @@
 defmodule Reader.EventQueue.Foreman do
   use GenServer
 
+  defmodule State do
+    @type topic_name_t :: String
+    @type partition_count_t :: integer
+    @type topic_tuple_t :: {topic_name_t, partition_count_t}
+    @type t :: %{
+      supervisor: pid,
+      known_topics: list(topic_tuple_t),
+      topic_workers: %{optional(topic_name_t) => list(pid)}
+    }
+    defstruct supervisor: nil, known_topics: [], topic_workers: %{}
+  end
+
   def start_link(opts \\ []) do
     GenServer.start_link(__MODULE__, opts, Keyword.take(opts, [:name]))
   end
@@ -12,27 +24,41 @@ defmodule Reader.EventQueue.Foreman do
 
     supervisor = Keyword.get(opts, :supervisor, Reader.EventQueue.Supervisor)
 
-    {:ok, {supervisor, []}}
+    {:ok, %State{supervisor: supervisor}}
   end
 
   def known_topics(name \\ __MODULE__) do
     GenServer.call(name, :known_topics)
   end
 
-  def handle_call(:known_topics, _from, {_s, known_topics} = state) do
-    {:reply, known_topics, state}
+  def consumers_for(name \\ __MODULE__, topic) do
+    GenServer.call(name, {:consumers_for, topic})
+  end
+
+  def handle_call(:known_topics, _from, state) do
+    {:reply, state.known_topics, state}
   end
 
-  def handle_info({:topics, new_topics}, {supervisor, known_topics}) do
-    new_topics
-    |> Enum.reject(&(&1 in known_topics))
-    |> Enum.each(&topic_added(supervisor, &1))
+  def handle_call({:consumers_for, topic}, _from, state) do
+    {:reply, state.topic_workers |> Map.get(topic, []), state}
+  end
+
+  def handle_info({:topics, new_topics}, state) do
+    state = new_topics
+    |> Enum.reject(&(&1 in state.known_topics))
+    |> Enum.reduce(state, fn {topic, partitions}, state ->
+      workers = topic_added(state.supervisor, {topic, partitions})
+      %{state | topic_workers: state.topic_workers |> Map.put(topic, workers)}
+    end)
 
-    known_topics
+    state = state.known_topics
     |> Enum.reject(&(&1 in new_topics))
-    |> Enum.each(&topic_removed(supervisor, &1))
+    |> Enum.reduce(state, fn {topic, partitions}, state ->
+      topic_removed(state.supervisor, {topic, partitions})
+      %{state | topic_workers: state.topic_workers |> Map.delete(topic)}
+    end)
 
-    {:noreply, {supervisor, new_topics}}
+    {:noreply, %{state | known_topics: new_topics}}
   end
 
   def handle_info(:topic_subscribe, state) do
@@ -43,9 +69,11 @@ defmodule Reader.EventQueue.Foreman do
   def handle_info(_msg, state), do: {:noreply, state}
 
   defp topic_added(supervisor, {topic, partitions}) do
-    for n <- 1..partitions do
-      Reader.EventQueue.Supervisor.start_child(supervisor, topic, n - 1)
-    end
+    1..partitions
+    |> Enum.map(fn n ->
+      {:ok, pid} = Reader.EventQueue.Supervisor.start_child(supervisor, topic, n - 1)
+      pid
+    end)
   end
 
   def topic_removed(supervisor, {topic, partitions}) do

lib/reader/event_queue/supervisor.ex

@@ -23,10 +23,8 @@ defmodule Reader.EventQueue.Supervisor do
         Logger.info("Reader.EventQueue.Supervisor already started child for #{topic}")
         {:ok, pid}
       error ->
-        Logger.error("Reader.EventQueue.Supervisor error #{topic}: #{inspect error}")
-        :timer.sleep(5)
-        System.halt(1)
-        error
+        msg = "Reader.EventQueue.Supervisor error #{topic}: #{inspect error}"
+        raise msg
     end
   end