Watch feature (#1527)

* Add part 1 watch functionality. (squashed)

* add funcs for service/endpoint fqdns

* add endpoints watch

* document exposed funcs

* only send subset deltas

* locking for watch map

* tests and docs

* add pod watch

* remove debugs prints

* feedback part 1

* add error reporting to proto

* inform clients of server stop+errors

* add grpc options param

* use proper context

* Review feedback:
 * Removed client (will move to another repo)
 * Use new log functions
 * Change watchChan to be for string not []string
 * Rework how k8s plugin stores watch tracking info to simplify
 * Normalize the qname on watch request

* Add blank line back

* Revert another spurious change

* Fix tests

* Add stop channel.
Fix tests.
Better docs for plugin interface.

* fmt.Printf -> log.Warningf

* Move from dnsserver to plugin/pkg/watch

* gofmt

* remove dead client watches

* sate linter

* linter omg

1 files changed, 53 insertions, 0 deletions

diff --git a/plugin/kubernetes/controller_test.go b/plugin/kubernetes/controller_test.go
new file mode 100644
index 000000000..02915fb51
--- /dev/null
+++ b/plugin/kubernetes/controller_test.go
@@ -0,0 +1,53 @@
+package kubernetes
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+
+	api "k8s.io/api/core/v1"
+)
+
+func endpointSubsets(addrs ...string) (eps []api.EndpointSubset) {
+	for _, ap := range addrs {
+		apa := strings.Split(ap, ":")
+		address := apa[0]
+		port, _ := strconv.Atoi(apa[1])
+		eps = append(eps, api.EndpointSubset{Addresses: []api.EndpointAddress{{IP: address}}, Ports: []api.EndpointPort{{Port: int32(port)}}})
+	}
+	return eps
+}
+
+func TestEndpointsSubsetDiffs(t *testing.T) {
+	var tests = []struct {
+		a, b, expected api.Endpoints
+	}{
+		{ // From a->b: Nothing changes
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")},
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")},
+			api.Endpoints{},
+		},
+		{ // From a->b: Everything goes away
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")},
+			api.Endpoints{},
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")},
+		},
+		{ // From a->b: Everything is new
+			api.Endpoints{},
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")},
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80", "10.0.0.2:8080")},
+		},
+		{ // From a->b: One goes away, one is new
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.2:8080")},
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.1:80")},
+			api.Endpoints{Subsets: endpointSubsets("10.0.0.2:8080", "10.0.0.1:80")},
+		},
+	}
+
+	for i, te := range tests {
+		got := endpointsSubsetDiffs(&te.a, &te.b)
+		if !endpointsEquivalent(got, &te.expected) {
+			t.Errorf("Expected '%v' for test %v, got '%v'.", te.expected, i, got)
+		}
+	}
+}