Showing posts with label ES. Show all posts
Showing posts with label ES. Show all posts

Tuesday, November 22, 2016

Elasticsearch pipeline aggreagtions - monitoring used capacity

Lets say I want to setup a simple monitoring system for my desktop. The desktop uses LVM and has three volumes v1, v2 and v3, all belonging to vg1 volume group. I would like to monitor used capacity of these volumes, and the whole system, over time. It's easy to write a script that samples used capacity of the volumes and pushes it to ElasticSearch. All I need to store is:

{
  "name": "v1",
  "ts": 1479762877,
  "used_capacity": 1288404287488
}

OK, so I've put the script into cron to run every 5 minutes and the data starts pouring in. Lets do some BI on it! First thing to find out is how full my desktop is, i.e. the total capacity of all volumes. Sounds like a easy job for Kibana, isn't it? Well, not really.

Part 1: Naive failure

Let's say each of my volumes is ~1TB full. Trying to chart area viz in Kibana with Average aggregation over used_capacity returns useless results (click the below image to enlarge):

The real total system capacity is ~3TB, but Kibana, rightfully, shows that AVG(v1, v2, v3) => AVG(1TB, 1TB, 1TB) => 1TB. So may be I need Sum? Not good either:

I got ~17TB capacity number which not even close to reality. This happens because Kibana uses simple Date Histogram with nested Sum aggregation, i.e.

  • Divide selected date range into ts buckets. 30 minutes in my example.
  • Calculate Sum of used_capacity values of all documents that fall in bucket.
That's why the larger is the bucket, the more weird the results would look.

This happens because Kibana is only capable of either: $$ \underbrace{\text{SUM}\left(\begin{array}{c}v1, v1, v1,...\\ v2, v2, v2,...\\ v3, v3, v3,...\\ \end{array}\right)}_{ts\ bucket} \quad\text{or}\quad \underbrace{\text{AVG}\left(\begin{array}{c}v1, v1, v1,...\\ v2, v2, v2,...\\ v3, v3, v3,...\\ \end{array}\right)}_{ts\ bucket} $$ While what I need is: $$ \underbrace{\text{SUM}\left(\begin{array}{c}\text{AVG}(v1, v1, v1,...)\\ \text{AVG}(v2, v2, v2,...)\\ \text{AVG}(v3, v3, v3,...)\\ \end{array}\right)}_{ts\ bucket} $$ So how to achieve this?

Part 2: Poor man's solution

The post title promised pipeline aggregations and I'll get there. The problem with pipeline aggregations is that they are not supported in Kibana. So, is there still a way to get along with Kibana? - sort of. I can leverage on the fact that my sampling script takes capacity values of all volumes at exactly the same time, i.e. each bunch of volume metrics is pushed to ES with the same ts value. Now, if I force Kibana to use ts bucket length of 1 minute, I can guarantee that in any given bucket, I will only have documents belonging to a single sample batch (that's because I send measurements to ES every 5 minutes, which is much larger than the 1m bucket size).

One can argue that it generates LOTS of buckets - and he is right, but there is one optimization point to consider. ES Date histogram aggregation supports automatic pruning of buckets that do not have a minimum number of documents. The default is 0, which means empty buckets are returned, but Kibana wisely sets it to 1. Now lets say I want to see capacity data chart for last 7 days, which is 7*24*60=10080 points (buckets); however since I take measurements only every 5 minutes, most of the buckets will be pruned and we are left only with 2000, which is fare enough for Full HD screen. The nice side-effect of this is that it forces Kibana to draw really smooth charts :) Let's see it in action:

The above graph shows capacity data for last 7 days. The key point is to open and Advanced section of X-Axis dialog and put {"interval": "1m"} in JSON Input field - this overrides Kibana's automatic interval. The bottom legend, that says "ts per 3 hours", is lying, but it's the least of evils. Also note how smooth is the graph line.

Part 3: Pipeline aggregations!

The above solution works, but does not scale well beyond a single system - getting measurements from multiple systems at exactly the same time is tricky. Another drawback is that trying to looks at several months of data will result in tens of thousands of buckets which will burden both on ES, on the network and Kibana.

The right solution is to implement the correct formula. I need something like this:

SELECT AVG(used_capacity), ts FROM
    (SELECT SUM(used_capacity) AS used_capacity, DATE(ts) AS ts FROM capacity_history GROUP BY DATE(ts), name)
GROUP BY ts

Elasticsearch supports this since version 2.0 with Pipeline aggregations:

GET capacity_history/_search
{
  "size": 0,
  "aggs": {
    "ts": {
      "date_histogram": {"interval": "1h", "field": "ts"},
      "aggs": {
        "vols": {
          "terms": {"field": "name.raw", "size": 0},
          "aggs": {
            "cap": {
              "avg": {"field": "logical_capacity"}
            }
          }
        },
        "total_cap": {
          "sum_bucket": {
            "buckets_path": "vols>cap"
}}}}}}

Response

  "aggregations": {
    "ts": {
      "buckets": [
        {
          "key_as_string": "1479600000",
          "key": 1479600000000,
          "doc_count": 36,
          "vols": {
            "doc_count_error_upper_bound": 0,
            "sum_other_doc_count": 0,
            "buckets": [
              {
                "key": "v1",
                "doc_count": 12,
                "cap": {
                  "value": 1073741824000
                }
              },
              {
                "key": "v2",
                "doc_count": 12,
                "cap": {
                  "value": 1073741824000
                }
              },
              {
                "key": "v3",
                "doc_count": 12,
                "cap": {
                  "value": 1072459894784
                }
              }
            ]
          },
          "total_cap": {
            "value": 3219943542784
          }
        },
        ...
Since we only need ts bucket key and value of total_cap aggregation, we can ask ES to filter the results to include only the data we need. In case we have lots of volumes it can reduce the amount of returned data by orders of magnitude!
GET capacity_history/_search?filter_path=aggregations.ts.buckets.key,aggregations.ts.buckets.total_cap.value,took,_shards,timed_out
...
{
  "took": 92,
  "timed_out": false,
  "_shards": {
    "total": 70,
    "successful": 70,
    "failed": 0
  },
  "aggregations": {
    "ts": {
      "buckets": [
        {
          "key": 1479600000000,
          "total_cap": {
            "value": 3219943542784
          }
        },
        {
          "key": 1479603600000,
          "total_cap": {
            "value": 3220228083712
          }
        },
        ...
NOTE: I suggest always to return meta timed_out and _shards fields to make sure you do not get partial data.

This method is generic and will work regardless of time alignment of the samples; bucket size can be adjusted to return a same amount of data points. The major drawback is that it is not supported by stock Kibana and thus you will need your own custom framework to visualize this.

Tuesday, April 5, 2016

Persistent storage for ElasticSearch master nodes?

ElasticSearch master nodes hold cluster state. I was trying to understand whether these nodes are required to have persistent storage or they can recover from whatever exists on data nodes? The short answer is: probably yes.

Update

You better have persistent disk for your ES data nodes - read here why.

Below I describe tests what I've done. But before that - some background on how did I get to this question at first place.

ElasticSearch on Kubernetes

We are working on running ElasticSearch 2.x on Kubernetes on Google Container Engine. There are two options to store a data for a container:
EmptyDir
Part of the local storage on Kubernetes node is allocated for the pod. If pod's controller restarts, the data survives. If pod is killed - the data is lost.
gcePersistentDisk
Compute engine persistent disk (created in advance) can be attached to a pod. The data persists. However there is a limitation - as of Kubernetes 1.2, a ReplicaSet can not attach a different disks to each pods that it creates, thus to run ElasticSearch data nodes, for example, you need to create a separate ReplicaSet (with size of 1) for each ES data node.

ES data nodes should have persistent disk - this is no brainer. However with regards to ES master nodes it's not clear. I've tried to understand where master nodes persist cluster state, and this thread states "on every node including client nodes". There is also a resolved issue about storing index metadata on data nodes.

Run, Kill, Repeat

So lets how it behaves in reality.

I created two node Kubernetes 1.2 cluster running n1-standard-2 instances (2 CPUs, 7.5GB RAM). And used Paulo Pires Kubernetes setup for ElasticSearch:

$ git clone https://github.com/pires/kubernetes-elasticsearch-cluster.git
$ cd kubernetes-elasticsearch-cluster
$ vim es-data-rc.yaml  # set replicas to 2
$ vim es-master-rc.yaml  # set replicas to 3
$ vim es-svc.yaml  # set type to ClusterIP

Lets launch it in the air:

$ for i in *.yaml; do kubectl create -f $i; done
$ sleep 1m; kubectl get pods
NAME              READY     STATUS    RESTARTS   AGE
es-client-ats2b   1/1       Running   0          1h
es-data-teodq     1/1       Running   0          1h
es-data-zwml2     1/1       Running   0          1h
es-master-3bosq   1/1       Running   0          1h
es-master-a47om   1/1       Running   0          1h
es-master-c1dy1   1/1       Running   0          1h

We are all good. Lets ingest some data and alter cluster settings:

$ CLIENTIP=$(kubectl describe pods es-client |grep '^IP' |head -n 1|awk '{print $2}')
$ curl -XPUT $CLIENTIP:9200/_cluster/settings?pretty -d '{"transient": {"discovery.zen.minimum_master_nodes": 2}}'
{
  "acknowledged" : true,
  "persistent" : { },
  "transient" : {
    "discovery" : {
      "zen" : {
        "minimum_master_nodes" : "2"
      }
    }
  }
}
$ curl -XPUT $CLIENTIP:9200/tweets/tweet/1 -d '{"foo": "bar"}'          
{
  "_id": "1",
  "_index": "tweets",
  "_shards": {
    "failed": 0,
    "successful": 2,
    "total": 2
  },
  "_type": "tweet",
  "_version": 1,
  "created": true
}
$ curl $CLIENTIP:9200/_cluster/health?pretty
{
  "cluster_name" : "myesdb",
  "status" : "green",
  "timed_out" : false,
  "number_of_nodes" : 6,
  "number_of_data_nodes" : 2,
  "active_primary_shards" : 0,
  "active_shards" : 0,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 0,
  "delayed_unassigned_shards" : 0,
  "number_of_pending_tasks" : 0,
  "number_of_in_flight_fetch" : 0,
  "task_max_waiting_in_queue_millis" : 0,
  "active_shards_percent_as_number" : 100.0
}

The data is there and the cluster is green. Now lets kill all of the masters and recreate them (their data disk will be lost):

$ kubectl delete -f es-master-rc.yaml 
replicationcontroller "es-master" deleted
$ kubectl create -f es-master-rc.yaml       
replicationcontroller "es-master" created

After dozens of seconds, new masters will be up again and we'll see the following in the leader's log:

[2016-04-04 15:57:44,880][INFO ][cluster.service          ] [Elaine Grey] new_master {Elaine Grey}{5NIL5jBbTYadefGzjDLb5A}{10.224.1.7}{10.224.1.7:9300}{data=false, master=true}, added {{Lyja}{CUyGl7w-R86qcOsNSj0xPA}{10.224.1.4}{10.224.1.4:9300}{master=false},{Typhoid Mary}{cWmlEtHuSdCImjHMNM6FsA}{10.224.1.3}{10.224.1.3:9300}{master=false},{Slug}{fQPe2C1FSH2UkuveBFJtbw}{10.224.1.5}{10.224.1.5:9300}{data=false, master=false},}, reason: zen-disco-join(elected_as_master, [0] joins received)
[2016-04-04 15:57:45,056][INFO ][node                     ] [Elaine Grey] started
[2016-04-04 15:57:45,058][INFO ][cluster.service          ] [Elaine Grey] added {{Stacy X}{U4sn1pkWRlGV-zVMW1OeAA}{10.224.1.8}{10.224.1.8:9300}{data=false, master=true},}, reason: zen-disco-join(pending joins after accumulation stop [election closed])
[2016-04-04 15:57:45,711][INFO ][gateway                  ] [Elaine Grey] recovered [0] indices into cluster_state
[2016-04-04 15:57:45,712][INFO ][cluster.service          ] [Elaine Grey] added {{Kiss}{MaNkKlQWR82QHFVhz38Ohg}{10.224.1.6}{10.224.1.6:9300}{data=false, master=true},}, reason: zen-disco-join(join from node[{Kiss}{MaNkKlQWR82QHFVhz38Ohg}{10.224.1.6}{10.224.1.6:9300}{data=false, master=true}])
[2016-04-04 15:57:46,077][INFO ][gateway                  ] [Elaine Grey] auto importing dangled indices [tweets/OPEN] from [{Lyja}{CUyGl7w-R86qcOsNSj0xPA}{10.224.1.4}{10.224.1.4:9300}{master=false}]
[2016-04-04 15:57:47,073][INFO ][cluster.routing.allocation] [Elaine Grey] Cluster health status changed from [RED] to [YELLOW] (reason: [shards started [[tweets][3]] ...]).
[2016-04-04 15:57:47,567][INFO ][cluster.routing.allocation] [Elaine Grey] Cluster health status changed from [YELLOW] to [GREEN] (reason: [shards started [[tweets][3]] ...]).
[2016-04-04 15:58:16,355][INFO ][io.fabric8.elasticsearch.discovery.kubernetes.KubernetesDiscovery] [Elaine Grey] updating discovery.zen.minimum_master_nodes from [-1] to [2]

So we see that:

  • The new master recovered 0 indices from cluster state - i.e. the cluster state was indeed lost.
  • The new master auto imported existing indices, which were "dangling" in ES terminology.
  • It also restored our transient quorum setting

And our cluster is green:


$ curl $CLIENTIP:9200/_cluster/health?pretty
{
  "cluster_name" : "myesdb",
  "status" : "green",
  "timed_out" : false,
  "number_of_nodes" : 6,
  "number_of_data_nodes" : 2,
  "active_primary_shards" : 5,
  "active_shards" : 10,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 0,
  "delayed_unassigned_shards" : 0,
  "number_of_pending_tasks" : 0,
  "number_of_in_flight_fetch" : 0,
  "task_max_waiting_in_queue_millis" : 0,
  "active_shards_percent_as_number" : 100.0
}

Now lets kill master without destroying their disks and see it behaves any differently:

$ for pod in $(kubectl get pods |grep es-master |awk '{print $1}'); do do kubectl exec $pod killall java & done
The master log shows the following:
[2016-04-04 16:33:55,695][INFO ][cluster.service          ] [Neptune] new_master {Neptune}{51QU1wf4T2Ky9QUzK5MkEw}{10.224.1.7}{10.224.1.7:9300}{data=false, master=true}, added {{Slug}{fQPe2C1FSH2UkuveBFJtbw}{10.224.1.5}{10.224.1.5:9300}{data=false, master=false},{Typhoid Mary}{cWmlEtHuSdCImjHMNM6FsA}{10.224.1.3}{10.224.1.3:9300}{master=false},{Lyja}{CUyGl7w-R86qcOsNSj0xPA}{10.224.1.4}{10.224.1.4:9300}{master=false},{Comet Man}{cOLRxsOKTC2OYR6Kuiplxw}{10.224.1.6}{10.224.1.6:9300}{data=false, master=true},}, reason: zen-disco-join(elected_as_master, [1] joins received)
[2016-04-04 16:33:55,867][INFO ][node                     ] [Neptune] started
[2016-04-04 16:33:56,428][INFO ][gateway                  ] [Neptune] recovered [1] indices into cluster_state
[2016-04-04 16:33:57,233][INFO ][cluster.routing.allocation] [Neptune] Cluster health status changed from [RED] to [YELLOW] (reason: [shards started [[tweets][0], [tweets][0]] ...]).
[2016-04-04 16:33:57,745][INFO ][cluster.routing.allocation] [Neptune] Cluster health status changed from [YELLOW] to [GREEN] (reason: [shards started [[tweets][4]] ...]).
[2016-04-04 16:34:01,417][INFO ][cluster.service          ] [Neptune] added {{Nicole St. Croix}{5VcQT4H8RHeAf3R0PW3K4A}{10.224.1.8}{10.224.1.8:9300}{data=false, master=true},}, reason: zen-disco-join(join from node[{Nicole St. Croix}{5VcQT4H8RHeAf3R0PW3K4A}{10.224.1.8}{10.224.1.8:9300}{data=false, master=true}])
[2016-04-04 17:01:55,811][INFO ][io.fabric8.elasticsearch.discovery.kubernetes.KubernetesDiscovery] [Neptune] updating discovery.zen.minimum_master_nodes from [-1] to [2]

So 1 index was recovered from the cluster state and there are no dangling indices this time.

Conclusions

While master nodes were able to recover both indices and cluster transient settings, I was testing only the most simple scenario. This is not enough to take a decision on whether we should maintain persistent disks for master nodes. On the other hand, if we do have persistent disks for master - do we need to backup the metadata? And what about ES 5.x? - one of the promised features is that master will hold an ID of the latest index change to prevent stale data nodes becoming primaries during network partitioning. This kind of metadata can not be stored on data nodes.

I'll update this post when I'll have the answers.

Friday, March 25, 2016

Caveat with ElasticSearch nGram tokenizer

Finally got some time to blog about ElasticSearch. I use it extensively during the last two years, but my findings are rather lengthy. Finally I've got something small to share.

ElasticSearch nGram tokernizer is very useful for efficient substring matching (at cost of index size of course). For example, I have an event message field like this

/dev/sda1 has failed due to ...
and I would like to find all events of failure for all SCSI disks. One option is to store message field as not analyzed string (i.e. a one single term) and use wildcard query:
GET /events
{
  "query": {
    "wildcard": {
      "message.raw": {
        "value": "/dev/sd?? has failed*"
      }
    }
  }
}
This will do the work perfectly, but to complete it, ElasticSearch will scan every value of message field looking for the pattern during search time. Once number of documents gets big enough, it will become slow.

One solution is to split message to substrings during indexing time, with (2,20) for (min, max) in our example:

# Analyzer definition in settings
"analysis": {
    "analyzer": {
        "substrings": {
            "tokenizer": "standard",
            "filter": ["lowercase", "thengram"]
        }   
    },  
    "filter": {
        "thengram": {
            "type": "nGram",
            "min_gram": 2,
            "max_gram": 20
        }   
    }   
} 

# message field definition in mappings
"message": {
    "type": "string",
    "index": "analyzed",
    "analyzer": "substrings"
}
and use match_phrase query:
GET events/_search
{
  "query": {
    "match": {
      "message": {
        "query": "dev sd has failed",
        "type": "phrase",
      }
    }
  }
}

The caveat

The above query will return weirdly unrelevant results and, at first glance, it's not obvious why. The caveat is, that our custom analyzer is applied both during indexing and search. So instead of searching for sequence of terms "dev", "sd", "has", "failed"; we are searching for sequence "de", "ev", "dev", "sd", "ha", "as", "has", etc. To fix this we need to tell Elastic to use different tokernizer during search (and search only). This can be done either by adding "analyzer": "standard" to query itself (which is error phone, since can be easily forgotten) or specified in mapping definition:
"message": {
    "type": "string",
    "index": "analyzed",
    "analyzer": "substrings",
    "search_analyzer": "standard"
}

Worth it?

I took 1,000,000 events sample data and run both wildcard and phrase queries that match 1,000 doc subset out of it. While for such a small data set, both are fast, the difference it quite striking nevertheless:
  • wildcard query - 30ms
  • phrase query - 5ms

Times 6 speed up! Another bonus for using phrase query is that you can get results highlighting (that is not supported for wildcard queries).