Class: Async::Service::Supervisor::UtilizationMonitor

Inherits:
Monitor
  • Object
show all
Defined in:
lib/async/service/supervisor/utilization_monitor.rb,
lib/metrics/provider/async/service/supervisor/utilization_monitor.rb

Overview

Monitors worker utilization metrics aggregated by service name.

Uses shared memory to efficiently collect utilization metrics from workers and aggregates them by service name for monitoring and reporting.

Defined Under Namespace

Classes: SegmentAllocator

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Monitor

#run, #status, #to_json

Constructor Details

#initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZE * 8, segment_size: 512) ⇒ UtilizationMonitor

Initialize a new utilization monitor.



197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 197

def initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZE * 8, segment_size: 512)
	super(interval: interval)
	@path = path
	@segment_size = segment_size
	
	@allocator = SegmentAllocator.new(path, size: size, segment_size: segment_size)
	
	# Track workers: worker_id => supervisor_controller
	@workers = {}
	
	@guard = Mutex.new
end

Class Method Details

.monitor_type ⇒ Object

The key used when this monitor's status is aggregated with others.



276
277
278
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 276

def self.monitor_type
	:utilization_monitor
end

Instance Method Details

#as_json ⇒ Object

Serialize utilization data for JSON.



332
333
334
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 332

def as_json
	self.sample
end

#emit(metrics) ⇒ Object

Emit the utilization metrics.



339
340
341
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 339

def emit(metrics)
	Console.info(self, "Utilization:", metrics: metrics)
end

#register(supervisor_controller) ⇒ Object

Register a worker with the utilization monitor.

Allocates a segment of shared memory and instructs the worker to map the shared memory file and expose utilization information at the allocated offset. The worker maps the file independently and returns its schema.



217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 217

def register(supervisor_controller)
	@guard.synchronize do
		worker_id = supervisor_controller.id
		return unless worker_id
		
		# Allocate a segment first (we'll get schema from worker)
		offset = @allocator.allocate(worker_id, [])
		
		unless offset
			Console.warn(self, "Failed to allocate utilization segment", worker_id: worker_id)
			return
		end
		
		# Inform worker of the shared memory path, size, and allocated offset
		# The worker will map the file itself and return its schema
		begin
			worker = supervisor_controller.worker
			
			if worker
				# Pass the segment size - observer will handle page alignment and file mapping
				schema = worker.setup_utilization_observer(@path, @segment_size, offset)
				
				# Update the allocation with the actual schema
				if schema && !schema.empty?
					@allocator.update_schema(worker_id, schema)
					@workers[worker_id] = supervisor_controller
					
					Console.info(self, "Registered worker utilization", worker_id: worker_id, offset: offset, schema: schema)
				else
					# Worker didn't provide schema, free the allocation
					@allocator.free(worker_id)
					Console.info(self, "Worker did not provide utilization schema", worker_id: worker_id)
				end
			end
		rescue => error
			Console.error(self, "Error setting up worker utilization", worker_id: worker_id, exception: error)
			@allocator.free(worker_id)
		end
	end
end

#remove(supervisor_controller) ⇒ Object

Remove a worker from the utilization monitor.

Returns the allocated segment back to the free list.



263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 263

def remove(supervisor_controller)
	@guard.synchronize do
		worker_id = supervisor_controller.id
		return unless worker_id
		
		@workers.delete(worker_id)
		@allocator.free(worker_id)
		
		Console.debug(self, "Freed utilization segment", worker_id: worker_id)
	end
end

#run_once ⇒ Object

Run one iteration of the utilization monitor.



344
345
346
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 344

def run_once
	self.emit(self.as_json)
end

#sample ⇒ Object

Sample aggregated utilization data.



283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 283

def sample
	aggregated = {}
	
	sample_by_worker.each_value do |worker|
		service_name = worker[:state][:name] || "unknown"
		
		data = worker[:utilization]
		
		# Initialize service aggregation if needed
		aggregated[service_name] ||= {}
		
		# Sum up all numeric fields
		data.each do |key, value|
			if value.is_a?(Numeric)
				aggregated[service_name][key] ||= 0
				aggregated[service_name][key] += value
			else
				# For non-numeric values, we could handle differently
				# For now, just store the last value
				aggregated[service_name][key] = value
			end
		end
		
		# Count workers per service (for utilization denominator)
		aggregated[service_name][:worker_count] = (aggregated[service_name][:worker_count] || 0) + 1
	end
	
	aggregated
end

#sample_by_worker ⇒ Object

Sample utilization data for each registered worker.



316
317
318
319
320
321
322
323
324
325
326
327
 
# File 'lib/async/service/supervisor/utilization_monitor.rb', line 316

def sample_by_worker
	@guard.synchronize do
		@workers.each_with_object({}) do |(worker_id, supervisor_controller), workers|
			if utilization = @allocator.read(worker_id)
				workers[worker_id] = {
					state: supervisor_controller.state.dup.freeze,
					utilization: utilization.freeze,
				}.freeze
			end
		end.freeze
	end
end