View Source Plausible.Ingestion.Counters.Buffer (Plausible v0.0.1)

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.

Summary

Types

bucket_fn_opt()
t()
unix_timestamp()

Functions

aggregate(buffer, metric, domain, timestamp)
bucket_10s(datetime)
flush(buffer, now \\ DateTime.utc_now())
new(buffer_name, opts \\ [])
previous_10s(datetime)

Types

Link to this type

bucket_fn_opt()

View Source 
@type bucket_fn_opt() ::
  {:aggregate_bucket_fn, (NaiveDateTime.t() -> unix_timestamp())}
  | {:flush_boundary_fn, (DateTime.t() -> unix_timestamp())}
Link to this type

t()

View Source 
@type t() :: %Plausible.Ingestion.Counters.Buffer{
  aggregate_bucket_fn: term(),
  buffer_name: term(),
  flush_boundary_fn: term()
}
Link to this type

unix_timestamp()

View Source 
@type unix_timestamp() :: pos_integer()

Functions

Link to this function

aggregate(buffer, metric, domain, timestamp)

View Source 
@spec aggregate(t(), binary(), binary(), timestamp :: NaiveDateTime.t()) :: t()
Link to this function

bucket_10s(datetime)

View Source 
@spec bucket_10s(NaiveDateTime.t()) :: unix_timestamp()
Link to this function

flush(buffer, now \\ DateTime.utc_now())

View Source 
@spec flush(t(), now :: DateTime.t()) :: [Plausible.Ingestion.Counters.Record.t()]
Link to this function

new(buffer_name, opts \\ [])

View Source 
@spec new(atom(), [bucket_fn_opt()]) :: t()
Link to this function

previous_10s(datetime)

View Source 
@spec previous_10s(DateTime.t()) :: unix_timestamp()