added varnish invalidation code

app.js

@@ -30,7 +30,6 @@ _.extend(global.settings, global.environment);
 
 var Windshaft = require('windshaft');
 var serverOptions = require('./lib/cartodb/server_options');
-var Cache       = require('./lib/cartodb/tile_cache');
 
 ws = CartodbWindshaft(serverOptions);
 ws.listen(global.environment.windshaft_port);

config/environments/development.js

@@ -5,7 +5,7 @@ module.exports.redis            = {host: '127.0.0.1',
                                    idleTimeoutMillis: 1,
                                    reapIntervalMillis: 1};
 module.exports.windshaft_port   = 8181;
-module.exports.lru_cache = false;
-module.exports.lru_cache_size = 10000;
 module.exports.enable_cors = true;
-module.exports.ttl_timeout = 5;
+module.exports.varnish_host = 'localhost';
+module.exports.varnish_port = 6082;
+module.exports.cache_enabled = true; 

config/environments/production.js

@@ -2,6 +2,7 @@ module.exports.name             = 'production';
 module.exports.postgres         = {user: 'tileuser', host: '127.0.0.1', port: 6432, max_size: 4};
 module.exports.redis            = {host: '127.0.0.1', port: 6379};
 module.exports.windshaft_port   = 8181;
-module.exports.lru_cache = false;
-module.exports.lru_cache_size = 10000;
 module.exports.ttl_timeout = 600; // 10 minutes
+module.exports.varnish_host = 'localhost';
+module.exports.varnish_port = 6082
+module.exports.cache_enabled = true; 

config/environments/test.js

@@ -5,6 +5,7 @@ module.exports.redis            = {host: '127.0.0.1',
                                    idleTimeoutMillis: 1,
                                    reapIntervalMillis: 1};
 module.exports.windshaft_port   = 8080;
-module.exports.lru_cache = true;
-module.exports.lru_cache_size = 10000;
 module.exports.enable_cors = true;
+module.exports.varnish_host = '';
+module.exports.varnish_port = null;
+module.exports.cache_enabled = false;

lib/cartodb/cache_validator.js

@@ -1,76 +1,17 @@
-//
-// this module allows to invalidate cache for a table or check if the cache for a table
-// is valid
-//
-// usage:
-//
-// Cache.getTimestamp('table_name', function(err, timestamp) {
-// });
-//
-// Cache.setTimestamp('table_name', timestamp, function(err, result) {
-// });
-//
-//
-
 var _          = require('underscore'),
-    RedisPool  = require('./redis_pool'),
+    Varnish    = require('./varnish')
-    Step       = require('step')
 
+var varnish_queue = null;
 
-function Cache(redis_opts) {
+function init(host, port) {
-
+    varnish_queue = new Varnish.VarnishQueue(host, port);
-    var redis_pool = new RedisPool(redis_opts);
-
-    var me = {}
-
-    var redis_options = {
-        db: 0
 }
 
-    var redisCommand = function(fn, callback) {
+function invalidate_db(dbname) {
-        var redis_client;
+    varnish_queue.run_cmd('purge obj.http.X-Cache-Channel == ' + dbname);
-        Step(
-            function getRedisClient(){
-                redis_pool.acquire(redis_options.db, this);
-            },
-            function getDataForTable(err, redis) {
-                if (err) throw err;
-                redis_client = redis;
-                fn(redis_client, this);
-            },
-            function exit(err, data) {
-                if (!_.isUndefined(redis_client))
-                    redis_pool.release(redis_options.db, redis_client);
-                if (callback) {
-                    callback(err, data);
-                }
-            }
-        );
 }
 
-    // get the timestamp for the table
+module.exports = {
-    me.getTimestamp = function(database, table, callback) {
+    init: init,
-        redisCommand(function(redis, step) {
+    invalidate_db: invalidate_db
-            redis.GET("cache:" + database + ":" + table + ":last_updated_at", step);
-        }, function(err, t) {
-            if(t != null)
-                callback(err, parseFloat(t));
-            else
-                callback(err, t);
-        });
-    }
-
-    me.setTimestamp = function(database, table, timestamp, callback) {
-        timestamp = timestamp || new Date().getTime();
-        redisCommand(function(redis, step) {
-            redis.SET("cache:" + database + ":" + table + ":last_updated_at", timestamp, step);
-        }, callback);
-    }
-
-    return me;
-};
-
-
-module.exports = function(redis_opts) {
-    return new Cache(redis_opts);
 }

lib/cartodb/cartodb_windshaft.js

@@ -2,29 +2,27 @@
 var _ = require('underscore')
     , Step       = require('step')
     , Windshaft = require('windshaft')
-    , Cache       = require('./tile_cache');
+    , Cache = require('./cache_validator')
 
 var CartodbWindshaft = function(serverOptions) {
 
-    // set cache if requested
-    if(serverOptions.lru_cache) {
-        var lru_cache = Cache.LRUcache(serverOptions.lru_cache_size || 10000,
-                serverOptions.redis,
-                serverOptions.ttl_timeout);
-        _.extend(serverOptions, {
-            beforeTileRender: lru_cache.beforeTileRender,
-            afterTileRender: lru_cache.afterTileRender,
-            cacheStats: lru_cache.getStats,
-            afterStateChange: lru_cache.afterStateChange
-        });
-    }
-
     // set the cache chanel info to invalidate the cache on the frontend server
     serverOptions.afterTileRender = function(req, res, tile, headers, callback) {
         res.header('X-Cache-Channel', req.params.dbname);
         callback(null, tile, headers);
     };
 
+    if(serverOptions.cache_enabled) {
+        console.log("cache invalidation enabled, varnish on " serverOptions.varnish_host, ' ', serverOptions.varnish_port);
+        Cache.init(serverOptions.varnish_host, serverOptions.varnish_port);
+        serverOptions.afterStateChange = function(req, data, callback) {
+            Cache.invalidate_db(req.params.dbname);
+            callback(null, data);
+        }
+    }
+
+
+
     // boot
     var ws = new Windshaft.Server(serverOptions);
 
@@ -46,6 +44,7 @@ var CartodbWindshaft = function(serverOptions) {
         );
     });
 
+
     /**
      * Helper to allow access to metadata to be used in embedded maps.
      */
@@ -64,17 +63,6 @@ var CartodbWindshaft = function(serverOptions) {
         );
     });
 
-    /**
-     * tile cache stats
-     */
-    ws.get('/cache', function(req, res){
-        if(serverOptions.cacheStats) {
-            res.send(serverOptions.cacheStats(req.query.tile_info, req.query.sort_by));
-        } else {
-            res.send("Cache no enabled")
-        }
-    });
-    
     return ws;
 }
 

lib/cartodb/server_options.js

@@ -8,9 +8,9 @@ module.exports = function(){
         grainstore: {datasource: global.environment.postgres},
         redis: global.environment.redis,
         enable_cors: global.environment.enable_cors,
-        lru_cache: global.environment.lru_cache,
+        varnish_host: global.environment.varnish_host,
-        lru_cache_size: global.environment.lru_cache_size,
+        varnish_port: global.environment.varnish_port,
-        ttl_timeout: global.environment.ttl_timeout
+        cache_enabled: global.environment.cache_enabled
 };
 
 /**

lib/cartodb/varnish.js

@@ -0,0 +1,166 @@
+
+/**
+ * this module implements varnish telnet management protocol
+ * https://www.varnish-cache.org/trac/wiki/ManagementPort
+ */
+
+var net = require('net')
+
+function VarnishClient(host, port, ready_callback) {
+
+    var ready = false;
+    var cmd_callback = null;
+    var client = null;
+
+    function connect() {
+        ready = false;
+        client = net.createConnection(port, host);
+    }
+
+    connect();
+
+    client.on('data', function(data) {
+        data = data.toString();
+        lines = data.split('\n', 2);
+        if(lines.length == 2) {
+            var tk = lines[0].split(' ')
+            var code = parseInt(tk[0], 10);
+            var body_length = parseInt(tk[1], 10);
+            var body = lines[1];
+            if(!ready) {
+                ready = true;
+                ready_callback();
+            } else if(cmd_callback) {
+                var c = cmd_callback
+                cmd_callback = null;
+                c(null, code, body);
+            }
+        }
+
+    });
+
+    // sends the command to the server
+    function _send(cmd, callback) {
+      cmd_callback = callback;
+      client.write(cmd + '\n');
+    }
+
+    // run command if there is no peding response
+    // fist param of the callback are the error, null
+    // if all went ok
+    this.run_cmd = function(cmd, callback) {
+       if(!cmd_callback) {
+         _send(cmd, callback);
+       } else {
+         callback('response pending');
+       }
+    }
+
+    // close the connection
+    this.close = function() {
+       client.end();
+       ready = false; 
+    }
+
+}
+
+
+function VarnishPool(opts, ready) {
+    var resources = [];
+    var available = [];
+
+    for(var i = 0; i < opts.pool_size; ++i) {
+        var v = new VarnishClient(opts.host, opts.port, function() {
+            resources.push(v);
+            available.push(v);
+        });
+    }
+
+    this.run_cmd = function(cmd, callback) {
+        var v = available.pop()
+        if(v) {
+            v.run_cmd(cmd, function(err, status_code, body) {
+                callback(err, status_code, body);
+                available.push(v);
+                ready();
+            });
+        } else {
+            callback('no clients available');
+        }
+    }
+
+    this.close = function() {
+        for(var i = 0; i < resources.length; ++i) {
+            resources[i].close();
+        }
+    }
+}
+
+function VarnishQueue(host, port) {
+
+    var queue = [];
+    var ready = false;
+
+    var client = new VarnishClient(host, port, function(err) {
+        ready = true;
+        _send_pending();
+    });
+
+    function _send_pending(empty_callback) {
+        var c = queue.pop();
+        if(!c) return;
+        client.run_cmd(c, function() {
+            if(queue.length > 0) {
+                process.nextTick(_send_pending);
+            } else {
+                if(empty_callback) {
+                    empty_callback();
+                }
+            }
+        });
+    }
+
+    this.run_cmd = function(cmd) {
+        queue.push(cmd);
+        if(ready) {
+            _send_pending();
+        }
+    }
+
+    this.end = function() {
+        _send_pending(function() {
+            client.close();
+        });
+    }
+
+}
+
+/*
+var queue = new VarnishQueue('localhost', 6082)
+setInterval(function() {
+    queue.run_cmd('purge obj.http.url == /')
+}, 10)
+*/
+/*
+v = new VarnishClient('localhost', 6082, function(err) {
+    console.log('connected');
+    v.run_cmd('purge obj.http.url == /', function(err, code, body) {
+        console.log(code);
+        console.log(body);
+        v.close();
+    });
+});
+
+pool = new VarnishPool({
+    host: 'locahost',
+    port: 6082,
+    pool_size: 5
+});
+/*
+v.close();
+*/
+
+module.exports = {
+    VarnishClient: VarnishClient,
+    VarnishQueue: VarnishQueue
+}