3.4. Reporting Activity Event Information

The Activity Collector accumulates information from the Activity events and provides this information to the Activity Server. The Activity Server then records it and provides support for querying this information. This section describes the role of Activity Collector and Activity Server in detail.

3.4.1. Activity Collector

The Activity Collector component is embedded in the activity execution environment. It accumulates activity information from the infrastructure used in the execution of a business transaction. It then reports the activity information to the Activity Server implicitly, using an appropriate Activity Logger implementation. The default Activity Logger implementation operates efficiently by providing a batching capability to send activity information to the server based either on a regular time interval, or a maximum number of activity units, whichever occurs first.
Locating the Activity Collector depends upon your run-time environment. For example, in a JEE environment, you can locate the Activity Collector using the following code: 
import org.overlord.rtgov.activity.collector.ActivityCollector;
import org.overlord.rtgov.activity.collector.ActivityCollectorAccessor;
....

ActivityCollector activityCollector = ActivityCollectorAccessor.getActivityCollector();
The accessor is initialized with an instance of the ActivityCollector when it is instantiated by the system such as a CDI. If an instance is not initialized when this method is invoked, then the client is blocked for a short period of time as it waits for the instance. If the instance is not initialized after this period, then a null is returned.

3.4.1.1. Processing the Activity Event Information

It is necessary to process certain events in order to enable the Runtime Governance infrastructure, and the user policies and rules that are defined within it, to make most effective use of activities that are reported. The relevant information extracted from these activity events are used in:
  • Correlating events to a particular business transaction instance.
  • Highlighting important properties that may be useful for business policies.
The Activity Collector has an ActivityCollector API that provides a method to enable information associated with the activity event to be preprocessed. The ActivityCollector API does this with the help of information processors.
These extracted properties can be used in further event analysis, to correlate the events and enable business relevant queries. The signature for this method is: 
public String processInformation(String processor, String type,
        Object info, java.util.Map<String, Object> headers,
                ActivityType actType);
Here,
  • The processor parameter is an optional value that you can use to explicitly name the information processor. If not specified, then Activity Collector checks all registered information processors to determine if they are relevant for the supplied information type.
  • The type parameter represents the information type. This can be in any form, as long as it matches the registered type defined in the information processor configuration.
  • The info parameter represents the actual information that the Activity Collector will process.
  • The headers parameter represents any header information that may have accompanied the information.
  • The actType parameter represents the activity event that any extracted properties must be recorded against.

3.4.1.2. Validating the Activity Events

The Activity Collector provides a validate method for pre-processing the activity event, using configured Activity Validators, before it is submitted to the activity server. This mechanism can be used to process activity events in the execution environment, prior to it being distributed to the activity server which may be located on a separate server. It can also be used to identify invalid situations, resulting in an exception. The execution environment can handle these exceptions and use them to block the business transaction associated with the activity event.

3.4.1.3. Managing the Activity Scope

An Activity Scope is a way of grouping a range of different activity types, that are reported to the activity server, into a single logical unit. It represents the same scope as an eXtended Architecture (XA) transaction, which means, it encompasses all the activity types within the range or discards it completely if the transaction is rolled back.
When the first activity is reported within the scope of a XA transaction, then the scope automatically starts. When that transaction subsequently commits, the Activity Unit (that is, the collection of activities accumulated during that scope) is reported to the Activity Server. However, if the activities are performed outside the scope of an XA transaction, then the component reporting the activity information can either explicitly start a scope, or just report the activity information. If no scope exists, and an activity type is reported, then it is reported to the activity server as a single event. The disadvantage of this approach is that it is less efficient, both in terms of reporting due to the duplication of certain header information, and for subsequent analysis. Having multiple activity events defined in a single unit related to the transaction, provides added value to interrelating the different event. This also provides some implied correlation that would not exist if the events were independently reported to the Activity Server.

3.4.1.4. Starting and Ending the Activity Scope

  1. Starting an Activity Scope

    To start the scope, invoke the startScope method on the Activity Collector as shown below: 
    activityCollector.startScope();
    If the application does not know whether a scope has already been started, and only wishes to start a single scope, then use the following condition: 
    boolean started=false;
if (!activityCollector.isScopeActive()) {
    activityCollector.startScope();
    started = true;
}
    Here, the isScopeActive method returns a boolean value to indicate whether the scope was previously started. If it returns true, then this component is also responsible for stopping the scope. If it returns false, then it means the scope has already been started, and therefore the component must not invoke the endScope method.

  2. Ending an Activity Scope

    To stop the scope, invoke the endScope method on the Activity Collector as shown below: 
    if (started) {
    activityCollector.endScope();
}

3.4.1.5. Reporting an Activity Type

The Activity Collector reports the activity event information to the server as an Activity Unit. An Activity Unit contains one or more actual activity events. The activity event is generically known as an Activity Type.
The Activity Collector mechanism removes the need for each component to report general information associated with the Activity Unit, and instead is only responsible for reporting the specific details associated with the situation that has occurred.

Procedure 3.2. Reporting an Activity Event

  1. To report an event, create the specific Activity Type and invoke the record method as shown below: 
    org.overlord.rtgov.activity.model.RequestSent sentreq=new org.overlord.rtgov.activity.model.soa.RequestSent();

sentreq.setServiceType(serviceType);
sentreq.setOperation(opName);
sentreq.setContent(content);
sentreq.setMessageType(mesgType);
sentreq.setMessageId(messageId);

activityCollector.record(sentreq);
  2. For certain types of event, it may also be appropriate to invoke an information processor(s) to extract relevant context and property information, that can then be associated with the activity event. To achieve this, use the processInformation() method as shown below: 
    Object modifiedContent=_activityCollector.processInformation(null, mesgType, content, headers, sentreq);

sentreq.setContent(modifiedContent);

3.4.1.6. Configuring an Activity Unit Logger

When the Activity Collector invokes the endScope method, the Activity Unit Logger component logs the activity unit that is generated as a result. The Activity Unit Logger interface has three methods:
  • init: This method initializes the activity unit logger implementation
  • log: This method is used for the Activity Unit logging.
  • close: This method closes the activity unit logger implementation.
The Batched Activity Unit Logger abstract base class implements the Activity Unit Logger interface. It provides the functionality to batch Activity Unit instances, and forwards them based on two properties:
  • Maximum Time Interval: If the time interval expires, then it sends the set of Activity Units to the server.
  • Maximum Unit Count: If the number of Activity Units reaches this maximum value, then it sends the batch to the server.
You can initialize this implementation explicitly in an embedded environment. In a JEE environment, use the PostConstruct and PreDestroy annotations to enable it for implicit initialization.
Additionally, Activity Unit Logger interface implements the Activity Server Logger, which is derived from the Batched Activity Unit Logger. The Activity Server Logger, therefore, sends activity information in a batch periodically based on the configured properties. When the batch of Activity Units are sent, this implementation forwards them to an implementation of the Activity Server interface, injected explicitly or implicitly into the logger. This can be used to either send the events directly to the Activity Server component, if it is located within the same server, or via a remote binding.
The following example shows a situation where an embedded Activity Collector is being initialized with an Activity Server Logger, which uses the REST Activity Server client implementation: 
import org.overlord.rtgov.activity.collector.ActivityCollector;
import org.overlord.rtgov.activity.collector.activity.server.ActivityServerLogger;
import org.overlord.rtgov.activity.server.rest.client.RESTActivityServer;

.....

        RESTActivityServer restc=new RESTActivityServer();
        restc.setServerURL(_activityServerURL);

        ActivityServerLogger activityUnitLogger=new ActivityServerLogger();
        activityUnitLogger.setActivityServer(restc);

        activityUnitLogger.init();

        _collector.setActivityUnitLogger(activityUnitLogger);

3.4.1.7. Collector Context Configuration

The Collector Context component of the Collector architecture is an interface that provides the Activity Collector with information about the environment such as principal, host, node, and port. This information can be used to complete the Origin information within an Activity Unit, as well as providing access to capabilities required such as the Transaction Manager, from the environment. Each type of environment in which the collector may be used provides an implementation of this interface. Depending upon the environment, this is either implicitly injected into the Activity Collector, or set explicitly using the setter method.

3.4.1.8. ActivityReporter Component for Simplified Reporting

Apart from using the Activity Collector mechanism for collective activity events, SwitchYard also provides an injectable component called ActivityReporter. The ActivityReporter component enables applications to perform simple activity reporting tasks. If injection is not possible, you can instantiate a default implementation of the interface. For example, the sample SwitchYard order management application uses this approach as shown below: 
@Service(InventoryService.class)
public class InventoryServiceBean implements InventoryService {

    private final Map<String, Item> _inventory = new HashMap<String, Item>();

    private org.overlord.rtgov.client.ActivityReporter _reporter=
                new org.overlord.rtgov.client.DefaultActivityReporter();

    public InventoryServiceBean() {
        ....
    }

    @Override
    public Item lookupItem(String itemId) throws ItemNotFoundException {
        Item item = _inventory.get(itemId);

        if (item == null) {

            if (_reporter != null) {
                _reporter.logError("No item found for id '"+itemId+"'");
            }

            throws new ItemNotFoundException("No item found for id " + itemId);
        }

        ....

        return item;
    }
}
The ActivityReporter enables the application to perform the following tasks:

Table 3.15. ActivityReporter Methods

Method Description
logInfo(String mesg) Log some information
logWarning(String meg) Log a warning
logError(String mesg) Log an error
report(String type, Map</String,String> props) Record a custom activity with a particular type and associated properties
report(ActivityType activity) Record an activity
Note that, you can not use this API to control the scope of an ActivityUnit. This API is purely intended to simplify the approach used for reporting additional incidental activities from within an application and you must use other parts of the infrastructure to handle the scope control of ActivityUnit.
Here is the maven dependency required to access the ActivityReporter: 
<dependency>
      <groupId>org.overlord.rtgov.integration</groupId>
      <artifactId>rtgov-client</artifactId>
      <version>${rtgov.version}</version>
</dependency>

3.4.2. Activity Server

Activity Units describe the activities that occur during the execution of business transactions in a distributed environment. An Activity Server records these Activity Units and also provides query support to retrieve previously recorded Activity Units.

3.4.2.1. Recording Activity Units

The Activity Units represent the logical grouping of individual situations that occur within a transaction boundary. The Activity Server records a list of Activity Units using the following approaches:
  • Direct Injection
    This is the simplest approach is to leverage CDI (for example within a JEE container) to directly inject the Activity Server implementation. Here is how you can use direct injection to get reference to the Activity Server: 
    import org.overlord.rtgov.activity.server.ActivityServer;

....

@Inject
private ActivityServer _activityServer=null;
    Once you obtain the reference to the Activity Server, you can call the store method to record a list of Activity Units as shown below: 
    import org.overlord.rtgov.activity.model.soa.RequestSent;
import org.overlord.rtgov.activity.model.ActivityUnit;

....

java.util.List<ActivityUnit> list=new .....;

RequestSent act=new RequestSent();
act.setServiceType(...);
...

list.add(act);

_activityServer.store(list);
  • REST Service
    You can access the Activity Server as RESTful service. here is an example: 
    
import org.codehaus.jackson.map.ObjectMapper;
import org.overlord.rtgov.activity.model.ActivityUnit;


.....

        java.util.List<ActivityUnit> activities=........
        java.net.URL storeUrl = new java.net.URL(....);
        // <host>/overlord-rtgov/activity/store

        java.net.HttpURLConnection connection = (java.net.HttpURLConnection) storeUrl.openConnection();

        String userPassword = username + ":" + password;
        String encoding = org.apache.commons.codec.binary.Base64.encodeBase64String(userPassword.getBytes());

        connection.setRequestProperty("Authorization", "Basic " + encoding);

        connection.setRequestMethod("POST");
        connection.setDoOutput(true);
        connection.setDoInput(true);
        connection.setUseCaches(false);
        connection.setAllowUserInteraction(false);
        connection.setRequestProperty("Content-Type", "application/json");

        java.io.OutputStream os=connection.getOutputStream();

        ObjectMapper mapper=new ObjectMapper();
        // Use jackson to serialize the activity units
        mapper.writeValue(os, activities);

        os.flush();
        os.close();

        java.io.InputStream is=connection.getInputStream();

        byte[] result=new byte[is.available()];

        is.read(result);
        is.close();
Note
Use the Activity Collector mechanism to aggregate and record the activity information. It is very efficient as each system individually reports events to the server.

3.4.2.2. Querying Activity Units

You can use the Activity Server to query a list of Activity Units that meet a supplied query specification. The Activity Server queries Activity Units using the following approaches:
  • Direct Injection
    The is the simplest approach is to leverage CDI to obtain a reference to the Activity Server. Once you obtain a reference to the Activity Server, then build the query specification based on the relevant criteria, and call the query method to retrieve the list of Activity Units as shown below: 
    import org.overlord.rtgov.activity.model.ActivityUnit;
import org.overlord.rtgov.activity.model.Context;
import org.overlord.rtgov.activity.server.QuerySpec;

....

QuerySpec qs=new QuerySpec()
                .setFromTimestamp(...)
                .setToTimestamp(...)
                .addContext(new Context(Context.CONVERSATION_ID,"txnId","123"));


java.util.List<ActivityUnit> list=_activityServer.query(qs);
  • REST Service
    You can access the Activity Server as RESTful service. here is an example: 
    import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.type.TypeReference;
import org.overlord.rtgov.activity.server.QuerySpec;
import org.overlord.rtgov.activity.model.ActivityUnit;


.....

        QuerySpec qs=........
        java.net.URL queryUrl = new java.net.URL(....);
        // <host>/overlord-rtgov/activity/query

        java.net.HttpURLConnection connection = (java.net.HttpURLConnection) queryUrl.openConnection();

        String userPassword = username + ":" + password;
        String encoding = org.apache.commons.codec.binary.Base64.encodeBase64String(userPassword.getBytes());

        connection.setRequestProperty("Authorization", "Basic " + encoding);

        connection.setRequestMethod("POST");
        connection.setDoOutput(true);
        connection.setDoInput(true);
        connection.setUseCaches(false);
        connection.setAllowUserInteraction(false);
        connection.setRequestProperty("Content-Type", "application/json");

        java.io.OutputStream os=connection.getOutputStream();

        ObjectMapper mapper=new ObjectMapper();
        // Use jackson to serialize the query spec
        mapper.writeValue(os, qs);

        os.flush();
        os.close();

        java.io.InputStream is=connection.getInputStream();

        java.util.List<ActivityUnit> activities = mapper.readValue(is, new TypeReference<java.util.List<ActivityUnit>>() {});

        is.close();