Pages in this section should be aimed toward developers working on OpenIDM.

The data store (repository) is used to add new tasks that any node can pick up (poll based model) and claim. Upon claiming a task, the node "leases" the task for a specified amount of time (e.g. 5 minutes). As it processes the task it renews the lease, i.e. regularly asks again for the lease to expire in 5 minutes. If the node for any reason stops processing (e.g. crashed) the lease expires, and another node is free to claim the task. The node can store a result in the task. This mechanism operates similar to a map/reduce mechanism.

The lease mechanism provides for automatic fail-over if a task does not make any progress anymore.

There is no leader and hence no single point of failure or need to re-elect leaders.

A node can split out a task (such as a recon batch) into many pieces (such as 1000 records to recon each), and create individual tasks that can be claimed and processed by many nodes in parallel. By grouping the tasks a subsequent "join" task can be triggered that can process the completion of the whole batch.

Nodes in the cluster upon start/up configuration should register all task handlers they are willing to handle. Nodes will only pick up work for task types it has handlers for, this is primarily intended to remove start-up race conditions, but can also be used to disable certain kinds of tasks (e.g. reconciliation) on a given node.

Each node has a scheduled activity that periodically checks for new work, and scans the tasks for abandoned tasks.

The underlying store must provide for some way to allow a node to uniquely claim a task, e.g. optimistic concurrency to check that the "claim" update succeeded rather than conflicted with another node already claiming the task.

There is no guaranteed ordering amongst the distributed tasks; they can happen in any order and concurrently.

When there is a requirement that a given functionality is only executed by one node at any given time in the cluster, this mechanism can be used as well. If the nodes know of a unique name/id for the functionality (e.g. "check-database-consistency", they can assign that id as the taskId and ask the submit to ignore if the tasks entry already exists in the DB. This way even if multiple nodes ask to submit the same task, only one task will be stored and hence picked up.

Queing serialized execution of tasks is currently not supported, i.e. note that the task is only executed once. Multiple submits are not queued for serialized execution, duplicate submits are discarded/ignored. We could add a submitSerialExecution() capability going forward if required (which would queue the tasks and only serially process each in the cluster).

The task is processed on an at-least-once approach; i.e. until the task handler says it completed the task successfully, or it says there is something wrong in the task and it should not be re-tried - triggering an error handler.

A group name (in the context of task type)  provides for a way to define a "join" on the group - a task that gets triggered when the whole group finished (either successfully or not).

A group task's unique identifier consists of the task type (e.g. "reconcile-set-of-records"), a group name ("reconcile-system-x"), task number in group(10), group total tasks (100). The group task generation/submission should be done in a fashion that re-doing the generation results in the same identifier.

When the total number of task that will be submitted for a group is unknown whilst generating the tasks, only the last record submitted for group has both the task number in group and group total tasks, which must be the same (100/100). This is necessary for the system to detect when the whole group has finished. Previous tasks only have the task number in group set. The generation of tasks within a group should be idempotent, i.e. it should be able to re-do the task generation in a fashion that duplicates are ignored because the taskid is the same again.

A task can submit sub-tasks through the handler provided context. This implies that the spawning task is the parent of these sub-tasks. This allows for an enclosing task to a) ensure only one enclosing task of the same type (controlled through task id) runs at the same time in the cluster, and b) there is a place to handle a "join" (group finished handler) of the overall task, after all the sub-tasks are finished.

Only when the sub-tasks all finished will the parent task's "group finished handler" be triggered. For example, if only one reconciliation for a given system should run at any given time, and it is broken into sub-tasks a Handler for Single task "[recon-parent][recon-system-x] will submit 100 sub-tasks as follows:

Rather than having the scheduler be cluster aware, every node will have its own local scheduler - configured with a homogenous schedule. Where there are tasks that should be triggered only once cluster-wide this can be achieved by inserting a task with a taskId that is the same for all nodes, and for example encodes the configured trigger time. The submit should be configured to ignore if the task already exists. Hence, examples of some variations upon a scheduler triggering:

package org.forgerock.openidm.distributedtask;


/**
 * Distribute tasks across the cluster nodes for both scaling and HA/fail-over
 */

public interface DistributedTask {

    /**
     * @param failGroupFast if true a single task failure within the group
     * aborts further processing of the group and eventually triggers the
     * group handler
     * @param taskData the data associated with the task. Content must be
     * either 1. String, 2. JSON based object model and/or 3. Serializable
     * @return Task ID
     */

    String submit(TaskIdentifier taskId, int priority, Object taskData);

    /**
     * Register the implementation for a given task type
     * Will get invoked when a node claims a task and hands it to this handler.
     * The handler is responsible for executing the task, maintaining the
     * lease as it is processing, and for returning results if any
     */

    void registerTaskHandler(String taskType, TaskHandler taskHandler);

    /**
     * Handle if a group is finished processing, either successfully
     * (all tasks) or not.
     */

    void registerGroupFinishedHandler(String taskType, TaskHandler taskHandler);

    /**
     * Optional handler to customize how failed tasks are treated. Default
     * is to log, remove individual task and trigger group complete handling
     * if appropriate.
     */

    void registerInvalidTaskHandler(TaskHandler taskHandler);

}


public interface TaskIdentifier {

    String getStringifiedTaskId();

}



public class SingleTaskId implements TaskIdentifier {

    public SingleTaskId(String taskType, String taskName) {...};

}


public class GroupedTaskId implements TaskIdentifier {

    public GroupedTaskId(String taskType, String groupName,
        boolean failGroupFast, int taskNumber, int totalGroupTasks) {...};

}


public abstract class TaskHandler {

    /**
     * @param leaseSeconds sets the initial number of seconds the task
     * handler should own the task, it should be
     * plenty of time to either process the task, or to renew the lease;
     * but at the same time be reasonably short to achieve the desired
     * fail-over times when a handler/node has a problem and stops
     * processing/renewing the lease.
     */

    public void setInitialLeaseTime(int leaseSeconds) {

    }

    /**
     * @param taskData The data for the task to process. For a regular task
     * this data is what was submitted, for group finished tasks this
     * contains the success/failure and results of all the tasks that ran in
     * the group.
     * @return optional result, which can be used in the group complete
     * handler
     */

    Object process(TaskIdentifier taskId, int priority, Object taskData,
        LeaseManager leaseManager) throws InvalidTaskException,
        RetryLaterException, LeaseExpiredException;

}


public interface LeaseManager {

    /**
     * Renew the lease (the time allotted for the task handler to process the
     * next step and before someone else should assume that this taskHandler
     * has an issue and another node should take over. The lease time should
     * be selected so that during normal operation it is plenty of time for
     * the taskHandler to renew the lease for continous processing.
     * An example would be a lease of 5 minutes, and the taskHandler expects
     * to renew the lease every ~30 seconds.
     * To allow for frequent lease updates without unnecessary performance
     * overhead, the lease manager may batch/time writes to the persistence
     * store, but it must first check continued ownership of the task with
     * sufficient time.
     * @param leaseSeconds how many seconds from current time the lease
     * should expire
     * @throws LeaseExpiredException if the lease renewal did not succeed.
     * The task handler is required to abort the task processing as it now
     * does not own the task anymore.
     */

    boolean renewLease(int leaseSeconds) throws LeaseExpiredException {}

}

Repository Query

Implementation Phases

Assumptions

Query Configuration

Queries can be configured either on the OrientDB service, i.e. the queries section of the file

conf/org.forgerock.repo.orientdb.json

Or they can be declared in the context of the calling module if it passes the expression to the repository as part of the query.

Query Types

The default is to express the query in the native language of the underlying store.

An optional querytype can be added going foward to explicitly declare the query type, e.g.

- Native query languages(s) - Languages/expressions directly understood by the underlying persistence store

- Common query languages(s) - NOTE: format TBD (*)  

examples:

QueryType NATIVE

QueryType COMMON_JSON_PATH (*)

QueryType COMMON_SQL (*)

Common Query Formats

To be decided

Query Tokens

Queries can contain tokens of the format ${token-name}, which will get replaced from parameters passed to the query.

Queries can be of type String or List.

At token of type List typically will be converted into a repository specific string representation to insert into the query, e.g. for an orientdb "query of "select ${_fields} from managed/user" a List of fields with firstname, lastname, email will be expanded to "select firstname,lastname,email from managed/user"

Query API

Object query(String id, Map params)

id:

identifies the resource set to query against, such as a relative uri for all "user" objects

params:

set of parameters that can be substituted as tokens into the query. Reserved property keys are defined in QueryConstants.

The system also populates the OrientDB document class name in the property "_resource" (QueryConstants.RESOURCE_NAME), which can be resulved as a query token, e.g.

select * from ${_resource}

return value:

a Map<String, Object> JSON based object model

The actual result set "records" are in the "result" map entry (QueryConstants.QUERY_RESULT) and of format List<Map<String, Object>>. Each list entry is a (potentially parts of) an object matching the query. If not all fields of an object were selected, only the selected parts of the object will be present.

Additional meta-data is available at the top level of the returned map, e.g. QueryConstants.STATISTICS_QUERY_TIME (time to execute the query in ms) or  QueryConstants.STATISTICS_CONVERSION_TIME (time to convert the results to the object model in ms)

The metadata in the future can contain more information about the query, such as paging token for the first and last result to page backwards/forwards.

The paging mechanism does not guarantee isolation or consistency across the consecutive paging calls, i.e. changing the underlying data in between/during paging calls(s) will show up in the results.  

Query Definition

The query definition happens based on structure of the stored object, e.g.

The query can contain tokens (e.g. ${firstname}) that will get substituted at runtime by the value supplied in the params map argument

For example, assuming a managed object "managed/user"

{
    "_id" : "999",
    "_rev" : "0",
    "firstname" : "John",
    "lastname" : "Doe",
    "address":
    {
         "street":"Some ln",
         "city":"Somewhere",
         "zip":"99999"
    }
}

Example associated queries

queries: {
    "my-query" : "select firstname, lastname from user where address.city = '${city}'"
}

The call to Managed Object API can look something equivalent to the following (e.g. from the REST API, or from recon code)

Map<String, String> params = new HashMap<String, String>();
params.put(QueryConstants.QUERY_ID, "my-query");
params.put("city", "Some");

Map<String, Object> result = query("managed/user", params);

List<Map<String, Object>> records = (List<Map<String, Object>>) result.get(QueryConstants.QUERY_RESULT);
for (Map<String, Object> entry : records.entrySet()) {
    entry.get("firstname");
    entry.get("lastname");
    entry.get("_id");
    entry.get("_rev");
}
Long queryTimeMs = (String) result.get(QueryConstants.STATISTICS_QUERY_TIME);
Long conversionTimeMs = result.get(QueryConstants.STATISTICS_CONVERSION_TIME);

// Not yet supported
// result.get(QueryConstants.PAGING_TOKEN_FIRST);
// result.get(QueryConstants.PAGING_TOKEN_LAST);
        

It may pass this query directly to the repository service, "as is". The specific repository service implementation, e.g. the repo-orientdb would have read in the query definitions. If necessary it can translate a common query format into something it natively understands, or - as in this example - use the native query directly.

Performance Considerations

Care needs to be taken to create appropriate indexes on queried fields.

When creating queries with tokens, note that OrientDB can only make prepared statements with tokens in the "where" clause. Hence if tokens are used elsewhere, the repository will create queries on demand with the replaced tokens.

Comments

TODO: make sure the API aligns with the query API specified on the managed object

An object modeled to represent the identity-related data managed by OpenIDM and potentially across multiple resources.

An artifact that defines the mappings between the properties of a managed object type and the properties of a resource object type.

An object, per the object model. Addressable objects have identifiers that can be used to establish a location where the object can be accessed, per OpenIDM's resource oriented architecture.

An application, service or data store with which OpenIDM synchronizes data with its own internal managed objects.

An artifact that is stored in a resource, which OpenIDM accesses during provisioning, synchronization and reconciliation.

An artifact that defines the structure and semantics of an object.