Agent job control

Overview

API to manage subordinate jobs at the agent side More…

Configurator API for Agent job control (tapi_cfg_job)
Configurator job factory control (tapi_job_factory_cfg)
Helper functions for handling options
convenience macros for option
functions for argument formatting
Library for Agent job control on agent side (ta_job)
RPC server job factory control (tapi_job_factory_rpc)
TAPI Job Methods (tapi_job_methods)
TAPI Job internal functions (tapi_job_internal)

// typedefs

typedef struct tapi_job_factory_t tapi_job_factory_t;
typedef struct tapi_job_t tapi_job_t;
typedef struct tapi_job_wrapper_t tapi_job_wrapper_t;
typedef struct tapi_job_channel_t tapi_job_channel_t;
typedef tapi_job_channel_t* tapi_job_channel_set_t[];
typedef struct tapi_job_simple_filter_t tapi_job_simple_filter_t;
typedef struct tapi_job_simple_desc_t tapi_job_simple_desc_t;
typedef struct tapi_job_status_t tapi_job_status_t;
typedef struct tapi_job_buffer_t tapi_job_buffer_t;
typedef enum tapi_job_exec_param_type tapi_job_exec_param_type;
typedef enum tapi_job_sched_param_type tapi_job_sched_param_type;
typedef struct tapi_job_exec_param tapi_job_exec_param;
typedef tapi_job_exec_param tapi_job_sched_param;
typedef struct tapi_job_exec_affinity_param tapi_job_exec_affinity_param;
typedef tapi_job_exec_affinity_param tapi_job_sched_affinity_param;
typedef struct tapi_job_exec_priority_param tapi_job_exec_priority_param;
typedef tapi_job_exec_priority_param tapi_job_sched_priority_param;
typedef struct tapi_job_exec_workdir_param tapi_job_exec_workdir_param;

// enums

enum tapi_job_exec_param_type;
enum tapi_job_sched_param_type;
enum tapi_job_status_type_t;
enum tapi_job_wrapper_priority_t;

// structs

struct tapi_job_buffer_t;
struct tapi_job_exec_affinity_param;
struct tapi_job_exec_param;
struct tapi_job_exec_priority_param;
struct tapi_job_exec_workdir_param;
struct tapi_job_simple_desc_t;
struct tapi_job_simple_filter_t;
struct tapi_job_status_t;

// global functions

const char* tapi_job_factory_ta(const tapi_job_factory_t* factory);
te_errno tapi_job_factory_set_path(tapi_job_factory_t* factory);
void tapi_job_factory_destroy(tapi_job_factory_t* factory);
tapi_job_factory_t* tapi_job_get_factory(tapi_job_t* job);
te_errno tapi_job_create(tapi_job_factory_t* factory, const char* spawner, const char* program, const char** argv, const char** env, tapi_job_t** job);
te_errno tapi_job_create_named(tapi_job_factory_t* factory, const char* name, const char* spawner, const char* program, const char** argv, const char** env, tapi_job_t** job);
te_errno tapi_job_recreate(tapi_job_factory_t* factory, const void* identifier, tapi_job_t** job);
te_errno tapi_job_simple_create(tapi_job_factory_t* factory, tapi_job_simple_desc_t* desc);
te_errno tapi_job_start(tapi_job_t* job);
te_errno tapi_job_kill(tapi_job_t* job, int signo);
te_errno tapi_job_killpg(tapi_job_t* job, int signo);
unsigned int tapi_job_get_timeout(void);
te_errno tapi_job_wait(tapi_job_t* job, int timeout_ms, tapi_job_status_t* status);
bool tapi_job_is_running(tapi_job_t* job);
te_errno tapi_job_alloc_input_channels(tapi_job_t* job, unsigned int n_channels, tapi_job_channel_t* channels[n_channels]);
te_errno tapi_job_alloc_output_channels(tapi_job_t* job, unsigned int n_channels, tapi_job_channel_t* channels[n_channels]);
te_errno tapi_job_dealloc_channels(tapi_job_channel_set_t channels);
te_errno tapi_job_attach_filter(tapi_job_channel_set_t channels, const char* filter_name, bool readable, te_log_level log_level, tapi_job_channel_t** filter);
te_errno tapi_job_attach_simple_filter(const tapi_job_simple_desc_t* desc, tapi_job_simple_filter_t* filter);
te_errno tapi_job_filter_add_regexp(tapi_job_channel_t* filter, const char* re, unsigned int extract);
te_errno tapi_job_filter_add_channels(tapi_job_channel_t* filter, tapi_job_channel_set_t channels);
te_errno tapi_job_filter_remove_channels(tapi_job_channel_t* filter, tapi_job_channel_set_t channels);
te_errno tapi_job_send(tapi_job_channel_t* channel, const te_string* str);
void tapi_job_simple_send(tapi_job_channel_t* channel, const te_string* str);
te_errno tapi_job_poll(const tapi_job_channel_set_t wait_set, int timeout_ms);
void tapi_job_simple_poll(const tapi_job_channel_set_t wait_set, int timeout_ms);
te_errno tapi_job_receive(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t* buffer);
te_errno tapi_job_receive_last(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t* buffer);
te_errno tapi_job_receive_many(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t** buffers, unsigned int* count);
void tapi_job_buffers_free(tapi_job_buffer_t* buffers, unsigned int count);
bool tapi_job_filters_have_data(const tapi_job_channel_set_t filters, int timeout_ms);
void tapi_job_simple_receive(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t* buffer);
te_errno tapi_job_receive_single(tapi_job_channel_t* filter, te_string* val, int timeout_ms);
te_errno tapi_job_clear(const tapi_job_channel_set_t filters);
te_errno tapi_job_stop(tapi_job_t* job, int signo, int term_timeout_ms);
te_errno tapi_job_destroy(tapi_job_t* job, int term_timeout_ms);
te_errno tapi_job_set_workdir(tapi_job_t* job, const char* dir);
te_errno tapi_job_wrapper_add(tapi_job_t* job, const char* tool, const char** argv, tapi_job_wrapper_priority_t priority, tapi_job_wrapper_t** wrap);
te_errno tapi_job_wrapper_delete(tapi_job_wrapper_t* wrapper);
te_errno tapi_job_add_exec_param(tapi_job_t* job, tapi_job_exec_param* exec_param);
static te_errno tapi_job_add_sched_param(tapi_job_t* job, tapi_job_exec_param* exec_param);
te_errno tapi_job_set_autorestart(tapi_job_t* job, unsigned int value);
te_errno tapi_job_get_autorestart(tapi_job_t* job, unsigned int* value);
void tapi_job_set_tracing(tapi_job_t* job, bool trace);

// macros

#define TAPI_JOB_BUFFER_INIT
#define TAPI_JOB_CHANNEL_SET(...)
#define TAPI_JOB_CHECK_STATUS(_status)
#define TAPI_JOB_SIMPLE_FILTERS(...)

Detailed Documentation

API to manage subordinate jobs at the agent side

Typedefs

typedef struct tapi_job_factory_t tapi_job_factory_t

An opaque type to represent factory that creates job instances

typedef struct tapi_job_t tapi_job_t

An opaque type to represent job instances

typedef struct tapi_job_wrapper_t tapi_job_wrapper_t

An opaque type to represent wrapper instances

typedef struct tapi_job_channel_t tapi_job_channel_t

Job channel handle

typedef tapi_job_channel_t* tapi_job_channel_set_t[]

Job channel set (a NULL terminated vector)

typedef struct tapi_job_simple_filter_t tapi_job_simple_filter_t

A simplified description of an output filter. The caller is expected to fill the fields one is interested in and leave others NULL or 0.

typedef struct tapi_job_simple_desc_t tapi_job_simple_desc_t

A simplified description of a job. The caller is expected to fill the fields one is interested in and leave others NULL.

typedef struct tapi_job_status_t tapi_job_status_t

A structure that represents completed job status

typedef struct tapi_job_buffer_t tapi_job_buffer_t

A structure to store messages read by tapi_job_receive() or tapi_job_receive_last().

typedef enum tapi_job_exec_param_type tapi_job_exec_param_type

TAPI job kinds of process parameters

typedef enum tapi_job_sched_param_type tapi_job_sched_param_type

Use tapi_job_exec_param_type in the new code

Deprecated

typedef struct tapi_job_exec_param tapi_job_exec_param

TAPI job process parameter

typedef tapi_job_exec_param tapi_job_sched_param

Use tapi_job_exec_param in the new code

Deprecated

typedef struct tapi_job_exec_affinity_param tapi_job_exec_affinity_param

Data specific for CPU affinity process type

typedef tapi_job_exec_affinity_param tapi_job_sched_affinity_param

Use tapi_job_exec_affinity_param in the new code

Deprecated

typedef struct tapi_job_exec_priority_param tapi_job_exec_priority_param

Data specific for priority process type

typedef tapi_job_exec_priority_param tapi_job_sched_priority_param

Use tapi_job_exec_priority_param in the new code

Deprecated

typedef struct tapi_job_exec_workdir_param tapi_job_exec_workdir_param

Data specific for workdir process type

Global Functions

const char* tapi_job_factory_ta(const tapi_job_factory_t* factory)

Get name of the test agent that hosts job instances created by factory.

The returned pointer should not be used after factory destruction

Parameters:

factory

Job factory

Returns:

Test Agent name (or NULL on failure)

te_errno tapi_job_factory_set_path(tapi_job_factory_t* factory)

Set path in factory’s environment to /agent:/env:PATH. The environment is inherited by created job if the environment is not specified in tapi_job_create().

This function supports only RPC factories

Job TAPI does not automatically inherit the environment from /agent:env

Parameters:

factory

Job factory

Returns:

Status code

void tapi_job_factory_destroy(tapi_job_factory_t* factory)

Destroy a job factory

Parameters:

factory

Job factory

tapi_job_factory_t* tapi_job_get_factory(tapi_job_t* job)

Get factory used to create the job.

Parameters:

job	Job instance handle

Returns:

Job factory handle or NULL if job is NULL

te_errno tapi_job_create(tapi_job_factory_t* factory, const char* spawner, const char* program, const char** argv, const char** env, tapi_job_t** job)

Create a job controlled in the way specified by factory. The job will be managed by spawner plugin.

The created job is not started automatically, use tapi_job_start() to actually run it.

The first element of args, by convention of exec family functions, should point to the filename associated with the file being executed.

Unnamed jobs created by CFG factory are not supposed to live longer than one test, so they should be destroyed at the end of the test where they were created. Otherwise, trying to create an unnamed job by CFG factory in a subsequent test will lead to an error.

Parameters:

factory	Job factory
spawner	Spawner plugin name (may be `NULL` for the default plugin)
program	Program path to run
argv	Program arguments (last item is `NULL`)
env	Program environment (last item is `NULL`). May be `NULL` to keep the current environment.
job	Job handle

Returns:

Status code

te_errno tapi_job_create_named(tapi_job_factory_t* factory, const char* name, const char* spawner, const char* program, const char** argv, const char** env, tapi_job_t** job)

Same as tapi_job_create(), but allows to set a name for the job being created

Parameters:

factory	Job factory
name	Name of the job. If CFG job factory is used, this parameter is mandatory. If RPC job factory is used, it should be `NULL`.
spawner	Spawner plugin name (may be `NULL` for the default plugin)
program	Program path to run
argv	Program arguments (last item is `NULL`)
env	Program environment (last item is `NULL`). May be `NULL` to keep the current environment.
job	Job handle

Returns:

Status code

See also:

tapi_job_set_autorestart

bool tapi_job_is_running(tapi_job_t* job)

Check whether a job is running

Parameters:

job	Job instance handle
TEST_FAIL

Returns:

true if job is running

te_errno tapi_job_alloc_input_channels(tapi_job_t* job, unsigned int n_channels, tapi_job_channel_t* channels[n_channels])

Allocate n_channels input channels. The first channel is expected to be connected to the job’s stdin; the wiring of others is spawner-dependant.

This function supports only jobs created by RPC factory

Parameters:

job	Job instance handle
n_channels	Number of channels
channels	A vector of obtained channel handles

Returns:

Status code

te_errno tapi_job_alloc_output_channels(tapi_job_t* job, unsigned int n_channels, tapi_job_channel_t* channels[n_channels])

Allocate n_channels output channels. The first and the second output channels are expected to be connected to stdout and stderr resp., the wiring of others is spawner-dependant.

This function supports only jobs created by RPC factory

The implementation need not support any other operations on primary output channels other than attaching and removing filters to/from them (see tapi_job_attach_filter(), tapi_job_filter_add_channels(), and tapi_job_filter_remove_channels())

Parameters:

job	Job instance handle
n_channels	Number of channels
channels	A vector of obtained channel handles

Returns:

Status code

te_errno tapi_job_dealloc_channels(tapi_job_channel_set_t channels)

Deallocate primary channels.

The channels are freed by the function.

Parameters:

channels

Channels to be deallocated.

Returns:

Status code

te_errno tapi_job_attach_filter(tapi_job_channel_set_t channels, const char* filter_name, bool readable, te_log_level log_level, tapi_job_channel_t** filter)

Create a secondary output channel applying a filter to an existing channel.

The implementation is only required to support filter channels over primary channels, so filter chaining may not be supported.

Parameters:

channels	Output channels to attach the filter to.
filter_name	Filter name (may be `NULL`, then the default filter is used) *
readable	If `true`, the output of the filter is sent to the test, that is, it can be read with tapi_job_receive(); otherwise, it is discarded, possibly after being logged.
log_level	If non-zero, the output of the filter is logged with a given log level
filter	Filter channel (may be `NULL` for trivial filters)
TE_EPERM	Some of `channels` are input channels
TE_EINVAL	Some of `channels` are filter output channels if the implementaion does not support filters over filters.
TE_EXDEV	`channels` are on different RPC servers

Returns:

Status code

te_errno tapi_job_attach_simple_filter(const tapi_job_simple_desc_t* desc, tapi_job_simple_filter_t* filter)

Attach a simple filter to a job created by tapi_job_simple_create()

Parameters:

desc	Job description (it must have job_loc, stdin_loc and stderr_loc filled in)
filter	Filter to attach
TE_ENOTCONN	tapi_job_simple_create() has not been called on `desc`

Returns:

Status code

te_errno tapi_job_filter_add_regexp(tapi_job_channel_t* filter, const char* re, unsigned int extract)

Add a regular expression for filter

Multi-segment matching is performed. Thus, PCRE_MULTILINE option is set.

Parameters:

filter	Filter handle
re	PCRE-style regular expression to match
extract	A substring to extract as an output of the filter (0 meaning the whole match)

Returns:

Status code

te_errno tapi_job_filter_add_channels(tapi_job_channel_t* filter, tapi_job_channel_set_t channels)

Attach an existing filter to additional output channels

Parameters:

filter	Filter to attach
channels	Output channels to attach the filter to

Returns:

Status code

te_errno tapi_job_filter_remove_channels(tapi_job_channel_t* filter, tapi_job_channel_set_t channels)

Remove filter from specified output channels. For instance, if the filter is attached to stdout and stderr and the function is called with only stdout specified, the filter will continue working with stderr. Either the function succeeds and the filter is removed from all specified channels, or the function fails and the filter is not removed from any channel.

If the filter is removed from every channel it was attached to, all associated resources including the memory for tapi_job_channel_t structure will be freed, so filter variable must not be used again.

Parameters:

filter	Filter to remove
channels	Channels from which to remove the filter

Returns:

Status code

te_errno tapi_job_send(tapi_job_channel_t* channel, const te_string* str)

Send data to a job input channel.

Parameters:

channel	Output channel handle
str	A pointer to string buffer
TE_EPERM	if `channel` is not an input channel

Returns:

Status code

void tapi_job_simple_send(tapi_job_channel_t* channel, const te_string* str)

The same as tapi_job_send(), but fails the test if an error happens, instead of returning an error

Parameters:

channel	Input channel handle
str	A pointer to string buffer
TEST_FAIL

te_errno tapi_job_poll(const tapi_job_channel_set_t wait_set, int timeout_ms)

Poll the job’s channels/filters for readiness. from wait_set

Parameters:

wait_set	Set of channels to wait
timeout_ms	Timeout in ms to wait (negative means tapi_job_get_timeout())
TE_EPERM	if some channels from `wait_set` are neither input channels nor pollable output channels
TE_EXDEV	if channels from `wait_set` are from different RPC servers

Returns:

Status code

void tapi_job_simple_poll(const tapi_job_channel_set_t wait_set, int timeout_ms)

The same as tapi_job_poll(), but fails the test if an error happens, instead of returning an error.

Parameters:

wait_set	Set of channels to wait
timeout_ms	Timeout in ms to wait (negative means tapi_job_get_timeout())
TEST_FAIL

te_errno tapi_job_receive(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t* buffer)

Read the next message from one of the available filters.

The data being read are appended to data buffer. At most one message is read.

Parameters:

filters	Set of filters to read from.
timeout_ms	Timeout to wait (negative means tapi_job_get_timeout())
buffer	Data buffer pointer. If `NULL`, the message is silently discarded.
TE_ETIMEDOUT	if there’s no data available within `timeout`
TE_EPERM	if some of the `filters` are input channels or primary output channels and the implementation does not support reading from them or unreadable output channels
TE_EXDEV	if `filters` are on different RPC servers

Returns:

Status code

te_errno tapi_job_receive_last(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t* buffer)

Read the last non-eos message (if there is one) from one of the available filters. If the only message contains eos, read it. The message is not removed from the queue, it can still be read with tapi_job_receive().

The data being read are appended to data buffer. At most one message is read.

Parameters:

filters	Set of filters to read from.
timeout_ms	Timeout to wait (negative means tapi_job_get_timeout())
buffer	Data buffer pointer. If `NULL`, the message is silently discarded.
TE_ETIMEDOUT	if there’s no data available within `timeout`
TE_EPERM	if some of the `filters` are input channels or primary output channels and the implementation does not support reading from them or unreadable output channels
TE_EXDEV	if `filters` are on different RPC servers

Returns:

Status code

te_errno tapi_job_receive_many(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t** buffers, unsigned int* count)

Obtain multiple messages at once from the specified filters. This function may retrieve some messages and return error if attempt to read the next message failed.

Parameters:

filters	Set of filters to read from
timeout_ms	Timeout to wait (negative means tapi_job_get_timeout()) until some messages are available
buffers	Where to save pointer to array of buffers with messages (should be released by caller with tapi_job_buffers_free())
count	On input, maximum number of messages to retrieve. If zero, all available messages will be retrieved. On output - number of actually retrieved messages

Returns:

Status code.

void tapi_job_buffers_free(tapi_job_buffer_t* buffers, unsigned int count)

Release array of message buffers.

Parameters:

buffers	Pointer to the array
count	Number of buffers in the array

bool tapi_job_filters_have_data(const tapi_job_channel_set_t filters, int timeout_ms)

Check if there is at least one non-eos message on one of the available filters. The message is not removed from the queue, it can still be read with tapi_job_receive().

The function is useful to check if a job produced some particular output.

Parameters:

filters	Set of filters to read from.
timeout_ms	Timeout to wait (negative means tapi_job_get_timeout())

Returns:

true if there is a non-eos message

void tapi_job_simple_receive(const tapi_job_channel_set_t filters, int timeout_ms, tapi_job_buffer_t* buffer)

The same as tapi_job_receive() but fails the test if an error happens. Also, unlike tapi_job_receive(), the function resets the buffer contents.

Parameters:

filters	Set of filters to read from.
timeout_ms	Timeout to wait (negative means tapi_job_get_timeout())
buffer	Data buffer pointer. If `NULL`, the message is silently discarded.
TEST_FAIL

te_errno tapi_job_receive_single(tapi_job_channel_t* filter, te_string* val, int timeout_ms)

Get contents of a single message. This function will fail if there is more than one message. It is useful for filters looking for something unique in the output.

Parameters:

filter	From where to read the message
val	Where to save the contents
timeout_ms	Timeout to wait (negative means tapi_job_get_timeout())

Returns:

Status code.

te_errno tapi_job_clear(const tapi_job_channel_set_t filters)

Remove all pending messages from filters, they are lost completely.

Parameters:

filters	Set of filters to clear.
TE_EPERM	if some of the `filters` are input channels or primary output channels and the implementation does not support reading from them or unreadable output channels
TE_EXDEV	if `filters` are on different RPC servers

Returns:

Status code

te_errno tapi_job_stop(tapi_job_t* job, int signo, int term_timeout_ms)

Stop a job. It can be started over with tapi_job_start(). The function tries to terminate the job with the specified signal. If the signal fails to terminate the job, the function will send SIGKILL.

Parameters signo and term_timeout_ms are supported only for jobs created by RPC factory. Use -1 to avoid warnings.

For autorestart jobs this function will stop the job and prevent the autorestart subsystem from starting the job over

Parameters:

job	Job instance handle
signo	Signal to be sent at first. If signo is `SIGKILL`, it will be sent only once.
term_timeout_ms	The timeout of graceful termination of a job, if it has been running. After the timeout expiration the job will be killed with `SIGKILL`. (negative means default timeout)

Returns:

Status code

Macros

#define TAPI_JOB_BUFFER_INIT

Static initializer for tapi_job_buffer_t

#define TAPI_JOB_CHANNEL_SET(...)

A convenience vector constructor to use with polling functions

#define TAPI_JOB_CHECK_STATUS(_status)

Check if the status is a success status (i.e. the job exited successfully)

Parameters:

_status

Job exit status

#define TAPI_JOB_SIMPLE_FILTERS(...)

A helper to create vectors of job simple filter descriptions

factory	Job factory that was used when the job was first created
identifier	Data that allows to identify the job. For jobs created by CFG factory it must be a C-string with a name that was assigned to the job when it was first create.
job	Location where to put the job handle
TE_ENOENT	The job with the `identifier` was never created

job	Job instance handle
timeout_ms	Timeout in ms. Meaning of negative timeout depends on the factory type that was used to create the job. See rpc_job_wait() and cfg_job_wait() for details.
status	Exit status (may be `NULL`)
TE_EINPROGRESS	Job is still running. For autorestart jobs, this means that the autorestart subsystem is working with the job and it will be restarted when needed.
TE_ECHILD	Job was never started