Hadoop for DevOps

Graphing like it's 1998!

This was awesome when it first came out, and still serves its purpose. The techniques use internally led to… RRDTool.

TSDB = Time Series Database.

Round Robin Database. What does that mean? For the use cases of MRTG and the ilk, you want to see what your load looked like today, but you also want to capture long term trends. Is my network becoming more busy? Is my disk slowly filling up. Problem is that can cause unbounded expansion of data. (And back in the megabyte scale era, this was a bigger deal.)

RRD is a Time Series Database; TSDB. Basically, when you create it, you specify your sample periods. Suppose you capture data once a minute. With RRD, you can roll up the data in 5 minute increments for a day, and 1 hour increments for a week. And it's constant space. Which is really cool. One less thing to think about. (You don't want your system monitoring to cause you extra system monitoring issues.) And it gives you long term trends.

But what if you want a 1 month period? And you didn't specify it when you created it? Bummer…

Or this happens: a prod issue lands on your desk. You need to look into a problem. Requires 1 minute granularity from last week? LOL!

Some of these are all-in-one products that are extensible, have plugins, training, etc. Some you have to combine with other products.

Problems: data locked in there. For example, web access logs. Those are handy, and have their own tools. But suppose you have other things… diagnostics. Errors. Might want to track those. Want to cross correlate different things. Web response time, network traffic, and database IO.

Also… traditional tools can have scalability problems. Use traditional databases, for example.

Textual stuff you want to search or act on. Numeric information, or information you can turn into numbers, that you want to graph.

Syslog? Grep? Logging like it's 1988! Seriously…

Splunk? A good solution, from what I've heard, for certain issues.

What do I want? Well… ideally not have have server or app problems. But that's not going to happen.

So let's talk about solutions. Since this is a Hadoop meetup, we'll focus on things in the Hadoop ecosystem, or that work around the edges.

Since we've got big data technologies, ideally I'd want more data. No downsampling. More flexible graphing. I want to see my web request latency vs my database latency. And maybe some other metric combo I just thought of. Basically, more cowbell!

Ultimately, this becomes a message passing, processing, and recording problem. Events are generated on a server in a cluster, and we need to capture it and store it somewhere. So some of the solutions ultimately are message queue products, albeit ones targeting this space.

Meta-irony: if you have sufficiently large Hadoop, or other, cluster it might be driving your needs to track lots of things…

And as an aside, one of the core use cases for Hadoop was log processing to assist with ad placement.

Quite a few companies use it. Box. Tumblr. StumbleUpon of course.

HBase manual even mentions OpenTSDB as a way to monitor your HBase.

tsd = Time Series Daemon. Writes the metrics to HBase. Also provides web interface for querying.

Other components: tcollector. Talks to the time series daemon. Will run scripts, de-dup, send to TSD, handle disconnections. Scripts just have to write output to standard output. It can be as simple as a 1 or 2 line shell script!

Anything that can be expressed as a number can be a metric. Databases, number of users, network statistics, etc.

Tags can have anything, like host, environment (prod vs qa), etc.

How big are they? How much space? 100 billion points is about 1 TB. (Using LZO compressed HBase tables.) About 12 bytes per data point.

There is support for purging old data if you don't want to store data forever. But there are people out there storing a trillion data points.

Quick start once HBase and OpenTSDB are installed.

Start (single node) HBase (in HBase directory):

./bin/start-hbase.sh 

Create tables if necessary (in opentsdb directory):

env COMPRESSION=none HBASE_HOME=../hbase-0.94.4 ./src/create_table.sh

Examine tables:

../hbase-0.94.4/bin/hbase shell
describe 'tsdb'
describe 'tsdb-uid'

Start tsdb:

tsdtmp=${TMPDIR-'/tmp'}/tsd
mkdir -p "$tsdtmp"
./build/tsdb tsd --port=4242 --staticroot=build/staticroot --cachedir="$tsdtmp"

Look at web interface:

open http://localhost:4242

Ok… now add some metrics:

./build/tsdb mkmetric some.stat1 some.stat2

(Show auto completion in web interface)

Or, alternatively, add a bunch for self monitoring:

echo stats | nc -w 1 localhost 4242 \
| awk '{ print $1 }' | sort -u \
| xargs ./build/tsdb mkmetric

Now load up data (script to be provided in github):

../hbase-examples/genstats.pl | nc -w 2 $me 4242

Look at interface for past 10 minutes. Set to auto refresh every 5 seconds. Now, load up a lot, then start loading continuously.

../hbase-examples/genstats2.pl| wc -l
1,209,602

../hbase-examples/genstats2.pl| nc -w 1 localhost 4242

../hbase-examples/genstats.pl|nc -w 1 localhost 4242

while true                                          
do
 ../hbase-examples/genstats.pl|nc -w 1 localhost 4242
 sleep 4
done

Example from interweb:

CREATE EXTERNAL TABLE events (
  id bigint, received_at string, generated_at string, source_id int, source_name string, source_ip string, facility string, severity string, program string, message string
)
PARTITIONED BY (
  dt string
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION 's3://your-s3-bucket.yourdomain.com/papertrail/logs';

It works, it's simple, but it's a one shot thing. Wouldn't want to continuously do this necessarily. It works, but not the most scalable thing.

Facebook, Twitter, Zygna.

WWFD (What would Facebook Do?) They wrote Scribe.

"Scribe is a server for aggregating log data that's streamed in real time from clients. It is designed to be scalable and reliable."

May be abandonware. (No updates in over a year.)

Not out of box solution.

Competes with Scribe and Flume, but has messaging semantics.

Persistent messaging.

Can load into HDFS. However, out of box, needs code or log4j config to generate messages. (If you don't use log4j… well… sorry.)

Useful if you control your applications' code.

"Apache Flume is a distributed, reliable, and available service for efficiently collecting, aggregating, and moving large amounts of log data. Its main goal is to deliver data from applications to Apache Hadoop's HDFS. It has a simple and flexible architecture based on streaming data flows. It is robust and fault tolerant with tunable reliability mechanisms and many failover and recovery mechanisms. It uses a simple extensible data model that allows for online analytic applications."

Since sources and sinks support Avro, you can chain them and create a hierarchy of agents.

Centralized configuration? Centralized liveness monitoring?

The following example loads some Apache web server logs into HBase. No extraction of data from the logs is attempted: each line is just stored as a literal value in HBase. There is an "interceptor" that can use regular expressions to parse and pick apart structured data and store it into separate HBase columns.

First, created a simple HBase table:

create 'log_table',
  {NAME => 'some_cf', VERSIONS => 1, COMPRESSION => 'NONE', BLOOMFILTER => 'ROW'}

Then, used the following simple agent configuration:

# flume-hbase.conf: A single-node Flume configuration that writes to HBase.  Reads from Syslog.
#
# To start:
#
#   ./bin/flume-ng agent --conf-file ~/examples/demo-flume/flume-hbase.conf --name a2 --conf ./conf
#

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1

#
# spooldir source type
#
# Renames files to .COMPLETED
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /var/log/apache/flumeSpool
a1.sources.r1.fileHeader = true

#
# Define a source interceptor
#
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp
a1.sources.r1.interceptors.i1.preserveExisting = true

#
# Logger sync
#
# a1.sinks.k1.type = logger

#
# HBase sync
#
a1.sinks.k1.type = org.apache.flume.sink.hbase.HBaseSink
a1.sinks.k1.table = log_table
a1.sinks.k1.columnFamily = some_cf
a1.sinks.k1.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Doesn't directly deal with Hadoop ecosystem, but I'm highlighting it for its log shipping abilities, and its ability to interoperate with a wide variety of inputs and outputs. In the case of outputs, it can send information to OpenTSDB. (In which case you'll want to use a filter to extract information from the logs that can be turned into a metric suitable for OpenTSDB.)

It's a Swiss Army Knive… maybe too much of a Swiss Army Knife!

Logstash example is a little too chatty for notes; see github for details. I ran it with an embedded Elastic search. Once it starts up (which can take a minute or 3), you can point browser at http://localhost:9292/ and start searching away…

These don't necessarily directly interop with HDFS, HBase, or OpenTSDB, but have plugins or otherwise can feed data. A whole lot of it is role your own.

Facebook does it. But when you're dealing with petabytes of data, you're probably used to rolling your own stuff.