lechindianer/kafkaex-lag-exporter
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Return more data for members

Browse Source
This commit is contained in:
Pascal Schmid
2024-03-28 23:45:40 +01:00
parent 9fbf7a98b8
commit 4922da165c
2 changed files with 25 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
lib/kafkaex_lag_exporter/behaviours/kafka_utils_behaviour.ex
View File

@@ -15,7 +15,7 @@ defmodule KafkaexLagExporter.KafkaUtils.Behaviour do
@callback get_consumer_group_names({host :: atom, port :: non_neg_integer}) :: list(binary) @callback get_consumer_group_names({host :: atom, port :: non_neg_integer}) :: list(binary)
@callback topic_names_for_consumer_groups( @callback get_consumer_group_info(
{host :: atom, port :: non_neg_integer}, {host :: atom, port :: non_neg_integer},
list(binary), list(binary),
list(binary) list(binary)
@@ -35,8 +35,8 @@ defmodule KafkaexLagExporter.KafkaUtils.Behaviour do
def get_consumer_group_names({host, port}), do: impl().get_consumer_group_names({host, port}) def get_consumer_group_names({host, port}), do: impl().get_consumer_group_names({host, port})
def topic_names_for_consumer_groups(endpoint, list, consumer_group_names), def get_consumer_group_info(endpoint, list, consumer_group_names),
do: impl().topic_names_for_consumer_groups(endpoint, list, consumer_group_names) do: impl().get_consumer_group_info(endpoint, list, consumer_group_names)
defp impl, defp impl,
do: Application.get_env(:kafkaex_lag_exporter, :kafka_utils, KafkaexLagExporter.KafkaUtils) do: Application.get_env(:kafkaex_lag_exporter, :kafka_utils, KafkaexLagExporter.KafkaUtils)

34
lib/kafkaex_lag_exporter/kafka_utils.ex
View File

@@ -82,25 +82,35 @@ defmodule KafkaexLagExporter.KafkaUtils do
[{_, groups} | _] = KafkaWrapper.list_all_groups([{host, port}], []) [{_, groups} | _] = KafkaWrapper.list_all_groups([{host, port}], [])
groups groups
|> Enum.filter(fn {_, _, protocol} -> protocol == "consumer" end) |> Enum.filter(fn {_, _, protocol} -> protocol === "consumer" end)
|> Enum.map(fn {_, group_name, "consumer"} -> group_name end) |> Enum.map(fn {_, group_name, _} -> group_name end)
end end
@impl true @impl true
def topic_names_for_consumer_groups(endpoint, list \\ [], consumer_group_names) do def get_consumer_group_info(endpoint, list \\ [], consumer_group_names) do
KafkaWrapper.describe_groups(endpoint, list, consumer_group_names) {:ok, group_descriptions} = KafkaWrapper.describe_groups(endpoint, list, consumer_group_names)
|> get_topic_names_for_consumer_groups
end
defp get_topic_names_for_consumer_groups({:ok, group_descriptions}) do
group_descriptions group_descriptions
|> Enum.map(fn %{group_id: consumer_group, members: members} -> [consumer_group, members] end) |> Enum.flat_map(fn %{group_id: consumer_group, members: members} ->
|> Enum.map(fn [consumer_group, members] -> {consumer_group, get_topic_names(members)} end) get_member_info(members)
|> Enum.map(fn {topic_names, consumer_id, member_host} ->
{consumer_group, topic_names, consumer_id, member_host}
end)
end)
end end
defp get_topic_names(members) do @spec get_member_info(
Enum.flat_map(members, fn member -> list(%{client_host: binary, member_assignment: binary, member_id: binary})
KafkaexLagExporter.TopicNameParser.parse_topic_names(member.member_assignment) ) ::
list({topic_names :: list(binary), consumer_id :: binary, member_host :: binary})
defp get_member_info(members) do
Enum.map(members, fn %{
client_host: member_host,
member_assignment: member_assignment,
member_id: consumer_id
} ->
topic_names = KafkaexLagExporter.TopicNameParser.parse_topic_names(member_assignment)
{topic_names, consumer_id, member_host}
end) end)
end end
end end