File size: 3,301 Bytes
8da2481 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | defmodule Plausible.Ingestion.Counters.Buffer do
@moduledoc """
A buffer aggregating counters for internal metrics, within 10 seconds time buckets.
See `Plausible.Ingestion.Counters` for integration.
Flushing is by default possible only once the 10s bucket is complete
(its window has moved). This is to avoid race conditions
when clearing up the buffer on dequeue - because there is no atomic "get and delete",
and items are buffered concurrently, there is a gap between get and delete
in which items written may disappear otherwise.
`aggregate_bucket_fn` and `flush_boundary_fn` control that semantics and
are configurable only for test purposes.
"""
defstruct [:buffer_name, :aggregate_bucket_fn, :flush_boundary_fn]
alias Plausible.Ingestion.Counters.Record
@type t() :: %__MODULE__{}
@type unix_timestamp() :: pos_integer()
@type bucket_fn_opt() ::
{:aggregate_bucket_fn, (NaiveDateTime.t() -> unix_timestamp())}
| {:flush_boundary_fn, (DateTime.t() -> unix_timestamp())}
@ets_opts [
:public,
:ordered_set,
:named_table,
write_concurrency: true
]
@spec new(atom(), [bucket_fn_opt()]) :: t()
def new(buffer_name, opts \\ []) do
^buffer_name = :ets.new(buffer_name, @ets_opts)
aggregate_bucket_fn = Keyword.get(opts, :aggregate_bucket_fn, &bucket_10s/1)
flush_boundary_fn = Keyword.get(opts, :flush_boundary_fn, &previous_10s/1)
%__MODULE__{
buffer_name: buffer_name,
aggregate_bucket_fn: aggregate_bucket_fn,
flush_boundary_fn: flush_boundary_fn
}
end
@spec aggregate(
t(),
binary(),
binary(),
timestamp :: NaiveDateTime.t(),
tracker_script_version :: integer()
) :: t()
def aggregate(
%__MODULE__{buffer_name: buffer_name, aggregate_bucket_fn: bucket_fn} = buffer,
metric,
domain,
timestamp,
tracker_script_version
) do
bucket =
bucket_fn.(timestamp)
:ets.update_counter(
buffer_name,
{bucket, metric, domain, tracker_script_version},
{2, 1},
{{bucket, metric, domain, tracker_script_version}, 0}
)
buffer
end
@spec flush(t(), now :: DateTime.t()) :: [Record.t()]
def flush(
%__MODULE__{buffer_name: buffer_name, flush_boundary_fn: flush_boundary_fn},
now \\ DateTime.utc_now()
) do
boundary = flush_boundary_fn.(now)
match = {{:"$1", :"$2", :"$3", :"$4"}, :"$5"}
guard = {:"=<", :"$1", boundary}
select = {{:"$1", :"$2", :"$3", :"$4", :"$5"}}
match_specs_read = [{match, [guard], [select]}]
match_specs_delete = [{match, [guard], [true]}]
case :ets.select(buffer_name, match_specs_read) do
[] ->
[]
data ->
:ets.select_delete(buffer_name, match_specs_delete)
data
end
end
@spec bucket_10s(NaiveDateTime.t()) :: unix_timestamp()
def bucket_10s(datetime) do
datetime
|> DateTime.from_naive!("Etc/UTC")
|> DateTime.truncate(:second)
|> Map.replace(:second, div(datetime.second, 10) * 10)
|> DateTime.to_unix()
end
@spec previous_10s(DateTime.t()) :: unix_timestamp()
def previous_10s(datetime) do
datetime
|> DateTime.add(-10, :second)
|> DateTime.to_unix()
end
end
|