appbaseio/streams-preview

Streams

Streams is Appbase's Elasticsearch streaming architecture. It comprises features such as streaming of documents based on queries to TTL and timed notifications. All features aim to be compatible with Elasticsearch's API. Streams also offers HTTP basic authentication and SSL. Our setup is Dockerized and super easy to configure. Streams is compatible with Elasticsearch v2 and v5.

Stream on!

Table of Contents

• Overview
• Installation
• SSL configuration
• Elasticsearch API support
• Architecture & Deployment
• Sample use cases
• Dependencies
• Development
• Copyright

Overview

This is a transparent streaming layer on top of Elasticsearch. It supports streaming on documents and on queries.

Elasticsearch requests can be split into two categories:

1. Data ingestion, which takes the form of indexing or update operations
2. Data retrieval, which takes the form of searching or getting documents

For requests in category 1, Streams checks if there are any open channels, subscribers listening to queries; if there aren't, it shouldn't spend time doing much more than proxying the request to Elasticsearch.

If there are active streams, it checks if the document being indexed or updated matches a subset of them; if so, the document is sent to the matching channels or streams before being proxied to Elasticsearch. Elasticsearch can be bypassed by passing a query parameter store=false — store=true is the default.

Channels or streams themselves can be of two types:

1. Document streams, which apply only to a single document, and will receive notifications about updates and deletions. Updates are notified by a new field _updated in the returned document; deletes are notified by a new field _deleted.
2. Query streams, which apply to a query, and will receive any new document (or updates) matching the respective query, which is simply an Elasticsearch DSL query

Data retrieval is where a stream begins its lifecycle; a data retrieval request is to be a streaming request if it has a URL parameter stream=true — stream=false is the default. For a document stream, simply adding that parameter at the end of the URL will have the effect of making it a stream, that is, the request does not return any result immediately; only if changes happen to that document — that is, reindex, updates, deletion.

Streams on queries have a similar behavior, expect that it only applies new documents matching existing queries.

Streams timeout after 2 hours — this value can only be changed programatically for now.

Installation

To install Streams, run the following Docker command:

docker run -d --name streams -p 80:80 -p 443:443 \
  -e ES_NODES=${ES_NODES} \
  -e DOMAIN_NAME=${DOMAIN_NAME} \
  -e AUTH_USER=${AUTH_USER} -e AUTH_PASS=${AUTH_PASS} \
  -v ${SOME_DIR}/ssl

Environment variables are used to pass in essential arguments or enable/disable certain behaviors.

-e ES_NODES: Comma-separated list of Elasticsearch nodes, as in
172.18.0.3:9200,172.18.0.4:9200
-e DOMAIN_NAME: Domain name to setup in Streams
-e SSL_OFF: Disable SSL usage; not recommended for production
-e AUTH_OFF: disable HTTP basic authentatication
-e AUTH_USER: username for HTTP basic auth
-e AUTH_PASS: password for HTTP basic auth
-e DNS_SERVER: address for a name server

The volume mapped onto /ssl refers to SSL configuration, as described in SSL configuration.

SSL configuration

Streams allows for generic SSL certificate configuration; the directory /ssl in the container is expected to contain two files: CERTIFICATE.crt and PRIVATE_KEY.key; the only requirement is that the certificate name ends with crt and the key with key.

-v some_dir_with_pair:/ssl

some_dir_with_pair is some directory containing the certificate-key pair.

Elasticsearch APIs involved in Streams

Streams covers the following Elasticsearch APIs — in the sense that it captures requests falling under the respective API, but does not interfere with its expected behavior; if it does, it is a bug.

• Get
• Index
• Update
• Search

This version targets only Elasticsearch version 2.x and 5.x and has been tested only in 2.4.3 and 5.1.

Architecture & Deployment Scenarios

Below is a block diagram of Streams architecture.

                     .---------------------.
                     |      OpenResty      |
                     |                     |
                     |    .-----------.    |
                     |    |           |    |
       Req 1         |    |           |    |
     <------------------> |  streams  |    |
       Req 2 ?stream |    |  port 80  |    |
     -------------------> |   / 443   |    |
                     |    |           |    |
                     |    .-----------.    |     .-----------------.
                     |          .          |     |  Elasticsearch  |
                     |          |       <------> |    upstream     |
                     |          .          |     .-----------------.
                     |    .-----------.    |
                     |    |           |    |
       Req 2         |    |  pub/sub  |    |
     <------------------- | localhost |    |
                     |    | port 5678 |    |
                     |    |           |    |
                     |    .-----------.    |
                     |                     |
                     |                     |
                     .---------------------.

Here, Req 1 is a normal, no-streaming, request; it's proxied to Elasticsearch. Req 2, on the other hand, opens a streaming channel, keeping the request open for matching documents; it's proxied to a virtual server.

As a proxy to Elasticsearch, Streams can be deployed in front of each cluster node or any subset of them.

Sample use cases

To test Streams locally, use run_dev.sh to provision a local environment for tests. This will run a container for Streams itself, and one for Elasticsearch. Running run_dev.sh with no arguments will create an Elasticsearch v2 container; running run_dev.sh -2 will have the same effect, and run_dev.sh -5 will run an Elasticsearch v5 container.

Running Streams

shell
# Start Elasticsearch
$ docker run -d --name=es -p 9200:9200 elasticsearch:2.4

$ docker run -d --name=streams -p 80:80 streams -e ES_NODES=172.17.0.2:9200 -e SSL_OFF=true -e AUTH_OFF=true appbaseio/streams-preview:0.1

# Set the IP address; 'localhost' here
$ streams=localhost

Basic features & Streaming

shell
# Create a document and store
$ curl -sLk -XPUT "$streams/blog/post/1" -d '{"content":"Hello, world!"}'

# Get the newly created document
$ curl -sLk -XGET "$streams/blog/post/1"

# Delete the document
$ curl -sLk -XDELETE "$streams/blog/post/1"

# Create the document again, this time with POST
$ curl -sLk -XPOST "$streams/blog/post" -d '{"content":"Hello, world!"}'

# Register a query and listen
$ curl -sLk -XGET "$streams/blog/_search?stream=true" -d '{"query":{"match":{"content":"appbase"}}}'

# Post something matching that query
$ curl -sLk -XPOST "$streams/blog/post" -d '{"content":"appbase for the realtime web"}'

# Post something matching that query - don't store this time
$ curl -sLk -XPOST "$streams/blog/post?store=false" -d '{"content":"this is appbase Streams"}'

# Query all blog posts and see that previous post hasn't been stored
$ curl -sLk -XPOST "$streams/blog/post/_search?pretty" -d '{"query":{"match_all":{}}}'

# Post something else and see that it is not streamed
$ curl -sLk -XPOST "$streams/blog/post" -d '{"content":"another post"}'

# Now query all blog posts again to see which ones were stored
$ curl -sLk -XGET "$streams/blog/post/_search" -d '{"query":{"match_all":{}}}'

# Let's insert a new document
$ curl -sLk -XPOST "$streams/blog/post/123" -d '{"content":"a blog post"}'

# Now let's listen to changes in that document
$ curl -sLk -XGET "$streams/blog/post/123?stream=true"

# Now let's update that document and watch the streamed update
$ curl -sLk -XPOST "$streams/blog/post/123/_update" -d '{"doc":{"content":"a new blog post"}}'

# Now let's delete that document and watch the deletion report
$ curl -sLk -XDELETE "$streams/blog/post/123"

# Now get info about active streams (subscribers) and messages state (requires exposing port 5678 during docker run)
$ curl -sLk -XGET "$streams:5678/_streams/debug"

Time to Live (TTL)

TTL enables defining a Time to Live for indices; it works on both PUT and POST requests at index creation time. It's enabled with a URL query argument named ttl, whose values obey the format Xu, where X is an integer and u is one of s, m, h or d, for seconds, minutes, hours or days.

shell
# Issue a GET request on a non-existent index and check the 404 status
$ curl -sLk -XGET "$streams/someindex"

# Create the index with a 10-second TTL
$ curl -sLk -XPUT "$streams/someindex?ttl=10s"

# Check quickly that the index still exists
$ curl -sLk -XGET "$streams/someindex"

# After 10 seconds, a 404 is returned
$ curl -sLk -XGET "$streams/someindex"

Time Queries

Time queries work in a similar fashion to streaming queries, but they allow setting up two variables: interval and count. Interval defines when new documents should be streamed and count defines how many times the cycle is to be repeated. As with streaming queries, it's mandatory that the query refers only to fields that are present in the type mapping.

shell
# Index a document to make sure the mapping exists
$ curl -sLk -XPOST "$streams/blog/post" -d '{"content":"Hello, world!"}'

# Create a new time query which should deliver results 50 times, every 5 seconds; these results are documents containing "appbase" in the field "content"
$ curl -sLk -XPOST "$streams/blog/post/_timequery?interval=5s&count=50" -d '{"query":{"match":{"content":"appbase"}}}'

# The listening endpoint looks similar, except that the request is a GET and only the query is specified
$ curl -sLk -XGET "$streams/blog/post/_timequery" -d '{"query":{"match":{"content":"appbase"}}}'

# Now, any new document matching the query will be streamed to listeners every 5 seconds, 50 times
$ curl -sLk -XPOST "$streams/blog/post" -d '{"content":"hello appbase"}'

Copyright

© 2017 Appbase, Inc