TemporalSdk.Worker (temporal_sdk v0.2.0)
View SourceTask worker module.
The task worker is a supervised group of processes that manages processing of the Temporal task executions. Temporal task executions are managed by the Temporal server. Task worker processes are supervised by the SDK cluster supervisor.
Task worker responsibilities:
- configure and start the task worker statistics telemetry poller,
- configure worker-level fixed window rate limiter time windows,
- configure worker-level options, such as worker Temporal
:namespace, - configure Temporal task poller rate limiter,
- configure, start and supervise Temporal task pollers pool,
- configure and enforce rate limiters limits,
- setup and spawn task executors for Temporal task executions polled by the task pollers.
Task Worker Configuration
Summary
Functions
Updates the dynamic configuration of the rate limiter on the Erlang nodes list.
Functions
@spec count( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :activity | :nexus | :workflow ) :: {:ok, non_neg_integer()} | :temporal_sdk_worker.invalid_error()
@spec list( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :activity | :nexus | :workflow ) :: {:ok, [:temporal_sdk_worker.worker_id()]} | :temporal_sdk_worker.invalid_error()
@spec options( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :temporal_sdk_worker.worker_type(), worker_id :: :temporal_sdk_worker.worker_id() ) :: {:ok, :temporal_sdk_worker.opts()} | :temporal_sdk_worker.invalid_error()
@spec set_limiter_config( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :temporal_sdk_worker.worker_type(), worker_id :: :temporal_sdk_worker.worker_id(), new_limiter_config :: :temporal_sdk_worker.limiter_config() | :temporal_sdk_worker.limiter_config_as_list(), nodes :: [node()] ) :: :ok | [ {:ok, :temporal_sdk_worker.set_limiter_config_ret()} | {:error, {:erpc, reason :: term()}} | term() ]
Updates the dynamic configuration of the rate limiter on the Erlang nodes list.
Function applies set_limiter_config/4 on the Erlang nodes provided as a list using
:erpc.multicall/4.
Returns :ok if the multiple call configuration update operation was successful on all nodes.
Returns an :erpc.multicall/4 formatted error if the call operation fails on any of the nodes.
Successful example:
Elixir
iex(a@host)1> n = Node.list() ++ [Node.self()]
[:b@host, :a@host]
iex(a@host)2> TemporalSdk.Worker.set_limiter_config(:cluster_1, :activity, "worker",
[{:limits, %{}}], n)
:okErlang
(a@host)1> N = nodes() ++ [node()].
[b@host,a@host]
(a@host)2> temporal_sdk_worker:set_limiter_config(cluster_1, activity, "worker", #{limits => #{}}, N).
okExample with 3 nodes, where the first node update is successful, the second node does not have a worker started, and the third node is not accessible:
Elixir
iex(a@host)1> n = [:b@host, :a@host, :c@host]
[:b@host, :a@host, :c@host]
iex(a@host)2> TemporalSdk.Worker.set_limiter_config(:cluster_1, :activity, "worker",
[{:limits, %{}}], n)
[ok: :ok, ok: :invalid_worker, error: {:erpc, :noconnection}]Erlang
(a@host)1> N = [b@host, a@host, c@host].
[b@host,a@host,c@host]
(a@host)3> temporal_sdk_worker:set_limiter_config(cluster_1, activity, "worker", #{limits => #{}}, N).
[{ok,ok},{ok,invalid_worker},{error,{erpc,noconnection}}]@spec start( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :activity | :nexus | :workflow, worker_opts :: :temporal_sdk_worker.opts() | :temporal_sdk_worker.opts_as_list() ) :: {:ok, :temporal_sdk_worker.opts()} | {:invalid_opts, map()} | :temporal_sdk_worker.invalid_error() | :supervisor.startchild_ret()
@spec start( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :activity | :nexus | :workflow, worker_opts :: :temporal_sdk_worker.opts() | :temporal_sdk_worker.opts_as_list(), nodes :: [node()] ) :: :ok
@spec stats( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :temporal_sdk_worker.worker_type(), worker_id :: :temporal_sdk_worker.worker_id() ) :: {:ok, :temporal_sdk_limiter.stats()} | :temporal_sdk_worker.invalid_error()
@spec terminate( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :activity | :nexus | :workflow, worker_id :: :temporal_sdk_worker.worker_id() ) :: :ok | {:error, :invalid_cluster | :not_found | :simple_one_for_one}
@spec terminate( cluster :: :temporal_sdk_cluster.cluster_name(), worker_type :: :activity | :nexus | :workflow, worker_id :: :temporal_sdk_worker.worker_id(), nodes :: [node()] ) :: :ok