Showing posts with label WSO2. Show all posts
Showing posts with label WSO2. Show all posts

Saturday, March 12, 2016

Sensing the world with Data of Things


Henry Ford once said “Any customer can have a car painted any colour that he wants so long as it is black!” which it’s now long gone. In the current context people seek for personalized treatment. Imagine calling customer service, every time you call you have to go through all the standard questions, and they don’t have a clue why you might be calling? Or whether you called before? and in the case of shopping, even if you are a regular customer and have a platinum or gold membership card you will not get any special treatment at the store, may be presenting the card at the cashier can get you a discount. 

What’s missing here? They don’t know anything about the customer to give a better service. Hence the simple remedy for the above issue is building customer profiles, this can be done with the historical data you might have about the customer, next you need to understand and react to the context the customer evolves such as whether he is in an urgency, has he contacted you before, etc, and finally you have to  react in real time to give the best customer satisfaction. Therefore to provide the best customer satisfaction identifying the context is a key element, and in the present world the best way of identifying the customer context is via the devices your customer has and via the sensors that’s around him which indeed the Internet of Things (IoT)


IoT is not a new thing, we have had lots of M2M systems that have monitored and controlled devices in the past, and when it comes to IoT we have more devices having sensors and a single device having more sensors. IoT it’s an ecosystem where IoT devices should be manufactured, app for those devices should be developed (e.g apps for phone), users 
should be using those devices and finally they should be monitored and managed. WSO2’s IoT Platform plays a key role in managing and providing analytics for the IoT devices in the ecosystem.




Data Types in IoT Analytics


Data from IoT devices are time bound because these devices do continuous monitoring and reporting. With this we can do time series processing such as energy consumption over time. OpenTSDB is a specialised DB implemented for time based processing.


Further since IoT devices are deployed in various geological locations and since some of those devices move, location is also becomes another important data type for IoT Devices. IoT devices are usually tacked with GPS and currently iBeacons are used when the devices are  within a building. Location based data enables geospatial processing such as traffic planning and better route suggestion for vehicles. Geospatially optimised processing engines such as GeoTrellis are especially developed for these type of usecases.


IoT is Distributed


Since IoT is distributed by nature, components of the IoT network constantly get added and removed. Further since IoT devices get connected to the IoT network through all type of communication networks such as from weak 3G networks to Ad-hoc peer-to-peer networks, and they also use various communication protocols such as Message Queuing Telemetry Transport (MQTT), Common Open Source Publishing Platform (CoApp) and ZigBee or Bluetooth low energy (BLE). Due to these the data flow of the IoT network continuously get modified and repurposed. As data load varies dynamically in the IoT network, on-premise deployment will not be suitable, and hence we have to move towards public or hybrid cloud based deployment. IoT has an event driven architecture to accommodate its distributed nature where its sensors reports data as continuous event streams working in an asynchronous manner.


Analytics for IoT


IoT usually produces perishable data where their value drastically degrades over time. This imposes the importance of Realtime Analytics in IoT. With Realtime Analytics temporal patterns, logical patterns, KPIs and thresholds can be detected and immediately alerted to respective stakeholders, such as alarming when temperature sensor hits a limit and notifying  via car dashboard if the tire pressure is low. Systems such as Apache Storm, Google Cloud DataFlow & WSO2 CEP are build for implementing such Realtime Analytics usecases.


Realtime alone is not enough! We should be able to understand how current situation deviates from the usual behaviour, to do so we have to process historical data. With Batch Analytics, periodic summarisation and analytics can be performed on historic data against which we can compare at realtime. Average temperature in a room last month, and total power usage of the factory last year are some example summarizations that can be done using systems like Apache Hadoop & Apache Spark on the data stored in scalable databases such as Apache Cassandra and Apache HBase.


Ok, with Batch Analytics we defined the thresholds and with Realtime Analytics we detected and alerted threshold violations. Notifying violations may results in preventing disasters but it does not help stopping similar issues arising again. To do so we need to investigate the historical data and identify the root course of the issue and eliminate that. This can be done through Interactive Analytics and with it’s Ad-Hoc Queries, it enables us to search the data set how system and all related entities have behaved before the alert was raised. Apache Drill, Apache Lucene and indexed storage systems such as Couchbase are some systems provides Interactive Analytics.


Than being reactive, staying a step ahead predicting issues & opportunities bring great value. This can be achieved through Predictive Analytics which helps in scenarios such as proactive maintenance, fraud detection and health warnings. Systems such as Apache Mahout, Apache Spark MLlib, Microsoft Azure Machine Learning, WSO2 ML & Skytree are systems that can help us build Predictive Analytics models.


An Integrated Solution for IoT Analytics


From the about technologies by selecting WSO2 Siddhi, Apache Storm, Apache Spark, Apache Lucene, Apache HBase, Apache Spark MLLib and with many other open source softwares WSO2 has built and integrated Data Analytics solutions that support Realtime, Batch, Interactive and Predictive analytics solution called WSO2 Data Analytics Server.




Issues in IoT Analytics



Extreme Load


With compared with the scale of the data produced by sensors, distributed centralised analytic platforms cannot scale and even if they can - it will not be cost effective. Hence we should look at whether we need to process and/or store all the data produced by the sensors? In most cases we only need the aggregations over time, trends that exceed thresholds, outliers, event matching a rare condition, and when the system is unstable or changing. For example from a temperature sensor we only need to send readings when there is a change in temperature and its no point periodically sending same value. This directs us to optimise sensors or data collection points to focus on doing local optimisations before publishing data. This helps in quick detection of issues as part of the data is already processed locally and instant notifications since decisions are also taken at edges. Taking decision at the edge can be implemented with the help of complex event processing libraries such as WSO2 Siddhi and Esper.


Uncertainty


Due to the distributed nature of IoT, data produced can be duplicated, arrives out of order, missing or even be wrong.

Redundant sensors & network latency can introduce duplicated events and out of order event arrival. This can impose difficulty doing temporal event processing, such as Time Windows & Pattern Matching. These are very useful for usecases such as Fraud detection, and Realtime Soccer Analytics (based on DEBS 2013 dataset) https://goo.gl/c2gPrQ where we build a system that monitors the soccer players and the ball and identified ball kicks, ball possession, shot on goal & offside. Algorithms based on K-Slack can help to order events before processing them on realtime.

Due to network outages data produced by the IoT sensors can go missing, and in these situations using complimenting sensor reading are very important where one of those sensor value will be some sort of an aggregation done at the edge which will help us to approximate the missing sensor values based on its aggregation. Such as publishing Load and Work reading when monitoring electricity where when some events are missed, from a later Work Event reading we will be able to approximate the Load reading that should have arrived during the outage. The other alternative is using fault-tolerant data streams such as Google Millwheel.

Further at times sensor reading won't be correct, this can be due to various reasons such as sensor quality and environment noise, in such situations we can use kalman filtering to smoothen consecutive sensor readings for better approximation. These type of issues are quite common when we use iBeacons for location sensing.


Visualisation of IoT data


Visualisation is one of the most important aspect of effective analytics and hence with Big Data and IoT visualisation becomes even complicated. Per-device & Summarization Views are essential and more than that users should be able to visualize as device groups based on various categories such as device type, location, device owner types, deployed zone and many more. Since these categories are dynamic and each person monitoring the system have various personal preferences, therefore composable & customisable dashboard is essential. Further charts and graphs should be able to visualise the huge stored data, where sampling & indexing techniques can be used for better responsiveness.  


Communicating with devices


In IoT sending a command/alert to a devices is complicated, to do so we have to use client side pooling based techniques. Here we store the data the need to be pushed to the client in a database or queue and expose them via secured APIs (through systems like WSO2 API Manager).


Reference Architecture for IoT Analytics




Here data is collected through message broker such as MQTT, immediately written to the disk by WSO2 Data Analytics Server (DAS), at the meantime the collected data is cleaned in realtime, this cleaned data is also persisted, and parallely the cleaned data is fed into realtime event processing which in deed sends alerts and provides realtime visualisations. Stored clean data is used by WSO2 Machine Learner (ML) to build machine learning models and deploy them at WSO2 DAS for realtime predict ions. Further the stored clean data is also used by Spark to run Batch analytics producing sumarisation, which are then visualised in dashboards.



It’s a pleasure for me presenting “Sensing the world with Data of Things” at Structure Data 2016, San Francisco. Please find the slides below. 




Sunday, February 28, 2016

WSO2 Complex Event Processor 4.1

WSO2 Complex Event Processor  4.1 released  23rd February 2016.

This release mainly focuses on improving and stabilizing the product and enhancing its capabilities.


One of the main features that was included in this release is instrumenting and monitoring support for WSO2CEP  as well as Siddhi.  This enable users to identify throughput and memory consumption of each and every component of WSO2CEP and Siddhi. Through this users can identify possible bottlenecks in their queries and optimize CEP for better performance.




The same is also exposed via JMX  such that It can be monitored to third party JMX consumers  such as jconsole.




Another important feature CEP introduced in this release is visualizing Siddhi queries.




One of the notable improvement of this release is its improved high-availability support. now CEP can support high availability with more than two nodes providing zero downtime with no data loss, and as it also stores its state as a periodic snapshot to a database even during full cluster restart CEP has the capability of restoring it state from its last available snapshot. For more information refer the documentation here.

Further WSO2CEP has introduced several improvements to it’s core runtime complex event processor engine Siddhi. They are as follows:


  • Hazelcast Event Table - Allowing events to be stored and  manipulated in  Hazelcast based In-Memory Data Grid.
  • Minima and Maxima detection - Detecting maxima and minima in an event pattern, this allows detecting complex  stock market patterns using combinations of maxima and minima.
  • Map extension -  This is introduced to support arbitrary key value pairs in Siddhi which was not supported for a long time. This function allows users to create a map, add remove and check for keys and values within a hash map.
  • InsertOrUpdate to Event Table -  This facilitate doing and insert or an update as an atomic operation on Even Tables.
  • Outer and left joins in Siddhi - In addition to inner joins Siddhi now supports outer and left joins
  • Time length window - The eviction policy on Siddhi window is triggered both by time and length properties.
  • External Batch window -  Allowing batch window to get triggered by event time rather by getting triggered by the system time.

You can download WSO2 Complex Event processor 4.1 from here, its documentation from here,  and find the latest Siddhi documentation here.

Wednesday, September 30, 2015

What to expect in WSO2 CEP 4.0

WSO2 Complex Event Processor 4.0 is released! A long waited CEP release from WSO2. This is a major break through among current CEP products.


Here are some notable features it has introduced :

Scale and Performance 

  1. Support for distributed processing with Apache Storm - WSO2 CEP runs Siddhi CEP on Apache Storm in a distributed mode while providing an SQL like Siddhi Query Language to define queries on Storm
  2. Faster CEP engine with WSO2 Siddhi 3.0 - With LMAX Disruptor and improved memory management, Siddhi was able to run even faster than its predecessors doing ~ 4M Events/Sec
  3. HA Support - Supporting Hot-Worm High Availability with zero downtime
  4. More effective native thrift/binary transports 
Usability 
  1. Realtime Dashboard support - Support for gadget generation wizard and customisable realtime dashboard
  2. Easy to use management console for business users - Ability to tweak queries with a form based UI without knowing the implementation and complexities of SiddhiQL, Storm and  Java 
  3. Event simulation capability - Try sending events easily 
  4. Siddhi query try-it - Try writing queries 
  5. Event flow visualisation - See how you the event flow pipeline will look like

Enhancements  
  1. New protocols introduced such as MQTT, Kafka, Websocket - Support for many new and improved event receiver and publisher capabilities
  2. Enhanced query language with more extensions -  Such as Function, Stream Processor, Window and Aggregate Function Processor
  3. Support for machine learning via ml, pmml (Predictive Model Markup Language) extensions - To run machine learning model on realtime
  4. Support for Natural Language processing via nlp (Natural Language Processing) extension - Supported via Stanford NLP
  5. Geo processing capabilities via geo extension and Geo Dashboard - Dashboard having the capability to monitor and apply queries on geo spaces. 
  6. Support for scripting using JavaScript, Scala and R - Can use scripts to write CEP logic!
This release not only overcomes the scaling bottleneck of previous WSO2 CEP versions, but it also provides a seamless query language that allows uses to run the same queries both on single node and on distributed environment, making WSO2 CEP 4.0 as a unique product in the market. 

You can download WSO2 CEP 4.0 from here 
and a quick getting started guide can be found here 

Try it out, and for questions please post on StackOverflow with tag 'wso2cep'. The community and the WSO2 team will be very happy to answer :)

You can contribute your code e.g. extensions and adopters, such that it will be maintained and released on future product releases:
For Extensions and Siddhi improvements please send the pull request to Siddhi repo and 
For adopters and other contributions please send to carbon-analytics-common carbon-event-processing and product-cep repos :) 

Be Realtime !!

Friday, July 25, 2014

Without restart: Enabling WSO2 ESB as a JMS Consumer of WSO2 MB

WSO2 ESB 4.8.1 & WSO2 MB 2.2.0 documentations have information on how to configure WSO2 ESB as a JMS Consumer of WSO2 MB queues and topics. But they do not point out a way to do this without restarting ESB server.

In this blog post we'll solve this issue.

With this method we will be able to create new queues in WSO2 MB and consume them from WSO2 ESB without restarting it.

Configure the WSO2 Message Broker

  • Offset the port of WSO2 MB to '1'  
  • Copy andes-client-*.jar and geronimo-jms_1.1_spec-*.jar from $MB_HOME/client-lib to $ESB_HOME/repository/components/lib 
  • Start the WSO2 MB

Configure the WSO2 Enterprise Service Bus

  • Edit the $ESB_HOME/repository/conf/jndi.properties file (comment or delete any existing configuration)
connectionfactory.QueueConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'
connectionfactory.TopicConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5673'
  • Edit the $ESB_HOME/repository/conf/axis2.xml file and uncomment the JMS Sender and JMS Listener configuration for WSO2 Message Broker 
  • Start the WSO2 ESB 

Create Proxy Service

The Proxy Service name will become the queue name in WSO2 MB. If you already have a queue in MB and if you want to listen to that queue, then set that queue name as the proxy service name. Here I'm using 'JMSConsumerProxy' as the queue name and the proxy service name.

<?xml version="1.0" encoding="UTF-8"?> 
<proxy xmlns="http://ws.apache.org/ns/synapse" 
       name="JMSConsumerProxy" 
       transports="jms" 
       statistics="disable" 
       trace="disable" 
       startOnLoad="true"> 
   <target> 
      <inSequence> 
         <property name="Action" 
                   value="urn:placeOrder" 
                   scope="default" 
                   type="STRING"/> 
         <log level="full"/> 
         <send> 
            <endpoint> 
               <address uri="http://localhost:9000/services/SimpleStockQuoteService"/> 
            </endpoint> 
         </send> 
      </inSequence> 
      <outSequence> 
         <drop/> 
      </outSequence> 
   </target> 
   <description/> 
</proxy> 

Testing the scenario

  • Inside $ESB_HOME/samples/axis2Server/src/SimpleStockQuoteService run ant 
  • Now start the Axis2 Server inside $ESB_HOME/samples/axis2Server (run the relevant command line script
  • Log into the WSO2 Message Broker Management Console and navigate to Browse Queues 
  • Find a Queue by the name JMSConsumerProxy 
  • Publish 1 message to the JMSConsumerProxy with payload (this has to be done in the Message Broker Management Console) 
<ser:placeOrder xmlns:ser="http://services.samples" xmlns:xsd="http://services.samples/xsd"> 
    <ser:order> 
        <xsd:quantity>4</xsd:quantity> 
    </ser:order> 
</ser:placeOrder>
  • Observe the output on the Axis2 Server and WSO2 ESB console.
Hope this helped you :) 

Friday, February 28, 2014

Getting inside out of Social Data using WSO2 BAM & WSO2 CEP

We engage in Social Media, but have we ever analysed the data it contains?

Why we need to analyse?
Can it help us to make prompt business decisions?
How we can do that?

I have done a screencast to answer all these,
using WSO2 Business Activity Monitor (2.4.0) and WSO2 Complex Event Processor (3.0.0).


Wednesday, April 10, 2013

Integrating WSO2 BAM & WSO2 CEP

This post demonstrate how we can integrate both WSO2 BAM and WSO2 CEP and run them as a single instance.

To integrate both WSO2 BAM and WSO2 CEP we need to download BAM and install the CEP features form the WSO2 p2repo. Here I have used WSO2 BAM 2.2.0 and 4.0.7 p2repo. Follow the wiki link to get to know how to install new features to existing products.

Testing the installation.
To test the instillation add broker-manager-config.xml to repository/conf/, and add axis2_client.xml to repository/conf/axis2/.
Here axis2_client.xml has the email sender information, and I'm using a demo email account for this.

Then create a directory called 'cepbuckets' in repository/deployment/server/ and add bam-cep-kpi-analyzer.xml there.
In bam-cep-kpi-analyzer.xml, you will be able to find a line
<cep:output brokerName="emailBroker" topic="wso2cep.demo@gmail.com/Phone Purchase Notification">
Change "wso2cep.demo@gmail.com" to the endpoint use email ID, who needs to receive the notification.

Now if you run the KIP-analyser sample of BAM all the events that have totalPrice>350000 and quantity>3 will be notified to the user.

Monday, February 18, 2013

Using Apache Qpid with WSO2 CEP

Though WSO2 CEP 2.x.x can work with Qpid, it does not have samples that demonstrates this, where  CEP only have samples for ActiveMQ.

Here I'm providing steps to configure, and run some samples using Apache Qpid.

First step is downloading and running Apache Qpid.
You can download Qpid from the project web site. Note, I have used Qpid java 0.18 for these samples, you can download that from here.

Next, unzip and start the broker.

Now to configure WSO2 CEP with Qpid JMS broker, follow the steps provided at Configuring JMS-Qpid Broker.

Test publishing and subscribing to JMS Map messages


Edit the jms-twitter-stockquote-analyser.xml provided at /samples/cep-samples/conf/buckets directory by changing the input and output broker names as, 
<bucket name="TwitterAndStockQuoteAnalyzer" xmlns="http://wso2.org/carbon/cep">
     ...
     <input topic="AllStockQuotes" brokerName="qpidJmsBroker">
     ...
     <input topic="TwitterFeed" brokerName="qpidJmsBroker">
     ...
     ...
     <output topic="PredictedStockQuotes" brokerName="qpidJmsBroker">
     ...
</bucket>

Here we are using the brokerName as "qpidJmsBroker", because we have define it so, when Configuring JMS-Qpid Broker.

Next copy past this file to /repository/deployment/server/cepbuckets directory. 

To test the broker download the client code from here, and unzip cep-210-jms-client-qpid.zip 

To subscribe to the PredictedStockQuotes topic, in the terminal go to cep-210-jms-client-qpid directory and run:
ant mapPredictedStockQuotesSubscriber
To publish to the AllStockQuotes topic, in another terminal from the same directory, run:
ant mapAllStockQuotesPublisher
To publish to the AllStockQuotes topic, in another terminal from the same directory, run:
ant mapTwitterFeedPublisher
Then you will be able to observe the outputs. Here the output will be similar to the output of Twitter and StockQuote Analyzer sample.

Test publishing and subscribing to JMS XML messages


Edit the xml-stockquote-anlyzer.xml provided at /samples/cep-samples/conf/buckets directory by changing the input and output broker names as, 

<cep:bucket name="XMLStockQuoteAnalyzer" xmlns="http://wso2.org/carbon/cep">
     ...
     <cep:input topic="AllStockQuotes" brokerName="qpidJmsBroker">
     ...
     ...
     <cep:output topic="FastMovingStockQuotes" brokerName="qpidJmsBroker">
     ...
</cep:bucket>

Here we are using the brokerName as "qpidJmsBroker", because we have define it so, when Configuring JMS-Qpid Broker.

Next copy past this file to /repository/deployment/server/cepbuckets directory, and make sure to delete other bucket configurations in order to avoid conflicts.

If you don't have the client code (used for the previous sample), download the client code from here, and unzip cep-210-jms-client-qpid.zip, 

To subscribe to the FastMovingStockQuotes topic, in the terminal go to cep-210-jms-client-qpid directory and run:
ant xmlFastMovingStockQuotesSubscriber
To publish to the AllStockQuotes topic, in another terminal from the same directory, run:
ant xmlAllStockQuotesPublisher
You will be able to observe the outputs. Here the output will be similar to the output of Stock Quote Analyzer sample.

Saturday, February 16, 2013

Writing Custom Broker for WSO2 CEP

In this post I'll explain how to create a simple Custom Broker in WSO2 CEP 2.1.x.

You can find a sample template Custom Broker project from here.

The first step is to create a Java project (here I have used Apache Maven),
then create an appropriate Broker Type by extending "org.wso2.carbon.broker.core.BrokerType" and an appropriate Broker Type Factory by extending the  "org.wso2.carbon.broker.core.BrokerTypeFactory" from the jar org.wso2.carbon.broker.core-4.0.7.jar

org.wso2.carbon.broker.core-4.0.7.jar will be available in the WSO2 public Maven Repository, which you can add to your project pom.xml as:

<repositories>
    <repository>
        <id>wso2-maven2-repository</id>
        <name>WSO2 Maven2 Repository</name>
        <url>http://dist.wso2.org/maven2</url>
    </repository>
</repositories>

else you can also find that at CEP_HOME/repository/components/plugins/ as org.wso2.carbon.broker.core_4.0.7.jar

When extending the BrokerType, the BrokerTypeImpl has to implement the following methods.

package org.wso2.carbon.broker.core;

import org.apache.axis2.engine.AxisConfiguration;
import org.wso2.carbon.broker.core.exception.BrokerEventProcessingException;

/**
 * This is a broker type. these interface let users to publish subscribe messages according to
 * some type. this type can either be local, jms or ws
 */
public interface BrokerType {

    /**
     * object which describes this type. it contains the name and available properties.
     *
     * @return - type dto
     */
    BrokerTypeDto getBrokerTypeDto();

    /**
     * subscribe to the connection specified in the broker configuration.
     *
     * @param topicName           - topic name to subscribe
     * @param brokerListener      - broker type will invoke this when it receive events
     * @param brokerConfiguration - broker configuration details
     * @throws BrokerEventProcessingException - if can not subscribe to the broker
     */
    String subscribe(String topicName,
                     BrokerListener brokerListener,
                     BrokerConfiguration brokerConfiguration,
                     AxisConfiguration axisConfiguration) throws BrokerEventProcessingException;

    /**
     * publish a message to a given connection with the broker configuration.
     *
     * @param topicName           - topic name to publish messages
     * @param object              - message to send
     * @param brokerConfiguration - broker configuration to be used
     * @throws BrokerEventProcessingException - if the message can not publish
     */
    void publish(String topicName,
                 Object object,
                 BrokerConfiguration brokerConfiguration) throws BrokerEventProcessingException;


    /**
     * publish test message to check the connection with the broker configuration.
     *
     * @param brokerConfiguration - broker configuration to be used
     * @throws BrokerEventProcessingException - if the message can not publish
     */
    void testConnection(BrokerConfiguration brokerConfiguration) throws BrokerEventProcessingException;

    /**
     * this method unsubscribes the subscription from the broker.
     *
     * @param topicName
     * @param brokerConfiguration
     * @throws BrokerEventProcessingException
     */
    void unsubscribe(String topicName,
                     BrokerConfiguration brokerConfiguration,
                     AxisConfiguration axisConfiguration, String subscriptionId)
            throws BrokerEventProcessingException;

}

Here you have to write the appropriate client code for both sending and receiving events.

The publish() method in the BrokerType will be called when CEP is sending an output, your implementaion to publish the message needs to go under the publish() method. The input parameter message Object, of the publish() method can be either of the following,
  • java.lang.String
  • java.util.Map
  • org.apache.axiom.om.OMElement
  • org.wso2.carbon.databridge.commons.Event
In your implementation you don't need to support all the message types, and supporting the appropriate type will be adequate.  This is because at the bucket output mapping, user can select the appropriate message type mapping when sending the events.

At the same time when a subscription was made for the subscribe() method the appropriate BrokerListener will be also provided, which is associated with the topic. BrokerListener also have a method onEvent(Object message) to which your client code receiving  the incoming events has to pass the received messages. Here the events passed to the onEvent() must have one of the following format,
  • java.util.Map
  • org.apache.axiom.om.OMElement
  • org.wso2.carbon.databridge.commons.Event
When extending the BrokerTypeFactory, the BrokerTypeFactoryImpl has to implement the getBrokerType() method.

package org.wso2.carbon.broker.core;

public interface BrokerTypeFactory {
    BrokerType getBrokerType();
}

Then you have to build the jar and deploy that to the CEP server.

If its a plain jar you have to add that to  CEP_HOME/repository/components/lib or
If its a OSGi bundle then you can add that to CEP_HOME/repository/components/dropings

Next, you have to create a file called "broker.xml" at wso2cep-2.0.1/repository/conf directory, and add the implantation class as in the following XML:

<brokerTypes xmlns="http://wso2.org/carbon/broker">
    <brokerType class="org.test.cep.broker.BrokerTypeFactoryImpl"/>
</brokerTypes>

Note: In the above code snippet change the "org.test.cep.broker.BrokerTypeFactoryImpl" with your implementation class name.

Finally, restart & login to the CEP server,  go to Configure -> Broker -> Add, and there you will be able configure your custom broker.



You will be able to download a sample template project from : https://dl.dropbox.com/u/17922825/blog/cep-ext-broker.zip

Saturday, December 8, 2012

Single Sign On across Web Applications and Web Services

Here the requirement is to have a single sign on across Web application and Web Services. In more detail once the user is  authenticated to access a WepApp he should not only be able to access other WepApps but those WebApps also should be able to access a set of back-end services with the logged in user's access rights with no further authentication. Here the back-end services can also be configured to authorize the user based on different user claims (E.g. email).


In the situation we can assume:
•  trust relation between Identity Server and Resource Server (ESB Proxy),
•  trust relation between Identity Server and Resource Client (Web App)
The communications are:

  1. User connect to the Web App with is the “Resource Client”
  2. WebApp finds out user is not authenticated and redirects to the SAML2 IdP.
  3. SAML2 Idp checks whether the user has an authenticated session - if not will prompt for credentials, once authenticated there ,user will be redirected back to WebApp (Resource Client)  with a SAML token, with the set of claims requested by the WebApp.
  4. Now, the WebApp (Resource Client) needs to access a back-end web service with the logged in user's access rights. WebApp process the “SAML assertion token” and creates the STSToken from which it calls the Proxy Service (Resource server).
  5. Proxy Service (Resource server) checks the “SAML assertion token” and allows access to the backend Web Service.
This is tested with;
•  WSO2 Identity Server 4.0.0
•  WSO2 Enterprise Service Bus 4.5.1
•  Apache Tomcat 7.0.25
•  Apache Subversion 1.6.12
•  Apache Maven 3.2.3
•  Java 1.6.0_26
Steps,

Step 1,

Download and deploy WSO2 IS 4.0.0 from http://wso2.org/products/identity-server/

Step 2
Download WSO2 ESB 4.5.1 from http://wso2.com/products/enterprise-service-bus/.
When deploying ESB change the port offsets by 1, by editing

<ESB_HOME>/repository/conf/carbon.xml file, and change the Offset element as follows
<Offset>1</Offset>


Step 3,

Download and deploy Apache Tomcat

Step 4,
Checkout https://svn.wso2.org/repos/wso2/people/suho/sso-webapp-ws sso-webapp-ws
This contains all the artifacts needed for this sample

Step 5,
Setup WSO2 IS for SSO for Web Apps
  1. Login to IS
  2. Click SAML SSO from the management console
  3. Add SSO details

    For the avis.com WebApp
    Issuer: http://localhost:8081/avis.com/consumer
    Assertion Consumer URL: http://localhost:8081/avis.com/consumer
    Check “Enable Attribute Profile” and add the following claims
              http://wso2.org/claims/givenname
              http://wso2.org/claims/emailaddress
    For the travelocity.com WebApp
    Issuer: http://localhost:8081/travelocity.com/consumer
    Assertion Consumer URL: http://localhost:8081/travelocity.com/consumer
    Check “Enable Attribute Profile” and add the following claims
              http://wso2.org/claims/givenname
              http://wso2.org/claims/emailaddress
    When successfully added you will get Consumer Index for each entry

    Now the SSO setup in the WSO2 IS is done.

Step 6,
Configuring the WebApps
  1. You can add these by editing the war file or edit in the src-dist and build that.
  2. Edit the WEB-INF/web.xml of each WebApp and add the appropriate “Consumer Index” returned by the WSO2IS as the param-value for AttributeConsumingServiceIndex.
  3. Edit the WEB-INF/classes/filepath.properties of each WebApp and add the appropriate
    properties. You can find all the necessary files in the resources directory.
    E.g.
    keystore.file=/<path>/sso-webapp-ws/resources/keystore/wso2carbon.jks
    repo.directory=/<path>/sso-webapp-ws/resources/repo
    bearerpolicy.file=/<path>/sso-webapp-ws/resources/bearer-policy.xml
    service.endpoint=http://localhost:8281/services/echoProxy

Step 7,
Add the proxy service
  1. Login to ESB
  2. Click on Add Proxy Service from the management console
  3. Create a Pass Through Proxy
  4. Add the proxy name and endpoint
    Proxy Service Name: echoProxy
    Target Endpoint: “Enter URL”
    Target URL: http://localhost:8281/services/echo
    And click “Create”

    Here the echo service (Endpoint service) is deployed on ESB by default

Step 8,
Add the policy to the registry to secure the Proxy Service

  1. Click on Registry from the ESB management console
  2. Navigate to /_system/config/repository and click on the repository
  3. Under “Entries” click on the add recourses to add policy file
  4. Now browse and add the bearer-policy.xml. This will be in the resources folder you
    checked out. Click “Add” to add the file.

Step 9,
Securing the proxy service
  1. Click on List under services from the ESB management console, and click on “Unsecured” link next to the echoProxy service entry.
  2. Select “yes”
  3. From the “Policy From Registry” section, click in Configuration registry to add policy from the cofig registry.
  4. Browse for the added “bearer-policy.xml” and click “OK”
  5. Click “Next”, then click on the “admin” and “wso2carbon.jks” checkboxes, select the private key store as “wso2carbon.jks” and click on “Finish” to Activate Security.
Step 10,
Deploy the webapps in Tomcat

Step 11,
  1. Enter following in the redirected page
    User Name: admin
    Password: admin

    With successful login you will be redirected to the home page, where the Web App will be call the proxy using the “SAML assertion token” received at the SSO login.
  2. With successful login at the Proxy Service you will get the following page
    Note: Here the Web Service Response will be as “Hi WSO2”
  3. If the login at the Proxy Service fails you will get the following page
    Note: Here the Web Service Response will be as “’Hi WSO2’ expected but, no response from the web service”

I would like to thank the WSO2 forks; Suresh AttanayakePrabath Siriwardena, & Balachandiran Ajanthan for helping me on this.


Monday, August 27, 2012

Distributed Processing Sample for WSO2 CEP


Today we released WSO2 Complex Event Processor 2.0.0 Milestone 2.
This is available at  https://svn.wso2.org/repos/wso2/people/suho/packs/cep/wso2cep-2.0.0-M2.zip

One of the key feature of this CEP is its support for distributed processing via Siddhi CEP Engine. To demonstrate its capability I came up with a sample on distributed Processing. This sample uses Siddhi CEP Engine for processing and JMS ActiveMQ Broker to publish and subscribe events.

To run the Distributed Processing Sample follow the steps give below;
 
Step 01: Configure and run ActiveMQ in your local machine
Download the ActiveMQ from "http://activemq.apache.org/activemq-543-release.html".
unzip the distribution and run the ActiveMQ server using the command “./activemq console” from apache-activemq-xxx/bin (in Linux)
Note: WSO2CEP has been tested with ActiveMQ 5.4.3

For each CEP node in the cluster follow the steps from 02 to 08.

Step 02: Deploy CEP server 
Unzip the CEP server (Do not start the server).
      
Step 03: Change the CEP server Offset.
If you are running multiple servers in the same machine change the offset
from file "wso2cep-2.0.0-2/repository/conf/carbon.xml" to different numbers. 
E.g.
<offset>1</offset>

This is to overcome server port conflicts.
E.g. If three WSO2 servers are going to be deployed in the same machine, they can have offsets as 1, 2, and 3.

Step 04: Copy paste ActiveMQ jars.
Copy paste activemq-all-xxx.jar from the ActiveMQ home directory to 
wso2cep-2.0.0/samples/lib directory.
Copy paste activemq-core-xxx.jar and geronimo-j2ee-management_1.1_spec-1.0.1.jar
from apache-activemq-xxx/lib to wso2cep-2.0.0/repository/components/lib directory
 
Step 06: Deploy required broker management configuration
This allows CEP to receive and send messages via JMS Broker.
To deploy run "ant deploy-broker-manager" from wso2cep-2.0.0/samples/cep-samples directory.

Step 07: Configure input, output and queries
For this sample, the configurations of input, output and queries are at wso2cep-2.0.0-1/samples/cep-samples/conf/buckets/purchase-analyser-bucket.xml
When running multiple server nodes each server need to receive the input events through different topics. To enable this behaviour, open the above file and change the input topics to have different names. The default input topic name is “PurchaseTopic”.

E.g changing the names to “PurchaseTopic1”.
<input brokername="activemqJmsBroker" topic="PurchaseTopic1" />

Note : To enable distributed processing (already enabled for this sample) the change we have to do is to make the "siddhi.enable.distributed.processing" property to "true".

<engineproviderconfiguration engineprovider="SiddhiCEPRuntime">
     <property name="siddhi.persistence.snapshot.time.interval.minutes">0</property>
     <property name="siddhi.enable.distributed.processing">true</property>
</engineproviderconfiguration>

Step 08: Deploy bucket configuration
Deploy the bucket by adding the configuration purchase-analyser-bucket.xml to the directory wso2cep-2.0.0/repository/deployment/server/cepbuckets/

Step 09: Start all the servers
Start CEP servers by running "./wso2server.sh" from wso2cep-2.0.0/bin directory 

Step 10:Subscribing to output topic
Start the subscriber of the output topic "PurchaseResults" by running "ant purchaseResultsSubscriber" in a separate terminal,
from wso2cep-2.0.0/samples/cep-samples directory

Step 11:Publishing events
To publish events to all the servers (to their input topics as we defined in Step 07 ) run the publisher from wso2cep-2.0.0/samples/cep-samples directory
with the command  "ant purchasePublisher -Dtopics=xxx,xxx,..."
The client will publish events to all the given topics in a round robin manner.

E.g If we we have configured the buckets to receive the inputs via the input topics as PurchaseTopic1, PurchaseTopic2 and PurchaseTopic3
"ant purchasePublisher -Dtopics=PurchaseTopic1,PurchaseTopic2,PurchaseTopic3."


Step 12: Observation
You will be able to observe how the counts in the results steadily increases when messages are sent to any of the servers, and how the results remain consistent even when some servers goes down and comes back.