Add integration on travis and integration test for controller resource events handler

.travis.yml

@@ -1,5 +1,24 @@
+sudo: required
+dist: trusty
+services:
+  - docker
+
 language: go
 go:
   - "1.10"
+
+# This will allow forks to be build with travis.
+go_import_path: github.com/spotahome/kooper
+
+before_install:
+  - curl -Lo minikube https://storage.googleapis.com/minikube/releases/${MINIKUBE_VERSION}/minikube-linux-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/
+  - curl -Lo kubectl https://storage.googleapis.com/kubernetes-release/release/${KUBERNETES_VERSION}/bin/linux/amd64/kubectl && chmod +x kubectl && sudo mv kubectl /usr/local/bin/
+
+
 script:
-  - make ci
+  - make ci
+
+env:
+  global:
+    - MINIKUBE_VERSION=v0.25.2
+    - KUBERNETES_VERSION=v1.9.4

Makefile

@@ -20,6 +20,7 @@ UID := $(shell id -u)
 # cmds
 UNIT_TEST_CMD := ./hack/scripts/unit-test.sh
 INTEGRATION_TEST_CMD := ./hack/scripts/integration-test.sh 
+CI_INTEGRATION_TEST_CMD := ./hack/scripts/integration-test-minikube.sh
 MOCKS_CMD := ./hack/scripts/mockgen.sh
 DOCKER_RUN_CMD := docker run -v ${PWD}:$(DOCKER_GO_SERVICE_PATH) --rm -it $(SERVICE_NAME)
 RUN_EXAMPLE_POD_ECHO := go run ./examples/echo-pod-controller/cmd/* --development
@@ -52,19 +53,20 @@ unit-test: build
 	$(DOCKER_RUN_CMD) /bin/sh -c '$(UNIT_TEST_CMD)'
 .PHONY: integration-test
 integration-test: build
-	$(DOCKER_RUN_CMD) /bin/sh -c '$(INTEGRATION_TEST_CMD)'
+	echo "[WARNING] Requires a kubernetes cluster configured (and running) on your kubeconfig!!"
+	$(INTEGRATION_TEST_CMD)
 .PHONY: test
-test: integration-test
+test: unit-test
 
 # Test stuff in ci
 .PHONY: ci-unit-test
 ci-unit-test: 
 	$(UNIT_TEST_CMD)
 .PHONY: ci-integration-test
 ci-integration-test:
-	$(INTEGRATION_TEST_CMD)
+	$(CI_INTEGRATION_TEST_CMD)
 .PHONY: ci
-ci: ci-integration-test
+ci: ci-unit-test ci-integration-test
 
 # Mocks stuff in dev
 .PHONY: mocks

hack/scripts/integration-test-minikube.sh

@@ -0,0 +1,38 @@
+#!/bin/bash
+
+set -o errexit
+set -o nounset
+
+KUBERNETES_VERSION=${KUBERNETES_VERSION:-1.9.4}
+current_dir="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+
+SUDO=''
+if [[ $(id -u) -ne 0 ]]
+then
+    SUDO="sudo"
+fi
+
+function cleanup {
+    echo "=> Removing minikube cluster"
+    $SUDO minikube delete
+}
+trap cleanup EXIT
+
+echo "=> Preparing minikube for running integration tests"
+$SUDO minikube start \
+    --vm-driver=none \
+    --feature-gates=CustomResourceSubresources=true \
+    --kubernetes-version=${KUBERNETES_VERSION}
+
+echo "=> Waiting for minikube to start"
+sleep 30
+
+# Hack for Travis. The kubeconfig has to be readable
+if [[ -v TRAVIS ]]
+then
+    $SUDO chmod a+r ${HOME}/.kube/config
+    $SUDO chmod a+r ${HOME}/.minikube/client.key
+fi
+
+echo "=> Running integration tests"
+${current_dir}/integration-test.sh

hack/scripts/integration-test.sh

@@ -3,4 +3,4 @@
 set -o errexit
 set -o nounset
 
-go test `go list ./... | grep -v vendor` -v -tags='integration'
+go test `go list ./... | grep test/integration` -v -tags='integration'

test/integration/controller/controller_test.go

@@ -0,0 +1,168 @@
+// +build integration
+
+package controller_test
+
+import (
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/spotahome/kooper/log"
+	"github.com/spotahome/kooper/operator/controller"
+	"github.com/spotahome/kooper/operator/handler"
+	"github.com/spotahome/kooper/operator/retrieve"
+	"github.com/spotahome/kooper/test/integration/helper/cli"
+	"github.com/spotahome/kooper/test/integration/helper/prepare"
+)
+
+// TestControllerHandleEvents will test the controller receives the resources list and watch
+// events are received and handled correctly.
+func TestControllerHandleEvents(t *testing.T) {
+	tests := []struct {
+		name               string
+		addServices        []*corev1.Service
+		updateServices     []string
+		delServices        []string
+		expAddedServices   []string
+		expDeletedServices []string
+	}{
+		{
+			name: "If a controller is watching services it should react to the service change events.",
+			addServices: []*corev1.Service{
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "svc1"},
+					Spec: corev1.ServiceSpec{
+						Type: "ClusterIP",
+						Ports: []corev1.ServicePort{
+							corev1.ServicePort{Name: "port1", Port: 8080},
+						},
+					},
+				},
+				{
+					ObjectMeta: metav1.ObjectMeta{Name: "svc2"},
+					Spec: corev1.ServiceSpec{
+						Type: "ClusterIP",
+						Ports: []corev1.ServicePort{
+							corev1.ServicePort{Name: "port1", Port: 8080},
+						},
+					},
+				},
+			},
+			updateServices:     []string{"svc1"},
+			delServices:        []string{"svc1", "svc2"},
+			expAddedServices:   []string{"svc1", "svc2", "svc1"},
+			expDeletedServices: []string{"svc1", "svc2"},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			require := require.New(t)
+			assert := assert.New(t)
+			resync := 30 * time.Second
+			stopC := make(chan struct{})
+			var gotAddedServices []string
+			var gotDeletedServices []string
+
+			// Create the kubernetes client.
+			k8scli, err := cli.GetK8sClient("")
+			require.NoError(err, "kubernetes client is required")
+
+			// Prepare the environment on the cluster.
+			prep := prepare.New(k8scli, t)
+			prep.SetUp()
+			defer prep.TearDown()
+
+			// Create the reitrever.
+			rt := &retrieve.Resource{
+				ListerWatcher: cache.NewListWatchFromClient(k8scli.CoreV1().RESTClient(), "services", prep.Namespace().Name, fields.Everything()),
+				Object:        &corev1.Service{},
+			}
+
+			// Call times are the number of times the handler should be called before sending the termination signal.
+			stopCallTimes := len(test.addServices) + len(test.updateServices) + len(test.delServices)
+			calledTimes := 0
+			var mx sync.Mutex
+
+			// Create the handler.
+			hl := &handler.HandlerFunc{
+				AddFunc: func(obj runtime.Object) error {
+					mx.Lock()
+					calledTimes++
+					mx.Unlock()
+
+					svc := obj.(*corev1.Service)
+					gotAddedServices = append(gotAddedServices, svc.Name)
+					if calledTimes >= stopCallTimes {
+						close(stopC)
+					}
+					return nil
+				},
+				DeleteFunc: func(id string) error {
+					mx.Lock()
+					calledTimes++
+					mx.Unlock()
+
+					// Ignore namespace.
+					id = strings.Split(id, "/")[1]
+					gotDeletedServices = append(gotDeletedServices, id)
+					if calledTimes >= stopCallTimes {
+						close(stopC)
+					}
+					return nil
+				},
+			}
+
+			// Create a Pod controller.
+			ctrl := controller.NewSequential(resync, hl, rt, nil, log.Dummy)
+			require.NotNil(ctrl, "controller is required")
+			go ctrl.Run(stopC)
+
+			// Create the required services.
+			for _, svc := range test.addServices {
+				_, err := k8scli.CoreV1().Services(prep.Namespace().Name).Create(svc)
+				assert.NoError(err)
+				time.Sleep(1 * time.Second)
+			}
+
+			for _, svc := range test.updateServices {
+				origSvc, err := k8scli.CoreV1().Services(prep.Namespace().Name).Get(svc, metav1.GetOptions{})
+				if assert.NoError(err) {
+					// Change something
+					origSvc.Spec.Ports = append(origSvc.Spec.Ports, corev1.ServicePort{Name: "updateport", Port: 9876})
+					_, err := k8scli.CoreV1().Services(prep.Namespace().Name).Update(origSvc)
+					assert.NoError(err)
+					time.Sleep(1 * time.Second)
+				}
+			}
+
+			// Delete the required services.
+			for _, svc := range test.delServices {
+				err := k8scli.CoreV1().Services(prep.Namespace().Name).Delete(svc, &metav1.DeleteOptions{})
+				assert.NoError(err)
+				time.Sleep(1 * time.Second)
+			}
+
+			// Wait until we have finished.
+			select {
+			// Timeout.
+			case <-time.After(20 * time.Second):
+			// Finished.
+			case <-stopC:
+			}
+
+			// Check.
+			assert.Equal(test.expAddedServices, gotAddedServices)
+			assert.Equal(test.expDeletedServices, gotDeletedServices)
+		})
+	}
+}

test/integration/controller/doc.go

@@ -0,0 +1 @@
+package controller

test/integration/helper/cli/cli.go

@@ -0,0 +1,33 @@
+package cli
+
+import (
+	"fmt"
+	"path/filepath"
+
+	"k8s.io/client-go/kubernetes"
+	_ "k8s.io/client-go/plugin/pkg/client/auth/oidc" // Load oidc authentication when creating the kubernetes client.
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/homedir"
+)
+
+// GetK8sClient returns a k8s client.
+func GetK8sClient(kubehome string) (kubernetes.Interface, error) {
+	// Fallback to default kubehome.
+	if kubehome == "" {
+		kubehome = filepath.Join(homedir.HomeDir(), ".kube", "config")
+	}
+
+	// Load kubernetes local connection.
+	config, err := clientcmd.BuildConfigFromFlags("", kubehome)
+	if err != nil {
+		return nil, fmt.Errorf("could not load configuration: %s", err)
+	}
+
+	// Get the client.
+	k8sCli, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	return k8sCli, nil
+}