Malachi.
View Source
A named storage policy: what an administrator says about a topic's segments, rather than something a vnode decides.
A policy carries a retention override and a placement spread attribute, and a topic points at one by
name. This module is only the shape and the rule for accepting one; where the definitions live is
Malachi.Cluster.PolicyRegistry, and which topic points at which name stays in Malachi.Metadata,
because that part is per-topic state and travels with the topic when it moves between vnodes.
Why validation is here and not at the edge
A policy that is stored is a policy every node will read, so the control plane is where the answer to "is this a policy" has to be the same on every replica. Keeping the predicate pure and in one module means the wire, a mix task and the replicated command cannot drift into three different opinions of what a policy is.
What is refused, and why each one is worth refusing:
- An unknown key.
%{retention_ms: 1000}is a plausible typo for:retentionand would be stored, read back, and silently do nothing forever. There is no valid reason to keep a key the code does not read. - A bound that is not a non-negative integer.
max_bytes: "10GB"is the shape an operator reaches for first.Malachi.Cluster.Retentioncompares it with>, which never matches against a binary, so the rule would simply never fire. - A negative bound. An age or a byte budget below zero expires everything the rule can see, which is the opposite of what anyone typing a negative number wants.
- A bound below its field's floor.
retention.segment_max_age_msrolls a topic's active segment, and a roll every few seconds would multiply the topic's segments for no retention a sweep running once a minute could deliver.
nil is allowed for a bound and means that rule is off, which is how a policy overrides one of the
two limits without inheriting the other.
Which fields exist, and since when
The settable fields are one table (fields/0), each with the machine version that introduced it, and
validation reads that table at the group's effective version (validate/3). See the comment above
the table for why a field is never added any other way.
Summary
Types
One settable field: its flat name (what the wire, the CLI and the dashboard say), its path in the
policy map, its type, the machine version that introduced it (since), and, for a bound, the
smallest value it accepts other than nil (min, default 0).
How a field's value is checked and carried: a :bound is a non-negative integer or nil (the rule is
off); an :attribute is a non-empty string or nil.
A policy's name, as an administrator chose it.
A named storage policy: per-topic retention overrides and a placement spread attribute. Both keys are optional; a policy applies only the ones it sets, falling back to the global defaults otherwise.
Functions
A topic's effective placement spread attribute and where it comes from: the policy's :spread_by
when the policy sets the key (nil included, which turns spreading off for that topic), otherwise
global. A topic with no policy, or bound to a name nothing defines, passes nil and gets global,
which is what placement does (Malachi.Broker): placement fails open.
Builds a policy from flat {name, value} pairs, the form the wire, the CLI and the dashboard carry.
The fields policy sets, as {name, value} pairs in table order: the inverse of from_pairs/2.
Whether policy is a policy this build can act on: validate/3 at code_version/0.
Whether policy is valid at machine version: validate/3 answering :ok.
Whether name can name a policy: a non-empty binary.
Checks policy against the fields of fields introduced at or below version.
Types
@type field() :: %{ :name => String.t(), :path => [atom(), ...], :type => field_type(), :since => non_neg_integer(), optional(:min) => non_neg_integer() }
One settable field: its flat name (what the wire, the CLI and the dashboard say), its path in the
policy map, its type, the machine version that introduced it (since), and, for a bound, the
smallest value it accepts other than nil (min, default 0).
@type field_type() :: :bound | :attribute
How a field's value is checked and carried: a :bound is a non-negative integer or nil (the rule is
off); an :attribute is a non-empty string or nil.
@type name() :: String.t()
A policy's name, as an administrator chose it.
@type t() :: %{ optional(:retention) => %{ optional(:max_age_ms) => non_neg_integer() | nil, optional(:max_bytes) => non_neg_integer() | nil, optional(:segment_max_age_ms) => pos_integer() | nil }, optional(:spread_by) => String.t() | nil }
A named storage policy: per-topic retention overrides and a placement spread attribute. Both keys are optional; a policy applies only the ones it sets, falling back to the global defaults otherwise.
Functions
A topic's effective placement spread attribute and where it comes from: the policy's :spread_by
when the policy sets the key (nil included, which turns spreading off for that topic), otherwise
global. A topic with no policy, or bound to a name nothing defines, passes nil and gets global,
which is what placement does (Malachi.Broker): placement fails open.
Examples
iex> Malachi.Cluster.Policy.effective_spread_by(%{spread_by: "rack"}, "dc")
{"rack", :policy}
iex> Malachi.Cluster.Policy.effective_spread_by(%{retention: %{}}, "dc")
{"dc", :global}field(name, fields \\ [%{name: "retention.max_age_ms", path: [:retention, :max_age_ms], since: 3, type: :bound}, %{name: "retention.max_bytes", path: [:retention, :max_bytes], since: 3, type: :bound}, %{min: 60000, name: "retention.segment_max_age_ms", path: [:retention, :segment_max_age_ms], since: 6, type: :bound}, %{name: "spread_by", path: [:spread_by], since: 3, type: :attribute}])
The field named name in fields, or nil.
@spec fields() :: [field()]
Every field a policy can set, in table order (see field/0).
from_pairs(pairs, fields \\ [%{name: "retention.max_age_ms", path: [:retention, :max_age_ms], since: 3, type: :bound}, %{name: "retention.max_bytes", path: [:retention, :max_bytes], since: 3, type: :bound}, %{min: 60000, name: "retention.segment_max_age_ms", path: [:retention, :segment_max_age_ms], since: 6, type: :bound}, %{name: "spread_by", path: [:spread_by], since: 3, type: :attribute}])
@spec from_pairs([{String.t(), term()}], [field()]) :: {:ok, t()} | {:error, {:unknown_policy_field | :invalid_policy_field | :duplicate_policy_field, String.t()}}
Builds a policy from flat {name, value} pairs, the form the wire, the CLI and the dashboard carry.
A field absent from pairs is absent from the policy (it inherits the global value), nil turns its
rule off, and 0 is a real budget: the three stay distinct. Refuses an unknown name
({:unknown_policy_field, name}), a value its type does not allow ({:invalid_policy_field, name}) and
a name given twice ({:duplicate_policy_field, name}). Checks the table of this build only: whether the
cluster's effective version admits each field is the store's decision (validate/3).
to_pairs(policy, fields \\ [%{name: "retention.max_age_ms", path: [:retention, :max_age_ms], since: 3, type: :bound}, %{name: "retention.max_bytes", path: [:retention, :max_bytes], since: 3, type: :bound}, %{min: 60000, name: "retention.segment_max_age_ms", path: [:retention, :segment_max_age_ms], since: 6, type: :bound}, %{name: "spread_by", path: [:spread_by], since: 3, type: :attribute}])
The fields policy sets, as {name, value} pairs in table order: the inverse of from_pairs/2.
Whether policy is a policy this build can act on: validate/3 at code_version/0.
Examples
iex> Malachi.Cluster.Policy.valid?(%{retention: %{max_age_ms: 604_800_000}, spread_by: "rack"})
true
iex> Malachi.Cluster.Policy.valid?(%{retention: %{max_bytes: "10GB"}})
false
iex> Malachi.Cluster.Policy.valid?(%{retention_ms: 1_000})
false
iex> Malachi.Cluster.Policy.valid?(%{retention: %{max_bytes: nil}})
true
iex> Malachi.Cluster.Policy.valid?(%{spread_by: :rack})
falsevalid?(policy, version, fields \\ [%{name: "retention.max_age_ms", path: [:retention, :max_age_ms], since: 3, type: :bound}, %{name: "retention.max_bytes", path: [:retention, :max_bytes], since: 3, type: :bound}, %{min: 60000, name: "retention.segment_max_age_ms", path: [:retention, :segment_max_age_ms], since: 6, type: :bound}, %{name: "spread_by", path: [:spread_by], since: 3, type: :attribute}])
@spec valid?(term(), non_neg_integer(), [field()]) :: boolean()
Whether policy is valid at machine version: validate/3 answering :ok.
Whether name can name a policy: a non-empty binary.
Examples
iex> Malachi.Cluster.Policy.valid_name?("durable")
true
iex> Malachi.Cluster.Policy.valid_name?("")
falsevalidate(policy, version, fields \\ [%{name: "retention.max_age_ms", path: [:retention, :max_age_ms], since: 3, type: :bound}, %{name: "retention.max_bytes", path: [:retention, :max_bytes], since: 3, type: :bound}, %{min: 60000, name: "retention.segment_max_age_ms", path: [:retention, :segment_max_age_ms], since: 6, type: :bound}, %{name: "spread_by", path: [:spread_by], since: 3, type: :attribute}])
@spec validate(term(), non_neg_integer(), [field()]) :: :ok | {:error, :invalid_policy | {:unsupported_policy_field, String.t(), non_neg_integer()}}
Checks policy against the fields of fields introduced at or below version.
Answers {:error, {:unsupported_policy_field, name, since}} when the policy sets a field this build
knows but version does not yet admit, so a caller can say "finish the rolling upgrade" rather than
"invalid", and {:error, :invalid_policy} for anything else that is not a policy: an unknown key, a
bound that is neither nil nor a non-negative integer at or above its field's min, an attribute
that is not a non-empty string or nil, or a value that is not a map where the table has a nested map.