Monitoring, as any system administrator will tell you, is ideal for getting to the root of performance issues. Monitoring can help you understand why a system is out of CPU or RAM resources, for example, and notify you when CPU or RAM usage nears a specified percent. What your system administrator may not know (but you can explain after reading this chapter) is that monitoring is equally well suited for ferreting out security issues.

Consider a scenario: You manage a Hadoop cluster (as system administrator) and are concerned about two specific users: Bob, a confirmed hacker, and Steve, who loves to run queries that access volumes of data he is not supposed to access! To stop password loss and avoid server crashes, you would like to be notified when Bob is trying to read the /etc/password file and when Steve is running a huge query that retrieves the whole database. Hadoop monitoring can provide the information you need. Specifically, Hadoop provides a number of Metrics to gain useful security details, which the leading monitoring systems can use to alert you to trouble. In addition, these monitoring systems let you define thresholds (for generating alerts) based on specific Metric values and also let you define appropriate actions (in case thresholds are met) Thus, Hadoop monitoring offers many features you can use for performance monitoring and troubleshooting.

In this chapter’s detailed overview of monitoring, I will discuss features that a monitoring system needs, with an emphasis on monitoring distributed clusters. Thereafter, I will discuss the Hadoop Metrics you can use for security purposes, and introduce Ganglia and Nagios, the two most popular monitoring applications for Hadoop. Last, I will discuss some helpful plug-ins for Ganglia and Nagios that provide integration between the two programs, as well as plug-ins that provide security-related functionality.

Overview of a Monitoring System

Monitoring a distributed system is always challenging. Not only are multiple processes interacting with users and each other, but you must monitor the system without affecting the performance of those processes in any way. A system like Hadoop presents an even greater challenge, because the monitoring software has to monitor individual hosts and then consolidate that data in the context of the whole system. It also needs to consider the roles of various components in context of the whole system. For example, the CPU usage on a DataNode is not as important as the CPU usage on NameNode. So, how will the system process CPU consumption alerts or identify separate threshold levels for hosts with different roles within the distributed system? Also, when considering CPU or storage usage for DataNodes, the monitoring system must consider combined usage for all the DataNodes within a cluster. Subsequently, the monitoring system needs to have capability of summarizing monitoring thresholds by role as well.

In addition to the complex resource monitoring capabilities, a monitoring system for distributed systems needs to have access to details of processes executing at any time. This is necessary for generating alerts (e.g., a user process resulting in 90% CPU usage) or performing any preventive action (e.g., a user is accessing critical system files).

Before you can effectively meet the challenges of monitoring a Hadoop system, you need to understand the architecture of a simple monitoring system. In the next section, I’ll discuss the components, processing, and features that you need for monitoring a distributed system effectively, as well as how this simple architecture can be adapted to be better suited for monitoring a Hadoop cluster.

Simple Monitoring System

A simple monitoring system needs four key components: a server or coordinator process, connections to poll distributed system hosts and gather the necessary information, a repository to store gathered information, and a graphical user interface as a front-end (Figure 7-1).

As you can see, the monitoring server consolidates input received by polling the distributed system hosts and writes detailed (as well as summarized) output to a repository. A console provides display options for the gathered data, which can be summarized using various parameters, such as monitoring event, server, type of alert, and so on.

Unfortunately, simple monitoring system architecture like this doesn’t scale well. Consider what would happen if Figure 7-1’s system had to monitor thousands of hosts instead of three. The monitoring server would have to manage polling a thousand connections, process and consolidate output, and present it on the console within a few seconds! With every host added to the monitoring system, the load on the monitoring server will increase. After a certain number of hosts, you won’t be able to add any more, because the server simply won’t be able tosupport them. Also, the large volume of polling will add to network traffic and impact overall system performance.

Add to that the complexities of a Hadoop cluster where you need to consider a node’s role while consolidating data for it, as well as summarizing data for multiple nodes with the same role. The simplistic design just won’t suffice, but it can be adapted for monitoring a Hadoop cluster.

Monitoring System for Hadoop

A simple monitoring system follows the same processing arrangement as the traditional client-server design: a single, centralized monitoring server does all the processing, and as the number of hosts increase, so does the processing load. Network traffic also weighs down the load, as polled data from hosts consolidates on the monitoring server.

Just as Hadoop’s distributed architecture is a marked improvement in efficiency over traditional client-server processing, a distributed processing model can improve a simple monitoring system as well. If a localized monitoring process captures and stores monitoring data for each node in a Hadoop cluster, for example, there is no longer a centralized server to become a processing bottleneck or a single point of failure. Every node is an active participant performing part of the processing in parallel. Each of these localized processes can then transmit data to other nodes in the cluster and also receive copies of data from other nodes in the cluster. A polling process can poll monitoring data for the whole cluster from any of the nodes within the cluster at any predetermined frequency. The data can be written to a repository and stored for further processing or displayed by a graphical or web based frontend. Figure 7-2 shows a possible design.

With this architecture, even adding 1000 hosts for monitoring would not adversely affect performance. No additional load burdens any of the existing nodes or the polling process, because the polling process can still poll from any of the nodes and doesn’t have to make multiple passes. The cluster nodes transmit data to a common channel that is received by all other nodes. So, increasing the number of nodes does not impact polling process or system performance in any way, making the architecture highly scalable. Compared to traditional monitoring systems, the only extra bit of work that you need to do is to apply the monitoring process configuration to all the nodes.

Taking a closer look at Figure 7-2, notice that the monitoring processes on individual nodes compute “local monitoring data.” The monitoring data needs to be computed locally; because Hadoop is a multi-node distributed system where data is spread onto its numerous DataNodes and as per the Hadoop philosophy of “taking processing to data,” the data is processed locally (where it resides—on the DataNodes). This “local monitoring data” is actually Metric output for individual nodes; it can tell you a lot about your system’s security and performance, as you’ll learn next.

Hadoop Metrics

Hadoop Metrics are simply information about what’s happening within your system, such as memory usage, number of open connections, or remaining capacity on a node. You can configure every Hadoop daemon to collect Metrics at a regular interval and then output the data using a plug-in. The collected data can contain information about Hadoop daemons (e.g., the resources used by them), events (e.g., MapReduce job executions), and measurements (e.g., number of files created for NameNode). The output plug-in you use determines the Metric’s destination. For example, FileContext writes the Metric to a file, GangliaContext passes the Metric passed on to the Ganglia monitoring system for display and consolidation, and NullContext discards the Metric.

Depending on the information they contain Metrics are classified into four contexts: jvm, dfs, rpc, and mapred. Metrics for jvm contain basic statistics for JVM (Java Virtual Machine) such as memory usage or thread counts etc. This context is applicable for all Hadoop daemons. The dfs (distributed file system) context is applicable to NameNode and DataNode. Some of the Metrics for this context output information such as capacity or number of files (for NameNode), number of failed disk volumes, remaining capacity on that particular worker node (for DataNode), et cetera. JobTracker and TaskTracker use the mapred context for their counters. These Metrics contain pre-job counter data, job counters, and post-job counters. The rpc context is used for remote procedure call (RPC) Metrics such as average time taken to process an RPC, number of open connections, and the like, and is applicable to all Hadoop daemons. Table 7-1 summarizes the contexts.

Early versions of Hadoop managed Metrics through a system named Metrics, while the current version of Hadoop uses Metrics2. The management systems have two major differences. Metrics relies on a one-to-one relationship of one context per plug-in, while Metrics2 enables you to output Metrics to multiple plug-ins. The Metrics2 system also uses a slightly different terminology; the Metrics data output by Hadoop daemons is referred to as sources and the plug-ins are called sinks. Sources produce the data, and sinks consume or output the data. Let me discuss a few Metrics for each of the contexts.

The jvm Context

The jvm Metrics focus on basic JVM statistics. Table 7-2 lists some of these Metrics.

You can infer how dynamic your JVM process is by looking at GcCount and GcTimeMillis Metrics; larger numbers indicate a lot of memory-based activity. A large number of fatal errors indicate a problem with your system or application, and you need to consult your logs immediately. The memory counter MemHeapUsedM tells you about total memory usage, and if you see a large number for ThreadsWaiting, you know you need more memory.

The dfs Context

The dfs (distributed file system) Metrics focus on basic file operations (create, delete) or capacity, transactions, and the like. Table 7-3 lists some of these Metrics.

The dfs Metrics can be used for security purposes. You can use them to spot unusual activity or sudden change in activity for your cluster. You can store the daily Metric values (in a Hive table), and calculate an average for last 30 days. Then, if the daily value for a Metric varies by, say, 50% from the average, you can generate an alert. You can also direct the Metrics output to Ganglia, use Ganglia for aggregation and averaging, and then use Nagios to generate alerts based on the 50% variation threshold.

The rpc Context

The rpc (remote procedure call) Metrics focus on process details of remote processes. Table 7-4 lists some important rpcMetrics.

The rpc Metrics can also be used for security purposes. You can use them to spot unusual RPC activity or sudden changes in RPC activity for your cluster. Again, you can store the daily Metric values in a Hive table (or use Ganglia) and maintain averages for last 30 days. Then, if the daily value for a Metric varies by a certain percentage from the average, such as 50%, you can generate an alert. Metrics such as RpcAuthenticationFailures orRpcAuthorizationFailures are especially important from the security perspective.

The mapred Context

Metrics for mapred (MapReduce) context provide job-related details (for JobTracker/TaskTracker). Table 7-5 lists some important mapred Metrics.

Metrics for mapred context provide valuable information about the jobs that were executed on your cluster. They can help you determine if your cluster has any performance issues (from a job execution perspective). You can use a monitoring system (like Ganglia) to make sure that you have enough map and reduce slots available at any time. Also, you can make sure that you don’t have any long-running jobs—unless you know about them in advance! You can use Nagios with Ganglia to generate appropriate alerts. Just like the other contexts, mapred Metrics can also be monitored for unusual job activity (against average job activity).

You can find Hadoop Metrics listed in Appendix D, “Hadoop Metrics and Their Relevance to Security.”Appendix D also includes an example that explains use of specific Metrics and pattern searches for security (I included the security-specific configuration for that example, too).

Metrics and Security

Several Metrics can provide useful security information, including the following:

• Activity statistics for NameNode: It’s important to monitor the activity on NameNode, as it can provide a lot of information that can alert you to security issues. Being the “brain” of a Hadoop cluster, NameNode is hub of all the file creation activity. If the number of newly created files changes drastically or the number of files whose permissions are changed increases drastically, the Metrics can trigger alerts so you can investigate.
• Activity statistics for a DataNode: For a DataNode, if the number of reads or writes by a local client increases suddenly, you definitely need to investigate. Also, if the number of blocks added or removed changes by a large percentage, then Metrics can trigger alerts to warn you.
• Activity statistics for RPC-related processing: For the NameNode (or a DataNode), you need to monitor closely the RPCMetrics, such as the number of processed RPC requests, number of failed RPC authentication calls, or number of failed RPC authorization calls. You can compare the daily numbers with weekly averages and generate alerts if the numbers differ by a threshold percentage. For example, if the number of failed RPC authorization calls for a day is 50 and the weekly average is 30, then if the alert threshold is 50% or more of the weekly average, an alert will be generated (50% of 30 is 15, and the daily number (50) is greater than 45).
• Activity statistics for sudden change in system resources: It is beneficial to monitor for sudden changes in any of the major system resources, such as available memory, CPU, or storage. Hadoop provides Metrics for monitoring these resources, and you can either define a specific percentage (for generating alerts) or monitor for a percent deviation from weekly or monthly averages. The later method is more precise, as some of the clusters may never hit the target alert percentage even with a malicious attack (e.g., if average memory usage for a cluster is 20% and a malicious attack causes the usage to jump to 60%). If you have defined an alert threshold of 80% or 90%, then you will never get an alert. Alternatively, if you have defined your alert threshold for 50% or more (of average usage), then you will definitely get an alert.

You can use a combination of Ganglia and Nagios to monitor sudden changes to any of your system resources or Metrics values for any of the Hadoop daemons. Again, Appendix D has an example that describes this approach.

If you don’t want to use a monitoring system and want to adopt the “old-fashioned” approach of writing the Metrics data to files and using Hive or HBase to load that data in tables, that will work, too. You will of course need to develop shellscripts for scheduling your dataloads, perform aggregations, generate summary reports and generate appropriate alerts.

Metrics Filtering

When you are troubleshooting a security breach or a possible performance issue, reviewing a large amount of Metrics data can take time and be distracting and error-prone. Filtering the Metrics data helps you focus on possible issues and save valuable time. Hadoop allows you to configure Metrics filters by source, context, record, and Metrics. The highest level for filtering is by source (e.g., DataNode5) and the lowest level of filtering is by the Metric name (e.g., FilesCreated). Filters can be combined to optimize the filtering efficiency.

For example, the following file sink accepts Metrics from context dfs only:

bcl.sink.file0.class=org.apache.hadoop.metrics2.sink.FileSink
bcl.sink.file0.context=dfs

To set up your filters, you first need to add a snippet like the following in your $HADOOP_INSTALL/hadoop/conf/hadoop-metrics2.properties file:

# Syntax: <prefix>.(source|sink).<instance>.<option>

*.sink.file.class=org.apache.hadoop.metrics2.sink.FileSink
*.source.filter.class=org.apache.hadoop.metrics2.filter.GlobFilter
*.record.filter.class=${*.source.filter.class}
*.metric.filter.class=${*.source.filter.class}

After this, you can include any of the following configuration options that will set up filters at various levels:

# This will filter out sources with names starting with Cluster2

jobtracker.*.source.filter.exclude=Cluster2*

# This will filter out records with names that match localhost in the source dfs

jobtracker.source.dfs.record.filter.exclude=localhost*

# This will filter out Metrics with names that match cpu* for sink instance file only

jobtracker.sink.file.metric.filter.exclude=cpu*
jobtracker.sink.file.filename=MyJT-metrics.out

So, to summarize, you can thus filter out Metric data by source, by a pattern within a source, or by Metric names or patterns within an output file for a sink.

Please remember, when you specify an “include” pattern only, the filter only includes data that matches the filter condition. Also, when you specify an “exclude” pattern only, the matched data is excluded. Most important, when you specify both of these patterns, sources that don’t match either pattern are included as well! Last, include patterns have precedence over exclude patterns.

Capturing Metrics Output to File

How do you direct output of NameNode or DataNode Metrics to files? With Metrics2, you can define a sink (output file) into which to direct output from your Metric source by adding a few lines to the hadoop-metrics2.properties configuration file in the directory /etc/Hadoop/conf or $HADOOP_INSTALL/hadoop/conf. In the following example, I am redirecting the NameNode and DataNode Metrics to separate output files as well as the Ganglia monitoring system (remember, Metrics2 can support output to two sinks at once):

# Following are entries from configuration file hadoop-metrics2.properties
# collectively they output Metrics from sources NameNode and DataNode to
# a sink named 'tfile' (output to file) and also to a sink named 'ganglia'
# (output to Ganglia)

# Defining sink for file output
*.sink.tfile.class=org.apache.hadoop.metrics2.sink.FileSink

# Filename for NameNode output

namenode.sink.tfile.filename = namenode-metrics.log

# Output the DataNode Metrics to a separate file
datanode.sink.tfile.filename = datanode-metrics.log

# Defining sink for Ganglia 3.1
*.sink.ganglia.class=org.apache.hadoop.metrics2.sink.ganglia.GangliaSink31

# Default polling period for GangliaSink
*.sink.ganglia.period=10

# Directing output to ganglia servers
namenode.sink.ganglia.servers=gangliahost_1:8649,gangliahost_2:8649
datanode.sink.ganglia.servers=gangliahost_1:8649,gangliahost_2:8649

Now that you have all the Metric data in files, you need to make effective use of it. If you don’t plan to use a monitoring system, you will have to define file sinks (as output) for all the Hadoop daemons and manually analyze the huge output files or aggregate them as required! At the most, you can define Hive external tables and ease the processing. Alternatively, you can direct the Metrics output to a JMX console for reviewing it.

Please note that with either of these approaches, you won’t be able to display the Metric data or aggregations graphically for a quick review. Also, you will need to set up interface with alerting mechanism via shellscripts (accessing the Hive data) and set up interfaces for paging the system administrators (in case of critical events) as well.

However, if you plan to use Ganglia, sending your Metrics to the Ganglia monitoring system is as simple as sending them to a file and provides many more advantages, as you’ll learn in the next section.

Security Monitoring with Ganglia and Nagios

The best security monitoring system for your Hadoop cluster is a system that matches your environment and needs. In some cases, making sure that only authorized users have access may be most important, while in other cases, you may need to monitor the system resources and raise an immediate alert if a sudden change in their usage occurs. Some cluster administrators solely want to monitor failed authentication requests. The leaders in Hadoop security monitoring, Ganglia (http://ganglia.sourceforge.net) and Nagios (www.nagios.org), meet this challenge by providing flexibility and varied means of monitoring the system resources, connections, and any other part of your Hadoop cluster that’s technically possible to monitor.

Both are open source tools with different strengths that complement each other nicely. Ganglia is very good at gathering Metrics, tracking them over time, and aggregating the results; while Nagios focuses more on providing an alerting mechanism. Since gathering Metrics and alerting are both equally essential aspects of monitoring, Ganglia and Nagios work best together. Both these tools have agents running on all hosts for a cluster and gather information via a polling process that can poll any of the hosts to get the necessary information.

Ganglia

Ganglia was designed at the University of California, Berkeley and started as an open source monitoring project meant to be used with large distributed systems. Ganglia’s open architecture makes it easy to integrate with other applications and gather statistics about their operations. That’s the reason Ganglia can receive and process output data from Hadoop Metrics with ease and use it effectively.

For a monitored cluster, each host runs a daemon process called gmond that collects and broadcasts the local Metrics data (like CPU usage, memory usage, etc.) to all the hosts within the cluster. A polling process (gmetad) can then query any of the hosts, read all the Metrics data and route it to a central monitoring server. The central host can display the Metrics, aggregate them, or summarize them for further use. Gmond has little overhead and hence can easily be run on every machine in the cluster without affecting user performance. Ganglia’s web interface can easily display the summary usage for last hour, day, week, or month as you need. Also, you can get details of any of these resource usages as necessary.

Ganglia Architecture

Broadly, Ganglia has four major components: gmond, gmetad, rrdtool and gweb. gmond runs on all the nodes in a cluster and gathers Metrics data, gmetad polls the data from gmond, rrdtool stores the polled data, and gweb is the interface that provides visualization and analysis for the stored data. Figure 7-3 illustrates how Ganglia’s components fit into the basic Hadoop distributed monitoring system shown in Figure 7-2.

Take a closer look at what each of the Ganglia components does:

• gmond: gmond needs to be installed on every host you want monitored. It interacts with the host operating system to acquire Metrics such as load Metrics (e.g., average cluster load), process Metrics (e.g., total running processes) or rpc Metrics (e.g., RpcAuthenticationFailures). It is modular and uses operating system–specific plugins to take measurements. Since only the necessary plugins are installed at compile time, gmond has a very small footprint and negligible overhead.

  gmond is not invoked as per request from an external polling engine (for measurement), but rather polls according to a schedule defined by a local configuration file. Measurements are shared with other hosts (from the cluster) via a simple listen/announce protocol broadcasted at the same multicast address. Every gmond host also records the Metrics it receives from other hosts within the cluster.

  Therefore, every host in a Ganglia cluster knows the current value of every Metric recorded by every other host in the same cluster. That’s the reason only one host per cluster needs to be polled to get Metrics of the entire cluster, and any individual host failures won’t affect the system at all! Also, this design reduces the number of hosts that need to be polled exponentially, and hence is easily scalable for large clusters.

• gmetad: gmetad is the polling process within the Ganglia monitoring system. It needs a list of hostnames that specifies at least one host per cluster. An XML-format dump of the Metrics for a cluster can be requested by gmetad from any host in the cluster on port 8649, which is how gmetad gets Metrics data for a cluster.
• RRDtool: RRDtool is the Ganglia component used for storing the Metrics data polled by gmetad from any of the cluster hosts. Metrics are stored in “round-robin” fashion; when no space remains to store new values, old values are overwritten. As per the specified data retention requirements, RRDtool aggregates the data values or “rolls them up.” This way of data storage allows us to quickly analyze recent data as well as maintain years of historical data using a small amount of disk space. Also, since all the required disk space is allocated in advance, capacity planning is very easy.
• gweb: gweb is the visualization interface for Ganglia. It provides instant access to any Metric from any host in the cluster without specifying any configuration details. It visually summarizes the entire grid using graphs that combine Metrics by cluster and provides drop-downs for additional details. If you need details of a specific host or Metric, you can specify the details and create a custom graph of exactly what you want to see.

  gweb allows you to change the time period in graphs, supports extracting data in various textual formats (CSV, JSON, and more), and provides a fully functional URL interface so that you can embed necessary graphs into other programs via specific URLs. Also, gweb is a PHP program, which is run under the Apache web server and is usually installed on the same physical hardware as gmetad, since it needs access to the RRD databases created by gmetad.

Configuring and Using Ganglia

With a clearer understanding of Ganglia’s major components, you’re ready to set it up and put it to work for security-related monitoring and outputting specific Hadoop Metrics.

To install Ganglia on a Hadoop cluster you wish to monitor, perform the following steps:

1. Install Ganglia components gmetad, gmond, and gweb on one of the cluster nodes or hosts. (For my example, I called the host GMaster).
2. Install Ganglia component gmond on all the other cluster nodes.

The exact command syntax or means of install will vary according to the operating system you use. Please refer to the Ganglia installation instructions for specifics. In all cases, however, you will need to modify configuration files for Ganglia to work correctly and also for Hadoop to output Metrics through Ganglia as expected (the configuration files gmond.conf, gmetad.conf, and hadoop-metrics2.properties need to be modified).

To begin, copy gmond.conf (with the following configuration) to all the cluster nodes:

/* the values closely match ./gmond/metric.h definitions in 2.5.x */
globals {
  daemonize = yes
  setuid = yes
  user = nobody
  debug_level = 0
  max_udp_msg_len = 1472
  mute = no
  deaf = no
  allow_extra_data = yes
  host_dmax = 86400 /*secs. Expires hosts in 1 day */
  host_tmax = 20 /*secs */
  cleanup_threshold = 300 /*secs */
  gexec = no
  send_metadata_interval = 0 /*secs */
}

/*
 * The cluster attributes specified will be used as part of the <CLUSTER>
 * tag that will wrap all hosts collected by this instance.
 */
cluster {
  name = "pract_hdp_sec"
  owner = "Apress"
  latlong = "N43.47 E112.34"
  url = "http://www.apress.com/9781430265443"
}

/* The host section describes attributes of the host, like the location */
host {
  location = "Chicago"
}

/* Feel free to specify as many udp_send_channels as you like */
udp_send_channel {
  bind_hostname = yes #soon to be default
  mcast_join = 239.2.11.71
  port = 8649
  ttl = 1
}

/* You can specify as many udp_recv_channels as you like as well. */
udp_recv_channel {
  mcast_join = 239.2.11.71
  port = 8649
  bind = 239.2.11.71
  retry_bind = true
}

/* You can specify as many tcp_accept_channels as you like to share
   an xml description of the state of the cluster */
tcp_accept_channel {
  port = 8649
}

/* Each Metrics module that is referenced by gmond must be specified and
   loaded. If the module has been statically linked with gmond, it does
   not require a load path. However all dynamically loadable modules must
   include a load path. */
modules {
  module {name = "core_metrics"}
  module {name = "cpu_module"  path = "modcpu.so"}
  module {name = "disk_module" path = "moddisk.so"}
  module {name = "load_module" path = "modload.so"}
  module {name = "mem_module"  path = "modmem.so"}
  module {name = "net_module"  path = "modnet.so"}
  module {name = "proc_module" path = "modproc.so"}
  module {name = "sys_module"  path = "modsys.so"}
}

In the Globals section, the daemonize attribute, when true, will make gmond run as a background process. A debug_level greater than 0 will result in gmond running in the foreground and outputting debugging information. The mute attribute, when true, will prevent gmond from sending any data, and the deaf attribute, when true, will prevent gmond from receiving any data. If host_dmax is set to a positive number, then gmond will flush a host after it has not heard from it for host_dmax seconds. The cleanup_threshold is the minimum amount of time before gmond will cleanup any hosts or Metrics with expired data. The send_metadata_interval set to 0 means that gmond will only send the metadata packets at startup and upon request from other gmond nodes running remotely.

Several Ganglia Metrics detect sudden changes in system resources and are well suited for security monitoring:

• cpu_aidle (percentage of CPU cycles idle since last boot; valid for Linux)
• cpu_user (percentage of CPU cycles spent executing user processes)
• load_five (reported system load, averaged over five minutes)
• mem_shared (amount of memory occupied by system and user processes)
• proc_run (total number of running processes)
• mem_free (amount of memory free)
• disk_free (total free disk space)
• bytes_in (number of bytes read from all non-loopback interfaces)
• bytes_out (number of bytes written to all non-loopback interfaces)

You can add them to your gmond.conf file in the following format:

collection_group {
  collect_every = 40
  time_threshold = 300
  metric {
    name = "bytes_out"
    value_threshold = 4096
    title = "Bytes Sent"
  }
}

As you can see in the example, Metrics that need to be collected and sent out at the same interval can be grouped under the same collection_group. In this example, collect_every specifies the sampling interval, time_threshold specifies the maximum data send interval (i.e., data is sent out at that interval), and value_threshold is Metric variance threshold (i.e., value is sent if it exceeds the value_threshold value).

The second configuration file is gmetad.conf, which needs to reside on the host (GMaster) only. Keep in mind that the code that follows is only an example, and you can set up your own data sources or change settings as you need for round-robin archives:

# Format:
# data_source "my cluster" [polling interval] address1:port addreses2:port ...
#
data_source "HDPTaskTracker" 50 localhost:8658
data_source "HDPDataNode" 50 localhost:8659
data_source "HDPNameNode" 50 localhost:8661
data_source "HDPJobTracker" 50 localhost:8662
data_source "HDPResourceManager" 50 localhost:8664
data_source "HDPHistoryServer" 50 localhost:8666
#
# Round-Robin Archives
# You can specify custom Round-Robin archives here
#
RRAs "RRA:AVERAGE:0.5:1:244" "RRA:AVERAGE:0.5:24:244" RRA:AVERAGE:0.5:168:244" "RRA:AVERAGE:0.5:672:244" "RRA:AVERAGE:0.5:5760:374"
#
# The name of this Grid. All the data sources above will be wrapped in a GRID
# tag with this name.
# default: unspecified
gridname "HDP_GRID"
#
# In earlier versions of gmetad, hostnames were handled in a case
# sensitive manner. If your hostname directories have been renamed to lower
# case, set this option to 0 to disable backward compatibility.
# From version 3.2, backwards compatibility will be disabled by default.
# default: 1   (for gmetad < 3.2)
# default: 0   (for gmetad >= 3.2)
case_sensitive_hostnames 1

Last, you need to customize the hadoop-metrics2.properties configuration file in the directory /etc/Hadoop/conf or $HADOOP_INSTALL/hadoop/conf. You can define appropriate sources (in this case, either the dfs, jvm, rpc, or mapred Metrics), sinks (just Ganglia or a combination of Ganglia and output files), and filters (to filter out Metrics data that you don’t need).

To set up your sources and sinks, use code similar to the following:

# syntax: [prefix].[source|sink|jmx].[instance].[options]
# See package.html for org.apache.hadoop.metrics2 for details

*.period=60

*.sink.ganglia.class=org.apache.hadoop.metrics2.sink.ganglia.GangliaSink31
*.sink.ganglia.period=10

# default for supportsparse is false
*.sink.ganglia.supportsparse=true

.sink.ganglia.slope=jvm.metrics.gcCount=zero,jvm.metrics.memHeapUsedM=both
.sink.ganglia.dmax=jvm.metrics.threadsBlocked=70,jvm.metrics.memHeapUsedM=40

# Associate sinks with server and ports
namenode.sink.ganglia.servers=localhost:8661
datanode.sink.ganglia.servers=localhost:8659
jobtracker.sink.ganglia.servers=localhost:8662
tasktracker.sink.ganglia.servers=localhost:8658
maptask.sink.ganglia.servers=localhost:8660
reducetask.sink.ganglia.servers=localhost:8660
resourcemanager.sink.ganglia.servers=localhost:8664
nodemanager.sink.ganglia.servers=localhost:8657
historyserver.sink.ganglia.servers=localhost:8666
resourcemanager.sink.ganglia.tagsForPrefix.yarn=Queue

Setting supportsparse to true helps in reducing bandwidth usage. Otherwise the Metrics cache is updated every time the Metric is published and that can be CPU/network intensive. Ganglia slope can have values of zero (the Metric value always remains the same), positive (the Metric value can only be increased), negative (the Metric value can only be decreased), or both (the Metric value can either be increased or decreased). The dmax value indicates how a long a particular value will be retained. For example, the value for JVM Metric threadsBlocked (from the preceding configuration) will be retained for 70 seconds only.

As I discussed earlier in the “Metrics Filtering,” section, filters are useful in situations where you are troubleshooting or need to focus on a known issue and need specific Metric data only. Of course, you can limit the Metrics data you are capturing through settings in gmond.conf (as you learned earlier in this section), but filters can be useful when you need Metric data limited (or captured) temporarily—and quickly!

Monitoring HBase Using Ganglia

Ganglia can be used to monitor HBase just as you have seen it used for monitoring Hadoop. There is a configuration file called hadoop-metrics.properties located in directory $HBASE_HOME/conf (where $HBASE_HOME is the HBase install directory). You need to configure all the “contexts” for HBase to use Ganglia as an output:

# Configuration of the "hbase" context for Ganglia
hbase.class=org.apache.hadoop.metrics.ganglia.GangliaContext
hbase.period=60
hbase.servers=localhost:8649

# Configuration of the "jvm" context for Ganglia
jvm.class=org.apache.hadoop.metrics.ganglia.GangliaContext
jvm.period=60
hbase.servers=localhost:8649

# Configuration of the "rpc" context for Ganglia
rpc.class=org.apache.hadoop.metrics.ganglia.GangliaContext
rpc.period=60
hbase.servers=localhost:8649

For the hbase context, you can see values for metrics like averageLoad (average number of regions served by each region server) or numRegionServers (number of online region servers) on the HBase master server.

Also, for the jvm context, you can see Metrics like MemHeapUsedM (heap memory used, in MB) and MemHeapCommittedM (heap memory committed, in MB). If more than one jvm is running (i.e., more than one HBase process) Ganglia aggregates the Metrics values instead of reporting them per instance.

This concludes the HBase monitoring section. I have listed all the HBase Metrics in Appendix D for your reference.

Before I conclude the discussion about Ganglia, I want you to have a quick look at the Ganglia web interface. Please review Figure 7-4. It shows the Ganglia dashboard displaying summary graphs for the previous month. You can see the average and maximum load, CPU usage, memory usage, and network usage. From the dashboard you can select detailed graphs for any of these resources or create custom graphs for the specific Metrics you need.

Nagios

Nagios is a specialized scheduling and notification engine. It doesn’t monitor any processes or resources, but instead schedules execution of plug-ins (executable programs—Nagios plug-ins are not the same as Hadoop Metrics plug-ins) and takes action based on execution status. For example, status 0 is Success, 1 is Warning, 2 is Critical, and 3 is Unknown. You can configure the Nagios service to map specific actions for each of these outputs for all the plug-ins defined within the configuration files. In addition, you can define your own plug-ins and define the frequency for monitoring them as well as actions mapped to each of the possible outputs.

In addition to codes, the plug-ins can also return a text message, which can be written to a log and also be displayed on the web interface. If the text message contains a pipe character, the text after it is treated as performance data. The performance data contains Metrics from the monitored hosts and can be passed to external systems (like Ganglia) for use.

Most of the time, Nagios is used for monitoring along with Ganglia. The reason is that both these open source tools complement each other nicely, since they have different strengths. For example, Ganglia is more focused on gathering Metrics and tracking them over a time period, while Nagios focuses more on being an alerting mechanism. Since gathering Metrics and alerting are both essential aspects of monitoring, they work best in conjunction. Both Ganglia and Nagios have agents running on all hosts for a cluster and gather information.

Getting back to Nagios, let me start with Nagios architecture.

Architecture

The Nagios daemon or service runs on a host and has plug-ins running on all the remote hosts that need to be monitored. (To integrate Nagios with Ganglia, be sure the Ganglia process gmond is running on every host that has a Nagios plug-in running). The remote Nagios plug-ins send information and updates to the Nagios service, and the Nagios web interface displays it. When issues are detected, the Nagios daemon notifies predefined administrative contacts using email or page (text message sent to a phone). Historical log data is available in a log file defined in the configuration file. As you can see in Figure 7-5, the Nagios monitoring system has three major components:

• Server: The server is responsible for managing and scheduling plug-ins. At regular intervals, the server checks the plug-in status and performs action as per the status. In case of alerts, configured administrative resources are notified.
• Plug-ins: Nagios provides a standard set of user-configurable plug-ins, plus you can add more as required. Plug-ins are executable programs (mostly written in C, Java, Python, etc). that perform a specific task and return a result to the Nagios server.
• Browser interface of Nagios: These are web pages generated by CGI that display summary information about monitored resources.

Although using Ganglia and Nagios in conjunction is an effective approach to security monitoring, the applications are not integrated by default. You need to integrate them through plug-ins, as the next section explains.

Nagios Integration with Ganglia

Nagios has no built-in Metrics. Remote or local plug-ins are executed and their status compared by Nagios with user-specified status/notification mapping to perform any necessary notification tasks. Services like NRPE (Nagios Remote Plugin Executor) or NSCA (Nagios Service Check Acceptor) are used for remote executions. If you’re using Ganglia for monitoring, however, all the Metrics Nagios needs (for CPU, memory, disk I/O, etc.) are already available. You simply have to point Nagios at Ganglia to collect these Metrics! To help you, as of version 2.2.0 the Ganglia project started including a number of official Nagios plug-ins in its gweb versions (for details, see https://github.com/ganglia/ganglia-web/wiki/Nagios-Integration). In Nagios, you can then use these plug-ins to create commands and services to compare Metrics captured (or generated) by Ganglia against alert thresholds defined in Nagios.

Originally, five Ganglia plug-ins were available:

• check_heartbeat (check heartbeat to verify if the host is available)
• check_metric (check a single Metric on a specific host)
• check_multiple_metrics (check multiple Metrics on a specific host)
• check_host_regex (check multiple Metrics across a regex-defined range of hosts)
• check_value_same_everywhere (check value or values are the same across a set of hosts)

Now, the current Ganglia web tarball (version 3.6.2) contains 10 plug-ins for Nagios integration! You can download it at http://sourceforge.net/projects/ganglia/files/ganglia-web/3.6.2/ to check out the five new plugins.

Using Ganglia’s Nagios Plug-ins

When extracted, the Ganglia web tarball contains a subdirectory called nagios that contains the shellscripts as well as PHP scripts for each of the plug-ins. The shellscript for a plug-in accepts values for parameters and passes them on to the corresponding PHP script. The PHP script processes the values and uses an XML dump of the grid state (state of the cluster containing details of all the Metrics; obtained by gmetad) to acquire current Metric values as per the request. A return code (indicating the status of request) is passed back to Nagios. Figure 7-6 illustrates the process.

Remember to enable the server-side PHP script functionality before using it and to verify the following parameter values in configuration file conf.php (used by gweb):

$conf['nagios_cache_enabled'] = 1;
$conf['nagios_cache_file']= $conf['conf_dir']."/nagios_ganglia.cache";
$conf['nagios_cache_time'] = 45;

The location of conf.php varies as per the operating system, Hadoop distribution, and other factors. Your best option is to use the find command:

find / -name conf.php –print

The steps to follow for using Nagios as a scheduling and alerting mechanism for any of the five Ganglia plug-ins are very similar. Therefore, I will demonstrate the process with two of the plug-ins: check_heartbeat and check_multiple_metrics. I also will assume you have installed Ganglia, PHP, and Nagios and you are using the Hortonworks Hadoop distribution.

The check_heartbeat plug-in is a heartbeat counter used by Ganglia to make sure a host is functioning normally. This counter is reset every time a new Metric packet is received for the host. To use this plug-in with Nagios, first copy the check_heartbeat.sh script from the Nagios subdirectory in the Ganglia web tarball (in my case, /var/www/html/ganglia/nagios) to your Nagios plug-ins directory (in my case, /usr/lib64/nagios/plugins). Make sure that the GANGLIA_URL inside the script is correct. Substitute your localhost name and check if http://localhost/ganglia takes you to the Ganglia homepage for your installation. Then check if this is the setting in check_heartbeat.sh:

GANGLIA_URL=http://<localhost>/ganglia/nagios/check_heartbeat.php

At this point, you might also want to verify if PHP command line installation on your Nagios server is functional; you can do that by running the php –version command. You should see a response similar to the following:

PHP 5.3.3 (cli) (built: Aug  6 2014 05:54:27)
Copyright (c) 1997-2010 The PHP Group
Zend Engine v2.3.0, Copyright (c) 1998-2010 Zend Technologies

Run the plug-in script and verify it provides the heartbeat status correctly:

> ./check_heartbeat.sh host=pract_hdp_sec threshold=75

OK Last beacon received 0 days, 0:00:07

Next, define this plug-in as a command for Nagios (see the sidebar “Nagios Commands and Macros” for details). The threshold is the amount of time since the last reported heartbeat; that is, if the last packet received was 50 seconds ago, you would specify 50 as the threshold:

define command {
command_name check_ganglia_heartbeat
command_line $USER1$/check_heartbeat.sh host=$HOSTADDRESS$ threshold=$ARG1$
}

Note the use of the macros $HOSTADDRESS$ (substituted to IPaddress of the host), $USER1$ (user-defined macro defined in a resource file), and $ARG1$ (first argument to the command). Using macros provides the information contained in them automatically to a command (since the referenced value is available). So, the command check_ganglia_heartbeat can be used for checking the heartbeat on any host within your cluster. Similarly, the argument value passed to this command lets you change that parameter at runtime. Please refer to the sidebar “Nagios Commands and Macros” for further details about macros.

You can define the commands in the Nagios main configuration file called nagios.cfg. Most of the time the file resides in /etc/nagios, but location may vary for your install. The main configuration file defines individual object configuration files for commands, services, contacts, templates, and so forth. In addition, there may be a specific section for Hadoop servers. For example, the Hortonworks nagios.cfg has the following section:

# Definitions for hadoop servers
cfg_file=/etc/nagios/objects/hadoop-hosts.cfg
cfg_file=/etc/nagios/objects/hadoop-hostgroups.cfg
cfg_file=/etc/nagios/objects/hadoop-servicegroups.cfg
cfg_file=/etc/nagios/objects/hadoop-services.cfg
cfg_file=/etc/nagios/objects/hadoop-commands.cfg

I will define the command check_ganglia_heartbeat in configuration file /etc/nagios/objects/hadoop-commands.cfg. The last step is defining a service for Nagios. Within Nagios, use of the term service is very generic or nonspecific. It may indicate an actual service running on the host (e.g., POP, SMTP, HTTP, etc.) or some other type of Metric associated with the host (free disk space, CPUusage, etc.). A service is defined in configuration file /etc/nagios/objects/hadoop-services.cfg and has the following syntax:

define service {
        host_name               localhost
        use                     hadoop-service
        service_description     GANGLIA::Ganglia Check Heartbeat
        servicegroups           GANGLIA
        check_command           check_ganglia_heartbeat!50
        normal_check_interval   0.25
        retry_check_interval    0.25
        max_check_attempts      4
}

Please note that check_command indicates the actual command that would be executed on the specified host. The parameter normal_check_interval indicates the number of time units to wait before scheduling the next check of the service. One time unit is 60 seconds (that’s the default), and therefore 0.25 indicates 15 seconds. retry_check_interval defines the number of time units to wait before scheduling a recheck of the service if it has changed to a non-okaystate, and max_check_attempts indicates the number of retries in such a situation.

The command check_multiple_metrics checks multiple Ganglia Metrics and generates a single alert. To use it, copy the check_multiple_metrics.sh script from the Nagios subdirectory in the Ganglia web tarball to your Nagios plug-ins directory. Make sure that GANGLIA_URL inside the script is set to http://localhost/ganglia/nagios/check_heartbeat.php, and also remember to substitute localhost with the appropriate host name.

Define the corresponding command check_ganglia_multiple_metrics in the configuration file /etc/nagios/objects/hadoop-commands.cfg:

define command {
command_name check_ganglia_multiple_metrics
command_line $USER1$/check_multiple_metrics.sh host=$HOSTADDRESS$ checks='$ARG1$'
}

You can add a list of checks delimited with a colon. Each check consists of Metric_name,operator,critical_value. Next, define a corresponding service in the configuration file /etc/nagios/objects/hadoop-services.cfg:

define service {
        host_name               localhost
        use                     hadoop-service
        service_description     GANGLIA::Ganglia check Multiple Metric service
        servicegroups           GANGLIA
        check_command          check_ganglia_multiple_metrics!disk_free,less,10:load_one,more,5
        normal_check_interval   0.25
        retry_check_interval    0.25
        max_check_attempts      4
}

Note the check_command section that defines the command to be executed:

check_ganglia_multiple_metrics!disk_free,less,10:load_one,more,5.

This indicates that an alert will be generated if free disk space (for the host) falls below 10GB or if 1-minute load average goes over 5.

After successfully defining your Ganglia plug-ins, you can use the Nagios web interface to check and manage these plug-ins. As you can see in Figure 7-7, the new check_heartbeat and check_multiple_metrics plug-ins are already in place and being managed by Nagios.

If you’d like more practice, you can follow the same steps and add the other three plug-ins.

The Nagios Community

The real strength of Nagios is in its active user community that’s constantly working towards making a more effective use of Nagios and adding plug-ins to enhance its functionality. To see the latest plug-ins your fellow users have developed, visit the community page at http://exchange.nagios.org/directory/Plugins. For security purposes, you’ll find many plug-ins that you can use effectively, such as:

• check_ssh_faillogin: Monitors the ssh failed login attempts; available at http://exchange.nagios.org/directory/Plugins/Security/check_ssh_faillogin/details.
• show_users: Shows logged users. Can alert on certain users being logged in using a whitelist, blacklist, or both. Details at: http://exchange.nagios.org/directory/Plugins/System-Metrics/Users/Show-Users/details
• check_long_running_procs.sh: Checks long-running processes; available at http://exchange.nagios.org/directory/Plugins/System-Metrics/Processes/Check-long-running-processes/details.

You can use the same process as you followed for using the Ganglia plug-ins to use any new plug-in. You will need to copy it to the Nagios plug-ins directory, then define a command and service. Of course, follow any specific install instructions for individual plug-ins or install any additional packages that are required for their functioning.

Summary

In this chapter, I have discussed monitoring for Hadoop as well as popular open-source monitoring tools. Remember, monitoring involves a good understanding of both the resources that need to be monitored and the environment that you plan to monitor. Though I can tell you what needs to be monitored for a Hadoop cluster, you know your environment’s individual requirements best. I have tried to provide some general hints, but from my experience, monitoring is always as good as your own system administrator’s knowledge and understanding of your environment.

The “relevance” (how “up to date” or “state of the art” a system is) is also a very valid consideration. You have to be conscious on a daily basis of all the innovations in your area of interest (including the malicious attacks) and tune your monitoring based on them. Remember, the best system administrators are the ones who are most alert and responsive.

Last, please try to look beyond the specific tools and version numbers to understand the principles and intentions behind the monitoring techniques described in this chapter. You may not have access to the same tools to monitor, but if you follow the principles, you will be able to set up effective systems for monitoring—and in the end, that’s both our goal.