Data Passing and Data Structures

Overview

Data is passed between six data structures.

Work Unit Data Structure

'work_unit' is the structure of the data stored in the input queue.

work_unit = {  
	:created_at 	=> Time.now.utc.strftime('%Y-%m-%d %H:%M:%S %Z'),  
	:s3_download 	=> [File.join(jobspec[:bucket], filename)],  
	:worker_name 	=> jobspec[:worker_name],  
}  

The Work Unit should contain :created_at.  It must also include the s3_download element which contains one or more file names that you have placed in S3. You should include the :worker_name which corresponds to your worker class code on the worker instance. If you do not include a :worker_name, the worker daemon will use its default_worker_name setting from the rightworker.yml. You can also include any other key/value pair you would like to pass to the worker code.

Message Env Data Structure

message_env(for worker): {
	:created_at		=>	work_unit[:created_at] 
	:s3_download		=>	work_unit[:s3_download] 
	:worker_name		=>	work_unit[:worker_name] 
	:anykey  		=>	work_unit[:anyvalue] 
	message_id		=>	SQS message ID 
	:s3_in	        	=>	$s3_in as specified in the rightworker.yml
	:user		        =>	{key, value pairs} as specified in the rightworker.yml
	:controller		=>	RightWorkersDaemon 
	:tmp_dir		=>	/mnt/tmp/%{message_id} 
	:log_dir		=>	%{tmp_dir} /log
	:output_dir		=>	%{tmp_dir} /output
	:s3_downloaded_list	=>	{ files-that-were-downloaded, %{s3_in}/filename}
 	:log_file		=>	%{log_dir}/user_worker.log
	:s3_out		        =>	%{output_dir
} 
 

Error Data Structure

Error { 
	:s3_download 		=> %{work_unit[:s3_download]}
	:worker_name 	=> %{work_unit[:worker_name]}
	:serial			=> %{work_unit[:serial]}
	:conversion_type	=> %{work_unit[:conversion_type]}
	:id			=> %{work_unit[:id]}  
	:created_at		=> %{work_unit[: created_at]}
 	result			=> %{results[:result]}
 	right_grid_status	=> %{daemon[:right_grid_status]}
	startime		=> %{results[:starttime]}
	finishtime		=> %{results[: finishtime]}
	id			=> %{results[:id]}
	audit_info		=> %{results[:audit_info]} 
}

Audit Data Structure

Audit {	
	worker_result      	= 0 # integer/string value with exit result from RightWorker 
	result 	                = %{results[:result]}
	queue_url 	        = http://queue.amazonaws.com/ZZZZZZZZZZZZZ/queue-name
	work_item_id         	= message id
	result_item_id  	= message id

work_unit_created_at = %{work_unit[: created_at]} start_timestamp = %{results[:starttime]} end_timestamp = %{results[: finishtime]} secs_to_work = 3 # Amount of seconds right worker worked s3_dowloaded_list = %{message_env[:s3_downloaded_list]} s3_downloaded_bytes = integer (bytes) secs_to_download = integer (seconds) s3_uploaded_list = ["bucket1/key1/file1", ..., "bucketN/keyN/fileN"] s3_uploaded_bytes = integer (bytes) secs_to_upload = integer (seconds) s3_log_path = 'bucket/key' errors = {} audit_info ={ key pairs} }

Right Results Data Structure

This is the data structure passed back from the worker code to the worker daemon. If the result element begins with 'exception' or 'aborted', the worker daemon will consider this an error result from the worker code. The worker code is free to put in any additional key/value pairs that it wants to pass back.

results {
 	created_at	=> %{message_env[:created_at]
	id	        => %{message_env[:id]
	serial	        => %{message_env[:serial]
	result      	=> 0	
	startime	=> Time.now  //beginning of processing
	finishtime	=> Time.now  //end of processing
	audit_info'	=> {Any_Key_name => “any data to appear in the audit message” }
} 

Output Data Structure

Output {
 	s3_upload 	        => { local_path/filename, s3_bucket/filename}
 	result	                => %{results[:result]}
 	right_grid_status	=> %{daemon[:right_grid_status]}
	startime	        => %{results[:starttime]}
	finishtime	        => %{results[: finishtime]}
	created_at	        => %{results[: created_at]}
	work_item_id            => message id
	id              	=> %{results[:id]}
	audit_info         	=> %{results[:audit_info]} 
	serial	                => %{results[:serial]}
}