Malachi.Policies (malachi v0.22.7)

The administrative entry point for storage policies: define and delete a named policy, bind a topic to one, and read back what a topic's retention actually is. The wire keys, mix malachi.policy and the dashboard are thin adapters over this module, so validation, the error vocabulary, the audit trail and the log line are written once.

Where each half lives

A policy's definition is one object for the cluster, in Malachi.Cluster.PolicyStore. Which policy a topic uses is that topic's own state, in its vnode's Malachi.Metadata ({:bind_topic_policy, topic, name}), so it travels with the topic when a vnode split moves it.

The checks a replicated state machine cannot make

Neither side can see the other: a metadata vnode cannot read the policy store, and the store cannot read the topics. So this module checks what they cannot, before it submits:

bind/3 refuses a name the store does not define (:no_such_policy). A topic bound to an undefined name holds its data: nothing expires under it (Malachi.Cluster.Retention.effective/4).
delete/3 refuses a policy some topic is still bound to ({:policy_in_use, topics}) unless it is given force: true, for the same reason.

The delete asks the vnodes themselves, read linearizably, and refuses ({:bindings_unavailable, _}) when one of them does not answer or a split is moving topics between them. The bind reads this node's replica of the store. Both are still checks made before a submit, so a bind racing a delete can leave a topic bound to a name that is gone. That topic then shows up in malachi_retention_unresolved_policy_sweeps_total and in topic_policy/1 as resolution: :unresolved, and the operator's backstop bound (MALACHI_RETENTION_UNRESOLVED_POLICY_MAX_AGE_MS) is the only one that applies to it.

Upgrades

Binding needs machine version 4. Until every member of a topic's metadata group runs a release that implements it and the operator has lifted the version pin, the control plane refuses it identically on every member with {:unsupported_command, key, 4, effective}, which reason_string/1 turns into "finish the rolling upgrade". Defining and deleting policies need version 3 only.

Summary

Types

actor()

Who asked for a change, as the audit log records it: a username, or cli@<node>.

reason()

Why a request was refused. reason_string/1 renders every one of them.

topic_policy()

A topic's policy as read back by topic_policy/1.

Functions

bind(topic, name, actor)

Binds topic to the policy name, or detaches it with nil. Refuses a name the store does not define, so a typo is reported here rather than holding the topic's data forever.

define(name, policy, actor)

Defines, or replaces, the policy name.

delete(name, actor, opts \\ [])

Deletes the policy name. Refuses with {:policy_in_use, topics} while any topic is bound to it, unless opts has force: true. Deleting a name nothing defines is :ok.

effective_pairs(topic_policy)

The effective value of every field Malachi.Cluster.Policy.fields/0 lists, as {name, value, origin} in table order: the flat form a surface prints or encodes. A field with no effective value to report is left out.

list()

Every definition, as a map from name to policy, read from this node's replica of the store.

reason_string(reason)

A stable, human-readable rendering of a refusal, the same on the wire, in the CLI and on the dashboard. It starts with the reason's name, so a client can still branch on it.

topic_policy(topic)

What topic's retention and placement spread actually are, and where each value comes from.

Types

actor()

@type actor() :: String.t()

Who asked for a change, as the audit log records it: a username, or cli@<node>.

reason()

@type reason() ::
  :invalid_policy_name
  | :invalid_topic
  | :invalid_policy
  | :no_such_policy
  | :no_such_topic
  | :migrating
  | {:policy_in_use, [String.t()]}
  | {:bindings_unavailable, term()}
  | {:unknown_policy_field | :invalid_policy_field | :duplicate_policy_field,
     String.t()}
  | {:unsupported_policy_field, String.t(), non_neg_integer(),
     non_neg_integer()}
  | {:unsupported_command, term(), non_neg_integer(), non_neg_integer()}
  | term()

Why a request was refused. reason_string/1 renders every one of them.

topic_policy()

@type topic_policy() :: %{
  topic: String.t(),
  policy: Malachi.Cluster.Policy.name() | nil,
  definition: Malachi.Cluster.Policy.t() | nil,
  resolution: Malachi.Cluster.Retention.resolution(),
  retention: %{
    required(atom()) =>
      {non_neg_integer() | nil, Malachi.Cluster.Retention.origin()}
  },
  spread_by: {String.t() | nil, :policy | :global}
}

A topic's policy as read back by topic_policy/1.

Functions

bind(topic, name, actor)

@spec bind(String.t(), Malachi.Cluster.Policy.name() | nil, actor()) ::
  :ok | {:error, reason()}

Binds topic to the policy name, or detaches it with nil. Refuses a name the store does not define, so a typo is reported here rather than holding the topic's data forever.

define(name, policy, actor)

@spec define(
  Malachi.Cluster.Policy.name(),
  Malachi.Cluster.Policy.t() | [{String.t(), term()}],
  actor()
) :: :ok | {:error, reason()}

Defines, or replaces, the policy name.

policy is either a policy map or the flat {field, value} pairs every surface carries (Malachi.Cluster.Policy.from_pairs/2). The store validates the definition against the cluster's effective machine version.

delete(name, actor, opts \\ [])

@spec delete(Malachi.Cluster.Policy.name(), actor(), keyword()) ::
  :ok | {:error, reason()}

Deletes the policy name. Refuses with {:policy_in_use, topics} while any topic is bound to it, unless opts has force: true. Deleting a name nothing defines is :ok.

effective_pairs(topic_policy)

@spec effective_pairs(topic_policy()) :: [{String.t(), term(), atom()}]

list()

@spec list() ::
  {:ok,
   %{required(Malachi.Cluster.Policy.name()) => Malachi.Cluster.Policy.t()}}
  | {:error, reason()}

Every definition, as a map from name to policy, read from this node's replica of the store.

reason_string(reason)

@spec reason_string(reason()) :: String.t()

A stable, human-readable rendering of a refusal, the same on the wire, in the CLI and on the dashboard. It starts with the reason's name, so a client can still branch on it.

Examples

iex> Malachi.Policies.reason_string({:policy_in_use, ["audit", "events"]})
"policy_in_use: audit, events"

iex> Malachi.Policies.reason_string(:no_such_policy)
"no_such_policy"

topic_policy(topic)

@spec topic_policy(String.t()) :: {:ok, topic_policy()} | {:error, reason()}

What topic's retention and placement spread actually are, and where each value comes from.

The retention bounds are computed by Malachi.Cluster.Retention.effective/4, the function the sweep applies, over this node's configured global limits and its replica of the policy store, so what this reports is what the data sees. A store that cannot be read is an error here, never an empty store: answering "no policies" would report the global limits for a topic whose policy holds its data longer.