Dynamic Logging & the CORBA Notification Service

To avoid these hassles, I propose in this article an architecture that's based on the CORBA Notification service. For example, imagine a complex distributed system consisting of many applications. The system works fine until suddenly, the results aren't as expected. Now imagine being able to start up an application monitor so that each application in the system starts sending debug messages. The problem is found and the application monitor is stopped. The applications detect that the monitor is no longer available and stop sending these debug messages. There is no more overhead. Only when the monitor is plugged in does the overhead occur. This is essentially what my proposed architecture does. Furthermore, the same architecture can be used for logging any kind of message, such as error messages and warnings. Finally, this architecture places no inherent limitations on the number of applications and monitors supported simultaneously.

The CORBA Notification Service

The CORBA Notification service (http://www.omg.org/techprocess/meetings/schedule/Notification_Service_1.2_RTF.html) is intended to eventually replace the CORBA Event service (http://www .omg .org/ cgi-bin/doc?formal/97-12-11), which provides a loosely coupled method of communication between the providers of events (that is, publishers) and consumers of events (subscribers). This is achieved through the Event channel that handles the registering/unregistering of publishers/subscribers. In this model publishers do not care how many subscribers are waiting for events, or even if there are any subscribers at all. The same goes for subscribers. Publishers/subscribers of events can operate in push/pull mode, and can be combined in any combination on a channel.

Despite these benefits, several major shortcomings in the Event service have been identified:

• The Event service lacks explicit quality of service (QoS) control. For example, there are no policies to control how the channel queues and delivers events, such as being able to specify a maximum queue size, delivery order (FIFO, LIFO, priority based, and so on), or how to discard events in case of queue overflow. Another example pertains to event delivery reliability. Is the delivery simply the best effort, or does the channel persist in the event until it is delivered?
• All event data in the Event service is of type any. The channel has no way of performing any type checks as the event passes through it. The typed consumers and suppliers addendum to the Event service is rarely supported in Event service implementations due to its complexity.

• The Event service lacks event filtering. This means that any event delivered by any publisher on a channel will be delivered to all subscribers on that channel. The subscribers must decide for themselves whether they want to process the event. Besides making more work for the subscriber, this creates unnecessary network traffic.

• The Event service does not support sharing of event types being published or subscribed to on a channel. This information could be useful in reducing network traffic. For example, if no one is interested in event "A," why publish it?

The Notification service systematically addresses each of these shortcomings. In particular, a major change over the original Event service is the introduction of the "structured event type."

As Figure 1 illustrates, the structured event consists of an event header and event body. The event header is further composed of a fixed header and variable header. The fixed header consists of an event domain name, event type, and event name. The variable header consists of a sequence of name-value pairs. The name is of type string and the value is of type any. The variable header is used for specifying event-specific QoS properties.

The event body consists of a filterable body and a remaining data part. The filterable body consists of a sequence of name-value pairs. These name-value pairs are application/event-specific pairs that are completely user defined. Similar to the variable header part, the name is of type string and the value is of type any. The remaining data part is of type any and is used for transmitting the event payload; that is, the actual data of the event.

The Notification service provides a powerful constraint language that can be used to filter on any part of an event. This constraint language is an extension to the Trader service constraint language. Filters are specified in a special filter object, which is then attached to either the proxy or administration object. When a filter is attached to a proxy object, only subscribers connected through that proxy are affected. On the other hand, when a filter is attached to an administration object, it will apply to all proxies created through that administration object. Multiple filters can be attached to both proxies and administration objects. Either AND or OR semantics can be specified on a proxy or administration object to decide the outcome of all the filters applied to that proxy or administration object. For example, if AND semantics are specified for a proxy, then an event must satisfy the filter constraints of all the filters attached to that proxy in order to pass through.

Listing One creates a structured event with the domain name "example" and type "test." I have added a priority (=3) QoS name-value pair in the variable header. The filterable data consists of one name-value pair, event number, and a value of 11. The remaining data portion is set to "Hello World." Given this information, examples of specifying filter constraints using the extended trader constraint language include:

• All events of domain "example" and type "test:" $domain_name == 'example' and $type_name == 'test.' The event created in Listing One passes through this filter because it satisfies the constraints for both the domain_name and the type_name.
• All events that have a priority greater than 5: $Priority > 5. The event created in Listing One would not pass through this filter because the priority QoS value is set to 3, which is less than 5.

• All events that have an event number less than 200 or greater than 500: $event number > 200 and $event number < 500. The event created in Listing One would not pass through this filter since the event number value is 11, which is not between 200 and 500.

• All events that belong to the domain "example" and either have a priority equal to 3 or an event number less than 1000: $domain_name == 'example' and ($Priority==3 or $event number <1000). The event created in Listing One would pass through this filter since you satisfy the constraint for the domain_name and priority QoS is set to 3. (The event number constraint is also satisfied but will never get evaluated because only one match needs to occur in an OR condition.)

The introduction of the structured event type makes it possible to distinguish between different event types. Publishers/ subscribers can notify the event channel of event types either directly by registering the event type or indirectly by attaching filter objects to the proxy and/or administration objects. The event channel maintains a list of event types that are either being published or subscribed and notifies all publishers/subscribers whenever this list changes. Thus, if applications A and B both subscribe to event type X, and application C publishes that event, then the channel will notify application C when both A and B have unregistered for that event; that is, C will not be notified when either A or B unregisters but only when both have unregistered. Intelligent publishers can use this feature to publish only those events for which there are interested subscribers. Similarly, if C stops publishing the event, applications A and B will be notified of this change.

The Dynamic Logging Architecture

The architecture I present here relies on event filtering and sharing of event types. The architecture consists of: Tracer, which is used by the application interested in logging messages; and TracerMon, a stand-alone application that is used to monitor the messages being logged.

Listing Two (Tracer.java) presents the source code for Tracer. Several of the methods including the constructor throw a TraceException, which is a Java exception I define, which contains a message describing the error that occurred. Listing Three is the definition of TraceException (TraceException.java). Tracer has one constructor that takes a reference to a CORBA ORB. This ORB reference is required for resolving the Notification service; that is, obtaining a reference to the Notification service server, and for creating the type any variables while creating the structured event. The constructor calls the initialize method, which first calls the getChannel method to get a reference to a channel and then calls connect to connect itself to that channel. Both the getChannel and connect methods have boilerplate code for using the Notification service.

There are two noteworthy aspects of this boilerplate code. First, the required flag is set to True if there is at least one subscriber for the Tracer events. Second, Tracer informs the channel of the event it will publish. Since Tracer is a push supplier of structured events, it extends the org.omg.CosNotifyComm._StructuredPushSupplierImplBase class. Also, because Tracer is a push supplier, the channel does not poll Tracer for events; rather Tracer pushes events to the channel. I have overridden the subscription_change method. The channel calls this method when the first subscription request for any event type occurs. It is also called when an event type has no more subscribers. I use this information to toggle the required flag. The required flag is used by the trace method to decide whether to send an event or not. If this flag is set to True, then the event is sent.

The trace method takes two parameters: the message and a trace level (which is used by TracerMon for filtering events). The code for creating the event is similar to Listing One. As in that example, this event also has one name-value pair in its filterable body. The value of this pair is set equal to the trace level parameter. Also note that the event being published has the domain "Tracer" and a blank type. Tracer also has a destroy method that is to be called by all well-behaved users for a clean shutdown. I have also overridden the finalize method for those not so well- behaved users.

TracerMon is a standalone Java application used to monitor the messages logged by applications using Tracer. The program takes two arguments: a log filename (to which the log messages are dumped in addition to appearing on the screen); and an optional argument (a string specifying the filter, "$trace level==10," for example). If no filter is specified, then the monitor displays all messages. The constructor has some boilerplate code for getting a reference to the Notification service and a channel, and connecting to it.

If a filter string was specified, TracerMon installs a filter object corresponding to this filter string on the proxy it obtains from the channel. If no filter was specified, TracerMon notifies the channel of the event type it is interested in. Remember, TracerMon is interested in events published by Tracer, which are of domain "Tracer" and a blank type. Also, if a filter is specified, then TracerMon does not need to directly notify the channel of the event type it is interested in since the filter already does this on TracerMon's behalf. Because TracerMon is a push consumer of structured events, it extends the org.omg.CosNotifyComm._StructuredPushConsumerImplBase class. Also, since TracerMon is a push consumer, it does not poll the channel; rather the channel will push events to TracerMon as they become available. Here, I have overridden the offer_change method. The channel calls this method when the first intent to publish any event type is announced to the channel. It is also called when no more suppliers are publishing an event anymore. I use the knowledge gained from this method to display diagnostic information. Unlike Tracer, where the implementation of the subscription_change method is critical so that it may gain its intelligence, the implementation of the offer_change method in TracerMon is purely for informational purposes. When an event becomes available, the channel calls the push_structured_event method, which I have overridden to display the message from the event and dump it to the log file specified. To exit TracerMon, press the "q" key and then Enter.

Testing the Architecture

Test.java (available electronically; see "Resource Center," page 5) is a program that tests the architecture. This test program uses Tracer to log 20 messages. Each message has a different trace level, which is set equal to the number of the message being logged; for example, trace level is 0 for the first message, 1 for the next, and 19 for the last message. There is a delay of two seconds between each message sent. The test program also uses Tracer's "on" method to significantly improve performance in those cases when tracing is not on because no expensive string construction is performed.

I suggest you perform the following tests:

• Start the test program and then start the TracerMon program. Each time, specify a different filter (or no filter) while starting the TracerMon program. For example, specify the filter as "$trace level == 5 or $trace level== 0." You should only see two messages ("Hello...5" and "Hello...10") from the test program in the monitor output.
• The Tracer program has several System.outs commented out. Uncomment these and run the programs again. Make sure that the messages are only sent when the TracerMon application is plugged in. For example, while both programs are running, stop the TracerMon. No more messages will be sent, as can be verified by making sure that the System.out within the if(required) loop in the trace method is not being executed. Now, while the test program is still running, start the TracerMon program. The messages will start being sent again. Notice how the channel notifies both the Tracer and the TracerMon of any changes in the events being published or subscribed.

• Finally, try running multiple TracerMon and/or test programs.

Conclusion

This architecture is meant to be a base upon which to build and there are several ways to improve/extend upon it, including:

• The architecture uses the default channel in its implementation. However, a separate channel defined solely for this purpose may make more sense. It is always desirable to have multiple channels, dedicating each channel to a specific type of message(s). This way each channel has a well-defined purpose and does not get bogged down with every message going through the system.
• The channel used may be stored in the CORBA Naming service with a well-known name, such as "TracerChannel." The Tracer may be modified so that it can be configured to know which name to use to perform the lookup to get the channel. This isolates/abstracts the channel creating/obtaining process for maximum portability.

• The only filterable property specified currently is trace level. More may be added as required. For example, a "classname" property might be added that is set equal to the class name of the object using Tracer. Now we can monitor messages being logged by a specific class in the system.

• Currently after getting a message, TracerMon simply displays it to standard out. TracerMon may be configured with a callback class that is responsible for rendering the message any way desired, for example, in a fancy UI. Now instead of calling System.out, TracerMon calls a method on this callback class.

• The current implementation of TracerMon works with Visibroker 3.4. There are two lines of code specific to the Visibroker ORB; namely, the call to the BOA impl_is_ready and the Visibroker specific ORB shutdown call. With a few modifications, this implementation can be made independent of the CORBA implementation. For example, one way is to create a separate class that implements an interface with two methods — init and shutdown. Both these methods take a reference to an ORB. In our case, init would call the BOA impl_is_ready method and shutdown would call the Visibroker-specific shutdown method. One such class for each supported ORB would be required. TracerMon could then be started up with the appropriate class. For ORBs that are fully compliant with CORBA 2.3 and higher, I suggest using the POA for maximum portability.

• Currently, as implemented, only Java programs can use Tracer. Tracer itself may be made into a CORBA server that can be used by any other CORBA client, irrespective of language (as long as an IDL to language mapping is available), such as C, C++, Smalltalk, and the like.

DDJ

Listing One

// domain, type, event name
EventType type = new EventType ("example", "test");
FixedEventHeader fixed = new FixedEventHeader (type, "event");

// variable header, NV Pairs
// Let's add a priority QoS
Property variable[] = new Property[1];
variable[0] = new Property ();
variable[0].name = Priority.value;
variable[0].value = orb.create_any ();
variable[0].value.insert_short ((short) 3);

// Filterable data, NV Pairs
// Remember these are user defined and can be anything.
Property filterable[] = new Property[1];
filterable[0] = new Property ();
filterable[0].name = "eventnumber";
filterable[0].value = orb.create_any ();
filterable[0].value.insert_long ((long)11);

// Event Data
org.omg.CORBA.Any data = orb.create_any ();
data.insert_string("Hello World");

// We have finished creating all the distinct pieces
// Now let's create a structured event
StructuredEvent event = new StructuredEvent ();
event.header = new EventHeader (fixed, variable);
event.filterable_data = filterable;
event.remainder_of_body = data;

Back to Article

Listing Two

// This is the class that applications use for logging messages.
import org.omg.CosNotification.*;
import org.omg.CosNotifyComm.*;
import org.omg.CosNotifyChannelAdmin.*;
public class Tracer extends _StructuredPushSupplierImplBase
{
    private StructuredProxyPushConsumer sp;
    private boolean required = false;
    private org.omg.CORBA.ORB orb = null;

    public Tracer(org.omg.CORBA.ORB orb) throws TraceException
    {
            // save the orb reference for later
            this.orb = orb;
        initialize();
    }
    // shutdown this instance of tracer
    public void destroy()
     {
        // delegate to finalize()
        finalize();
        sp = null;
    }
    public void finalize()
    {
        // Disconnect from the channel if we need to.
        if(sp != null)
        sp.disconnect_structured_push_consumer();
    }
    // not implemented
    public void disconnect_structured_push_supplier ()
    {
    }
    // This is where Tracer gets its "smarts" from. This function is called 
    // any time the aggregate of event subscribed to on the channel.
    public void subscription_change (EventType added[], EventType removed[])
    {
        //System.out.println("Received Subscription change...");
        for( int i = 0; i < added.length; i++ )
        {
            if( added[i].domain_name.equals ("Tracer") &&
                added[i].type_name.equals ("") )
            {
                //System.out.println("Turning events on...");
                // Synchronize to serialize the "required" flag access
                synchronized(this)
                {
                    required = true;
                }
                break;
            }
        }
        for( int i = 0; i < removed.length; i++ )
        {
            if( removed[i].domain_name.equals ("Tracer") &&
                removed[i].type_name.equals ("") )
           {
                //System.out.println("Turning events off...");
                // Synchronize to serialize the "required" flag access
                synchronized(this)
                {
                    required = false;
                }
                break;
            }

        }
    }
    // Synchronize to serialize the "required" flag access
    synchronized public boolean on()
    {
        return required;
     }
    // This method takes the message and tracelevel parameters and
    // constructs a structured event.
    public void trace(String msg, int tracelevel) throws TraceException
    {
            if(sp == null)
        {
            throw new TraceException("Error: No connection");
        }
        boolean req = on();
        // If anyone is interested in these events the
        // required flag would be true.
        if( req )
        {
                //System.out.println("Logging message: " + msg);
            EventType type = new EventType ("Tracer", "");
            FixedEventHeader fixed = new FixedEventHeader (type, "event");
            // variable header, NV Pairs
            Property variable[] = new Property[0];
            // Filterable data, NV Pairs
            // Use the tracelevel here
            Property filterable[] = new Property[1];
            filterable[0] = new Property ();
            filterable[0].name = "tracelevel";
            filterable[0].value = orb.create_any ();
            filterable[0].value.insert_long(tracelevel);
            // Event Data
            org.omg.CORBA.Any data = orb.create_any ();
            data.insert_string(msg);
            // Put everything together...
            StructuredEvent event = new StructuredEvent ();
            event.header = new EventHeader (fixed, variable);
            event.filterable_data = filterable;
            event.remainder_of_body = data;
            try
            {
                // Off the event goes...
                sp.push_structured_event(event);
            }
           catch(org.omg.CosEventComm.Disconnected e)
            {
                throw new TraceException("Error: Reconnect required");
            }
        }
    }
    /////////////////////////
    // Private Methods...
    /////////////////////////
    private void initialize() throws TraceException
    {
            EventChannel channel = getChannel();
        connect(channel);
    }
    private EventChannel getChannel() throws TraceException
    {
             // Get a reference to the notification service
        org.omg.CORBA.Object obj = null;
        try
        {
            obj = orb.resolve_initial_references("NotificationService");

        }
        catch( org.omg.CORBA.ORBPackage.InvalidName ex )
        {
            throw new TraceException("Error: Failed to 
                                        resolve Notification Service");
        }
        // Get the event channel factory from the service
        EventChannelFactory factory = EventChannelFactoryHelper.narrow(obj);
        // Get the event channel from the factory
        EventChannel channel = null;
        try
        {
            channel = factory.get_event_channel(0);
        }
        catch(ChannelNotFound ex)
        {
            // Channel is not found, let's create a new one...
            // Create a channel with no QoS or admin properties.
            Property[] qosProp = new Property[0];
            Property[] admProp = new Property[0];
            org.omg.CORBA.IntHolder id = new org.omg.CORBA.IntHolder ();

            try
            {
                channel = factory.create_channel(qosProp, admProp, id);
            }
            catch( UnsupportedQoS exc )
            {
            }
            catch( UnsupportedAdmin exc )
            {
            }
        }

        return channel;
        }
   private void connect(EventChannel channel) throws TraceException
    {
        // Get the default supplier admin object for this channel
        SupplierAdmin admin = channel.default_supplier_admin ();
        // Now ask the admin for a reference to a push consumer proxy
        ClientType ctype = ClientType.STRUCTURED_EVENT;
        org.omg.CORBA.IntHolder pid = new org.omg.CORBA.IntHolder ();
        ProxyConsumer proxy = null;
        try
        {
            proxy = admin.obtain_notification_push_consumer (ctype, pid);
        }
        catch( AdminLimitExceeded ex )
        {
            throw new TraceException("Error: Max number of proxies exceeded");
        }
         sp = StructuredProxyPushConsumerHelper.narrow (proxy);
        try
        {
            // connect to the channel
            sp.connect_structured_push_supplier(this);
            // Inital "smarts"
            // See if any TracerMons are there listening for Tracer events
            EventType types[] =
sp.obtain_subscription_types(org.omg.CosNotifyChannelAdmin.ObtainInfoMode.ALL_NOW_UPDATES_ON);
            //System.out.println("Total events being 
                                         subscribed to = " + types.length);
            for( int i=0; i<types.length; i++ )
            {
                if( types[i].domain_name.equals("Tracer") 
                                         && types[i].type_name.equals("") )
                {
                    //System.out.println("Turning events on...");
                    required = true;
                    break;
                }
            }
            // Tell the channel which events I publish...
            NotifyPublish publisher = NotifyPublishHelper.narrow (proxy);
            EventType added[] = new EventType[1];
            added[0] = new EventType ("Tracer", "");
            EventType removed[] = new EventType[0];
            publisher.offer_change(added, removed);
        }
        catch( org.omg.CosEventChannelAdmin.AlreadyConnected ex )
        {
            throw new TraceException("Error: Unexpected 
                                     exception [AlreadyConnected]");
        }
        catch( org.omg.CosNotifyComm.InvalidEventType ex1 )
        {
            throw new TraceException("Error: Unexpected 
                                     Exception [InvalidEventType]");
        }
    }
}

Back to Article

Listing Three

 // Exception raised by the Tracer class in case of an error.
public class TraceException extends java.lang.Exception
{
    String message = null;
    public TraceException(String msg)
    {
    message = msg;
    }
    public String getMessage()
    {
    return message;
    }
}

Back to Article