Class: Hatchet::Clients::Grpc::Admin

Inherits:
Object
  • Object
show all
Defined in:
lib/hatchet/clients/grpc/admin.rb,
sig/hatchet/clients/grpc/admin.rbs

Overview

gRPC client for the Hatchet Admin service (workflow registration & triggering).

Uses two stubs on the shared channel:

  • V1::AdminService::Stub (v1) for: PutWorkflow, GetRunDetails, CancelTasks, ReplayTasks
  • WorkflowService::Stub (v0) for: TriggerWorkflow, BulkTriggerWorkflow, ScheduleWorkflow, PutRateLimit

The v0 WorkflowService is retained for triggering because it supports parent-child linking fields that the v1 TriggerWorkflowRun does not expose.

Examples:

admin = Admin.new(config: hatchet_config, channel: channel)
admin.put_workflow(workflow.to_proto(config))
ref = admin.trigger_workflow("MyWorkflow", input: { "key" => "value" })

Constant Summary collapse

BULK_TRIGGER_BATCH_SIZE =

Returns:

  • (Integer) 
1000
COMPARATOR_MAP = 
{
  equal: :EQUAL, not_equal: :NOT_EQUAL,
  greater_than: :GREATER_THAN, greater_than_or_equal: :GREATER_THAN_OR_EQUAL,
  less_than: :LESS_THAN, less_than_or_equal: :LESS_THAN_OR_EQUAL,
}.freeze

Instance Method Summary collapse

Constructor Details

#initialize(config:, channel:) ⇒ Admin

Returns a new instance of Admin.

Parameters:

  • config (Hatchet::Config) —

    The Hatchet configuration

  • channel (GRPC::Core::Channel) —

    Shared gRPC channel

  • config: (Config)
  • channel: (Object) 


28
29
30
31
32
33
34
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 28

def initialize(config:, channel:)
  @config = config
  @logger = config.logger
  @channel = channel
  @v0_stub = nil
  @v1_stub = nil
end

Instance Method Details

#bulk_trigger_workflow(workflow_name, items) ⇒ Array<String>

Trigger multiple workflow runs in bulk via v0 WorkflowService.

Automatically batches requests in groups of 1000.

Parameters:

  • workflow_name (String) —

    The workflow name (will be namespaced)

  • items (Array<Hash>) —

    Array of { input:, options: } items

Returns:

  • (Array<String>) —

    Array of workflow run IDs

Raises:



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 116

def bulk_trigger_workflow(workflow_name, items)
  ensure_connected!

  name = @config.apply_namespace(workflow_name)

  requests = items.map do |item|
    input = item[:input] || {}
    opts = item[:options] || {}

    request_args = {
      name: name,
      input: JSON.generate(input),
    }

    request_args[:parent_id] = opts[:parent_id] if opts[:parent_id]
    request_args[:parent_task_run_external_id] = opts[:parent_task_run_external_id] if opts[:parent_task_run_external_id]
    request_args[:child_index] = opts[:child_index] if opts[:child_index]
    request_args[:child_key] = opts[:child_key] if opts[:child_key]
    request_args[:desired_worker_id] = opts[:desired_worker_id] if opts[:desired_worker_id]
    request_args[:priority] = opts[:priority] if opts[:priority]

    if opts[:additional_metadata]
      request_args[:additional_metadata] = if opts[:additional_metadata].is_a?(String)
                                             opts[:additional_metadata]
                                           else
                                             JSON.generate(opts[:additional_metadata])
                                           end
    end

    request_args[:desired_worker_labels] = build_trigger_worker_labels(opts[:desired_worker_labels]) if opts[:desired_worker_labels]

    ::V1::TriggerWorkflowRequest.new(**request_args)
  end

  # Batch in groups of BULK_TRIGGER_BATCH_SIZE
  all_run_ids = []
  begin
    requests.each_slice(BULK_TRIGGER_BATCH_SIZE) do |batch|
      bulk_request = ::BulkTriggerWorkflowRequest.new(workflows: batch)
      response = @v0_stub.bulk_trigger_workflow(bulk_request, metadata: @config.auth_metadata)
      all_run_ids.concat(response.workflow_run_ids.to_a)
    end
  rescue ::GRPC::AlreadyExists => e
    collision = extract_bulk_trigger_idempotency_collision(e)

    raise collision if collision

    raise Error, "gRPC error triggering bulk workflow: #{e.class}: #{e.message}"
  rescue ::GRPC::ResourceExhausted => e
    raise ResourceExhaustedError, e.message
  rescue ::GRPC::BadStatus => e
    raise Error, "gRPC error triggering bulk workflow: #{e.class}: #{e.message}"
  end

  all_run_ids
end

#close ⇒ void

This method returns an undefined value.

Close the connection.



254
255
256
257
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 254

def close
  @v0_stub = nil
  @v1_stub = nil
end

#get_run_details(external_id:) ⇒ V1::GetRunDetailsResponse

Get run details via v1 AdminService.

Parameters:

  • external_id (String) —

    The workflow run external ID

  • external_id: (String)

Returns:



228
229
230
231
232
233
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 228

def get_run_details(external_id:)
  ensure_connected!

  request = ::V1::GetRunDetailsRequest.new(external_id: external_id)
  @v1_stub.get_run_details(request, metadata: @config.auth_metadata)
end

#put_rate_limit(key:, limit:, duration:) ⇒ PutRateLimitResponse

Put a rate limit via v0 WorkflowService.

Parameters:

  • key (String) —

    Rate limit key

  • limit (Integer) —

    Rate limit value

  • duration (Symbol) —

    Rate limit duration enum

  • key: (String)
  • limit: (Integer)
  • duration: (Symbol)

Returns:



241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 241

def put_rate_limit(key:, limit:, duration:)
  ensure_connected!

  request = ::PutRateLimitRequest.new(
    key: key,
    limit: limit,
    duration: duration,
  )

  @v0_stub.put_rate_limit(request, metadata: @config.auth_metadata)
end

#put_workflow(workflow_proto) ⇒ V1::CreateWorkflowVersionResponse

Register a workflow definition with the server via v1 AdminService.

Parameters:

Returns:



40
41
42
43
44
45
46
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 40

def put_workflow(workflow_proto)
  ensure_connected!

  response = @v1_stub.put_workflow(workflow_proto, metadata: @config.auth_metadata)
  @logger.debug("Registered workflow: #{workflow_proto.name}")
  response
end

#schedule_workflow(workflow_name, run_at:, input: {}, options: {}) ⇒ WorkflowVersion

Schedule a workflow for future execution via v0 WorkflowService.

Parameters:

  • workflow_name (String) —

    The workflow name (will be namespaced)

  • run_at (Time) —

    When to run

  • input (Hash) (defaults to: {}) —

    Workflow input

  • options (Hash) (defaults to: {}) —

    Trigger options

Returns:

Raises:



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 181

def schedule_workflow(workflow_name, run_at:, input: {}, options: {})
  ensure_connected!

  name = @config.apply_namespace(workflow_name)

  schedule_timestamp = Google::Protobuf::Timestamp.new(
    seconds: run_at.to_i,
    nanos: run_at.respond_to?(:nsec) ? run_at.nsec : 0,
  )

  request_args = {
    name: name,
    schedules: [schedule_timestamp],
    input: JSON.generate(input),
  }

  request_args[:parent_id] = options[:parent_id] if options[:parent_id]
  request_args[:parent_task_run_external_id] = options[:parent_task_run_external_id] if options[:parent_task_run_external_id]
  request_args[:child_index] = options[:child_index] if options[:child_index]
  request_args[:child_key] = options[:child_key] if options[:child_key]
  request_args[:priority] = options[:priority] if options[:priority]

  if options[:additional_metadata]
    request_args[:additional_metadata] = if options[:additional_metadata].is_a?(String)
                                           options[:additional_metadata]
                                         else
                                           JSON.generate(options[:additional_metadata])
                                         end
  end

  request = ::ScheduleWorkflowRequest.new(**request_args)

  begin
    @v0_stub.schedule_workflow(request, metadata: @config.auth_metadata)
  rescue ::GRPC::AlreadyExists => e
    raise DedupeViolationError, "Deduplication violation: #{e.message}"
  rescue ::GRPC::ResourceExhausted => e
    raise ResourceExhaustedError, e.message
  rescue ::GRPC::BadStatus => e
    raise Error, "gRPC error scheduling workflow: #{e.class}: #{e.message}"
  end
end

#trigger_workflow(workflow_name, input: {}, options: {}) ⇒ String

Trigger a workflow run via v0 WorkflowService.

Parameters:

  • workflow_name (String) —

    The workflow name (will be namespaced)

  • input (Hash) (defaults to: {}) —

    Workflow input

  • options (Hash) (defaults to: {}) —

    Trigger options

Options Hash (options:):

  • :parent_id (String) —

    Parent workflow run ID

  • :parent_task_run_external_id (String) —

    Parent step run ID

  • :child_index (Integer) —

    Child workflow index

  • :child_key (String) —

    Child workflow key

  • :additional_metadata (Hash) —

    Additional metadata

  • :desired_worker_id (String) —

    Desired worker for sticky dispatch

  • :priority (Integer) —

    Priority value

Returns:

  • (String) —

    The workflow run ID

Raises:



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/hatchet/clients/grpc/admin.rb', line 62

def trigger_workflow(workflow_name, input: {}, options: {})
  ensure_connected!

  name = @config.apply_namespace(workflow_name)

  request_args = {
    name: name,
    input: JSON.generate(input),
  }

  request_args[:parent_id] = options[:parent_id] if options[:parent_id]
  request_args[:parent_task_run_external_id] = options[:parent_task_run_external_id] if options[:parent_task_run_external_id]
  request_args[:child_index] = options[:child_index] if options[:child_index]
  request_args[:child_key] = options[:child_key] if options[:child_key]
  request_args[:desired_worker_id] = options[:desired_worker_id] if options[:desired_worker_id]
  request_args[:priority] = options[:priority] if options[:priority]

  if options[:additional_metadata]
    request_args[:additional_metadata] = if options[:additional_metadata].is_a?(String)
                                           options[:additional_metadata]
                                         else
                                           JSON.generate(options[:additional_metadata])
                                         end
  end

  if options[:desired_worker_labels]
    request_args[:desired_worker_labels] = build_trigger_worker_labels(options[:desired_worker_labels])
  end

  request = ::V1::TriggerWorkflowRequest.new(**request_args)

  begin
    response = @v0_stub.trigger_workflow(request, metadata: @config.auth_metadata)
    response.workflow_run_id
  rescue ::GRPC::AlreadyExists => e
    run_id = extract_idempotency_run_id(e)
    raise(IdempotencyCollisionError, run_id) if run_id

    raise DedupeViolationError, "Deduplication violation: #{e.message}"
  rescue ::GRPC::ResourceExhausted => e
    raise ResourceExhaustedError, e.message
  rescue ::GRPC::BadStatus => e
    raise Error, "gRPC error triggering workflow: #{e.class}: #{e.message}"
  end
end