{"id":33421,"date":"2020-03-17T13:47:59","date_gmt":"2020-03-17T08:17:59","guid":{"rendered":"https:\/\/www.wikitechy.com\/technology\/?p=33421"},"modified":"2020-03-17T13:47:59","modified_gmt":"2020-03-17T08:17:59","slug":"how-to-migrate-from-elasticsearch-1-7-to-6-8-with-zero-downtime","status":"publish","type":"post","link":"https:\/\/www.wikitechy.com\/technology\/how-to-migrate-from-elasticsearch-1-7-to-6-8-with-zero-downtime\/","title":{"rendered":"How to migrate from Elasticsearch 1.7 to 6.8 with zero downtime"},"content":{"rendered":"

My last task at BigPanda was to upgrade an existing service that was using Elasticsearch version 1.7 to a more moderen Elasticsearch version, 6.8.1.<\/p>\n

In this post, i will be able to share how we migrated from Elasticsearch 1.6 to 6.8 with harsh constraints like zero downtime, no data loss, and 0 bugs. I’ll also provide you with a script that does the migration for you.<\/p>\n

This post contains 6 chapters (and one is optional):<\/p>\n

What\u2019s in it for me?<\/strong> –> What were the new features that led us to upgrade our version?<\/p>\n

The constraints<\/strong> –> What were our business requirements?<\/p>\n

Problem solving<\/strong> –> How did we address the constraints?<\/p>\n

Moving forward<\/strong> –> The plan.<\/p>\n[Optional chapter] –> How did we handle the infamous mapping explosion problem?<\/p>\n

Finally<\/span><\/strong> –> the way to do data migration between clusters.<\/p>\n

Chapter 1 \u2014 What\u2019s in it for me?<\/h2>\n

What benefits were we expecting to unravel by upgrading our data store?<\/p>\n

There were a few of reasons:<\/p>\n

    \n
  • Performance and stability issues<\/strong> \u2014 We were experiencing an enormous number of outages with long MTTR that caused us tons of headaches. This was reflected in frequent high latencies, high CPU usage, and more issues.<\/li>\n
  • Non-existent support in old Elasticsearch versions<\/strong> \u2014 We were missing some operative knowledge in Elasticsearch, and once we looked for outside consulting we were encouraged to migrate forward to receive support.<\/li>\n
  • Dynamic mappings in our schema<\/strong> \u2014 Our current schema in Elasticsearch 1.7 used a feature called dynamic mappings that made our cluster explode multiple times. So we wanted to deal with this issue.<\/li>\n
  • Poor visibility on our existing cluster<\/strong> \u2014 We wanted a far better view under the hood and saw that later versions had great metrics exporting tools.<\/li>\n<\/ul>\n

    Chapter 2 \u2014 The constraints<\/h2>\n
      \n
    • ZERO downtime migration<\/strong> \u2014 we’ve active users on our system, and that we couldn’t afford for the system to be down while we were migrating.<\/li>\n
    • Recovery plan<\/strong> \u2014 We couldn’t afford to \u201close\u201d or \u201ccorrupt\u201d data, regardless of the value . So we would have liked to organize a recovery plan just in case our migration failed.<\/li>\n
    • Zero bugs<\/strong> \u2014 We couldn’t change existing search functionality for end-users.<\/li>\n<\/ul>\n

       <\/p>\n

      Chapter 3 \u2014 Problem solving and thinking of a plan<\/h2>\n

      Let\u2019s tackle the constraints from the only to the foremost difficult:<\/p>\n

      Zero bugs<\/h3>\n

      In order to deal with this requirement, I studied all the possible requests the service receives and what its outputs were. Then I added unit-tests where needed.<\/p>\n

      In addition, I added multiple metrics (to the\u00a0Elasticsearch<\/strong>\u00a0Indexer<\/strong> and therefore the new Elasticsearch Indexer<\/strong> ) to trace latency, throughput, and performance, which allowed me to validate that we only improved them.<\/p>\n

      Recovery plan<\/h3>\n

      This means that I needed to deal with the subsequent situation: I deployed the new code to production and stuff wasn’t working needless to say . What am i able to do about it then<\/p>\n

      Since i used to be working during a service that used event-sourcing, I could add another listener (diagram attached below) and begin writing to a replacement Elasticsearch cluster without affecting production status<\/p>\n

      Zero downtime migration<\/h3>\n

      The current service is in live mode and can’t be \u201cdeactivated\u201d for periods longer than 5\u201310 minutes. The trick to getting this right is this:<\/p>\n

        \n
      • Store a log of all the actions your service is handling (we use Kafka in production)<\/li>\n
      • Start the migration process offline (and keep track of the offset before you started the migration)<\/li>\n
      • When the migration ends, start the new service against the log with the recorded offset and catch up the lag<\/li>\n
      • When the lag finishes, change your frontend to question against the new service and you’re done<\/li>\n<\/ul>\n

        Chapter 4 \u2014 The plan<\/h2>\n

        Our current service uses the subsequent architecture (based on message passing in Kafka):<\/p>\n

        \"\"<\/p>\n

         <\/p>\n

          \n
        • Event topic<\/strong> contains events produced by other applications (for example, UserId 3 created<\/strong>)<\/li>\n
        • Command topic<\/strong> contains the interpretation of those events into specific commands employed by this application (for example: Add userId 3<\/strong>)<\/li>\n
        • Elasticsearch 1.7 \u2014 The datastore of the command Topic<\/strong> read by the Elasticsearch Indexer.<\/li>\n<\/ul>\n

          We planned to feature another consumer (new Elasticsearch Indexer)<\/strong> to the command topic<\/strong>, which can read an equivalent exact messages and write them in parallel to Elasticsearch 6.8.<\/p>\n

          \"\"<\/p>\n

           <\/p>\n

          Where should I start?<\/strong><\/h3>\n

          To be honest, I considered myself a newbie Elasticsearch user. To feel confident to perform this task, I had to believe the simplest thanks to approach this subject and learn it. a couple of things that helped were:<\/p>\n

            \n
          • Documentation<\/strong> \u2014 It\u2019s an insanely useful resource for everything Elasticsearch. Take the time to read it and take notes (don\u2019t miss: Mapping and QueryDsl).<\/li>\n
          • HTTP API<\/strong> \u2014 everything under CAT API. This was super useful to debug things locally and see how Elasticsearch responds (don\u2019t miss: cluster health, cat indices, search, delete index).<\/li>\n
          • Metrics (\u2764\ufe0f)<\/strong> \u2014 From the primary day, we configured a shiny new dashboard with many cool metrics (taken from elasticsearch-exporter-for-Prometheus) that helped and pushed us to know more about Elasticsearch.<\/li>\n<\/ul>\n

            The code<\/h2>\n

            Our codebase was employing a library called elastic4s<\/span> and was using the oldest release available within the library \u2014 a very good reason to migrate! therefore the very first thing to try to to was just to migrate versions and see what broke.<\/p>\n

            There are a couple of tactics on the way to do that code migration. The tactic we chose was to undertake and restore existing functionality first within the new Elasticsearch version without re-writing the all code from the beginning . In other words, to succeed in existing functionality but on a more moderen version of Elasticsearch.<\/p>\n

            Luckily for us, the code already contained almost full testing coverage so our task was much much simpler, which took around 2 weeks of development time.<\/p>\n

            \"\"<\/p>\n

            It’s important to note that, if that wasn’t the case, we would have had to invest some time in filling that coverage up. Only then would we be able to migrate since one of our constraints was to not break existing functionality.<\/em><\/p>\n

            Chapter 5 \u2014 The mapping explosion problem<\/h2>\n

            Let\u2019s describe our use-case in more detail. This is our model:<\/p>\n

            <\/div> 
            class InsertMessageCommand(tags: Map[String,String])<\/code><\/pre> <\/div>\n
            And for example, an instance of this message would be:<\/p>\n
             
             <\/div> 
            new InsertMessageCommand(Map("name"->"dor","lastName"->"sever"))<\/code><\/pre> <\/div>\n
            And given this model, we needed to support the following query requirements:<\/p>\n
              \n
            1. Query by value<\/li>\n
            2. Query by tag name and value<\/li>\n<\/ol>\n
              The way this was modeled in our Elasticsearch 1.7 schema was using a dynamic template schema (since the tag keys are dynamic, and cannot be modeled in advanced).<\/p>\n
              The dynamic template caused us multiple outages due to the mapping explosion problem, and the schema looked like this:<\/p>\n
               
               <\/div> 
              curl -X PUT "localhost:9200\/_template\/my_template?pretty" -H 'Content-Type: application\/json' -d '{    "index_patterns": [        "your-index-names*"    ],    "mappings": {            "_doc": {                "dynamic_templates": [                    {                        "tags": {                            "mapping": {                                "type": "text"                            },                            "path_match": "actions.tags.*"                        }                    }                ]            }        },    "aliases": {}}'  curl -X PUT "localhost:9200\/your-index-names-1\/_doc\/1?pretty" -H 'Content-Type: application\/json' -d'{  "actions": {    "tags" : {        "name": "John",        "lname" : "Smith"    }  }}'curl -X PUT "localhost:9200\/your-index-names-1\/_doc\/2?pretty" -H 'Content-Type: application\/json' -d'{  "actions": {    "tags" : {        "name": "Dor",        "lname" : "Sever"  }}}'curl -X PUT "localhost:9200\/your-index-names-1\/_doc\/3?pretty" -H 'Content-Type: application\/json' -d'{  "actions": {    "tags" : {        "name": "AnotherName",        "lname" : "AnotherLastName"  }}}'<\/code><\/pre> <\/div>\n
               
               <\/div> 
              curl -X GET "localhost:9200\/_search?pretty" -H 'Content-Type: application\/json' -d'{    "query": {        "match" : {            "actions.tags.name" : {                "query" : "John"            }        }    }}'# returns 1 match(doc 1)curl -X GET "localhost:9200\/_search?pretty" -H 'Content-Type: application\/json' -d'{    "query": {        "match" : {            "actions.tags.lname" : {                "query" : "John"            }        }    }}'# returns zero matches# search by valuecurl -X GET "localhost:9200\/_search?pretty" -H 'Content-Type: application\/json' -d'{    "query": {        "query_string" : {            "fields": ["actions.tags.*" ],            "query" : "Dor"        }    }}'<\/code><\/pre> <\/div>\n
              Nested documents solution<\/h2>\n
              Our first instinct in solving the mapping explosion problem was to use nested documents.<\/p>\n
              We read the nested data type tutorial within the Elastic docs and defined the subsequent schema and queries:<\/p>\n
               
               <\/div> 
              curl -X PUT "localhost:9200\/my_index?pretty" -H 'Content-Type: application\/json' -d'{        "mappings": {            "_doc": {            "properties": {            "tags": {                "type": "nested"                 }                            }        }        }}'curl -X PUT "localhost:9200\/my_index\/_doc\/1?pretty" -H 'Content-Type: application\/json' -d'{  "tags" : [    {      "key" : "John",      "value" :  "Smith"    },    {      "key" : "Alice",      "value" :  "White"    }  ]}'# Query by tag key and valuecurl -X GET "localhost:9200\/my_index\/_search?pretty" -H 'Content-Type: application\/json' -d'{  "query": {    "nested": {      "path": "tags",      "query": {        "bool": {          "must": [            { "match": { "tags.key": "Alice" }},            { "match": { "tags.value":  "White" }}           ]        }      }    }  }}'# Returns 1 documentcurl -X GET "localhost:9200\/my_index\/_search?pretty" -H 'Content-Type: application\/json' -d'{  "query": {    "nested": {      "path": "tags",      "query": {        "bool": {          "must": [            { "match": { "tags.value":  "Smith" }}           ]        }      }    }  }}'# Query by tag value# Returns 1 result<\/code><\/pre> <\/div>\n
              And this solution worked. However, once we tried to insert real customer data we saw that the amount of documents in our index increased by around 500 times.<\/p>\n
              We considered the subsequent problems and went on to seek out a far better solution:<\/p>\n
              The amount of documents we had in our cluster was around 500 million documents. This meant that, with the new schema, we were getting to reach 2 hundred fifty billion documents (that\u2019s 250,000,000,000 documents \ud83d\ude31).<\/p>\n
              We read this specialized blog post \u2014 https:\/\/blog.gojekengineering.com\/elasticsearch-the-trouble-with-nested-documents-e97b33b46194 which highlights that nested documents can cause high latency in queries and heap usage problems.<\/p>\n
              Testing \u2014 Since we were converting 1 document within the old cluster to an unknown number of documents within the new cluster, it might be much harder to trace if the migration process worked with none data loss. If our conversion was 1:1, we could assert that the count within the old cluster equalled the count within the new cluster.<\/p>\n
              Avoiding nested documents<\/h2>\n
              The real trick during this was to specialise in what supported queries we were running: search by tag value, and search by tag key and value.<\/p>\n
              The first query doesn’t require nested documents since it works on one field. For the latter, we did the subsequent trick. We created a field that contains the mixture of the key and therefore the value. Whenever a user queries on a key, value match, we translate their request to the corresponding text and query against that field.<\/p>\n
              Example:<\/p>\n
               
               <\/div> 
              curl -X PUT "localhost:9200\/my_index_2?pretty" -H 'Content-Type: application\/json' -d'{    "mappings": {        "_doc": {            "properties": {                "tags": {                    "type": "object",                    "properties": {                        "keyToValue": {                            "type": "keyword"                        },                        "value": {                            "type": "keyword"                        }                    }                }            }        }    }}'curl -X PUT "localhost:9200\/my_index_2\/_doc\/1?pretty" -H 'Content-Type: application\/json' -d'{  "tags" : [    {      "keyToValue" : "John:Smith",      "value" : "Smith"    },    {      "keyToValue" : "Alice:White",      "value" : "White"    }  ]}'# Query by key,value# User queries for key: Alice, and value : White , we then query elastic with this query:curl -X GET "localhost:9200\/my_index_2\/_search?pretty" -H 'Content-Type: application\/json' -d'{  "query": {        "bool": {          "must": [ { "match": { "tags.keyToValue": "Alice:White" }}]  }}}'# Query by value onlycurl -X GET "localhost:9200\/my_index_2\/_search?pretty" -H 'Content-Type: application\/json' -d'{  "query": {        "bool": {          "must": [ { "match": { "tags.value": "White" }}]  }}}'<\/code><\/pre> <\/div>\n
              Chapter 6 \u2014 The migration process<\/h2>\n
              We planned to migrate about 500 million documents with zero downtime. to try to to that we needed:<\/p>\n
                \n
              • A strategy on the way to transfer data from the old Elastic to the new Elasticsearch<\/li>\n
              • A strategy on the way to close the lag between the beginning of the migration and therefore the end of it<\/li>\n<\/ul>\n
                And our two options in closing the lag:<\/strong><\/p>\n
                  \n
                • Our messaging system is Kafka based. We could have just taken the present offset before the migration started, and after the migration ended, start consuming from that specific offset. This solution requires some manual tweaking of offsets and a few other stuff, but will work.<\/li>\n
                • Another approach to solving this issue was to start out consuming messages from the start of the subject in Kafka and make our actions on Elasticsearch idempotent \u2014 meaning, if the change was \u201capplied\u201d already, nothing would change in Elastic store.<\/li>\n<\/ul>\n
                  The requests made by our service against Elastic were already idempotent, so we elect option 2 because it required zero manual work (no got to take specific offsets, then set them afterward during a new consumer group).<\/p>\n
                  How can we migrate the data?<\/h3>\n
                  These were the choices we thought of:<\/p>\n
                    \n
                  • If our Kafka contained all messages from the start of your time , we could just play from the beginning and therefore the end state would be equal. But since we apply retention to out topics, this wasn’t an option.<\/li>\n
                  • Dump messages to disk then ingest them to Elastic directly \u2013 This solution looked quite weird. Why store them in disk rather than just writing them on to Elastic?<\/li>\n
                  • Transfer messages between old Elastic to new Elastic \u2014 This meant, writing some kind of \u201cscript\u201d (did anyone say Python? \ud83d\ude03) which will hook up with the old Elasticsearch cluster, query for items, transform them to the new schema, and index them within the cluster.<\/li>\n<\/ul>\n
                    We choose the last option. These were the planning choices we had in mind:<\/strong><\/p>\n
                      \n
                    • Let\u2019s not attempt to believe error handling unless we’d like to. Let\u2019s attempt to write something super simple, and if errors occur, let\u2019s attempt to address them. within the end, we didn’t got to address this issue since no errors occurred during the migration.<\/li>\n
                    • It\u2019s a one-off operation, so whatever works first \/ KISS.<\/li>\n
                    • Metrics \u2014 Since the migration processes can take hours to days, we wanted the power from day 1 to be ready to monitor the error count and to trace the present progress and replica rate of the script.<\/li>\n<\/ul>\n
                      \"\"<\/p>\n
                      We thought long and hard and choose Python as our weapon of choice. The final version of the code is below:<\/p>\n
                       
                       <\/div> 
                      dictor==0.1.2 - to copy and transform our Elasticsearch documentselasticsearch==1.9.0 - to connect to "old" Elasticsearchelasticsearch6==6.4.2 - to connect to the "new" Elasticsearchstatsd==3.3.0 - to report metrics<\/code><\/pre> <\/div>\n
                       
                       <\/div> 
                      from elasticsearch import Elasticsearchfrom elasticsearch6 import Elasticsearch as Elasticsearch6import sysfrom elasticsearch.helpers import scanfrom elasticsearch6.helpers import parallel_bulkimport statsdES_SOURCE = Elasticsearch(sys.argv[1])ES_TARGET = Elasticsearch6(sys.argv[2])INDEX_SOURCE = sys.argv[3]INDEX_TARGET = sys.argv[4]QUERY_MATCH_ALL = {"query": {"match_all": {}}}SCAN_SIZE = 1000SCAN_REQUEST_TIMEOUT = '3m'REQUEST_TIMEOUT = 180MAX_CHUNK_BYTES = 15 * 1024 * 1024RAISE_ON_ERROR = Falsedef transform_item(item, index_target):    # implement your logic transformation here    transformed_source_doc = item.get("_source")    return {"_index": index_target,            "_type": "_doc",            "_id": item['_id'],            "_source": transformed_source_doc}def transformedStream(es_source, match_query, index_source, index_target, transform_logic_func):    for item in scan(es_source, query=match_query, index=index_source, size=SCAN_SIZE,                     timeout=SCAN_REQUEST_TIMEOUT):        yield transform_logic_func(item, index_target)def index_source_to_target(es_source, es_target, match_query, index_source, index_target, bulk_size, statsd_client,                           logger, transform_logic_func):    ok_count = 0    fail_count = 0    count_response = es_source.count(index=index_source, body=match_query)    count_result = count_response['count']    statsd_client.gauge(stat='elastic_migration_document_total_count,index={0},type=success'.format(index_target),                        value=count_result)    with statsd_client.timer('elastic_migration_time_ms,index={0}'.format(index_target)):        actions_stream = transformedStream(es_source, match_query, index_source, index_target, transform_logic_func)        for (ok, item) in parallel_bulk(es_target,                                        chunk_size=bulk_size,                                        max_chunk_bytes=MAX_CHUNK_BYTES,                                        actions=actions_stream,                                        request_timeout=REQUEST_TIMEOUT,                                        raise_on_error=RAISE_ON_ERROR):            if not ok:                logger.error("got error on index {} which is : {}".format(index_target, item))                fail_count += 1                statsd_client.incr('elastic_migration_document_count,index={0},type=failure'.format(index_target),                                   1)            else:                ok_count += 1                statsd_client.incr('elastic_migration_document_count,index={0},type=success'.format(index_target),                                   1)    return ok_count, fail_countstatsd_client = statsd.StatsClient(host='localhost', port=8125)if __name__ == "__main__":    index_source_to_target(ES_SOURCE, ES_TARGET, QUERY_MATCH_ALL, INDEX_SOURCE, INDEX_TARGET, BULK_SIZE,                           statsd_client, transform_item)<\/code><\/pre> <\/div>\n
                      Conclusion<\/h2>\n
                      Migrating data during a live production system may be a complicated task that needs tons of attention and careful planning. i like to recommend taking the time to figure through the steps listed above and find out what works best for your needs.<\/p>\n
                      As a rule of thumb, always attempt to reduce your requirements the maximum amount as possible. for instance , may be a zero downtime migration required? are you able to afford data-loss?<\/p>\n
                      \"\"<\/p>\n
                      Upgrading data stores is typically a marathon and not a sprint, so take a deep breath and check out to enjoy the ride.<\/p>\n
                        \n
                      • The whole process listed above took me around 4 months of labor<\/li>\n
                      • All of the Elasticsearch examples that appear during this post are tested against version 6.8.1<\/li>\n<\/ul>\n","protected":false},"excerpt":{"rendered":"
                        My last task at BigPanda was to upgrade an existing service that was using Elasticsearch version 1.7 to a more moderen Elasticsearch version, 6.8.1. In this post, i will be able to share how we migrated from Elasticsearch 1.6 to 6.8 with harsh constraints like zero downtime, no data loss, and 0 bugs. I’ll also […]<\/p>\n","protected":false},"author":2,"featured_media":33501,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4148],"tags":[86894,86761],"class_list":["post-33421","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-python","tag-migrate-from-elasticsearch","tag-python"],"_links":{"self":[{"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/posts\/33421","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/comments?post=33421"}],"version-history":[{"count":0,"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/posts\/33421\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/media\/33501"}],"wp:attachment":[{"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/media?parent=33421"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/categories?post=33421"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.wikitechy.com\/technology\/wp-json\/wp\/v2\/tags?post=33421"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}