Feature/centralize varnish trigger

app/models/table.rb

@@ -444,10 +444,10 @@ def after_commit
         update_table_pg_stats
 
         # Set default triggers
-        add_python
-        set_trigger_update_updated_at
-        set_trigger_cache_timestamp
-        set_trigger_check_quota
+        #owner.add_python
+        #owner.create_function_invalidate_varnish
+        #owner.create_trigger_function_update_timestamp
+        set_triggers
       rescue => e
         self.handle_creation_error(e)
       end
@@ -1112,13 +1112,6 @@ def oid
   end
 
   # DB Triggers and things
-  # TODO: move to user (is db-wide, not table-wide)
-  def add_python
-    owner.in_database(:as => :superuser).run(<<-SQL
-      CREATE OR REPLACE PROCEDURAL LANGUAGE 'plpythonu' HANDLER plpython_call_handler;
-    SQL
-    )
-  end
 
   def has_trigger? trigger_name
     owner.in_database(:as => :superuser).select('trigger_name').from(:information_schema__triggers)
@@ -1137,6 +1130,24 @@ def pg_indexes
       .all.map { |t| t[:indexname] }
   end
 
+  def set_triggers
+
+    set_trigger_update_updated_at
+
+    # NOTE: We're dropping the cache_checkpoint here
+    #       as a poor man's migration path from 2.1.
+    #       Once an official 2.2 is out and a proper
+    #       migration script is written this line can
+    #       be removed. For the record, the actual cache
+    #       (varnish) management is now triggered indirectly
+    #       by the "track_updates" trigger.
+    #
+    drop_trigger_cache_checkpoint
+
+    set_trigger_track_updates
+    set_trigger_check_quota
+  end
+
   def set_trigger_the_geom_webmercator
     return true unless self.schema(:reload => true).flatten.include?(THE_GEOM)
     owner.in_database(:as => :superuser) do |user_database|
@@ -1177,62 +1188,38 @@ def set_trigger_update_updated_at
     )
   end
 
-  # move to C
-  def set_trigger_cache_timestamp
-
-    varnish_host = Cartodb.config[:varnish_management].try(:[],'host') || '127.0.0.1'
-    varnish_port = Cartodb.config[:varnish_management].try(:[],'port') || 6082
-    varnish_timeout = Cartodb.config[:varnish_management].try(:[],'timeout') || 5
-    varnish_critical = Cartodb.config[:varnish_management].try(:[],'critical') == true ? 1 : 0
-    varnish_retry = Cartodb.config[:varnish_management].try(:[],'retry') || 5
-    purge_command = Cartodb::config[:varnish_management]["purge_command"]
-
+  # Drop "cache_checkpoint", if it exists
+  # NOTE: this is for migrating from 2.1
+  def drop_trigger_cache_checkpoint
     owner.in_database(:as => :superuser).run(<<-TRIGGER
-    CREATE OR REPLACE FUNCTION update_timestamp() RETURNS trigger AS
-    $$
-        critical = #{varnish_critical}
-        timeout = #{varnish_timeout}
-        retry = #{varnish_retry}
-
-        client = GD.get('varnish', None)
-
-        while True:
-
-          if not client:
-              try:
-                import varnish
-                client = GD['varnish'] = varnish.VarnishHandler(('#{varnish_host}', #{varnish_port}, timeout))
-              except Exception as err:
-                plpy.warning('Varnish connection error: ' +  str(err))
-                # NOTE: we won't retry on connection error
-                if critical:
-                  plpy.error('Varnish connection error: ' +  str(err))
-                break
-
-          try:
-            table_name = TD["table_name"]
-            client.fetch('#{purge_command} obj.http.X-Cache-Channel ~ "^#{self.database_name}:(.*%s.*)|(table)$"' % table_name)
-            break
-          except Exception as err:
-            plpy.warning('Varnish fetch error: ' + str(err))
-            client = GD['varnish'] = None # force reconnect
-            if not retry:
-              if critical:
-                plpy.error('Varnish fetch error: ' +  str(err))
-              break
-            retry -= 1 # try reconnecting
-    $$
-    LANGUAGE 'plpythonu' VOLATILE;
+    DROP TRIGGER IF EXISTS cache_checkpoint ON "#{self.name}";
+TRIGGER
+    )
+  end
 
+  # Set a "cache_checkpoint" trigger to invalidate varnish
+  # TODO: drop this trigger, delegate to a trigger on CDB_TableMetadata
+  def set_trigger_cache_checkpoint
+    owner.in_database(:as => :superuser).run(<<-TRIGGER
+    BEGIN;
     DROP TRIGGER IF EXISTS cache_checkpoint ON "#{self.name}";
     CREATE TRIGGER cache_checkpoint BEFORE UPDATE OR INSERT OR DELETE OR TRUNCATE ON "#{self.name}" EXECUTE PROCEDURE update_timestamp();
+    COMMIT;
+TRIGGER
+    )
+  end
+
+  # Set a "track_updates" trigger to keep CDB_TableMetadata updated
+  def set_trigger_track_updates
 
+    owner.in_database(:as => :superuser).run(<<-TRIGGER
+    BEGIN;
     DROP TRIGGER IF EXISTS track_updates ON "#{self.name}";
     CREATE trigger track_updates
       AFTER INSERT OR UPDATE OR DELETE OR TRUNCATE ON "#{self.name}"
       FOR EACH STATEMENT
       EXECUTE PROCEDURE cdb_tablemetadata_trigger();
-
+    COMMIT;
 TRIGGER
     )
   end
@@ -1248,6 +1235,7 @@ def set_trigger_check_quota
     # (it'll always run before each statement)
     check_probability_factor = 0.001 # TODO: base on database usage ?
     owner.in_database(:as => :superuser).run(<<-TRIGGER
+    BEGIN;
     DROP TRIGGER IF EXISTS test_quota ON "#{self.name}";
     CREATE TRIGGER test_quota BEFORE UPDATE OR INSERT ON "#{self.name}"
       EXECUTE PROCEDURE CDB_CheckQuota(1, #{self.owner.quota_in_bytes});
@@ -1256,6 +1244,7 @@ def set_trigger_check_quota
       FOR EACH ROW
       EXECUTE PROCEDURE CDB_CheckQuota( #{check_probability_factor},
                                         #{self.owner.quota_in_bytes} );
+    COMMIT;
   TRIGGER
   )
   end

app/models/user.rb

@@ -654,8 +654,103 @@ def create_importer_schema
     end
   end #create_importer_schema
 
+  # Add plpythonu pl handler
+  def add_python
+    in_database(:as => :superuser).run(<<-SQL
+      CREATE OR REPLACE PROCEDURAL LANGUAGE 'plpythonu' HANDLER plpython_call_handler;
+    SQL
+    )
+  end
+
+  # Create a "cdb_invalidate_varnish()" function to invalidate Varnish
+  #
+  # The function can only be used by the superuser, we expect
+  # security-definer triggers OR triggers on superuser-owned tables
+  # to call it with controlled set of parameters.
+  #
+  # The function is written in python because it needs to reach out
+  # to a Varnish server.
+  #
+  # Being unable to communicate with Varnish may or may not be critical
+  # depending on CartoDB configuration at time of function definition.
+  #
+  def create_function_invalidate_varnish
+
+    add_python
+
+    varnish_host = Cartodb.config[:varnish_management].try(:[],'host') || '127.0.0.1'
+    varnish_port = Cartodb.config[:varnish_management].try(:[],'port') || 6082
+    varnish_timeout = Cartodb.config[:varnish_management].try(:[],'timeout') || 5
+    varnish_critical = Cartodb.config[:varnish_management].try(:[],'critical') == true ? 1 : 0
+    varnish_retry = Cartodb.config[:varnish_management].try(:[],'retry') || 5
+    purge_command = Cartodb::config[:varnish_management]["purge_command"]
+
+    in_database(:as => :superuser).run(<<-TRIGGER
+    BEGIN;
+    CREATE OR REPLACE FUNCTION cdb_invalidate_varnish(table_name text) RETURNS void AS
+    $$
+        critical = #{varnish_critical}
+        timeout = #{varnish_timeout}
+        retry = #{varnish_retry}
+
+        client = GD.get('varnish', None)
+
+        while True:
+
+          if not client:
+              try:
+                import varnish
+                client = GD['varnish'] = varnish.VarnishHandler(('#{varnish_host}', #{varnish_port}, timeout))
+              except Exception as err:
+                plpy.warning('Varnish connection error: ' +  str(err))
+                # NOTE: we won't retry on connection error
+                if critical:
+                  plpy.error('Varnish connection error: ' +  str(err))
+                break
+
+          try:
+            client.fetch('#{purge_command} obj.http.X-Cache-Channel ~ "^#{self.database_name}:(.*%s.*)|(table)$"' % table_name)
+            break
+          except Exception as err:
+            plpy.warning('Varnish fetch error: ' + str(err))
+            client = GD['varnish'] = None # force reconnect
+            if not retry:
+              if critical:
+                plpy.error('Varnish fetch error: ' +  str(err))
+              break
+            retry -= 1 # try reconnecting
+    $$
+    LANGUAGE 'plpythonu' VOLATILE;
+    REVOKE ALL ON FUNCTION cdb_invalidate_varnish(TEXT) FROM PUBLIC;
+    COMMIT;
+TRIGGER
+    )
+  end
+
+  # Create an "update_timestamp()" trigger function to invalidate Varnish
+  # This is currently invoked by the "cache_checkpoint" trigger attached
+  # to tables
+  #
+  # TODO: drop this and replace with a trigger on CDB_TableMetadata
+  #
+  def create_trigger_function_update_timestamp
+    in_database(:as => :superuser).run(<<-TRIGGER
+    CREATE OR REPLACE FUNCTION update_timestamp() RETURNS trigger AS
+    $$
+        table_name = TD["table_name"]
+        plan = plpy.prepare("SELECT cdb_invalidate_varnish($1)", ["text"])
+        plpy.execute(plan, [table_name])
+    $$
+    LANGUAGE 'plpythonu' VOLATILE SECURITY DEFINER;
+TRIGGER
+    )
+  end
+
+
   # Cartodb functions
   def load_cartodb_functions(files = [])
+    create_function_invalidate_varnish
+    create_trigger_function_update_timestamp
     in_database(:as => :superuser) do |user_database|
       user_database.transaction do
         if files.empty?

lib/sql/scripts-available/CDB_TableMetadata.sql

@@ -1,3 +1,4 @@
+
 CREATE TABLE IF NOT EXISTS
   public.CDB_TableMetadata (
     tabname regclass not null primary key,
@@ -47,10 +48,47 @@ BEGIN
   FROM nv LEFT JOIN updated USING(tabname)
   WHERE updated.tabname IS NULL;
 
+  RETURN NULL;
+END;
+$$
+LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+--
+-- Trigger invalidating varnish whenever CDB_TableMetadata
+-- record change.
+--
+CREATE OR REPLACE FUNCTION _CDB_TableMetadata_Updated()
+RETURNS trigger AS
+$$
+DECLARE
+  tabname TEXT;
+BEGIN
+
+  IF TG_OP = 'UPDATE' or TG_OP = 'INSERT' THEN
+    tabname = NEW.tabname;
+  ELSE
+    tabname = OLD.tabname;
+  END IF;
+
   -- Notify table data update
-  PERFORM pg_notify('cdb_tabledata_update', TG_TABLE_NAME);
+  PERFORM pg_notify('cdb_tabledata_update', tabname);
+
+  --RAISE NOTICE 'Table % was updated', tabname;
+
+  -- This will be needed until we'll have someone listening
+  -- on the event we just broadcasted:
+  --
+  --  LISTEN cdb_tabledata_update;
+  --
+  PERFORM cdb_invalidate_varnish(tabname);
 
   RETURN NULL;
 END;
 $$
 LANGUAGE plpgsql VOLATILE SECURITY DEFINER;
+
+DROP TRIGGER IF EXISTS table_modified ON CDB_TableMetadata;
+CREATE TRIGGER table_modified AFTER INSERT OR UPDATE OR DELETE
+ON CDB_TableMetadata FOR EACH ROW EXECUTE PROCEDURE
+ _CDB_TableMetadata_Updated();
+

lib/tasks/db_maintenance.rake

@@ -64,10 +64,8 @@ namespace :cartodb do
         begin
           user.tables.all.each do |table|
             begin
-              table.add_python
-              table.set_trigger_check_quota
-              table.set_trigger_update_updated_at
-              table.set_trigger_cache_timestamp
+              # set triggers
+              table.set_triggers
             rescue => e
               puts e
               next
@@ -101,10 +99,7 @@ namespace :cartodb do
           end
 
           # reset triggers
-          table.add_python
-          table.set_trigger_update_updated_at
-          table.set_trigger_cache_timestamp
-          table.set_trigger_check_quota
+          table.set_triggers
         end  
       end
     end