Slides for Cloud Hands-on Tutorial in IEEE Cloud 2012

Following are the slides I used for the tutorial. As the sample, we deployed a Web Service and a Web Application. When deployed, users access the web application, which will do a call to the Web Service and show the results.

Hands on deployed the service and webapp in three setup.

1. In the local machine
2. In the Amazon EC2 (IaaS)
3. In the WSO2 Stratos (PaaS)

The script for the demo can be found below. Please note it is only  a rough script. 

Abstract

When cloud computing was introduced a few years ago, it promised many advantages including self-service, elasticity, pay as you go, improved accessibility to computation resources, and full deployment automation. Few years down the line, now we can see many ideas, architectures, and systems shaping up providing more clarity and understanding into cloud computing landscape. Among them, there are many cloud computing systems that enable users to build systems in the cloud with minimal effort. Understanding those systems could provide great insights and understanding into both “cloud promise” and “cloud reality”. This tutorial will provide a brief introduction to the cloud, and discuss how to use some of the existing cloud computing systems with hands-on. We will demonstrate how to develop and run a SOA application in an Infrastructure as a Service platform (IaaS) as well as in a Platform as a Service Platform (PaaS).

Script 

As a demo, we will deploy a Web Service and a Web Application. When deployed, users access the web application, which will do a call to the Web Service and show the results. You can download the sample code from http://people.apache.org/~hemapani/dist/clouddemo/HelloStratos.zip. Download and unzip the distribution.

1. webapp folder will have the Web app, and you can build the HelloStratosWebapp.war file by running the ant command from the webapp folder.
2. services folder has the Web service, and you can build theHelloStratos-1.0.aar file by running mvn clean install

Local Hello Stratos Demo

1. Edit the webapp/src/main/resources/index.jsp to do the service call to 127.0.0.1. Rebuild the webapp by running ant from webapp directory.
2. Download WSO2 AS
3. Running following commands to install WSO2 AS

   unzip wso2as-4.1.2.zip
   cd /Users/srinath/playground/cloud-2012-tutorial/wso2as-4.1.0
   cp /Users/srinath/playground/cloud-2012-tutorial/HelloStratos/webapp/target/HelloStratosWebapp.war repository/deployment/server/webapps/
   cp /Users/srinath/playground/cloud-2012-tutorial/HelloStratos/service/target/HelloStratos-1.0.aar repository/deployment/server/axis2services/
   

4. Start the WSO2 AS via following commands.

   cd bin/
   ./wso2server.sh
   

5. Go to the admin console and login via https://127.0.0.1:9443/carbon/admin/login.jsp, username password are admin, admin.
6. Try out the service. 
7. Show soap traces.
8. Show Web app.
9. Try out the Web app 

EC2 Demo

1. Create AWS account
2. Go to management console
3. Create instance(Base AMI ami-2ab91843 and AMI with all setup AMI: Unnamed (ami-5805a731)) Create key pair in the process, save the key pair
4. Connect to instance (srinath-2012.pem is the keypair in this case)



ssh -i ~/.ssh/ec2/srinath-2012.pem ubuntu@ec2-23-22-125-63.compute-1.amazonaws.com



5. Create a folder ieee-clouddemo
6. Download WSO2 AS and install following command



wget http://people.apache.org/~hemapani/dist/wso2as-4.1.2.zip
sudo apt-get install unzip  

Unzip WSAS distribution

wget http://download.java.net/jdk6/6u34/promoted/b03/binaries/jdk-6u34-ea-bin-b03-linux-amd64-20_jun_2012.bin (direct link http://jdk6.java.net/download.html)

./jdk-6u34-ea-bin-b03-linux-amd64-20_jun_2012.bin

export JAVA_HOME=/home/ubuntu/jdk1.6.0_34



7. Edit Host name in carbon.xml
8. Access Admin console via https://ec2-107-20-54-230.compute-1.amazonaws.com:9443/carbon/admin/login.jsp
9. Upload the service via admin console
10. tryout the service
11. Download the HelloStratosWebapp.war from
12. Upload the war file
13. try out the war file
14. Save the AMI

Stratos Demo

1. Create an account from https://stratoslive.wso2.com/
2. Login to Application Service in Stratos https://stratoslive.wso2.com/t/ieeecloud1.org/carbon/ (replace the ieeecloud1.org with your tenant.)
3. Show services and compare with the standalone version
4. Upload the service to stratos
5. try it out
6. Login to windows VM and upload the Web app
7. Try the demo
8. Enable security via uncommenting commented parts in web.xml
9. Re-upload the services
10. Try the Web app, it will ask for passwords. Login via admin admin.

Few thought on debugging

If you a developer who works on Distributed system, there is one thing you learn well. That is how to debug, and how to avoid having to debug. Following are some of my thoughts and somethings I generally do.

Debugging a Local Java Program

1. If you write your program well, generally you will have a stack trace when you have a problem. (This does not apply well with performance problems and memory leaks. I will write a separate note about those. )
2. Look at the trace; go to where the error happened. Try to figure out what happened. The best way to do this is by walking through the logic again.
3. If that did not work, copy and paste your stack trace in to Google. About 80% of the time, you will find the answer there. Pay special attention to JIRA bug reports for the projects you are using and online forums like stackoverflow.com.
4. If that did not work, you will have to debug. You can debug by running the code from your IDE (e.g. Eclipse) or by connecting to a server through a remote debugger. Walk through the logic using the debugger that generally tells what happened. For example, the article explains how to debug with eclipse and how to connect to a remote server.
5. If none of these worked, now it is the time to go and ask for someone to help. There are some bugs that are very hard for the author of the code to see. However, you should go for help with problem recreated, debugger attached, and ready to let him step through the execution.
6. If you have trouble with a specific tool, you can ask for help as user lists, forums, or general developer forums like stackoverflow.com.

Debugging a Distributed System

1. Debugging distributed systems are hard. So best approach is to not to have to debug them. You can almost get there by writing unit tests and tests what you write in small steps. My preferred approach is to make one path work end to end, and do small changes while testing each change.
2. Distributed system will have multiple Processors (JVMs). So it is tricky to debug them. If it is at all possible, find a way to run whole your distributed system within the same process (JVM). This will need some imagination from your end, but it will save you lot of trouble later.
3. If you are debugging a distributed system, it is often useful to capture messages that are sent and received. You can do this via tools like TCP Monitor, SOAP Monitor, or Wireshark.
4. It is doubly important to log all exception that can happen in your code. Otherwise, you will have no idea whether system worked or not. 
5. I often append the timestamp and the name of process or host to each log line. One way to do this is by writing a Log4j appender. Time stamp and process or host name let me merge sort all the logs into one file and read the execution of the system in one read.
6. It is likely that your distributed system process lot of messages. So it is very hard to read and understand the log. One way out of this is to trace every 1000^th messages. I do this by having a message count and using.

   if(1000%messageCount=1){
      log.info(….);
   }
   

7. If you running a complex system that has more than five nodes, you should invest in some mechanism to collect the logs using something like FLUME and automate their processing to find stack traces etc.

Scaling WSO2 Stratos

WSO2 Stratos is a Platform as a Service (PaaS) that offers middleware technologies like Web Services, Workflows, Messaging etc as a Service.

PaaS environments bring together many users and could potentially attract a large number of users. Therefore, scaling up is a major consideration for a PaaS. This post explain our experiences and some thoughts on scaling Stratos.

Problem

Stratos is multi-tenanted. In other words, there are many tenants. Each tenant generally represents an organization and isolated from other tenants, where each tenant has his own users, resources, and permissions. Stratos supports multiple PaaS services. Each PaaS service is actually a WSO2 Products (e.g. AS, BPS, ESB etc.) offered as a service. Using those services, tenants may deploy their own Web Services, Mediation logic, Workflows, and Gadgets etc.

1. WSO2 Stratos runtime provides servers where each can support multiple tenants and provide a PaaS service. For example, there are multi-tenanted AS, BPS, ESB etc.
2. Stratos can provision (add/remove) resources (computing nodes) as needed on demand.
3. Problem is to build a system that scale up and down without end users realizing it.

Answers

Following describes a series of solutions while each solution adds a new feature to solve a specific problem. It explains the rationale and thought process behind the final design.

Solution 1
WSO2 Stratos consists of a multi-tenanted server of each type (e.g. ESB, BPS etc.). Users first talks to an Identity Server (IS), gets a SSO (single sign-on) token, and log in to any server. We stored all tenants data in a registry. Each server loads all tenants at the startup and can support any tenant when they receive a request.

Solution 2
Solution 1 does not scale at all. So we started running multiple instances of each server and put a load balancer (LB). LB load balances the requests to different servers. When load on the server instances are high, LB starts new server instances and when the load is low, LB shuts down some instances. We call this auto-scaling.

Solution 3
When Stratos had several hundred tenants and many tenants with tens of services, it took a long time to load all tenants at the startup. Start up took 15-30 minutes. Furthermore, most tenants stays inactive most of the time. However, since each node has to hold all tenants, Stratos spends resources for inactive tenants as well.

To avoid above problems, solution 3 added lazy loading. All information about tenants is stored in a central registry. Tenants are loaded into memory only when they are needed. Tenants get unloaded when they have been idle for more than a given timeout. You can find more information about Lazy loading from Azzez’s blog entry “Lazy Loading Deployment Artifacts in a PaaS Deployment”.

Solution 4
When tenants have several artifacts, loading them takes time. So if the tenant is accessed while it has not been loaded in solution 3, first request or two to the tenant will timeout.

Solution 4 added ghost deployer to solve the above problem. Ghost deployer does not load all information about tenants, but just loads the metadata. Actual artifacts are loaded on demand. As a result, loading a tenant has become a much simpler in Solution 4. So this avoids requests from timing out while loading the tenant. You can also find more information about Lazy loading from Azzez’s blog entry “Lazy Loading Deployment Artifacts in a PaaS Deployment”.

Solution 5
However in the solution 4, LB does not scale to handle a large number of requests. So we replace the LB with multiple LBs that have same metadata in all nodes. Therefore, all LB will take the same decision when it received a request. We can use this model to scale Stratos by placing a hardware Load balancer or setting up DNS round robin to distributed requests among LBs.

To synchronize the metadata across all LBs, we can use group communication. That is a MXM communication, which is heavy. Instead,  LBs in Stratos are designed to send updates as batches to a single decision service, and the decision service takes auto-scaling decisions. We enforce High Availability by running a replica of the decision service and keeping it up-to-date via state replication through group communication.

Solution 6
However, in Solution 5, LB instances are not aware of tenants. Due to lazy loading, all requests will work even through LBs route messages arbiterly. However, this might lead to a scenario where a single node has to load too many tenants.

To avoid this, in the solution 6, LBs are aware of tenants and allocate only a subset of tenants to each LB. You can find more information from the Sajeewa’s blog entry, WSO2 Tenant Aware Load balancer.

Solution 7
Upcoming Stratos release will follow the solution 6. However, the next potential problem is that Registry which holds the configurations and resources of all tenants could not scale to handle a large number of tenants. Hence the registry needs to be partitioned across multiple users.

Getting Captcha to do useful work

Getting Captcha to digitize books. Well solution is simple and elegant. show two words, one is  known  while other is one to digitize. Users do not know which is which. If the control (known word) is good, we can trust the second with high probability.

This is done by many sites, and they have digitized about 2.5 million books per year this way. See the above Ted talk by Louis Ahn for details.

They are also trying to use language learners to translate the web, which shows the samples to translate and combine translations to create the final version.

This, in my opinion, is a truly creative solution.

May be we can use the same idea to tag images. For example, show a picture and ask how many elephants are in this picture. It is not very easy for a program to answer such random questions on pictures. Now play the same tick with controls and without controls as above. (Well need some thinking to get it right .. but)

Debugging Hadoop Task tracker, Job tracker, Data Node or Name Node

Hadoop conf/hadoop-env.sh has following environment variables 

1. HADOOP_NAMENODE_OPTS
2. HADOOP_SECONDARYNAMENODE_OPTS, 
3. HADOOP_DATANODE_OPTS
4. HADOOP_BALANCER_OPTS
5. HADOOP_JOBTRACKER_OPTS
6. HADOOP_TASKTRACKER_OPTS

You can use them to start the remote debugger so that you can connection and debug any of the above servers. Unfortunately, Hadoop tasks are started through a separate JVM by the task tracker, and you cannot use this method to debug your map or reduce function as they run in separate JVMs. 

To debug task tracker, do following steps. 

1. Edit conf/hadoop-env.sh to have following

export HADOOP_TASKTRACKER_OPTS="-Xdebug -Xrunjdwp:transport=dt_socket,address=5000,server=y,suspend=n"

2. Start Hadoop (bin/start-dfs.sh and bin/start-mapred.sh)

3. It will block waiting for debug connection

4. Connect to the server using Eclipse "Remote Java Application" in the Debug configurations and add the break points

5. Run a map reduce Job 

How to scale Complex Event Processing (CEP) Systems?

What is CEP?

Complex event processing (CEP) systems query events on the fly without storing them.

• For an introduction and definition of CEP, please refer to CEP Blog and Wikipedia.) . 
• If you need a real comprehensive coverage of CEP, read the paper "Processing flows of information: from Data stream to Complex Event Processing" [1]. (or the slide deck). 

In CEP, we think in terms of event streams. Event stream is a logical sequence of events that become available over time. For example, stock event steam consists of events that notify changes to stock price. Users provide queries to the CEP engine, which implements the CEP logic, and the CEP engine matches those queries against events coming through event streams.

CEP differs from other paradigms like event processing, filtering etc., by its support for temporal queries that reason in terms of temporal concepts like "time windows" and "before and after relationships" etc. For example, a typical CEP query will say that

“If IBM stock value increased by more than 10% within an hour, please notify me".

Such a CEP query has few defining characteristics.

1. CEP queries generally keep running, and keep emitting events when events match the condition given in the query.
2. CEP query operates on the fly and stores only minimal amount of events into a storage.
3. CEP Engines responds to conditions generally within milliseconds range.

What is Scaling CEP?

There are many CEP Engine implementations (see CEP Players list 2012). However, mostly CEP engines run in a large box, scaling up horizontally. Vertically scaling CEP engines is still an open problem. Reminder of this post discusses what vertically scaling CEP engine means and some of the potential answers.

We use the term Scaling to describe the ability for a CEP system to hande larger or complex queries by adding more resources. Scaling CEP has several dimensions.

1. Handling Large number of queries
2. Queries that needs large working memory
3. Handling a complex query that might not fit within a single machine
4. Handling large number of events

How to Scale CEP?

Let us consider each dimension separately.

Handling Large Number of Queries

This is the easiest of the four since we can use the shared nothing architecture. Here we can run multiple CEP Engine instances (each instance runs in a single host) each running a subset of queries.

1. Trivial implementation will send all events (streams) to all the CEP engines
2. More optimized implementations can use a Publish/subscribe message broker network (like Narada Broker). Here each CEP engine should analyze the deployed queries and subscribe to required event streams in the broker network. Generally, we match each event stream to a topic in the publish/subscribe system.
3. Third option is to delegate the event distribution to a Stream Processing system (e.g. Apache S4 or Strom). For instance, links [4] and [5] describe such a scenario to run Esper within Strom.

Queries that need large working memory

For instance, a long running complex query that needs to maintain a large window and all events in the window would need a large working memory. Potential answer to this problem is to use a distributed cache to store the working memory. Reference [6] describes such a scenario.

Handling large number of events handling a complex query that might not fit within a single machine

We will handle the both scenarios together as both are two sides of the same coin. In both cases, we have trouble fitting a single query into a single host such that it can support the given event rate.
To handle such scenarios, we have to distribute the query across many computers.

We can do this by breaking the query into several steps in a pipeline that matches events against some conditions and republish the matching events to steps further in the pipeline. Then we can deploy different steps of the pipeline into different machines.

For example, lets consider the following query. This query matches if there are two events within 30 seconds from IBM stocks that having price greater than 70 and having prize increase more than 10%.

select a1.symbol, a1.prize, a2.prize from every 
    a1=StockStream[price > 70  symbol =’IBM’] -> 
    a2=StockStream[price > 70  symbol =’IBM’]
        [a1.price < 1.1*a2.prize][within.time=30]

As shown by the figure, we can break the query into three nodes, and each node will have to republish the matching events to the next node. (Option 1)

CEP Query as a Pipeline

However, queries often have other properties that allow further optimization. For example, although the last step of matching prize increase is stateful other two steps are stateless. Stateful operations remember information after processing an event so that earlier events affect the processing of later events while stateless operations only depends on the event being processed.

Therefore, we can add multiple instances in the place of those statless instances using a shared-nothing architecture. For example, we can break the query into five nodes as shown by the bottom part of the picture (Option 2).

Also another favorable fact is that CEP processing generally happens through filtering where amount of events reduce as we progress through the pipeline. Therefore, pushing stateless filter like operations (e.g. matching against symbol ="IBM") to the first parts of the pipeline and scaling them in shared nothing manner should allow us to scale up the system for much higher event rates. For example, lets say that the StockQuote event stream generates 100,000 events per seconds, but only 5% of them are about IBM. Therefore, only 5000 events will make it past the first filter, which we can handle much easier than 100k events.

However, it is worth noting that above method only works with some queries. For example, if we have a query that has a single stateful operation like window-based pattern, we cannot use this method.
Unfortunately, there is no framework that can do this out of the box (let me know if I am wrong). So if you want to do this, you will have to code it. If you choose to do that, using a pub/sub network or stream processing system might reduce most of the complexities.

Please shared your thoughts!

1. Alessandro Margara and Gianpaolo Cugola. 2011. Processing flows of information: from data stream to complex event processing. InProceedings of the 5th ACM international conference on Distributed event-based system(DEBS '11).
2. http://www.thetibcoblog.com/2009/08/21/cep-versus-esp-an-essay-or-maybe-a-rant/
3. http://www.slideshare.net/TimBassCEP/mythbusters-event-stream-processing-v-complex-event-processing-presentation
4. Run Esper with Storm -http://stackoverflow.com/questions/9164785/how-to-scale-out-with-esper
5. http://tomdzk.wordpress.com/2011/09/28/storm-esper/
6. Distributed Cache to scale CEP -http://magmasystems.blogspot.com/2008/02/cep-engines-and-object-caches.html

How to measure the Performance of a Server?

I have repeated following too many times in last few years and decide to write this up. If I have missed something, please add a comment.

Understanding Server Performance

Characteristic Performance Graph's of a Server

Above graphs capture the characteristic behavior of a server. As shown by the graph, server performance is gauged by measuring latency and throughput against latency.

• Latency measures the end-to-end time processing time. In a messaging environment, teams determine latency by measuring the time between sending a request and receiving the response. Latency is measured from the client machine and includes the network overhead as well.
• Throughput measures the amount of messages that a server processes during a specific time interval (e.g. per second). Throughput is calculated by measuring the time taken to processes a set of messages and then using the following equation.

Throughput = number of completed requests / time to complete the requests

It is worth noting that these two values are often loosely related. However, a we cannot directly derive one measurement from the other.

As shown by the figure, a server has an initial range where throughput increases at a roughly linear rate and latency either remains constant or linear. As concurrency increases, the approximately linear relationship decays, and system performance rapidly degrades. Performance tuning attempts to modify the relationship between concurrency and throughput and/or latency, and maintain a linear relationship as long as possible.

For more details about latency and throughput, read the following online resources:

1. Understanding Latency versus Throughput
2. Latency vs Throughput

Unlike static server capacity measurements (e.g. CPU processing speed, memory size), performance is a dynamic measurement. Latency and throughput are strongly influenced by concurrency and work unit size. Larger work unit size usually negatively influence latency and throughput. Concurrency is the number of aggregate work units (e.g. message, business process, transformation, or rule) processed in parallel (e.g. per second). Higher concurrency values have a tendency to increase latency (wait time) and decrease throughput (units processed).

To visualize server performance across the range of possible workloads, we draw a graph of latency or throughput against concurrency or work unit size as shown by the above graph.

Doing the Performance Test

Your goal of running a performance test is to draw a graph like above. To do that you have to run the performance test multiple times with different concurrency and for each test measure latency and throughput.

Following are some of the common steps and a checklist.

Workload and client Setup

1. Each concurrent client simulates a different user. Each run in a separate thread, and run a series of operations (requests) against the server.
2. First step is finding a workload. If there is a well-known benchmark for the particular server you are using, use that. If not, create a benchmark by simulating the real user operations as closely as possible.
3. Messages generated by the test must not be identical. Otherwise, caching might come to play and provide too optimistic results. Best method is to capture and replay a real workload. If that is not possible, generate a randomized workload. Use a known data set whenever it makes sense.
4. We measure latency and throughput from the client. For each test run we need to measure following.
5. End to end time taken by each operation. Latency is the AVERAGE of all end-to-end latencies.
6. For each client, we collect the test-started time, test-end time, and the number of completed messages. Throughout is the SUM of throughput measured at each client.
7. To measure the time, if you are in Java, you can use System.nanoTime() or System. currentTimeInMillis () and with other programing languages you should use equivalent methods.
8. For each test run, it is best to take readings for about 10,000 messages. For example, with concurrency 10, each client should send at least 1000 messages. Even if there are many clients, each client should at least send 200 messages.
9. They are many tools that can do the performance test. Examples are JMeter, LoadUI, javabench, ab. Use them when applicable.

Experimental Setup

1. You may need to tune the server for best perforce with settings like enough Heap memory, open file limits etc.
2. Do not run both client and the server on the same machine (they interfere with each other and results and affected)
3. You need at least 1GB network to avoid the interference of the network.
4. Generally, you should not run more than 200 clients from the same machine. For some cases, you might need multiple machines to run the client.
5. You have to note down and report the environment (Memory, CPU, number of cores, operating system of each machines) with the results. It is a good practice to measure the CPU usage and memory while test is running. You can use JConsole (if it is Java) and if you are in a linux machine run “watch cat /proc/loadavg” command to track load average. CPU usage is a very unreliable matrix as it changes very fast. However, load average is a very reliable matrix.

Running the test

1. Make sure you restart the server between each two test runs
2. When you start the server, first send it few hundred requests before starting the real test to warm up the server.
3. Automate as much as possible. Ideally running one command should run the test, collect results, verify the results, and print summery/ graphs.
4. Make sure nothing else is running in the machines at the same time.
5. After test run has finished, check the logs and results to make sure operations were really successful.

Verifying your results

1. You must catch and print the errors both at the server and the client. If there are too many errors, you results might be useless. Also it is a good idea to verify the results at the client side.
2. Often performance tests are the first time you stress your system, and more often than not you will run into errors. Leave time to fix them.
3. You might want to use a profiler to detect any obvious performance bottlenecks before you actually run the tests. (Java Profilers: JProfiler, YourKit)

Analyze your results and write it up

1. Data cleanup (Optional) – it is a common practice to remove the outliers. General method is to either remove anything that is more than 3 X stddev different from the mean or remove 1% of furthest data from mean.
2. Draw the graphs. Make sure you have a title, captions for both X and Y-axis with Units, and legend if you have more than one dataset in the same graph. (You can use Excel, OpenOffice, or GNU Plot). Rule of thumb is that reader needs to be able to understand the graph without reading the text).
3. Optionally, draw 90% or 95% confidence intervals (Error bars)
4. Try to interpret what results mean
• Generalize
• Understand the trends and explain them
• Look for any odd results and explain them
• Make sure to have a conclusion

Hope this was useful. If you enjoyed this post you might also like Mastering the 4 Balancing Acts in Microservices Architecture and Distributed Caching Woes: Cache Invalidation. 

I now write at https://medium.com/@srinathperera.