Add support for dynamic updates (RFC2136)

Makefile

@@ -1,30 +1,41 @@
-all::   clean build
+all::   clean build test
 build:: get compile
 
-clean:
+motd:
+	@echo
+	@echo '       ___                     __'
+	@echo '  ____/ (_)_____________  ____/ /___  _____'
+	@echo ' / __  / / ___/ ___/ __ \/ __  / __ \/ ___/'
+	@echo '/ /_/ / (__  ) /__/ /_/ / /_/ / / / (__  )'
+	@echo '\__,_/_/____/\___/\____/\__,_/_/ /_/____/'
+	@echo
+	@echo '© Copyright DueDil 2015. Licensed under MIT.'
+	@echo
+
+clean: motd
 	@echo "\033[34m●\033[39m Cleaning out the build folder ./build"
 	rm -rf build/*
 	@echo "\033[32m✔\033[39m Cleaned ./build"
 
-get:
+get: motd
 	@echo "\033[34m●\033[39m Downloading go packages"
 	go get github.com/tools/godep
 	go get -d
 	godep restore
 	@echo "\033[32m✔\033[39m Finished downloading packages"
 
-compile:
+compile: motd get
 	@echo "\033[34m●\033[39m Building into ./build"
 	mkdir -p build/bin
 	go build -o build/bin/discodns *.go
 	@echo "\033[32m✔\033[39m Successfully built into ./build"
 
-test:
+test: motd
 	@echo "\033[34m●\033[39m Running tests"
-	go test -race
+	go test -race ./
 	@echo "\033[32m✔\033[39m Tests passed"
 
-install:
+install: motd compile
 	@echo "\033[34m●\033[39m Installing into /usr/local/bin"
 	cp build/bin/discodns /usr/local/bin/
 	@echo "\033[32m✔\033[39m Successfully installed into /usr/local/bin/discodns"

README.md

@@ -1,6 +1,6 @@
 
 discodns
-======
+========
 
 [![Build Status](https://travis-ci.org/duedil-ltd/discodns.png?branch=master)](https://travis-ci.org/duedil-ltd/discodns)
 
@@ -18,7 +18,7 @@ An authoritative DNS nameserver that queries an [etcd](http://github.com/coreos/
 - Support for TTLs
     - Global default on all records
     - Individual TTL values for individual records
-- Runtime and application metrics are captured regularly for monitoring (stdout or grahite)
+- Runtime and application metrics are captured regularly for monitoring (stdout or graphite)
 - Incoming query filters
 
 #### Production Readyness

error.go

@@ -14,7 +14,7 @@ type NodeConversionError struct {
 
 func (e *NodeConversionError) Error() string {
     return fmt.Sprintf(
-        "Unable to convert etc Node into a RR of type %d ('%s'): %s. Node details: %+v",
+        "Unable to convert etcd Node into a RR of type %d ('%s'): %s. Node details: %+v",
         e.AttemptedType,
         dns.TypeToString[e.AttemptedType],
         e.Message,

filter.go

@@ -68,3 +68,32 @@ func (f *QueryFilterer) ShouldAcceptQuery(req *dns.Msg) bool {
 
     return accepted
 }
+
+// parseFilters will convert a string into a Query Filter structure. The accepted
+// format for input is [domain]:[type,type,...]. For example...
+// 
+// - "domain:A,AAAA" # Match all A and AAAA queries within `domain`
+// - ":TXT" # Matches only TXT queries for any domain
+// - "domain:" # Matches any query within `domain`
+func parseFilters(filters []string) []QueryFilter {
+    parsedFilters := make([]QueryFilter, 0)
+    for _, filter := range filters {
+        components := strings.Split(filter, ":")
+        if len(components) != 2 {
+            logger.Printf("Expected only one colon ([domain]:[type,type...])")
+            continue
+        }
+
+        domain := dns.Fqdn(components[0])
+        types := strings.Split(components[1], ",")
+
+        if len(types) == 1 && len(types[0]) == 0 {
+            types = make([]string, 0)
+        }
+
+        debugMsg("Adding filter with domain '" + domain + "' and types '" + strings.Join(types, ",") + "'")
+        parsedFilters = append(parsedFilters, QueryFilter{domain, types})
+    }
+
+    return parsedFilters
+}

lock.go

@@ -0,0 +1,151 @@
+package main
+
+import (
+    "code.google.com/p/go-uuid/uuid"
+    "errors"
+    "github.com/coreos/go-etcd/etcd"
+    "time"
+)
+
+const (
+    ETCD_LOCK_TTL       = 10
+    ETCD_LOCK_HEARTBEAT = 5
+)
+
+// EtcdKeyLock represents a lock on a single key. Its semantics: once asked to
+// Acquire, it will try to grab hold of the key in etcd, if it doesnt exist.
+// If it does exist, then the lock waits for the other party to release it.
+// Once Acquired, it will hold on to the lock indefinitely until Abandoned
+type EtcdKeyLock struct {
+    uuid       string
+    key        string
+    etcdClient *etcd.Client
+    killChan   chan bool
+    killed     bool
+}
+
+func NewEtcdKeyLock(etcdClient *etcd.Client, key string) *EtcdKeyLock {
+    uuid := uuid.New()
+    return &EtcdKeyLock{uuid: uuid, key: key, etcdClient: etcdClient, killChan: make(chan bool)}
+}
+
+// Start the process of trying to acquire a key lock. Returns a channel that
+// will be sent true when the lock is aquired then closed. Callers can use this
+// as their signal to proceed. The lock will kept indefinitely until abandoned.
+func (l *EtcdKeyLock) Acquire() chan bool {
+    inner_acq := make(chan bool)
+    acquired := make(chan bool)
+    go func() {
+        _, ok := <-inner_acq
+        if ok {
+            acquired <- true
+            go heartbeat(l, nil)
+            go removeWhenCancelled(l)
+        }
+        close(acquired)
+    }()
+    go tryCreate(l, inner_acq)
+    return acquired
+}
+
+// Abandons the lock. This just means closing the internal cancellation
+// channel, causing all the child goros to do whatever they need to do.
+func (l *EtcdKeyLock) Abandon() {
+    if !l.killed {
+        l.killed = true
+        close(l.killChan)
+    }
+}
+
+// Blocking version of Acquire, hiding the channels from callers who just want
+// to synchronously wait
+func (l *EtcdKeyLock) WaitForAcquire(timeout int) (bool, error) {
+    timeoutKiller := time.AfterFunc(time.Duration(timeout) * time.Second, func(){
+        l.Abandon()
+    })
+    acq := l.Acquire()
+    ok, open := <- acq
+    if ok && open {
+        stopped := timeoutKiller.Stop()
+        // Stopped == false means the timer already fired: this shouldn't be
+        // possible (we shouldn't have been able to get an OK message in that
+        // case). Erroring mostly out of paranoia: I'm positive this race can't
+        // happen (famous last words though)
+        if !stopped {
+            return false, errors.New("Acquired a lock that was also killed by a timeout: this should not be possible!")
+        }
+        return true, nil
+    } else {
+        return false, errors.New("Couldn't aqcuire lock in time")
+    }
+}
+
+// The internals of trying to get a lock: Try to PUT to the lock key iff it
+// doesn't exist. If that suceeds, the lock is owned; signal the chan and
+// return. If it fails, watch the etcd key until it changes. When it does
+// change, try again. Repeat indefinitely until cancelled.
+func tryCreate(l *EtcdKeyLock, acq chan bool) {
+    defer close(acq)
+    for {
+        select {
+        case _, chOpen := <-l.killChan:
+            if !chOpen {
+                return
+            }
+        default:
+            _, err := l.etcdClient.Create(l.key, l.uuid, ETCD_LOCK_TTL)
+            if err == nil {
+                acq <- true
+                return
+            } else {
+                err, cast := err.(*etcd.EtcdError)
+                if cast && err.ErrorCode == 105 {
+                    // Watch until it changes. (The current index is given to
+                    // make sure we don't miss any changes in between)
+                    _, err := l.etcdClient.Watch(l.key, err.Index+1, false, nil, l.killChan)
+                    if err == nil {
+                        // Skip the sleep and attempt a retry asap
+                        continue
+                    }
+                }
+                // if not created and not watching, pause briefly
+                time.Sleep(1 * time.Second)
+            }
+        }
+    }
+}
+
+func heartbeat(l *EtcdKeyLock, ping chan bool) {
+    if ping == nil {
+        ping = make(chan bool)
+    }
+    defer close(ping)
+    for {
+        time.Sleep(ETCD_LOCK_HEARTBEAT * time.Second)
+        select {
+        case _, chOpen := <-l.killChan:
+            if !chOpen {
+                return
+            }
+        default:
+            _, err := l.etcdClient.Set(l.key, l.uuid, ETCD_LOCK_TTL)
+            // non-blocking write on the ping channel
+            select {
+            case ping <- (err == nil):
+            default:
+            }
+        }
+    }
+}
+
+func removeWhenCancelled(l *EtcdKeyLock) {
+    defer func() {
+        l.etcdClient.CompareAndDelete(l.key, l.uuid, 0)
+    }()
+    for {
+        _, chOpen := <-l.killChan
+        if !chOpen {
+            return
+        }
+    }
+}

lock_test.go

@@ -0,0 +1,59 @@
+package main
+
+import (
+    "testing"
+    "time"
+)
+
+func TestSimpleLockUnlock(t *testing.T) {
+    testKey := "TestSimpleLockUnlock/.lock"
+    client.Delete(testKey, true)
+
+    lock := NewEtcdKeyLock(client, testKey)
+    locked, err := lock.WaitForAcquire(1)
+
+    if !locked || err != nil {
+        t.Error("Expected to acquire lock, failed")
+        t.Fatal()
+    }
+    _, err = client.Get(testKey, false, true)
+    if err != nil {
+        t.Error("Lock claimed to succeed but etcd record missing/broken")
+        t.Fatal()
+    }
+
+    lock.Abandon()
+    time.Sleep(500 * time.Millisecond)
+    _, err = client.Get(testKey, false, true)
+    if err == nil {
+        t.Error("Lock abandoned, but key exists")
+        t.Fatal()
+    }
+}
+
+func TestConflictingLock(t *testing.T) {
+    testKey := "TestConflictingLock/.lock"
+    client.Delete(testKey, true)
+
+    lock_a := NewEtcdKeyLock(client, testKey)
+    lock_a.WaitForAcquire(30)
+
+    lock_b := NewEtcdKeyLock(client, testKey)
+    b_locked, b_err := lock_b.WaitForAcquire(1)
+    if b_locked || b_err == nil {
+        t.Error("Expected second lock to timeout")
+        t.Fatal()
+    }
+
+    lock_c := NewEtcdKeyLock(client, testKey)
+    go func(){
+        time.Sleep(500 * time.Millisecond)
+        lock_a.Abandon()
+    }()
+
+    c_locked, c_err := lock_c.WaitForAcquire(5)
+    if !c_locked || c_err != nil {
+        t.Error("Expected third lock to succeed in time")
+        t.Fatal()
+    }
+}