Search This Blog


The next chapter

Every end is a new beginning

For the past 6 years I’ve been a part of the CM-Well development team. I’m writing this post with lot’s of mixed feelings. Working on CM-Well has been an awesome experience! I got a chance to work with so many amazing people. But now it’s time to move on, and I’m excited to start a new gig.

CM-Well - the early days

Last year we open-sourced CM-Well, and released the code on github. Doing so involved cleaning up the code, and getting rid off the commit history. Many team members who contributed a lot to the success of the project are not recognized. So, to get it out in the open, I opened the old repo, which has commits up until July last year, and I’m sharing some stats1.

$ git shortlog -sn --all
  2697  Gilad Hoch
  1015  yaakov
   954  michael
   914  israel
   739  Israel
   720  Vadim.Punski
   580  Gilad.Hoch
   450  Mark Zitnik
   310  Tzofia Shiftan
   234  Matan Keidar
   222  Mark.Zitnik
   170  Eli
   125  Yaakov Breuer
   102  Michael.Irzh
    95  Michael
    94  Israel Klein
    87  Tzofia.Shiftan
    58  Eli Orzitzer
    54  Michael Irzh
    44  Builder
    42  gilad
    31  DockerBuilder
    22  dudi
    14  Dudi
    14  Dudi Landau
    14  Yoni.Mataraso
    10  matan
     9  tserver
     8  Liorw
     5  Liya.Katz
     4  israel klein
     2  Shachar.Ben-Zeev
     2  Yoni Mataraso
     2  builder
     1  James Howlett
     1  Yaakov

These are just the commits in the old repo, not including any new commits in github. I also created visualizations using Gource for the old repo.

The first has files fading, and focuses on the contributors:

The second version gives an overview of the entire project:

The project is active, and has a lot of work invested in it, as you can see from the videos. But it doesn’t quite show how CM-Well is used internally. So I fetched some random access.log file from one of the servers, and wrote a little something to convert the log into a logstalgia acceptable format:

https://github.com/hochgi/logstalgia-access-log-converter

I took this opportunity to get to know mill build tool, and some libraries I wanted to experiment with. Long story short, I got 10 minutes of real CM-Well action on a single node (which is part of a cluster that has 20 web servers in it - so you’re only getting 1/20th of the action), and made a visualization using logstalgia with:

$ logstalgia -f -1280x720 --title "CM-Well access.log visualization"                \
    --screen 2 -p 0.2 -u 1 --background 75715e -x -g "meta,URI=/meta/.*$,10"        \
    -g "SPARQL,URI=/_sp?.*$,30" -g "_out,URI=/_out?.*$,10" -g "_in,URI=/_in?.*$,10" \
    -g "misc,URI=/.*,40" --paddle-mode single --paddle-position 0.75                \
    --disable-progress --font-size 25 --glow-duration 0.5 --glow-multiplier 2       \
    --glow-intensity 0.25 converted-access.log

And the output:

I gotta say, it came out pretty neat!2

Goodbye

I suck at goodbyes, so let me just say that I really loved working on CM-Well. It is a great project, and I hope to see it thrive. I will keep track of it, and plan to contribute occasionally on my spare time.


  1. In the early days, we used SVN, and we converted the repo to git at some point, which is why you see some duplicated names (that, and also we may have also committed from multiple users).↩︎

  2. Kinda get me into thinking I should write a logstash appender that streams real-time action directly into a logstalgia end point. It’s gotta be the coolest monitoring one can ask for…!↩︎

Unfolding streams like a boss (part 2)

Parallelizing resumable bulk consumes with CM-Well & akka-stream

In the previous post we introduced CM-Well’s consume API, and showed how it is tailored to be used with akka-http & akka-stream. This post will get into the gory details of how to squeeze the most out of CM-Well, using the bulk-consume API with akka-http & akka-stream.

The consume API in depth

There are two types of “consumes”, consume & bulk-consume. We will focus on the latter. But a few words on consume to not leave you hanging: consume just wraps a regular search, with a few extra filters. in terms of CM-Well’s qp, it translate to the following:

Given filters qp=$QP and an timestamp index-time=${ITIME:-0}, CM-Well generates the following (loosely) equivalent search parameters:

# set to current time minus 30 seconds
NOW=$(
  MILLIS=$(  date +%s%N | cut -b1-13 )
  calc $MILLIS - 30000
)

"?op=search
 &qp=system.indexTime>$ITIME,system.indexTime<$NOW,[$QP]
 &sort-by=system.indexTime
 &length=100"

It then fetches those (up to) 100 (by default) sorted results, and: if all results has the same index time SOME_ITIME, it will replace the previous op=search with op=stream, and previous qp=system.indexTime>$ITIME,system.indexTime<$NOW,[$QP] with qp=system.indexTime:$SOME_ITIME,[$QP]. else it will have multiple values, all sorted. it will drop all the tailing results with index time = $MAX_ITIME, and return the rest, with a token in the header setting the next $ITIME to be $MAX_ITIME - 1.

These are the basics, the are a few more concerns to take into consideration, and if thats interest you, go ahead and check the source code.

Understanding bulk-consume

In contrast to consume API, bulk-consume tries to be more efficient, and retrieve a lot more infotons per request à la stream style. Under the hood it uses Elasticsearch’s scroll API in a similar way to how we described stream API is made in the previous post. The problem is, that you can’t get sorted results with scroll from Elasticsearch. So, instead of advancing the timeline using sorted search, we filter results in advance.

This means there’s a pre-processing phase where we try to find a from and to timestamps, that are not “too far” apart, in terms of number of results, but enough to stream efficiently. CM-Well does it using a simple binary search to do so, and it tries to return a chunked response with O(1M) results (by default). There are many edge cases covered, like an early cutoff, if the binary search doesn’t converged fast enough, And dealing with near current time results, etc’…

Like consume, bulk-consume returns a position token in headers. In fact, the tokens are interchangeable between the 2 APIs, But those returned from a bulk-consume request, might contain some extra attributes. It turns out, that many pre-processing phases can be avoided if previous request stored an optional next “to” timestamp it might have encounterd during the binary search. So, what’s so great about the bulk-consume API? pipelined parallelization! You see, the next position token is given eagerly in the response headers, and a user can use it right away to fire up the next request. Since it will probably take some time to fetch all those O(1M) results, you could end up with as many parallel streams of data that you can handle.

But, you might ask: “What about failures? retrying?”, the answer is, that bulk consume also let’s you set upper bound timestamp explicitly. If your token was an optimized one, you can reuse it safely. If not, a new binary search might yield different time range, and you could end up with duplicate results, or worse, data gaps. To overcome this, you should supply a timestamp explicitly when retrying. But, what should you supply? Well, there’s another header for that. Other than X-CM-WELL-POSITION header, you also get X-CM-WELL-TO header, and the value is the upper bound timestamp found in the binary search. You should supply this timestamp using to-hint query parameter, and retry the bulk-consume request with it. Note that if the position token is optimized, to-hint will be ignored.

OK, got it. let’s write some code

As implied, we will show how to build an akka-stream Source of data from CM-Well, using unfoldFlow, Retry, and other cool constructs you can find on akka-stream & akka-stream-contrib libs.

The easy part (motivation)

Assuming we can somehow get:

type PositionToken = String
val initialPosition: PositionToken = ???
val consume: Flow[PositionToken,(PositionToken,ByteString),_] = ???

The work left is ridiculously easy thanks to unfoldFlow:

SourceGen.unfoldFlow(initialPosition)(consume)

And we’re done! OK… not really… It’s too simplified. unfoldFlow can’t unfold the next element until it gets the previous generated state. This means that all our fancy talk about pipelining parallelization isn’t being taken into consideration here. So let’s try and improve that. How ’bout:

val consume: Flow[PositionToken,(PositionToken,Source[ByteString,_]),_] = ???
SourceGen.unfoldFlow(initialPosition)(consume)
         .flatMapConcat(List.apply)

This is already much better. Each bulk-consume Source is being queried eagerly. But we still have a downside here… bulks are not evenly sized, and size is counted as the number of infotons in the bulk. Not their actual size… Moreover, we mentioned retries are supported using to-hint with X-CM-WELL-TO header’s value. So, if we are going to retry some streams, this means we need to buffer an entire chunk, and only emit once we know it is complete, so we don’t get duplicate results from retries. This implies a single bulk can get us “stuck” waiting for it. The 2 major problems are: * No matter how high we set up our parallelization factor, we could still end up back-pressuring our slow source (by slow, I mean that whatever the use-case, we must assume a fast consumer. e.g: flush to disk, which is much faster than our network calls). * Having $parallelization-factor × O(1M) all buffer into memory, makes our GC inefficient, due to objects kept in memory for long time. And also we cause our downstream to be in starvation until we “unstuck” the current bulk.

So, since bulk are not sorted according to timeline anyway, then no reason not to use merge instead of concat:

SourceGen.unfoldFlow(initialPosition)(consume)
         .flatMapMerge(identity)

Also, we will try to optimize even further. Our queries to bulk-consume are our “bottle-neck”. So, it is better to not pull in all the data with the bulk. let’s use a thin format, like tsv, which won’t return data itself, only a tuple consisting of infoton’s path, date, uuid, and indexTime. This way, we can at a later stage pull in data of small batches of infotons we got from the bulk consume. So our final higher level stream should either look like:

val consume: Flow[PositionToken,(PositionToken,Source[List[ByteString],_]),_] = ???
val addData: Flow[List[ByteString],ByteString,_] = ???
SourceGen.unfoldFlow(initialPosition)(consume)
         .flatMapMerge(identity)
         .mapConcat(identity)
         .via(addData)

where addData flow utilizes Balance to fan out and parallelize data fetching job, and then fan back in to construct a Flow shape which takes care of parallelization internally. Another option, is to use a simpler mapAsync(parallelizationFactor)(...) to get an easier way to parallelize the data fetching job. Or, it can look like:

val consume: Flow[PositionToken,(PositionToken,Source[List[ByteString],_]),_] = ???
val addData: Flow[List[ByteString],ByteString,_] = ???
SourceGen.unfoldFlow(initialPosition)(consume)
         .flatMapMerge(_.via(addData))

OK, armed with our goal in mind, let’s implement the consume flow: ### The detailed part Let’s try to break down & visualize the stream:

  • Retry:
    • bulk-consume ~> fold results into a single element
    • if #lines != X-CM-WELL-N header or any other failure:
      • retry with to-hint=${X-CM-WELL-TO}
    • else
      • emit results
  • Enrich with data:
    • batch paths
    • Balance vs. mapAsync parallelization of fetching (we’ll introduce both approaches)

Wrapping up

This might be covered in a future post. Currently, It has been sitting untouched for too long, and I’m “flushing” it. For now, implementation details are left as an excersize for the reader ;)