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

Replace Mox with Patch

Browse Source
This commit is contained in:
Pascal Schmid
2024-03-27 22:19:32 +01:00
parent 207926ed83
commit 7b40cbedd1
11 changed files with 66 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,7 +19,7 @@ defmodule KafkaexLagExporter.KafkaUtils.Behaviour do
{host :: atom, port :: non_neg_integer},
list(binary),
list(binary)
) :: list(binary)
) :: list({consumer_group :: binary, topics :: list(binary)})
def connection(client), do: impl().connection(client)

6
lib/kafkaex_lag_exporter/consumer_offset.ex
View File

@@ -29,10 +29,10 @@ defmodule KafkaexLagExporter.ConsumerOffset do
def handle_info(:tick, state) do
[endpoint | _] = state.endpoints
{consumer_lags, consumer_lag_sum} = KafkaexLagExporter.ConsumerOffsetFetcher.get(endpoint)
%{sum: lag_sum, lags: lags} = KafkaexLagExporter.ConsumerOffsetFetcher.get(endpoint)
KafkaexLagExporter.Metrics.group_lag_per_partition(endpoint, consumer_lags)
KafkaexLagExporter.Metrics.group_sum_lag(endpoint, consumer_lag_sum)
KafkaexLagExporter.Metrics.group_lag_per_partition(endpoint, lags)
KafkaexLagExporter.Metrics.group_sum_lag(endpoint, lag_sum)
Process.send_after(self(), :tick, @interval)

21
lib/kafkaex_lag_exporter/consumer_offset_fetcher.ex
View File

@@ -3,19 +3,24 @@ defmodule KafkaexLagExporter.ConsumerOffsetFetcher do
require Logger
# TODO: change return type
@spec get(KafkaexLagExporter.KafkaWrapper.endpoint()) :: {any(), any()}
alias KafkaexLagExporter.KafkaUtils
# TODO fix type
@spec get(KafkaexLagExporter.KafkaWrapper.Behaviour.endpoint()) :: %{
lags: list(binary),
sum: list(binary)
}
def get(endpoint) do
consumer_group_names = KafkaexLagExporter.KafkaUtils.get_consumer_group_names(endpoint)
consumer_group_names = KafkaUtils.get_consumer_group_names(endpoint)
consumer_lags =
KafkaexLagExporter.KafkaUtils.topic_names_for_consumer_groups(
KafkaUtils.topic_names_for_consumer_groups(
endpoint,
[],
consumer_group_names
)
|> Enum.map(fn [consumer_group, topics] ->
[consumer_group, get_lag_for_consumer(consumer_group, topics)]
|> Enum.map(fn {consumer_group, topics} ->
{consumer_group, get_lag_for_consumer(consumer_group, topics)}
end)
consumer_lag_sum = get_lag_for_consumer_sum(consumer_lags)
@@ -26,13 +31,13 @@ defmodule KafkaexLagExporter.ConsumerOffsetFetcher do
defp get_lag_for_consumer(consumer_group, topics) do
topics
|> Enum.flat_map(fn topic ->
KafkaexLagExporter.KafkaUtils.lag(topic, consumer_group, :client1)
KafkaUtils.lag(topic, consumer_group, :client1)
end)
end
defp get_lag_for_consumer_sum(lags_per_consumer_group) do
lags_per_consumer_group
|> Enum.map(fn [topic, lag_per_partition] -> [topic, sum_topic_lag(lag_per_partition, 0)] end)
|> Enum.map(fn {topic, lag_per_partition} -> {topic, sum_topic_lag(lag_per_partition, 0)} end)
end
defp sum_topic_lag([], acc), do: acc