How do you implement AWS Elasticache auto discovery for node.js

Question

I'm a node noob and trying to understand how one would implement auto discovery in a node.js application. I'm going to use the cluster module and want each worker process to be kept up to date (and persistently connected to) the elasticache nodes.

Since there is no concept of shared memory (like PHP APC) would you have to have code that runs in each worker, that wakes up every X seconds and somehow updates the list of IP's and re-connects the memcache client?

How do people solve this today? Example code would be much appreciated.

Answer 1

Note that at this time, Auto Discovery is only available for cache clusters running the memcached engine.

For Cache Engine Version 1.4.14 or Higher you need to create a TCP/IP socket to the Cache Cluster Configuration Endpoint (or any Cache Node Endpoint) and send this command:

config get cluster

With Node.js you can use the net.Socket class to to that.

The reply consists of two lines:

• The version number of the configuration information. Each time a node is added or removed from the cache cluster, the version number increases by one.

• A list of cache nodes. Each node in the list is represented by a hostname|ip-address|port group, and each node is delimited by a space.

A carriage return and a linefeed character (CR + LF) appears at the end of each line.

Here you can find a more thorough description of how to add Auto Discovery to your client library.

Using the cluster module you need to store the same information in each process (i.e. child) and I would use "setInterval" per child to periodically check (e.g. every 60 seconds) the list of nodes and re-connect only if the list has changed (this should not happen very often).

You can optionally update the list on the master only and use "worker.send" to update the workers. This could keep all the processes running in a single server more in sync, but it would not help in a multi server architecture, so it is very important to use consistent hashing in order to be able to change the list of nodes and loose the "minimum" amount of keys stored in the memcached cluster.

I would use a global variable to store this kind of configuration.

Thinking twice you can use the AWS SDK for Node.js to get the list of ElastiCache Nodes (and that works for the Redis engine as well).

In that case the code would be something like:

var util = require('util'),
    AWS = require('aws-sdk'),
    Memcached = require('memcached');

global.AWS_REGION = 'eu-west-1'; // Just as a sample I'm using the EU West region                                                                                     
global.CACHE_CLUSTER_ID = 'test';
global.CACHE_ENDPOINTS = [];
global.MEMCACHED = null;

function init() {

    AWS.config.update({
        region: global.AWS_REGION
    });
    elasticache = new AWS.ElastiCache();

    function getElastiCacheEndpoints() {

        function sameEndpoints(list1, list2) {
            if (list1.length != list2.length)
                return false;
            return list1.every(
                function(e) {
                    return list2.indexOf(e) > -1;
                });
        }

        function logElastiCacheEndpoints() {
            global.CACHE_ENDPOINTS.forEach(
                function(e) {
                    util.log('Memcached Endpoint: ' + e);
                });
        }

        elasticache.describeCacheClusters({
                CacheClusterId: global.CACHE_CLUSTER_ID,
                ShowCacheNodeInfo: true
            },
            function(err, data) {
                if (!err) {
                    util.log('Describe Cache Cluster Id:' + global.CACHE_CLUSTER_ID);
                    if (data.CacheClusters[0].CacheClusterStatus == 'available') {
                        var endpoints = [];

                        data.CacheClusters[0].CacheNodes.forEach(
                            function(n) {
                                var e = n.Endpoint.Address + ':' + n.Endpoint.Port;
                                endpoints.push(e);
                            });
                        if (!sameEndpoints(endpoints, global.CACHE_ENDPOINTS)) {
                            util.log('Memached Endpoints changed');
                            global.CACHE_ENDPOINTS = endpoints;
                            if (global.MEMCACHED)
                                global.MEMCACHED.end();
                            global.MEMCACHED = new Memcached(global.CACHE_ENDPOINTS);
                            process.nextTick(logElastiCacheEndpoints);
                            setInterval(getElastiCacheEndpoints, 60000); // From now on, update every 60 seconds                        
                        }
                    } else {
                        setTimeout(getElastiCacheEndpoints, 10000); // Try again after 10 seconds until 'available'                     
                    }
                } else {
                    util.log('Error describing Cache Cluster:' + err);
                }
            });
    }

    getElastiCacheEndpoints();

}

init();