One of the most common questions we get from our WordPress VIP clients, many of whom are large media companies that publish constantly, is how they can bias their search results towards more recent content when scoring and sorting them. This type of problem is extremely hard to solve with a traditional RDBMS but we provide most of our VIP clients their own dedicated Elasticsearch index and as it happens ES comes with some powerful scoring functions for just this purpose.
With Elasticsearch adding date based weighting of results via function scores is pretty straightforward. A query like the following will multiply the TF-IDF textual relevancy score of the content with a date based score that makes older content less and less important as time progresses:
Adding such a time based decay is simple however it’s not obvious which type of function scoring should be used (gauss, exp, or linear) much less what are good values to use when configuring the offset, scale, and decay of each function.
While leading a workshop on querying Elasticsearch last week I struggled to explain exactly how these scoring functions will effect the final ranking of documents. So instead of trying to workout the coefficients we are adding with function scores by hand I decided to build a simple visualization so that it’s easier to play around with different settings to see what will happen to our scores under various settings.
The CodePen at the top of this post allows you to adjust various settings then shows how scores decay over a 1 year period with each of the three types of function scorers. Note, both offset and scale are specified in days so 28 is the equivalent of 28d when used within the ES query DSL.
At Automattic we run a diverse array of systems and as with many companies Kafka is the glue that ties them together; letting us to shuffle data back and forth. Our experience with Kafka have thus far been fantastic, it’s stable, provides excellent throughput, and the simple API makes it trivial to hook any of our systems up to it. In fact it’s been so popular that we’ve been steadily piping more and more data through it over the past year. Now we’re starting to run out of disk space necessitating an expansion of the cluster.
The expansion plan was pretty straight forward. Spin up some more brokers, have them join the same cluster, then when they are all up simply use Kafka’s provided kafka-reassign-partitions.sh script to rebalance all topics across all brokers.
We then thought, with more available capacity why not also do a rolling upgrade of our cluster to get on the latest distro across all our servers? And seeing as we run most topics with a replication factor of 2 so it seemed better to use the reassign script to move data off each server as we upgrade as to not have partitions run with only a single copy of the data at any time during the upgrade process.
The Problem
The expansion and rolling restart method seemed like a a good plan however after the upgrade we discovered that while the reassign script balances partitions between servers, those servers do not balance each topic between the configured log.dirs. This lead to major imbalances on each broker causing some disks to be virtually empty while other disks to be almost entirely full. A quick search showed others have also run into similar issues with no good way of fixing it.
Why Did It Happen?
The problem stems from the fact that we run our Kafka cluster in JBOD mode, meaning we don’t RAID or combine the disks on the hardware or OS level. Instead we mount all disks then configure Kafka’s log.dirs to use all drives on the server. When run this way Kafka distributes partitions assigned to each broker evenly across all log dirs available to it.
Each new partition that is created will be placed in the directory which currently has the fewest partitions.
–Kafka Docs
Unfortunately, we have some quite unbalanced topics. Some topics are huge on the order of TBs, with long retention periods. Then there are other topics that perhaps are not yet fully productionized with only a couple MB or KB in each partition. By simply balancing the number of partitions across log.dirs we end up with some disks with mostly large partitions and some with mostly small partitions.
Hack Partition Reassignment
There is currently some talk about trying to make JBOD support in Kafka better but we don’t want to run with such unbalanced disk use nor did we want to bring brokers down to try and hack the metadata in order to move files around in the interim. Luckily how Kafka brokers assign partitions and moves data during partition reassignment operations is well defined which means we can force each broker to balance its data through the use of partition reassignment if we control it very carefully. The basic concept is pretty simple:
Ensure the topic we want to distribute has the right number of partitions. In order to achieve an even distribution of a topic we need to make sure we have enough partitions to evenly distribute in the first place. This means we’ll need to make sure that the number of partitions × replication factor is a multiple of the number of brokers × number of disks on each broker. If this is not done we will not end up with exactly the same number of partitions per disk across all brokers and disks.
Balance partition counts across disks on all brokers. We want every disks to have the same number of partitions on each broker before we move anything. Doing so ensures that when new partitions are created they will be round robined across every available disk on each broker evenly. When a broker is assigned a new partition Kafka assigns that partition to one of the directories configured in log.dirs that has the fewest partitions. Without first balancing partition counts across disks these assignment operations will not assign the topic being moved across disks evenly.
Move one topic at a time but move all partitions of that topic together. When Kafka reassigns partitions the new partition is created on the broker it’s being moved to and then synced with the leader. By moving every replica of every partition of a single topic together Kafka will be forced to create a new copy of each partition thus distributing all those partitions evenly.
Tips and Hints
The easiest way to make sure partition counts are balanced across all disks of all brokers is to simply create a new topic and manually assign the right number of partitions to each broker. To do this go to each of your brokers and count up how many partitions you have on each disk then figure out how many more partitions you need on each broker to bring the partition count of the broker up to be the max partitions on any one disk × number of disks. For example if you have 3 disks and with 10 partitions on disk A and 6 partitions on disk B & C you’ll need 8 more partitions to bring the total partition count of that broker up to 30.
Once you have all the counts simply create a new topic with the number of new partitions needed for each broker. Continuing with the above example if you’ve determined that broker “1” needs 8 more partitions and broker “2” needs 5 more partitions you would issue the following command to create a new topic with 13 partitions, each with a single replica.
Once this filler topic has been created the number of partitions per disk on each broker will be evened out.
Forcing a reassignment of every partition and replica for a topic is a bit harder to accomplish. First you must make sure that the current number of replicas and number of future replicas is less then the total number of brokers. (This is necessary because all brokers that you are assigning a partition to must not already be assigned that partition.) If this is not done then only some portion of partitions for a topic will actually be moved and it’s likely one of the partitions being moved will end up getting moving to the same disk as a partition not being moved leading to a unbalanced distribution once again.
With dozens of partitions and thousands of possible permutations trying to assign all partitions so that they are both balanced across all brokers and are not assigned to a broker that already contains a replica of it was not something I wanted to do by hand so I wrote a Python script to generate the necessary assignments for me.
To use this script you must first get the current partition replica assignment from Kafka for the topic that you are trying to move using the kafka-reassign-partitions.sh tool:
The param specified by broker-list does not actually matter as we don’t care about the plan generated by the tool. We’re just using this to print out the current assignment in JSON format.
The topic.json file should look something like the following. (Remember, only move one topic at a time otherwise balance is not guaranteed!)
The kafka-reassign-partitions.sh script will generate two JSON strings, one showing the current partition assignment and one showing a proposed partition assignment. Ignore the proposed assignment and instead pass the current assignment JSON string to kafka-rebalance.py to get a new proposed assignment, one that guarantees all partitions will relocate.
This is a pretty convoluted way to rebalance topics across all disks on each Kafka broker however it’s pretty safe as this method does not involve any metadata hacking nor does it necessitate bring any brokers down. Hopefully Kafka will become better at balancing topics across disks when run in JBOD mode by itself in the future.
At Automattic we see over 131M unique visitors per month from the US alone. As part of the data team we are responsible for taking in the stream of Nginx logs and turning them into counts of views and unique visitors per day, week, and month on both a per blog and global basis.
To do all that we have a near-realtime pipeline that uses a myriad of technologies including PHP, Kafka, and various components from the Hadoop ecosystem. Unfortunately this system broke down last month and caused us to lose a portion of uniques data. After resolving the initial issue it became clear to us we will need to reprocess data from original log files in order to recover all of the data we’ve lost.
Problem is with of billions of hits a day, our data volume is comparable to that of the IceCube South Pole Neutrino Observatory, a detector made of one cubic kilometer of ice. Combine that with the fact that we’ve been leaking data over the course of a couple of days means we are left with a lot of logs to reprocess. Certainly not something we want to try and churn through on a single server. Luckily we already have a Hadoop cluster in place so we can tap into the powers of MapReduce to solve this problem.
Hive to the rescue
Apache Hive is a data warehouse infrastructure built on top of Hadoop. It allows processing of data with SQL like queries and it’s very pluggable so we can configure it to process our logs quite easily.
First, we will create a Hive table that’s configured to read raw compressed Nginx logs. To do this we will instruct Hive to create an external table at the location where we will copy our logs to on HDFS. We specify that the table is to be stored as TEXTFILE which allows Hive to read the stored files in as either plain text or for those files with a .gz extension, gziped plain text. We also want to apply a custom serialization/deserialization (serde) format to each log line so that each line is parsed into columns of data in our table. To do that we will use the included RegEx serde, it’s slow but given it simply applies a RegEx expression to each line in order to extract data it’s highly configurable.
Our create table statement looks something like this:
Once we have done this all we need to do is to copy our gziped logs to the specified HDFS location and query that Hive table. Hive even offers some very helpful functions like parse_url() which we can use to extract parts of the URL or query parameters.
Of course there’s quite a bit of business logic associated with our log processing and not all of it can be replicated with the bundled Hive functions. Fortunately Hive’s very pluggable on the querying side as well. Hive has a concept of User Defined Functions (UDFs) where we can write a simple Java class that implements a common UDF interface and install it in our cluster at query time to be distributed out to all our mappers for processing.
Using these methods and the spare capacity of our Hadoop cluster we were able to reprocess our logs in a couple hours instead of having a single server take days or even weeks to churn through them. Just in time to make sure our bloggers get the fireworks they rightly deserve on their annual reports.
Hey, thanks for reading and if the above sounded at all interesting to you we’re hiring and would love to talk with you.🙂
At WordPress.com we process a lot of events including some some events that are batched and sent asynchronously sometimes days later. But when querying this data we are likely to care more about when the events occurred rather then when it was sent to our servers. Knowing this we store our event data in Hive partitioned by when the events occurred rather then when they are ingested.
Event Ingestion, Take One
The initial design of our ETL pipeline looks something like this:
Raw logs are aggregated and processed by a custom parser which functions as both an aggregator for high level stats as well as emitter of raw logs into the various Kafka topics.
A Flume agent then uses the Kafka source to pull from the appropriate topic.
The Flume Morphline interceptor is then used to do a series of transformations including annotating what type of event the log line represented.
Kite then handles interacting with Hive and persisting the events in HDFS.
With this ETL data is available for querying almost immediately and is stored in (close to) their final state within Hive tables. As an added bonus, because we are using Kite Datasets and the accompanying Flume sink Hive partitions are handled for us automatically. However due to the asynchronous nature of our event collection we end up having to write to multiple partitions at the same time which results in the formation of many small files. To get around this we simply run a compaction job with Oozie after some date cutoff for events.
This process worked very well until a couple days ago when some network issues and some bugs caused OOM errors which resulted in our Flume agent sporadically loosing all events buffered in memory. Fortunately, we persist our Kafka topic of important logs for about a month so we can just replay it then merge and dedupe the data in Hive. Not only that, but we have a lot of servers in Cloudera Manager so we can just add our backfilling Flume agent role to a bunch of them and we’ll be done in no time. #winning
Alas, the dream was not to be. Because we must write to many partitions at once sending events to the Kite Dataset sink with so many Flume agents caused our Hive Metastore to become unstable not only limiting the rate of our ingestion but also causing numerous query failures for our growth explorers. Sure, we could have just slowed ingestion down but being snake people that just won’t do. After all we have the technology; we can make our ETL…
Better… Stronger… Faster
With what we have learned about our first ETL pipeline we’ve decided to rewrite it but this time with an eye on durability and ingestion performance as well. We kept our custom log parser / aggregator / Kafka emitter as it’s doing an admirable job. In addition, we decided to stick with Flume (and its Kafka source) for our ETL process because of its ease of use and customizability.
We did not like the fact that when a Flume agent crashed it would just drop events in the memory channel on the floor so to make our process more durable we opted to use Flume’s Kafka channel instead. This allowed us to multiplex and publish records after they have been transformed by the Morphline interceptor to Kafka so that they are persisted when the Flume Agent dies. Doing this comes at the expense of possible duplicate events being emitted when anomalies happen however we figure it’s far easier to dedupe with Hive queries then it is to recreate missing data.
To make writes more performant we reconfigured Morphline to convert our events to Avro records with our predefined schemas and then serialize and compress those records making them ready for writing. Once the raw Avro byte array has been generated we multiplex it into the proper Kafka channel. Finally we use the HDFS sink to pull these events from Kafka in batches and write the raw Avro byte arrays to HDFS partitioned by when the record was written. We do this directly without touching Hive or its Metastore.
By partitioning on when the record was written we can ensure we only write to a single partition at a time. This results in fewer and larger files which is not only more performant but it also gives us the ability to know when partitions can be considered complete. With this knowledge we can now have Oozie jobs that merge, dedupe, and compact events from the intermediate table partitioned by the time an event is recorded into another table partitioned by when events occurred for optimal query performance.
Show Me The Code
So what does this look like? Here’s a simplified flume.conf example:
#
# Setup some names
#
agent.sources = sr-kafka
agent.channels = ch-kafka-type1 ch-kafka-type2
agent.sinks = sk-hdfs-type1 sk-hdfs-type2
#
# Configure same Kafka source for all channels
#
agent.sources.sr-kafka.channels = ch-kafka-type1 ch-kafka-type2
agent.sources.sr-kafka.type = org.apache.flume.source.kafka.KafkaSource
agent.sources.sr-kafka.zookeeperConnect = HOST1:PORT,HOST2:PORT,HOST3:PORT/PATH
agent.sources.sr-kafka.groupId = flume_source_20150712
agent.sources.sr-kafka.topic = kafka-topic
# Grabs in batches of 500 or every second
agent.sources.sr-kafka.batchSize = 500
agent.sources.sr-kafka.batchDurationMillis = 1000
# Read from start of topic
agent.sources.sr-kafka.kafka.auto.offset.reset = smallest
#
# Configure interceptors
#
agent.sources.sr-kafka.interceptors = in-morphline-etl in-host-set
agent.sources.sr-kafka.interceptors.in-morphline-etl.type = org.apache.flume.sink.solr.morphline.MorphlineInterceptor$Builder
agent.sources.sr-kafka.interceptors.in-morphline-etl.morphlineFile = /path/to/morphline.conf
agent.sources.sr-kafka.interceptors.in-morphline-etl.morphlineId = morphline_id
agent.sources.sr-kafka.interceptors.in-host-set.type = host
agent.sources.sr-kafka.interceptors.in-host-set.useIP = false
agent.sources.sr-kafka.interceptors.in-host-set.hostHeader = flume_host
#
# Multiplex our records into channels based on the value of `eventmarker` which comes from Morphline
#
agent.sources.sr-kafka.selector.type = multiplexing
agent.sources.sr-kafka.selector.header = eventmarker
agent.sources.sr-kafka.selector.default = ch-kafka-type1
agent.sources.sr-kafka.selector.mapping.type1 = ch-kafka-type1
agent.sources.sr-kafka.selector.mapping.type2 = ch-kafka-type2
agent.sources.sr-kafka.selector.mapping.type3 = ch-kafka-type1 ch-kafka-type2
#
# Configure the channels we multiplexed into
#
agent.channels.ch-kafka-type1.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.ch-kafka-type1.brokerList = HOST1:PORT,HOST2:PORT,HOST3:PORT
agent.channels.ch-kafka-type1.zookeeperConnect = HOST1:PORT,HOST2:PORT,HOST3:PORT/PATH
agent.channels.ch-kafka-type1.groupId = flume_channel_20150712
agent.channels.ch-kafka-type1.topic = kafka-topic-flume-type1
agent.channels.ch-kafka-type2.type = org.apache.flume.channel.kafka.KafkaChannel
agent.channels.ch-kafka-type2.brokerList = HOST1:PORT,HOST2:PORT,HOST3:PORT
agent.channels.ch-kafka-type2.zookeeperConnect = HOST1:PORT,HOST2:PORT,HOST3:PORT/PATH
agent.channels.ch-kafka-type2.groupId = flume_channel_20150712
agent.channels.ch-kafka-type2.topic = kafka-topic-flume-type2
#
# Configure sinks; We pull from Kafka in batches and write large files into HDFS.
#
agent.sinks.sk-hdfs-type1.channel = ch-kafka-type1
agent.sinks.sk-hdfs-type1.type = hdfs
agent.sinks.sk-hdfs-type1.hdfs.path = hdfs://path/to/database/etl_type1/record_ymdh=%Y%m%d%H
# Prefix files with the Flume agent's hostname so we can run multiple agents without collision
agent.sinks.sk-hdfs-type1.hdfs.filePrefix = %{flume_host}
# Hive needs files to end in .avro
agent.sinks.sk-hdfs-type1.hdfs.fileSuffix = .avro
# Roll files in HDFS every 5 min or at 255MB; don't roll based on number of records
# We roll at 255MB because our block size is 128MB, we want 2 full blocks without going over
agent.sinks.sk-hdfs-type1.hdfs.rollInterval = 300
agent.sinks.sk-hdfs-type1.hdfs.rollSize = 267386880
agent.sinks.sk-hdfs-type1.hdfs.rollCount = 0
# Write to HDFS file in batches of 500 records
agent.sinks.sk-hdfs-type1.hdfs.batchSize = 500
# We already serialized and encoded the record into Avro in Morphline so just write the byte array
agent.sinks.sk-hdfs-type1.hdfs.fileType = DataStream
# Give us a higher timeout because we are writing in batch
agent.sinks.sk-hdfs-type1.hdfs.callTimeout = 60000
# Use current time in UTC for the value of `record_ymdh=%Y%m%d%H` above
agent.sinks.sk-hdfs-type1.hdfs.timeZone = UTC
agent.sinks.sk-hdfs-type1.hdfs.useLocalTimeStamp = true
# Our record is serialized via Avro
agent.sinks.sk-hdfs-type1.serializer = org.apache.flume.sink.hdfs.AvroEventSerializer$Builder
agent.sinks.sk-hdfs-type2.channel = ch-kafka-type2
agent.sinks.sk-hdfs-type2.type = hdfs
agent.sinks.sk-hdfs-type2.hdfs.path = hdfs://path/to/database/etl_type2/record_ymdh=%Y%m%d%H
agent.sinks.sk-hdfs-type2.hdfs.filePrefix = %{flume_host}
agent.sinks.sk-hdfs-type2.hdfs.fileSuffix = .avro
agent.sinks.sk-hdfs-type2.hdfs.rollInterval = 300
agent.sinks.sk-hdfs-type2.hdfs.rollSize = 267386880
agent.sinks.sk-hdfs-type2.hdfs.rollCount = 0
agent.sinks.sk-hdfs-type2.hdfs.batchSize = 500
agent.sinks.sk-hdfs-type2.hdfs.fileType = DataStream
agent.sinks.sk-hdfs-type2.hdfs.callTimeout = 60000
agent.sinks.sk-hdfs-type2.hdfs.timeZone = UTC
agent.sinks.sk-hdfs-type2.hdfs.useLocalTimeStamp = true
agent.sinks.sk-hdfs-type2.serializer = org.apache.flume.sink.hdfs.AvroEventSerializer$Builder
The most important part of the above is that we set the HDFS sink use the Avro serializer and instruct that it should simply write the raw bytes as we've already serialized the Avro record and compressed it with Morphline. Speaking of which, here's our example morphline.conf:
morphlines : [
{
id : morphline_id
# Import the Kite SDK and any custom libs you may have and need
importCommands : [
"org.kitesdk.**",
"com.a8c.**"
]
commands : [
# Each command consumes the output record of the previous command
# and pipes another record downstream.
{
# Parse input attachment and emit a record for each input line
readLine {
charset : UTF-8
}
}
{
# More commands for your ETL process
}
{
# Say we set a field named `eventmarker` somewhere above to indicate the
# type of record this is and we have a different schemas
if {
conditions : [
{ equals { eventmarker : "type1" } }
]
then : [
{
# Set the schema for the Flume HDFS sink
setValues {
flume.avro.schema.url : "file:/path/to/schema/type1.avsc"
}
}
{
# Converts this to an Avro record according to schema
toAvro {
schemaFile : /path/to/schema/type1.avsc
}
}
]
else : [
{
setValues {
flume.avro.schema.url : "file:/path/to/schema/type2.avsc"
}
}
{
toAvro {
schemaFile : /path/to/schema/type2.avsc
}
}
]
}
}
{
# Serialize the Avro record into a byte array, compressed with snappy
writeAvroToByteArray : {
format : containerlessBinary
codec : snappy
}
}
]
}
]
With these configs Flume will write compressed Avro files directly to HDFS but we will need to let Hive know about where to look so we need to create the table in Hive.
-- Table name
CREATE TABLE IF NOT EXISTS etl_type1
-- We need to specify how we are partitioning this table with the Flume HDFS sink
PARTITIONED BY ( record_ymdh INT )
-- Files were written in Avro!
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
-- We are writing to this dir in HDFS from Flume
LOCATION 'hdfs://path/to/database/etl_type1'
-- We also store the Avro schema in a hidden dir on HDFS for convenience
TBLPROPERTIES (
'avro.schema.url'='hdfs://path/to/database/etl_type1/.schema/type1.avsc'
);
For convenience I also stored the Avro schema for the table in the .schema directory on HDFS but that schema can really be anywhere readable by Hive.
Of course as we ingest data the Flume HDFS sink with start creating new directories, a.k.a. partitions, in HDFS but Hive will know nothing about them. So you will need to let Hive repair its Metastore by scanning HDFS before you can query for new data:
MSCK REPAIR TABLE etl_type1;
SELECT *
FROM etl_type1
WHERE ...;
What’s Next?
We have not really pushed this new pipeline to see where the limits are however as I write this we are on track to ingest a month of data in less then 12 hours. In addition, scaling this pipeline by simply spinning up more Flume agents has thus far been linear. The one down side is that for the most up to date information we will now need to look at 2 separate Hive tables with different partition strategies making queries a bit more complicated.
We have in effect made what I like to call the “Iota Architecture” — a system that’s 1/3 of the way to a true lambda architecture. We currently have a system that emits a stream of events that can be read in batch or by a stream processor but we only have a batch process in place to allow for performant queries on “archival” data. Perhaps someday we’ll get the other 2/3 in place for our growth explorers to easily get a unified view.
With Heroku-WP I hoped to lower the bar in getting WordPress up and running on a more modern tech stack. But what are the performance implications of running WordPress on such a modern set of technologies? Surely it’s faster but by how much and is the performance gains worth the trouble?
To answer that I’ve conducted a quick and dirty stress test of a sample WordPress site running under PHP and HipHop VM (HHVM) and found that the HHVM version loaded almost twice as fast and was able to serve over twice as many requests.
Response Time
Ok Responses
Errors / Timeouts
Anon PHP
4.091
8,939
0.94%
Anon HHVM
2.122
18,308
0.00%
Change
48.1%
2.05X
Auth PHP
20.688
457
74.17%
Auth HHVM
14.359
1,242
43.45%
Change
30.6%
2.72X
In the numbers above anonymous requests represents hits to various pages without a WordPress logged in cookie which are eligible for Batcache caching whereas authorized requests are hits to the same pages with a login cookie thus bypassing page caching.
Test Methodology
To conduct this test I created a brand new Heroku instance based on the Heroku-WP template with the memcached addon installed. To populate the content I imported the WordPress Theme Unit Test after performing the initial setup. I then ran each permutation of the load test 5 times, flushing caches in between and averaged the results. When switching between PHP and HHVM I simply pushed a config change to composer.json and the boot script to toggle between the two interpreters.
To conduct the actual load test I used loader.io, (shameless referal link) a cloud based load generator with an insane 10,000 concurrent connection limit on the free account. (Quite generous when compared to the limit of 250 concurrent users on the free blitz.io account which I’ve used previously.) The free account of loader.io will only accept 2 URLs to test however with WordPress we can access most front-end pages via GET parameters passed to /index.php. Using this method and the following payload file I was able to have loader.io cycle through all posts, pages, categories, time based index pages, and the home page on the sample site for a more comprehensive sampling.
Finally, I ran all the tests for 60 seconds ramping up from 0 to 1,000 concurrent connections. I purposely excluded loading of static assets to place as much pressure as possible on PHP / HHVM for this synthetic stress test.
Conclusions
As expected, under high load MySQL is the real bottleneck, slowing down requests and causing errors due to hanging or killed queries. Caching was able to mitigate poor DB performance and bring page load times down by an order of magnitude for both PHP and HHVM.
Beyond that through simply turning on HHVM cut load times in half and appears to have allowed the server to better cope with multiple simultaneous slow requests piling up during times of high load. So if you have already installed a caching plugin and optimized your WordPress site running it on HHVM could potentially be an easy way to squeeze some more performance out of your setup.
If you’re running a multi-node Elasticsearch cluster checkout Automattic’s fork of the Elasticsearch StatsD Plugin for pushing cluster and node metrics to StatsD.
At Automattic we’ve been using a set of Munin scripts to collect and aggregate Elasticsearch metrics via its native node & stats REST APIs. This method works relatively well giving us enough longitudinal information about the cluster and nodes to diagnose issues or test optimizations. That said, cluster monitoring with Munin at a 5 minute resolution leaves a lot to be desired.
First and foremost, our Elasticsearch cluster is spread across 3 data centers each with it’s own Munin instance. This makes collecting and aggregating even simple metrics like cluster wide load quite difficult. In addition, due to the polling nature of metrics collection with Munin we are limited to a somewhat corse resolution of 5 minutes. While this is good enough for looking at time series data over the course of a day or week it’s quite time-consuming to wait 5 or 10 minutes for graphs to update when deploying changes or testing performance optimizations.
We’ve already had some good experience instrumenting our PHP stack with StatsD and building dashboards with Grafana so reusing that infrastructure for Elasticsearch metrics seems like a good fit. Our fearless leader Barry suggested we tryout the Elasticsearch StatsD Plugin from Swoop Inc. however upon closer inspection we found that it does not deal with clustered Elasticsearch environments well and have yet to be updated to work with ES 1.x. So over the past week we’ve forked and rewritten much of it to suit our needs.
The Automattic Elasticsearch StatsD Plugin is designed to run on all nodes of a cluster and push metrics to StatsD on a configurable interval. Once installed and configured each node will send system metrics (e.g. CPU / JVM / network / etc.) about itself. Data nodes can also be configured to send metrics about the portion of the index stored on itself. Finally, the elected master of the cluster is responsible for sending aggregate cluster metrics about the index (documents / indexing operations / cache sizes / etc.). By default the plugin will send the total cluster aggregate as well as per index metrics for indices however granularity can be configured to report down to the individual shard level if so desired.
It’s been a year since I last made any major changes to my WordPress on Heroku build and in tech years that’s a lifetime. Since then Heroku has released a new PHP buildpack with nginx and HHVM built in. Much progress have also been made both HHVM and WordPress to make both compatible with each other. So it seems like now is as good a time as any to update the stack this site is running on.
So without further ado I like to introduce: Heroku WP — A template for HHVM powered WordPress served by nginx.
The Goal
There are numerous other templates out there for running WordPress on Heroku and my main goals for this templates are:
It should be simple — use the default buildpack provided by Heroku so there’s no other 3rd party dependency to implicitly trust or to maintain.
It should be fast — use the latest technologies available to squeeze every last ounce of performance out of each Heroku Dyno.
It should be secure — security is not an add-on, admin pages should be secure by default and database connections needs to be encrypted.
It should scale — just because we can serve millions of page hits a day off a single Heroku Dyno does not mean we’ll stop there. The template should be made with cloud architecture in mind so that the number of Dynos can scale up and down without breaking.
The Stack
Standing on the shoulder of giants I was able to use the latest Heroku buildpack and get WordPress running on:
NGINX — An event driven web server that was engineered for the modern day to replace Apache. This high performance web server is preferred by more top 1,000 sites then any other and it’s what’s used by the largest WordPress install out there, WordPress.com.
HHVM — HipHop Virtual Machine, a JIT (just in time) compiler developed by Facebook to run PHP scripts which when tested with WordPress showed up to a 2x improvement.
I have yet to run any statical analysis on performance however antidotally it feels a lot faster navigating WP admin and page generation times looks much better. I’m looking forward to running more tests and performance tuning this build in the coming weeks.
Update:
While still not a head-to-head test looking at the response times as reported by StatusCake for this site running on Heroku-WP and a mirror of this site that is running on the old Heroku LAMP stack with no load other then StatusCake pings shows a dramatic improvement:
