Trees       Indices       Help   
Ganeti
Package ganeti :: Module jqueue :: Class JobQueue
[hide private]
[frames] | no frames]

Class JobQueue

source code

Queue used to manage the jobs.

Instance Methods [hide private]
 
__init__(self, context)
Constructor for JobQueue.		 source code
 
_InspectQueue(self)
Loads the whole job queue and resumes unfinished jobs.		 source code
 
AddNode(self, node)
Register a new node with the queue.		 source code
 
RemoveNode(self, node_name)
Callback called when removing nodes from the cluster.		 source code
(list, list)
_GetNodeIp(self)
Helper for returning the node name/ip list.		 source code
 
_UpdateJobQueueFile(self, file_name, data, replicate)
Writes a file locally and then replicates it to all nodes.		 source code
 
_RenameFilesUnlocked(self, rename)
Renames a file locally and then replicate the change.		 source code
str
_NewSerialsUnlocked(self, count)
Generates a new job identifier.		 source code
list
_GetJobIDsUnlocked(self, sort=True)
Return all known job IDs.		 source code
_QueuedJob or None
_LoadJobUnlocked(self, job_id)
Loads a job from the disk or memory.		 source code
_QueuedJob or None
_LoadJobFromDisk(self, job_id)
Load the given job file from disk.		 source code
_QueuedJob or None
SafeLoadJobFromDisk(self, job_id)
Load the given job file from disk.		 source code
 
_UpdateQueueSizeUnlocked(self)
Update the queue size.		 source code
 
SetDrainFlag(self, drain_flag)
Sets the drain flag for the queue.		 source code
_QueuedJob
_SubmitJobUnlocked(self, job_id, ops)
Create and store a new job.		 source code
 
SubmitJob(self, ops)
Create and store a new job.		 source code
 
SubmitManyJobs(self, jobs)
Create and store multiple jobs.		 source code
 
UpdateJobUnlocked(self, job, replicate=True)
Update a job's on disk storage.		 source code
tuple (job info, log entries)
WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial, timeout)
Waits for changes in a job.		 source code
 
CancelJob(self, job_id)
Cancels a job.		 source code
int
_ArchiveJobsUnlocked(self, jobs)
Archives jobs.		 source code
bool
ArchiveJob(self, job_id)
Archives a job.		 source code
 
AutoArchiveJobs(self, age, timeout)
Archives all jobs based on age.		 source code
list
QueryJobs(self, job_ids, fields)
Returns a list of jobs in queue.		 source code
 
Shutdown(self)
Stops the job queue.		 source code

Inherited from object: __delattr__, __format__, __getattribute__, __hash__, __new__, __reduce__, __reduce_ex__, __repr__, __setattr__, __sizeof__, __str__, __subclasshook__

Class Methods [hide private]
str
_GetArchiveDirectory(cls, job_id)
Returns the archive directory for a job.		 source code
str
_GetArchivedJobPath(cls, job_id)
Returns the archived job file for a give job id.		 source code
Static Methods [hide private]
 
_CheckRpcResult(result, nodes, failmsg)
Verifies the status of an RPC call.		 source code
str
_FormatJobID(job_id)
Convert a job ID to string format.		 source code
str
_GetJobPath(job_id)
Returns the job file for a given job id.		 source code
boolean
_IsQueueMarkedDrain()
Check if the queue is marked from drain.		 source code
Class Variables [hide private]
  _RE_JOB_FILE = re.compile(r"^job-(%s)$" % constants.JOB_ID_TEM...
regex matching the valid job file names
Properties [hide private]

Inherited from object: __class__

Method Details [hide private]

__init__(self, context)
(Constructor)

source code 

Constructor for JobQueue.

The constructor will initialize the job queue object and then start loading the current jobs from disk, either for starting them (if they were queue) or for aborting them (if they were already running).

Parameters:
  • context (GanetiContext) - the context object for access to the configuration data and other ganeti objects
Overrides: object.__init__

_InspectQueue(self)

source code 

Loads the whole job queue and resumes unfinished jobs.

This function needs the lock here because WorkerPool.AddTask() may start a job while we're still doing our work.

Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

AddNode(self, node)

source code 

Register a new node with the queue.

Parameters:
Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

RemoveNode(self, node_name)

source code 

Callback called when removing nodes from the cluster.

Parameters:
  • node_name (str) - the name of the node to remove
Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

_CheckRpcResult(result, nodes, failmsg)
Static Method

source code 

Verifies the status of an RPC call.

Since we aim to keep consistency should this node (the current master) fail, we will log errors if our rpc fail, and especially log the case when more than half of the nodes fails.

Parameters:
  • result - the data as returned from the rpc call
  • nodes (list) - the list of nodes we made the call to
  • failmsg (str) - the identifier to be used for logging

_GetNodeIp(self)

source code 

Helper for returning the node name/ip list.

Returns: (list, list)
a tuple of two lists, the first one with the node names and the second one with the node addresses

_UpdateJobQueueFile(self, file_name, data, replicate)

source code 

Writes a file locally and then replicates it to all nodes.

This function will replace the contents of a file on the local node and then replicate it to all the other nodes we have.

Parameters:
  • file_name (str) - the path of the file to be replicated
  • data (str) - the new contents of the file
  • replicate (boolean) - whether to spread the changes to the remote nodes

_RenameFilesUnlocked(self, rename)

source code 

Renames a file locally and then replicate the change.

This function will rename a file in the local queue directory and then replicate this rename to all the other nodes we have.

Parameters:
  • rename (list of (old, new)) - List containing tuples mapping old to new names

_FormatJobID(job_id)
Static Method

source code 

Convert a job ID to string format.

Currently this just does str(job_id) after performing some checks, but if we want to change the job id format this will abstract this change.

Parameters:
  • job_id (int or long) - the numeric job id
Returns: str
the formatted job id

_GetArchiveDirectory(cls, job_id)
Class Method

source code 

Returns the archive directory for a job.

Parameters:
  • job_id (str) - Job identifier
Returns: str
Directory name

_NewSerialsUnlocked(self, count)

source code 

Generates a new job identifier.

Job identifiers are unique during the lifetime of a cluster.

Parameters:
  • count (integer) - how many serials to return
Returns: str
a string representing the job identifier.

_GetJobPath(job_id)
Static Method

source code 

Returns the job file for a given job id.

Parameters:
  • job_id (str) - the job identifier
Returns: str
the path to the job file

_GetArchivedJobPath(cls, job_id)
Class Method

source code 

Returns the archived job file for a give job id.

Parameters:
  • job_id (str) - the job identifier
Returns: str
the path to the archived job file

_GetJobIDsUnlocked(self, sort=True)

source code 

Return all known job IDs.

The method only looks at disk because it's a requirement that all jobs are present on disk (so in the _memcache we don't have any extra IDs).

Parameters:
  • sort (boolean) - perform sorting on the returned job ids
Returns: list
the list of job IDs

_LoadJobUnlocked(self, job_id)

source code 

Loads a job from the disk or memory.

Given a job id, this will return the cached job object if existing, or try to load the job from the disk. If loading from disk, it will also add the job to the cache.

Parameters:
  • job_id - the job id
Returns: _QueuedJob or None
either None or the job object

_LoadJobFromDisk(self, job_id)

source code 

Load the given job file from disk.

Given a job file, read, load and restore it in a _QueuedJob format.

Parameters:
  • job_id (string) - job identifier
Returns: _QueuedJob or None
either None or the job object

SafeLoadJobFromDisk(self, job_id)

source code 

Load the given job file from disk.

Given a job file, read, load and restore it in a _QueuedJob format. In case of error reading the job, it gets returned as None, and the exception is logged.

Parameters:
  • job_id (string) - job identifier
Returns: _QueuedJob or None
either None or the job object

_IsQueueMarkedDrain()
Static Method

source code 

Check if the queue is marked from drain.

This currently uses the queue drain file, which makes it a per-node flag. In the future this can be moved to the config file.

Returns: boolean
True of the job queue is marked for draining

SetDrainFlag(self, drain_flag)

source code 

Sets the drain flag for the queue.

Parameters:
  • drain_flag (boolean) - Whether to set or unset the drain flag
Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

_SubmitJobUnlocked(self, job_id, ops)

source code 

Create and store a new job.

This enters the job into our job queue and also puts it on the new queue, in order for it to be picked up by the queue processors.

Parameters:
  • job_id (job ID) - the job ID for the new job
  • ops (list) - The list of OpCodes that will become the new job.
Returns: _QueuedJob
the job object to be queued
Decorators:
  • @_RequireOpenQueue
Raises:

SubmitJob(self, ops)

source code 

Create and store a new job.

Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

See Also: _SubmitJobUnlocked

SubmitManyJobs(self, jobs)

source code 

Create and store multiple jobs.

Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

See Also: _SubmitJobUnlocked

UpdateJobUnlocked(self, job, replicate=True)

source code 

Update a job's on disk storage.

After a job has been modified, this function needs to be called in order to write the changes to disk and replicate them to the other nodes.

Parameters:
  • job (_QueuedJob) - the changed job
  • replicate (boolean) - whether to replicate the change to remote nodes
Decorators:
  • @_RequireOpenQueue

WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial, timeout)

source code 

Waits for changes in a job.

Parameters:
  • job_id (string) - Job identifier
  • fields (list of strings) - Which fields to check for changes
  • prev_job_info (list or None) - Last job information returned
  • prev_log_serial (int) - Last job message serial number
  • timeout (float) - maximum time to wait in seconds
Returns: tuple (job info, log entries)
a tuple of the job information as required via the fields parameter, and the log entries as a list

if the job has not changed and the timeout has expired, we instead return a special value, constants.JOB_NOTCHANGED, which should be interpreted as such by the clients

CancelJob(self, job_id)

source code 

Cancels a job.

This will only succeed if the job has not started yet.

Parameters:
  • job_id (string) - job ID of job to be cancelled.
Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

_ArchiveJobsUnlocked(self, jobs)

source code 

Archives jobs.

Parameters:
Returns: int
Number of archived jobs
Decorators:
  • @_RequireOpenQueue

ArchiveJob(self, job_id)

source code 

Archives a job.

This is just a wrapper over _ArchiveJobsUnlocked.

Parameters:
  • job_id (string) - Job ID of job to be archived.
Returns: bool
Whether job was archived
Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

AutoArchiveJobs(self, age, timeout)

source code 

Archives all jobs based on age.

The method will archive all jobs which are older than the age parameter. For jobs that don't have an end timestamp, the start timestamp will be considered. The special '-1' age will cause archival of all jobs (that are not running or queued).

Parameters:
  • age (int) - the minimum age in seconds
Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

QueryJobs(self, job_ids, fields)

source code 

Returns a list of jobs in queue.

Parameters:
  • job_ids (list) - sequence of job identifiers or None for all
  • fields (list) - names of fields to return
Returns: list
list one element per job, each element being list with the requested fields

Shutdown(self)

source code 

Stops the job queue.

This shutdowns all the worker threads an closes the queue.

Decorators:
  • @locking.ssynchronized(_LOCK)
  • @_RequireOpenQueue

Class Variable Details [hide private]

_RE_JOB_FILE

regex matching the valid job file names
Value:
re.compile(r"^job-(%s)$" % constants.JOB_ID_TEMPLATE)

   Trees       Indices       Help   
Ganeti
Generated by Epydoc 3.0.1 on Thu Dec 9 15:55:44 2010 http://epydoc.sourceforge.net