TemporalSdk.Cluster (temporal_sdk v0.1.15)

SDK cluster configuration and management module.

The SDK cluster is a supervised group of processes that facilitates SDK interaction with the (clustered) Temporal service Temporal server(s). Interaction includes polling and processing Temporal task executions and handling gRPC Temporal API services. SDK cluster supervisors are supervised by the SDK node supervisor. SDK cluster supervisor supervises task workers. SDK cluster can only be started at runtime and cannot be started dynamically.

SDK cluster responsibilities:

configure and start the SDK cluster statistics telemetry poller,
configure SDK cluster-level fixed window rate limiter time windows,
configure SDK cluster-specific options, such as workflow_scope,
configure, start and supervise the gRPC client connections pool connecting to the Temporal server(s),
configure, start and supervise task worker processes.

SDK Cluster Configuration

cluster - cluster-specific configuration options.

client - SDK gRPC client configuration.

activities - list of runtime activity task workers.

nexuses - list of runtime nexus task workers.

workflows - list of runtime workflow task workers.

Example cluster_1 SDK cluster configuration:

Elixir

config :temporal_sdk,
  clusters: [
    cluster_1: [
      cluster: [telemetry_poll_interval: {20, :second}],
      client: %{
        :adapter =>
          {:temporal_sdk_grpc_adapter_gun_pool, [{:endpoints, [{{127, 0, 0, 1}, 7233}]}]}
      },
      activities: [[task_queue: "activity_tq1"], [task_queue: "activity_tq2"]],
      nexuses: [[task_queue: "nexus_tq"]],
      workflows: [[task_queue: "workflow_tq"]]
    ]
  ]

Erlang

{temporal_sdk, [
    {clusters, [
        {cluster_1, [
            {cluster, [{telemetry_poll_interval, {20, second}}]},
            {client, #{
                adapter =>
                    {temporal_sdk_grpc_adapter_gun_pool, [
                        {endpoints, [{{127, 0, 0, 1}, 7233}]}
                    ]}
            }},
            {activities, [[{task_queue, "activity_tq1"}], [{task_queue, "activity_tq2"}]]},
            {nexuses, [[{task_queue, "nexus_tq"}]]},
            {workflows, [[{task_queue, "workflow_tq"}]]}
        ]}
    ]}
]}

Example configuration above sets the following options for the cluster_1 SDK cluster:

telemetry_poll_interval is set to 20 seconds,
the SDK gRPC client will utilize the :temporal_sdk_grpc_adapter_gun_pool HTTP/2 adapter to connect to the Temporal server running on 127.0.0.1:7233,
two activity runtime task workers are started to poll Temporal server on "activity_tq1" and "activity_tq2" activity task queues,
nexus runtime task worker is started to poll Temporal server on "nexus_tq" nexus task queue,
workflow runtime task worker is started to poll Temporal server on "workflow_tq" workflow task queue.

Multiple virtual SDK clusters can be configured, each with its own cluster-specific configuration, connecting to the same Temporal server.

See GitHub: SDK samples repository "Payload Converter" sample for SDK virtual clusters example:

Elixir: lib/payload_converter,
Erlang: src/payload_converter.

Cluster-Specific Configuration

enable_single_distributed_workflow_execution - enables single workflow execution per SDK cluster, see SDK Architecture - Workflow Execution Scope for details. If not set, the value is inherited from the top-level SDK node enable_single_distributed_workflow_execution configuration option, which is set to true by default.

limiter_time_windows - SDK cluster-level fixed window rate limiter time windows configuration. Values specified here also serve as the time intervals at which telemetry events [temporal_sdk, task_counter, cluster] are emitted. By default, the fixed window rate limiter time window is set to 60 seconds. See also :temporal_sdk_node limiter_time_windows configuration option and :temporal_sdk_limiter.

telemetry_poll_interval - the time interval at which the SDK cluster telemetry poller polls for SDK cluster statistics and emits [temporal_sdk, cluster, stats] telemetry event. Default poll time interval is 10 seconds.

workflow_scope - the name of the workflow scope for the given cluster. Used in the SDK node scope_config configuration option. By default, it is set to the cluster name.

Summary

Functions

is_started(cluster)

Returns true if the given cluster is started, {error, invalid_cluster} error otherwise.

list()

Returns a list of names of started clusters.

stats(cluster)

Retrieves the current concurrency statistics from the SDK cluster-level concurrency rate limiter.

Functions

is_started(cluster)

Returns true if the given cluster is started, {error, invalid_cluster} error otherwise.

Example:

Elixir

iex(1)> TemporalSdk.Cluster.is_started(:cluster_1)
true
iex(2)> TemporalSdk.Cluster.is_started(:invalid)
{:error, :invalid_cluster}

Erlang

1> temporal_sdk_cluster:is_started(cluster_1).
true
2> temporal_sdk_cluster:is_started(invalid).
{error,invalid_cluster}

list()

Returns a list of names of started clusters.

Example:

Elixir

iex(1)> TemporalSdk.Cluster.list()
[:cluster_1]

Erlang

1> temporal_sdk_cluster:list().
[cluster_1]

stats(cluster)

Retrieves the current concurrency statistics from the SDK cluster-level concurrency rate limiter.

Function returns a t:temporal_sdk_limiter:stats/0 map, where the map key is the t:temporal_sdk_limiter:temporal_limitable/0 and the value is the number of currently running limitable task executions at the SDK cluster-level. The number of task executions at the SDK cluster-level corresponds to the sum of task executions across all runtime and dynamic task workers belonging to the given SDK cluster.

Elixir

iex(1)> TemporalSdk.Cluster.stats(:cluster_1)
{:ok,
 %{
   workflow: 1,
   nexus: 0,
   activity_regular: 5,
   activity_session: 0,
   activity_eager: 0,
   activity_direct: 0
 }}
iex(2)> TemporalSdk.Cluster.stats(:invalid)
{:error, :invalid_cluster}

Erlang

1> temporal_sdk_cluster:stats(cluster_1).
{ok,#{activity_direct => 0,activity_eager => 0,
      activity_regular => 5,activity_session => 0,nexus => 0,
      workflow => 1}}
2> temporal_sdk_cluster:stats(invalid).
{error,invalid_cluster}