Merge pull request #9 from YunoHost/queue_worker_to_avoid_useless_ios

[enh] queue/worker pattern to avoid hammering the server with ios

dynette.cron.py

@@ -2,6 +2,7 @@
 
 ### Configuration ###
 
+postgresql_dsn = "dbname=dynette user=dynette password=myPassword"
 conf_file = '/etc/bind/named.conf.local'    # Include this filename in '/etc/bind/named.conf'
 zone_dir  = '/var/lib/bind/'                # Do not forget the trailing '/'
 subs_urls = ['https://dyndns.yunohost.org']  # 127.0.0.1 if you install subscribe server locally
@@ -25,9 +26,12 @@ allowed_operations = {
 ### Script ###
 
 import os
+import sys
 import json
+import psycopg2
 from urllib import urlopen
 
+
 # Get master key
 master_key_path = os.path.join(os.path.dirname(__file__), 'master.key')
 master_key = open(master_key_path).read().rstrip()
@@ -35,93 +39,119 @@ master_key = open(master_key_path).read().rstrip()
 # Bind configuration
 lines = ['// Generated by Dynette CRON']
 
-# Loop through Dynette servers
+with psycopg2.connect(postgresql_dsn) as postgresql_connection:
-for url in subs_urls:
+    with postgresql_connection.cursor() as psql:
+        # look in the job queue if we have tasks to handle
+        need_rewrite = False
+        need_bind9_cache_flush = False
 
-    lines.extend([
+        # DataMapper convert table names to lower cases and add a "s" at the
-            'key dynette. {',
+        # end
-            '       algorithm hmac-md5;',
+        # consume all available tasks at once to merge them and avoir doing
-            '       secret "'+ master_key +'";',
+        # useless jobs
-            '};',
+        psql.execute("SELECT task FROM jobqueues ORDER BY id ASC;")
-    ])
 
-    # Get available DynDNS domains
+        for task in psql.fetchall():
-    domains = json.loads(str(urlopen(url +'/domains').read()))
+            task = task[0]
-    for domain in domains:
+            if task == "conf_rewrite":
+                need_rewrite = True
+            elif task == "bind9_cache_flush":
+                need_bind9_cache_flush = True
 
-        # Create zone database if not present
+        if not need_rewrite and not need_bind9_cache_flush:
-        if not os.path.exists(zone_dir + domain +'.db'):
+            sys.exit(0)
-            db_lines = [
-                '$ORIGIN .',
-                '$TTL 10 ; 10 seconds',
-                domain+'.   IN SOA  '+ ns0 +'. '+ rname +'. (',
-                '                                18         ; serial',
-                '                                10800      ; refresh (3 hours)',
-                '                                3600       ; retry (1 hour)',
-                '                                604800     ; expire (1 week)',
-                '                                10         ; minimum (10 seconds)',
-                '                                )',
-                '$TTL 3600       ; 1 hour',
-                '                        NS      '+ ns0 +'.',
-                '                        NS      '+ ns1 +'.',
-                '',
-                '$ORIGIN '+ domain +'.',
-            ]
-            with open(zone_dir + domain +'.db', 'w') as zone:
-                for line in db_lines:
-                    zone.write(line + '\n')
 
-        lines.extend([
+        # we have consume all the jobs, flush it
-                'zone "'+ domain +'" {',
+        # because we are in a SQL transaction we won't have situation where a
-                '   type master;',
+        # job could be added just after we read them all
-                '   file "'+ zone_dir + domain +'.db"; ',
+        psql.execute("DELETE FROM jobqueues;")
-                '   update-policy {',
-                '       grant dynette. wildcard *.'+ domain +'. ANY;',
-        ])
 
-        # Get registered sub-domains
+        # Loop through Dynette servers
-        result = json.loads(str(urlopen(url +'/all/'+ domain).read()))
+        for url in subs_urls:
-        for entry in result:
-            for subd, type in allowed_operations.items():
-                if subd == '.': subd = ''
-                lines.append('       grant '+ entry['subdomain'] +'. name '+ subd + entry['subdomain'] +'. ' + ' '.join(type) +';')
 
-        lines.extend([
-                '   };',
-                '};'
-                '',
-        ])
-
-        for entry in result:
             lines.extend([
-                'key '+ entry['subdomain'] +'. {',
+                    'key dynette. {',
-                '       algorithm ' + entry['key_algo'] + ';',
+                    '       algorithm hmac-md5;',
-                '       secret "'+ entry['public_key'] +'";',
+                    '       secret "'+ master_key +'";',
-                '};',
+                    '};',
             ])
 
-# Backup old Bind configuration file.
+            # Get available DynDNS domains
-os.system('cp '+ conf_file +' '+ conf_file +'.back')
+            domains = json.loads(str(urlopen(url +'/domains').read()))
+            for domain in domains:
 
-# Write Bind configuration file.
+                # Create zone database if not present
-with open(conf_file, 'w') as zone:
+                if not os.path.exists(zone_dir + domain +'.db'):
-    zone.write('\n'.join(lines) + '\n')
+                    db_lines = [
+                        '$ORIGIN .',
+                        '$TTL 10 ; 10 seconds',
+                        domain+'.   IN SOA  '+ ns0 +'. '+ rname +'. (',
+                        '                                18         ; serial',
+                        '                                10800      ; refresh (3 hours)',
+                        '                                3600       ; retry (1 hour)',
+                        '                                604800     ; expire (1 week)',
+                        '                                10         ; minimum (10 seconds)',
+                        '                                )',
+                        '$TTL 3600       ; 1 hour',
+                        '                        NS      '+ ns0 +'.',
+                        '                        NS      '+ ns1 +'.',
+                        '',
+                        '$ORIGIN '+ domain +'.',
+                    ]
+                    with open(zone_dir + domain +'.db', 'w') as zone:
+                        for line in db_lines:
+                            zone.write(line + '\n')
 
-# Restore ownership
+                lines.extend([
-os.system('chown -R bind:bind '+ zone_dir +' '+ conf_file)
+                        'zone "'+ domain +'" {',
+                        '   type master;',
+                        '   file "'+ zone_dir + domain +'.db"; ',
+                        '   update-policy {',
+                        '       grant dynette. wildcard *.'+ domain +'. ANY;',
+                ])
 
-# Reload Bind
+                # Get registered sub-domains
-if os.system('/usr/sbin/rndc reload') == 0:
+                result = json.loads(str(urlopen(url +'/all/'+ domain).read()))
-    exit(0)
+                for entry in result:
-else:
+                    for subd, type in allowed_operations.items():
-    os.system('cp '+ conf_file +' '+ conf_file +'.bad')
+                        if subd == '.': subd = ''
-    os.system('cp '+ conf_file +'.back '+ conf_file)
+                        lines.append('       grant '+ entry['subdomain'] +'. name '+ subd + entry['subdomain'] +'. ' + ' '.join(type) +';')
-    os.system('/usr/sbin/rndc reload')
-    print("An error occured ! Please check daemon.log and your conf.bad")
-    exit(1)
 
-# mein got this is so awful
+                lines.extend([
-if os.path.exists('/tmp/dynette_flush_bind_cache'):
+                        '   };',
-    os.system('/usr/sbin/rndc flush')
+                        '};'
-    os.system('/usr/sbin/rndc reload')
+                        '',
-    os.system('rm /tmp/dynette_flush_bind_cache')
+                ])
+
+                for entry in result:
+                    lines.extend([
+                        'key '+ entry['subdomain'] +'. {',
+                        '       algorithm ' + entry['key_algo'] + ';',
+                        '       secret "'+ entry['public_key'] +'";',
+                        '};',
+                    ])
+
+        # update bind9 zone
+        if need_rewrite:
+            # Backup old Bind configuration file.
+            os.system('cp '+ conf_file +' '+ conf_file +'.back')
+
+            # Write Bind configuration file.
+            with open(conf_file, 'w') as zone:
+                zone.write('\n'.join(lines) + '\n')
+
+            # Restore ownership
+            os.system('chown -R bind:bind '+ zone_dir +' '+ conf_file)
+
+            # Reload Bind
+            if os.system('/usr/sbin/rndc reload') != 0:
+                os.system('cp '+ conf_file +' '+ conf_file +'.bad')
+                os.system('cp '+ conf_file +'.back '+ conf_file)
+                os.system('/usr/sbin/rndc reload')
+                print("An error occured ! Please check daemon.log and your conf.bad")
+                sys.exit(1)
+
+        # flush bind9 cache (mostly because we got a hmac-sha512 key migration
+        if need_bind9_cache_flush:
+            os.system('/usr/sbin/rndc flush')
+            os.system('/usr/sbin/rndc reload')

dynette.rb

@@ -68,6 +68,25 @@ class Ipban
     property :ip_addr, String, :key => true
 end
 
+################
+### JobQueue ###
+################
+
+# JobQueue to communicate with the conf updater
+class Jobqueue
+    include DataMapper::Resource
+
+    property :id, Serial, :key => true
+    property :task, String
+end
+
+def schedule_conf_rewrite()
+    Jobqueue.create(:task => "conf_rewrite")
+end
+
+def schedule_bind9_cache_flush()
+    Jobqueue.create(:task => "bind9_cache_flush")
+end
 
 ################
 ### Handlers ###
@@ -198,6 +217,7 @@ post '/key/:public_key' do
     entry.ips << Ip.create(:ip_addr => request.ip)
 
     if entry.save
+        schedule_conf_rewrite
         halt 201, { :public_key => entry.public_key, :subdomain => entry.subdomain, :current_ip => entry.current_ip }.to_json
     else
         halt 412, { :error => "A problem occured during DNS registration" }.to_json
@@ -237,11 +257,8 @@ put '/migrate_key_to_sha512/' do
         halt 412, { :error => "A problem occured during key algo migration" }.to_json
     end
 
-    # I don't have any other way of communicating with this dynette.cron.py
+    schedule_conf_rewrite
-    # this is awful
+    schedule_bind9_cache_flush
-    File.open("/tmp/dynette_flush_bind_cache", "w").close
-    # let's try flusing here, hope that could help ... (this design is so awful)
-    `/usr/sbin/rndc flush`
 
     halt 201, { :public_key => entry.public_key, :subdomain => entry.subdomain, :current_ip => entry.current_ip }.to_json
 end
@@ -255,6 +272,7 @@ put '/key/:public_key' do
     end
     entry.current_ip = request.ip
     if entry.save
+        schedule_conf_rewrite
         halt 201, { :public_key => entry.public_key, :subdomain => entry.subdomain, :current_ip => entry.current_ip }.to_json
     else
         halt 412, { :error => "A problem occured during DNS update" }.to_json
@@ -270,6 +288,7 @@ delete '/key/:public_key' do
     if entry = Entry.first(:public_key => params[:public_key])
         Ip.first(:entry_id => entry.id).destroy
         if entry.destroy
+            schedule_conf_rewrite
             halt 200, "OK".to_json
         else
             halt 412, { :error => "A problem occured during DNS deletion" }.to_json
@@ -296,6 +315,7 @@ delete '/domains/:subdomain' do
 
         Ip.first(:entry_id => entry.id).destroy
         if entry.destroy
+            schedule_conf_rewrite
             halt 200, "OK".to_json
         else
             halt 412, { :error => "A problem occured during DNS deletion" }.to_json