Support for custom metrics in workflows and minor README fixes

Author: cretz

README.md

@@ -114,7 +114,7 @@ class SayHelloActivity < Temporalio::Activity::Definition
 end
 ```
 
-Now to create the workflow, put the following in `say_hello_workflow.rb`:
+Workflows are also classes. To create the workflow, put the following in `say_hello_workflow.rb`:
 
 ```ruby
 require 'temporalio/workflow'
@@ -178,7 +178,7 @@ client = Temporalio::Client.connect('localhost:7233', 'my-namespace')
 # Run workflow
 result = client.execute_workflow(
   SayHelloWorkflow,
-  'Temporal',
+  'Temporal', # This is the workflow argument
   id: 'my-workflow-id',
   task_queue: 'my-task-queue'
 )
@@ -641,6 +641,8 @@ from workflows including:
   be called outside of a workflow without raising an exception.
 * `info` - Immutable workflow information.
 * `logger` - A Ruby logger that adds contextual information and takes care not to log on replay.
+* `metric_meter` - A metric meter for making custom metrics that adds contextual information and takes care not to log
+  on replay.
 * `random` - A deterministic `Random` instance.
 * `memo` - A read-only hash of the memo (updated via `upsert_memo`).
 * `search_attributes` - A read-only `SearchAttributes` collection (updated via `upsert_search_attributes`).

temporalio/lib/temporalio/internal/worker/workflow_instance.rb

@@ -18,6 +18,7 @@
 require 'temporalio/internal/worker/workflow_instance/inbound_implementation'
 require 'temporalio/internal/worker/workflow_instance/outbound_implementation'
 require 'temporalio/internal/worker/workflow_instance/replay_safe_logger'
+require 'temporalio/internal/worker/workflow_instance/replay_safe_metric'
 require 'temporalio/internal/worker/workflow_instance/scheduler'
 require 'temporalio/retry_policy'
 require 'temporalio/scoped_logger'
@@ -64,6 +65,7 @@ def initialize(details)
           end
           @logger = ReplaySafeLogger.new(logger: details.logger, instance: self)
           @logger.scoped_values_getter = proc { scoped_logger_info }
+          @runtime_metric_meter = details.metric_meter
           @scheduler = Scheduler.new(self)
           @payload_converter = details.payload_converter
           @failure_converter = details.failure_converter
@@ -189,6 +191,18 @@ def patch(patch_id:, deprecated:)
           end
         end
 
+        def metric_meter
+          @metric_meter ||= ReplaySafeMetric::Meter.new(
+            @runtime_metric_meter.with_additional_attributes(
+              {
+                namespace: info.namespace,
+                task_queue: info.task_queue,
+                workflow_type: info.workflow_type
+              }
+            )
+          )
+        end
+
         private
 
         def run_in_scheduler(&)

temporalio/lib/temporalio/internal/worker/workflow_instance/context.rb

@@ -136,6 +136,10 @@ def memo
             @instance.memo
           end
 
+          def metric_meter
+            @instance.metric_meter
+          end
+
           def now
             @instance.now
           end

temporalio/lib/temporalio/internal/worker/workflow_instance/details.rb

@@ -6,16 +6,17 @@ module Worker
       class WorkflowInstance
         # Details needed to instantiate a {WorkflowInstance}.
         class Details
-          attr_reader :namespace, :task_queue, :definition, :initial_activation, :logger, :payload_converter,
-                      :failure_converter, :interceptors, :disable_eager_activity_execution, :illegal_calls,
-                      :workflow_failure_exception_types
+          attr_reader :namespace, :task_queue, :definition, :initial_activation, :logger, :metric_meter,
+                      :payload_converter, :failure_converter, :interceptors, :disable_eager_activity_execution,
+                      :illegal_calls, :workflow_failure_exception_types
 
           def initialize(
             namespace:,
             task_queue:,
             definition:,
             initial_activation:,
             logger:,
+            metric_meter:,
             payload_converter:,
             failure_converter:,
             interceptors:,
@@ -28,6 +29,7 @@ def initialize(
             @definition = definition
             @initial_activation = initial_activation
             @logger = logger
+            @metric_meter = metric_meter
             @payload_converter = payload_converter
             @failure_converter = failure_converter
             @interceptors = interceptors

temporalio/lib/temporalio/internal/worker/workflow_instance/replay_safe_metric.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+require 'temporalio/scoped_logger'
+
+module Temporalio
+  module Internal
+    module Worker
+      class WorkflowInstance
+        # Wrapper for a metric that does not log on replay.
+        class ReplaySafeMetric < SimpleDelegator
+          def record(value, additional_attributes: nil)
+            return if Temporalio::Workflow.in_workflow? && Temporalio::Workflow::Unsafe.replaying?
+
+            super
+          end
+
+          def with_additional_attributes(additional_attributes)
+            ReplaySafeMetric.new(super)
+          end
+
+          class Meter < SimpleDelegator
+            def create_metric(
+              metric_type,
+              name,
+              description: nil,
+              unit: nil,
+              value_type: :integer
+            )
+              ReplaySafeMetric.new(super)
+            end
+
+            def with_additional_attributes(additional_attributes)
+              Meter.new(super)
+            end
+          end
+        end
+      end
+    end
+  end
+end

temporalio/lib/temporalio/internal/worker/workflow_worker.rb

@@ -56,6 +56,7 @@ def initialize(worker:, bridge_worker:, workflow_definitions:)
             workflow_definitions:,
             bridge_worker:,
             logger: worker.options.logger,
+            metric_meter: worker.options.client.connection.options.runtime.metric_meter,
             data_converter: worker.options.client.data_converter,
             deadlock_timeout: worker.options.debug_mode ? nil : 2.0,
             # TODO(cretz): Make this more performant for the default set?
@@ -140,18 +141,19 @@ def apply_codec_on_payload_visit(payload_or_payloads, &)
         end
 
         class State
-          attr_reader :workflow_definitions, :bridge_worker, :logger, :data_converter, :deadlock_timeout,
+          attr_reader :workflow_definitions, :bridge_worker, :logger, :metric_meter, :data_converter, :deadlock_timeout,
                       :illegal_calls, :namespace, :task_queue, :disable_eager_activity_execution,
                       :workflow_interceptors, :workflow_failure_exception_types
 
           def initialize(
-            workflow_definitions:, bridge_worker:, logger:, data_converter:, deadlock_timeout:,
+            workflow_definitions:, bridge_worker:, logger:, metric_meter:, data_converter:, deadlock_timeout:,
             illegal_calls:, namespace:, task_queue:, disable_eager_activity_execution:,
             workflow_interceptors:, workflow_failure_exception_types:
           )
             @workflow_definitions = workflow_definitions
             @bridge_worker = bridge_worker
             @logger = logger
+            @metric_meter = metric_meter
             @data_converter = data_converter
             @deadlock_timeout = deadlock_timeout
             @illegal_calls = illegal_calls

temporalio/lib/temporalio/worker/workflow_executor/thread_pool.rb

@@ -203,6 +203,7 @@ def create_instance(initial_activation, worker_state)
                 definition:,
                 initial_activation:,
                 logger: worker_state.logger,
+                metric_meter: worker_state.metric_meter,
                 payload_converter: worker_state.data_converter.payload_converter,
                 failure_converter: worker_state.data_converter.failure_converter,
                 interceptors: worker_state.workflow_interceptors,

temporalio/lib/temporalio/workflow.rb

@@ -232,6 +232,12 @@ def self.memo
       _current.memo
     end
 
+    # @return [Metric::Meter] Metric meter to create metrics on. This metric meter already contains some
+    #   workflow-specific attributes and takes care not to apply metrics during replay.
+    def self.metric_meter
+      _current.metric_meter
+    end
+
     # @return [Time] Current UTC time for this workflow. This creates and returns a new {::Time} instance every time it
     #   is invoked, it is not the same instance continually mutated.
     def self.now

temporalio/sig/temporalio/internal/worker/workflow_instance.rbs

@@ -52,6 +52,8 @@ module Temporalio
 
         def patch: (patch_id: Symbol | String, deprecated: bool) -> bool
 
+        def metric_meter: -> Temporalio::Metric::Meter
+
         def run_in_scheduler: [T] { -> T } -> T
 
         def activate_internal: (untyped activation) -> untyped

temporalio/sig/temporalio/internal/worker/workflow_instance/context.rbs

@@ -59,6 +59,8 @@ module Temporalio
 
           def memo: -> ExternallyImmutableHash[String, Object?]
 
+          def metric_meter: -> Temporalio::Metric::Meter
+
           def now: -> Time
 
           def patched: (String patch_id) -> bool