Repository: influxdata/telegraf-operator
Pull request #63 (2021-07-29 11:46:59)

feat: allow telegraf to support hot reload
For this to fully work, telegraf image must include influxdata/telegraf#9485 which will be released soon.

- [x] Signed the [InfluxData CLA](https://www.influxdata.com/legal/cla/)
  ([if necessary](https://github.com/influxdata/telegraf-operator/blob/master/CONTRIBUTING.md#sign-the-influxdata-cla))
- [x] Tests pass (no build errors)
- [x] Rebased/mergeable



Patch:
diff --git a/Dockerfile b/Dockerfile
index a95bd4e..3678c2f 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -10,7 +10,7 @@ COPY go.sum go.sum
 RUN go mod download
 
 # Copy the go source
-COPY main.go sidecar.go handler.go class_data.go errors.go ./
+COPY main.go sidecar.go handler.go class_data.go errors.go watcher.go updater.go ./
 
 # Build
 RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 GO111MODULE=on go build -a -o manager *.go
diff --git a/class_data.go b/class_data.go
index c1c9288..fa2f0ad 100644
--- a/class_data.go
+++ b/class_data.go
@@ -24,12 +24,26 @@ import (
 	"github.com/influxdata/toml"
 )
 
-type classDataHandler struct {
+// directoryClassDataHandler provides a handler for getting class data from class name.
+type directoryClassDataHandler struct {
 	Logger                   logr.Logger
 	TelegrafClassesDirectory string
 }
 
-func (c *classDataHandler) validateClassData() error {
+// classDataHandler defines interface for validating class data and converting from class name to class data.
+type classDataHandler interface {
+	getData(className string) (string, error)
+	validateClassData() error
+}
+
+func newDirectoryClassDataHandler(logger logr.Logger, telegrafClassesDirectory string) *directoryClassDataHandler {
+	return &directoryClassDataHandler{
+		Logger:                   logger,
+		TelegrafClassesDirectory: telegrafClassesDirectory,
+	}
+}
+
+func (c *directoryClassDataHandler) validateClassData() error {
 	classDataValid := true
 	filesAvailable := false
 
@@ -72,7 +86,8 @@ func (c *classDataHandler) validateClassData() error {
 	return nil
 }
 
-func (c *classDataHandler) getData(className string) (string, error) {
+// getData returns class data for a given class name.
+func (c *directoryClassDataHandler) getData(className string) (string, error) {
 	data, err := ioutil.ReadFile(filepath.Join(c.TelegrafClassesDirectory, className))
 
 	if err != nil {
diff --git a/deploy/dev.yml b/deploy/dev.yml
index 09e980a..d2fd43b 100644
--- a/deploy/dev.yml
+++ b/deploy/dev.yml
@@ -21,6 +21,12 @@ rules:
   - apiGroups: [""]
     resources: ["secrets"]
     verbs: ["*"]
+  - apiGroups: [""]
+    resources: ["namespaces"]
+    verbs: ["get", "list"]
+  - apiGroups: [""]
+    resources: ["pods"]
+    verbs: ["get"]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
@@ -100,6 +106,9 @@ spec:
             # allow injecting telegraf-istio sidecar for pods with
             # istio sidecar annotations enabled
             - --enable-istio-injection=true
+            # if telegraf image supports it, telegraf-watch-config can be set so
+            #  telegraf-opeator and telegraf hot reload changes when classes change
+            # - --telegraf-watch-config=inotify
           env:
             - name: POD_NAMESPACE
               valueFrom:
diff --git a/go.mod b/go.mod
index 017d777..3df9958 100644
--- a/go.mod
+++ b/go.mod
@@ -3,6 +3,7 @@ module github.com/influxdata/telegraf-operator
 go 1.16
 
 require (
+	github.com/fsnotify/fsnotify v1.4.9
 	github.com/go-logr/logr v0.3.0
 	github.com/influxdata/toml v0.0.0-20180607005434-2a2e3012f7cf
 	github.com/naoina/go-stringutil v0.1.0 // indirect
diff --git a/handler.go b/handler.go
index 9911b9d..17a33eb 100644
--- a/handler.go
+++ b/handler.go
@@ -40,7 +40,7 @@ type podInjector struct {
 	decoder *admission.Decoder
 	names.NameGenerator
 	Logger                      logr.Logger
-	ClassDataHandler            *classDataHandler
+	ClassDataHandler            *directoryClassDataHandler
 	SidecarHandler              *sidecarHandler
 	RequireAnnotationsForSecret bool
 }
diff --git a/main.go b/main.go
index 95efa8d..983acfe 100644
--- a/main.go
+++ b/main.go
@@ -117,10 +117,7 @@ func main() {
 
 	logger := setupLog.WithName("podInjector")
 
-	classData := &classDataHandler{
-		Logger:                   logger,
-		TelegrafClassesDirectory: telegrafClassesDirectory,
-	}
+	classData := newDirectoryClassDataHandler(logger, telegrafClassesDirectory)
 
 	err = classData.validateClassData()
 	if err != nil {
@@ -151,6 +148,18 @@ func main() {
 		os.Exit(1)
 	}
 
+	updater, err := newSecretsUpdater(ctrl.Log.WithName("updater"), sidecar)
+	if err != nil {
+		setupLog.Error(err, "setting up secrets updater failed")
+		os.Exit(1)
+	}
+
+	_, err = newTelegrafClassesWatcher(ctrl.Log.WithName("watcher"), telegrafClassesDirectory, updater.onChange)
+	if err != nil {
+		setupLog.Error(err, "setting up watcher failed")
+		os.Exit(1)
+	}
+
 	hookServer.Register("/mutate-v1-pod", &webhook.Admission{Handler: &podInjector{
 		Logger:                      logger,
 		SidecarHandler:              sidecar,
diff --git a/sidecar.go b/sidecar.go
index f1d1c9f..59da1ca 100644
--- a/sidecar.go
+++ b/sidecar.go
@@ -64,8 +64,9 @@ const (
 	TelegrafSecretLabelPod        = "telegraf.influxdata.com/pod"
 )
 
+// sidecarHandler provides logic for handling telegraf sidecars and related secrets.
 type sidecarHandler struct {
-	ClassDataHandler            *classDataHandler
+	ClassDataHandler            classDataHandler
 	Logger                      logr.Logger
 	TelegrafDefaultClass        string
 	TelegrafImage               string
@@ -171,12 +172,7 @@ func (h *sidecarHandler) addTelegrafSidecar(result *sidecarHandlerResponse, pod
 		className = extClass
 	}
 
-	classData, err := h.ClassDataHandler.getData(className)
-	if err != nil {
-		return newNonFatalError(err, "telegraf-operator could not create sidecar container for unknown class")
-	}
-
-	telegrafConf, err := h.assembleConf(pod, classData)
+	telegrafConf, err := h.assembleConf(pod, className)
 	if err != nil {
 		return newNonFatalError(err, "telegraf-operator could not create sidecar container due to error in class data")
 	}
@@ -217,8 +213,17 @@ func (h *sidecarHandler) addContainerAndSecret(result *sidecarHandlerResponse, p
 	return nil
 }
 
+func (h *sidecarHandler) getClassData(className string) (string, error) {
+	return h.ClassDataHandler.getData(className)
+}
+
 // Assembling telegraf configuration
-func (h *sidecarHandler) assembleConf(pod *corev1.Pod, classData string) (telegrafConf string, err error) {
+func (h *sidecarHandler) assembleConf(pod *corev1.Pod, className string) (telegrafConf string, err error) {
+	classData, err := h.ClassDataHandler.getData(className)
+	if err != nil {
+		return "", newNonFatalError(err, "telegraf-operator could not create sidecar container for unknown class")
+	}
+
 	ports := ports(pod)
 	if len(ports) != 0 {
 		path := "/metrics"
diff --git a/updater.go b/updater.go
new file mode 100644
index 0000000..b903269
--- /dev/null
+++ b/updater.go
@@ -0,0 +1,115 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+)
+
+// secretsUpdater updates all secrets managed by telegraf-operator whose contents have changed in all namespaces.
+type secretsUpdater struct {
+	logger       logr.Logger
+	clientset    kubernetes.Interface
+	batchDelay   time.Duration
+	assembleConf func(*corev1.Pod, string) (string, error)
+}
+
+// newSecretsUpdater creates new instance of secretsUpdater.
+func newSecretsUpdater(logger logr.Logger, sidecar *sidecarHandler) (*secretsUpdater, error) {
+	cfg, err := config.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	clientset, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &secretsUpdater{
+		logger:       logger,
+		clientset:    clientset,
+		batchDelay:   10 * time.Second,
+		assembleConf: sidecar.assembleConf,
+	}, nil
+}
+
+// onChange updates secrets all namespaces, handling and logging errors internally
+func (u *secretsUpdater) onChange() {
+	u.logger.Info("checking secrets for updater")
+
+	ctx := context.Background()
+
+	// find all namespaces and find all telegraf-operator managed secrets in each namespace
+	namespaces, err := u.clientset.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		u.logger.Error(err, "unable to list namespaces")
+		return
+	}
+
+	// iterate over all namespaces, trying to update all telegraf-operator managed secrets
+	for _, namespace := range namespaces.Items {
+		err = u.updateSecretsInNamespace(ctx, namespace.Name)
+		if err != nil {
+			u.logger.Error(err, "unable to update secrets", "namespace", namespace)
+			return
+		}
+	}
+}
+
+// updateSecretsInNamespace updates secrets in a single namespace, returning errors if they occur
+func (u *secretsUpdater) updateSecretsInNamespace(ctx context.Context, namespace string) error {
+	secretsClient := u.clientset.CoreV1().Secrets(namespace)
+
+	// find all secrets having the label set by telegraf-operator, limiting results only to secrets
+	// that the telegraf-operator is managing
+	secrets, err := secretsClient.List(ctx, metav1.ListOptions{
+		LabelSelector: TelegrafSecretLabelClassName,
+	})
+	if err != nil {
+		return err
+	}
+
+	for _, secret := range secrets.Items {
+		// get the pod and class name labels
+		podName := secret.GetLabels()[TelegrafSecretLabelPod]
+		className := secret.GetLabels()[TelegrafSecretLabelClassName]
+
+		// if one of the labels was not present, throw an error
+		if podName == "" || className == "" {
+			return fmt.Errorf(`unable to get pod and class name for secret %s in namespace %s; podName="%s"; className="%s"`, secret.Name, secret.Namespace, podName, className)
+		}
+
+		// get the pod that the secret is used in
+		pod, err := u.clientset.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+
+		telegrafConf, err := u.assembleConf(pod, className)
+		if err != nil {
+			return err
+		}
+
+		// check whether secret should be updated, perform the update if needed
+		if string(secret.Data[TelegrafSecretDataKey]) != telegrafConf {
+			u.logger.Info("updating secret", "namespace", namespace, "name", secret.Name, "podName", podName, "class", className)
+			secret.Data[TelegrafSecretDataKey] = []byte(telegrafConf)
+
+			_, err = secretsClient.Update(ctx, &secret, metav1.UpdateOptions{})
+			if err != nil {
+				return err
+			}
+		} else {
+			u.logger.Info("not updating secret", "namespace", namespace, "name", secret.Name, "podName", podName, "class", className)
+		}
+	}
+
+	return nil
+}
diff --git a/watcher.go b/watcher.go
new file mode 100644
index 0000000..63b1ffb
--- /dev/null
+++ b/watcher.go
@@ -0,0 +1,129 @@
+package main
+
+import (
+	"io/ioutil"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+	"github.com/go-logr/logr"
+)
+
+type telegrafClassesOnChange func()
+
+// telegrafClassesWatcher allows monitoring a directory with telegraf classes using
+// fsnotify package and batching multiple events to reduce number of Kubernetes API calls.
+type telegrafClassesWatcher struct {
+	logger        logr.Logger
+	watcherEvents chan fsnotify.Event
+	onChange      telegrafClassesOnChange
+
+	eventCount   uint64
+	eventChannel chan struct{}
+	eventDelay   time.Duration
+}
+
+// newTelegrafClassesWatcher creates a new instance of telegrafClassesWatcher.
+func newTelegrafClassesWatcher(logger logr.Logger, telegrafClassesDirectory string, onChange telegrafClassesOnChange) (*telegrafClassesWatcher, error) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, err
+	}
+
+	// watching the contents of classes directory requires adding the directory as well as most child elements
+	logger.Info("adding directory to watcher", "directory", telegrafClassesDirectory)
+	watcher.Add(telegrafClassesDirectory)
+
+	items, err := ioutil.ReadDir(telegrafClassesDirectory)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, item := range items {
+		name := item.Name()
+
+		// Add all items in the classes directory except for current/previous secret contents that begin with "..", "." and ".."
+		// explicitly add "..data" directory as this is the directory that maps current state of the secret.
+		//
+		// Example listing of classes directory:
+		//
+		// drwxrwxrwt 3 root root  100 Jul 29 12:27 .
+		// drwxr-xr-x 1 root root 4096 Jul 29 12:26 ..
+		// drwxr-xr-x 2 root root   60 Jul 29 12:27 ..2021_07_29_12_27_39.113045998
+		// lrwxrwxrwx 1 root root   31 Jul 29 12:27 ..data -> ..2021_07_29_12_27_39.113045998
+		// lrwxrwxrwx 1 root root   20 Jul 29 12:26 app -> ..data/app
+		// lrwxrwxrwx 1 root root   20 Jul 29 12:26 basic -> ..data/basic
+		//
+		// in the above case, we want to match "..data", "app" and "basic", but skip ".", ".." and "..2021_07_29_12_27_39.113045998"
+		if name == "..data" || (name != "." && !strings.HasPrefix(name, "..")) {
+			p := filepath.Join(telegrafClassesDirectory, name)
+			logger.Info("adding item to watch", "path", p)
+			err = watcher.Add(p)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	w := &telegrafClassesWatcher{
+		watcherEvents: watcher.Events,
+		logger:        logger,
+		onChange:      onChange,
+
+		// allow large number of messages in the channel to avoid blocking
+		eventChannel: make(chan struct{}, 100),
+
+		// delay by 10 seconds to group multiple fsnotify events into single invocation of callback
+		eventDelay: 10 * time.Second,
+	}
+
+	w.createGoroutines()
+
+	return w, nil
+}
+
+// createGoroutines runs all required goroutines for the watcher.
+func (w *telegrafClassesWatcher) createGoroutines() {
+	go w.batchChanges()
+	go w.monitorForChanges()
+}
+
+// batchChanges is a goroutine that batches invocations of onChange()
+// based on events sent from monitorForChanges().
+func (w *telegrafClassesWatcher) batchChanges() {
+	var previousEventCount uint64
+	for range w.eventChannel {
+
+		currentEventCount := atomic.LoadUint64(&w.eventCount)
+
+		// check if counter is same as last time events were processed,
+		// only delay and batch if it is different
+		if currentEventCount != previousEventCount {
+			// delay processing of the event to batch multiple events from file
+			time.Sleep(w.eventDelay)
+
+			// update  the event counter again to latest, potentially different value
+			currentEventCount = atomic.LoadUint64(&w.eventCount)
+
+			w.onChange()
+
+			previousEventCount = currentEventCount
+		}
+	}
+}
+
+// monitorForChanges helps batch events from fsnotify by incrementing a counter and
+// sending events using an internal channel, then handled by batchChanges().
+func (w *telegrafClassesWatcher) monitorForChanges() {
+	for {
+		_, ok := <-w.watcherEvents
+		if ok {
+			// increase the event counter and send a message to goroutine
+			// that batches invocations of onChange()
+			atomic.AddUint64(&w.eventCount, 1)
+			w.eventChannel <- struct{}{}
+		}
+	}
+}


Test patch:
diff --git a/class_data_test.go b/class_data_test.go
index a6e5b0a..e5aaea4 100644
--- a/class_data_test.go
+++ b/class_data_test.go
@@ -46,7 +46,7 @@ func Test_classDataHandler_getData(t *testing.T) {
 			dir := createTempClassesDirectory(t, tt.classes)
 			defer os.RemoveAll(dir)
 
-			testClassDataHandler := &classDataHandler{
+			testClassDataHandler := &directoryClassDataHandler{
 				Logger:                   logger,
 				TelegrafClassesDirectory: dir,
 			}
@@ -95,7 +95,7 @@ func Test_classDataHandler_validateClassData(t *testing.T) {
 			dir := createTempClassesDirectory(t, tt.classes)
 			defer os.RemoveAll(dir)
 
-			testClassDataHandler := &classDataHandler{
+			testClassDataHandler := &directoryClassDataHandler{
 				Logger:                   logger,
 				TelegrafClassesDirectory: dir,
 			}
diff --git a/handler_test.go b/handler_test.go
index 46f1278..a2a1a71 100644
--- a/handler_test.go
+++ b/handler_test.go
@@ -839,7 +839,7 @@ func Test_podInjector_Handle(t *testing.T) {
 
 			logger := &logrTesting.TestLogger{T: t}
 
-			testClassDataHandler := &classDataHandler{
+			testClassDataHandler := &directoryClassDataHandler{
 				Logger:                   logger,
 				TelegrafClassesDirectory: dir,
 			}
diff --git a/sidecar_test.go b/sidecar_test.go
index 6f7844f..2e49656 100644
--- a/sidecar_test.go
+++ b/sidecar_test.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"bytes"
+	"fmt"
 	"os"
 	"reflect"
 	"strings"
@@ -54,6 +55,27 @@ stringData:
 type: Opaque`
 )
 
+type mockClassDataHandler struct {
+	classes map[string]string
+}
+
+func newMockClassDataHandler(classes map[string]string) *mockClassDataHandler {
+	return &mockClassDataHandler{classes: classes}
+}
+
+func (h *mockClassDataHandler) validateClassData() error {
+	return nil
+}
+
+func (m *mockClassDataHandler) getData(className string) (string, error) {
+	v, ok := m.classes[className]
+	if ok {
+		return v, nil
+	} else {
+		return "", fmt.Errorf("class %s not found", className)
+	}
+}
+
 func Test_skip(t *testing.T) {
 	handler := &sidecarHandler{
 		RequestsCPU:    defaultRequestsCPU,
@@ -260,6 +282,7 @@ func Test_assembleConf(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 
 			handler := &sidecarHandler{
+				ClassDataHandler:            newMockClassDataHandler(map[string]string{"class": tt.classData}),
 				EnableDefaultInternalPlugin: tt.enableDefaultInternalPlugin,
 				RequestsCPU:                 defaultRequestsCPU,
 				RequestsMemory:              defaultRequestsMemory,
@@ -267,7 +290,7 @@ func Test_assembleConf(t *testing.T) {
 				LimitsMemory:                defaultLimitsMemory,
 				Logger:                      &logrTesting.TestLogger{T: t},
 			}
-			gotConfig, err := handler.assembleConf(tt.pod, tt.classData)
+			gotConfig, err := handler.assembleConf(tt.pod, "class")
 			if (err != nil) != tt.wantErr {
 				t.Errorf("assembleConf() error = %v, wantErr %v", err, tt.wantErr)
 				return
@@ -592,7 +615,7 @@ status: {}
 				},
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
-						corev1.Container{
+						{
 							Name:  "telegraf",
 							Image: "alpine:latest",
 						},
@@ -690,7 +713,7 @@ status: {}
 				},
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
-						corev1.Container{
+						{
 							Name:  "telegraf-istio",
 							Image: "alpine:latest",
 						},
@@ -896,7 +919,7 @@ status: {}
 
 			logger := &logrTesting.TestLogger{T: t}
 
-			testClassDataHandler := &classDataHandler{
+			testClassDataHandler := &directoryClassDataHandler{
 				Logger:                   logger,
 				TelegrafClassesDirectory: dir,
 			}
diff --git a/updater_test.go b/updater_test.go
new file mode 100644
index 0000000..50387c2
--- /dev/null
+++ b/updater_test.go
@@ -0,0 +1,193 @@
+package main
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/go-logr/logr"
+	logrTesting "github.com/go-logr/logr/testing"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+// mockSidecarHandler mocks minimal interface of sidecar handler that updater needs, rendering strings and
+// returning sorted list of objects that assembleConf() method was called for.
+type mockSidecarHandler struct {
+	assembleConfResults []string
+}
+
+// assembleConf generates a mock result that is not a valid telegraf configuration, but namespace, name and class name separated by dot for testing purposes
+func (h *mockSidecarHandler) assembleConf(pod *corev1.Pod, className string) (string, error) {
+	val := fmt.Sprintf("%s.%s.%s", pod.Namespace, pod.Name, className)
+	h.assembleConfResults = append(h.assembleConfResults, val)
+	return val, nil
+}
+
+// get method returns sorted list of invocations that assempleConf() method was called for.
+func (h *mockSidecarHandler) get() []string {
+	sort.Strings(h.assembleConfResults)
+	return h.assembleConfResults
+}
+
+// secretsUpdaterTest is a helper structure to test secretsUpdater with fake objects.
+type secretsUpdaterTest struct {
+	logger      logr.Logger
+	updater     *secretsUpdater
+	fakeClient  *fake.Clientset
+	mockSidecar *mockSidecarHandler
+	ns1         *corev1.Namespace
+	pod1        *corev1.Pod
+	pod2        *corev1.Pod
+	secret1     *corev1.Secret
+	secret2     *corev1.Secret
+	secret3     *corev1.Secret
+}
+
+// newSecretsUpdaterTest creates a new instance of secretsUpdaterTest without initializing all objects.
+// The createObjects() method should be called before testing to create K8s client and other objects.
+func newSecretsUpdaterTest(t *testing.T, objects ...runtime.Object) *secretsUpdaterTest {
+	logger := &logrTesting.TestLogger{T: t}
+
+	ns1 := &corev1.Namespace{
+		TypeMeta: v1.TypeMeta{Kind: "Namespace", APIVersion: "v1"},
+		ObjectMeta: v1.ObjectMeta{
+			Name: "ns1",
+		},
+	}
+
+	pod1 := &corev1.Pod{
+		TypeMeta: v1.TypeMeta{Kind: "Pod", APIVersion: "v1"},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "pod1",
+			Namespace: "ns1",
+			Annotations: map[string]string{
+				TelegrafClass:       "test",
+				TelegrafMetricsPath: "/metrics",
+				TelegrafMetricsPort: "6060",
+			},
+		},
+	}
+
+	pod2 := &corev1.Pod{
+		TypeMeta: v1.TypeMeta{Kind: "Pod", APIVersion: "v1"},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "pod2",
+			Namespace: "ns1",
+			Annotations: map[string]string{
+				TelegrafClass:       "app",
+				TelegrafMetricsPath: "/metrics",
+				TelegrafMetricsPort: "6060",
+			},
+		},
+	}
+
+	secret1 := &corev1.Secret{
+		TypeMeta: v1.TypeMeta{Kind: "Pod", APIVersion: "v1"},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "telegraf-config-pod1",
+			Namespace: "ns1",
+			Labels: map[string]string{
+				TelegrafSecretLabelClassName: "test",
+				TelegrafSecretLabelPod:       pod1.GetObjectMeta().GetName(),
+			},
+		},
+		Data: map[string][]byte{
+			TelegrafSecretDataKey: []byte("ns1.pod1.test"),
+		},
+	}
+
+	secret2 := &corev1.Secret{
+		TypeMeta: v1.TypeMeta{Kind: "Pod", APIVersion: "v1"},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "telegraf-config-pod2",
+			Namespace: "ns1",
+			Labels: map[string]string{
+				TelegrafSecretLabelClassName: "app",
+				TelegrafSecretLabelPod:       pod2.GetObjectMeta().GetName(),
+			},
+		},
+		Data: map[string][]byte{
+			TelegrafSecretDataKey: []byte("ns1.pod2.app"),
+		},
+	}
+
+	secret3 := &corev1.Secret{
+		TypeMeta: v1.TypeMeta{Kind: "Pod", APIVersion: "v1"},
+		ObjectMeta: v1.ObjectMeta{
+			Name:      "unrelated1",
+			Namespace: "ns1",
+			Labels:    map[string]string{},
+		},
+		Data: map[string][]byte{
+			"key": []byte("value"),
+		},
+	}
+
+	return &secretsUpdaterTest{
+		logger:  logger,
+		ns1:     ns1,
+		pod1:    pod1,
+		pod2:    pod2,
+		secret1: secret1,
+		secret2: secret2,
+		secret3: secret3,
+	}
+}
+
+// createObjects creates Kubernetes fake clientset as well as other objects that depend on it.
+func (t *secretsUpdaterTest) createObjects() {
+	t.fakeClient = fake.NewSimpleClientset(
+		t.ns1,
+		t.pod1, t.pod2,
+		t.secret1, t.secret2, t.secret3,
+	)
+
+	t.mockSidecar = &mockSidecarHandler{}
+
+	t.updater = &secretsUpdater{
+		logger:       t.logger,
+		clientset:    t.fakeClient,
+		assembleConf: t.mockSidecar.assembleConf,
+	}
+
+}
+
+func Test_AssembleConfForSecretsWithLabels(t *testing.T) {
+	test := newSecretsUpdaterTest(t)
+
+	test.createObjects()
+	test.updater.onChange()
+	// validate that assembleConf() was called for secret1 and secret2, but not secret3
+	if want, got := "ns1.pod1.test;ns1.pod2.app", strings.Join(test.mockSidecar.get(), ";"); want != got {
+		t.Errorf("wrong configurations assembled; want=%q; got=%q", want, got)
+	}
+
+	// assume 4 actions called on Kubernetes client - list namespaces, list secrets, get 2 pods
+	if want, got := 4, len(test.fakeClient.Actions()); want != got {
+		t.Errorf("wanted %d actions to be invoked, got %d", want, got)
+	}
+}
+
+func Test_Secret1Updated(t *testing.T) {
+	test := newSecretsUpdaterTest(t, &corev1.Secret{})
+	// store the secret value to something different than what assembleConf() will return
+	test.secret2.Data[TelegrafSecretDataKey] = []byte("invalid")
+
+	test.createObjects()
+	test.updater.onChange()
+
+	// assume 5 actions called on Kubernetes client - list namespaces, list secrets, get 2 pods, update secret
+	if want, got := 5, len(test.fakeClient.Actions()); want != got {
+		t.Errorf("wanted %d actions to be invoked, got %d", want, got)
+	}
+
+	// verify that last action is an update of a secret
+	lastAction := test.fakeClient.Actions()[4]
+	if !lastAction.Matches("update", "secrets") {
+		t.Errorf("last action mismatch: %v", lastAction)
+	}
+}
diff --git a/watcher_test.go b/watcher_test.go
new file mode 100644
index 0000000..4d1d174
--- /dev/null
+++ b/watcher_test.go
@@ -0,0 +1,84 @@
+package main
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+	logrTesting "github.com/go-logr/logr/testing"
+)
+
+type mockOnChange struct {
+	count int64
+}
+
+func (m *mockOnChange) onChange() {
+	atomic.AddInt64(&m.count, 1)
+}
+
+func (m *mockOnChange) get() int {
+	return int(atomic.LoadInt64(&m.count))
+}
+
+func testWatcher(t *testing.T, onChange telegrafClassesOnChange) *telegrafClassesWatcher {
+	logger := &logrTesting.TestLogger{T: t}
+
+	w := &telegrafClassesWatcher{
+		watcherEvents: make(chan fsnotify.Event, 100),
+		logger:        logger,
+		onChange:      onChange,
+		eventChannel:  make(chan struct{}, 100),
+		eventDelay:    50 * time.Millisecond,
+	}
+
+	w.createGoroutines()
+
+	return w
+}
+
+func sendTestWatcherEvent(w *telegrafClassesWatcher) {
+	w.watcherEvents <- fsnotify.Event{Name: "dummy", Op: fsnotify.Write}
+}
+
+func Test_Watcher_SingleEvent(t *testing.T) {
+	mock := &mockOnChange{}
+	watcher := testWatcher(t, mock.onChange)
+	sendTestWatcherEvent(watcher)
+	time.Sleep(watcher.eventDelay * 2)
+
+	if want, got := 1, mock.get(); want != got {
+		t.Errorf("want %v, got %v", want, got)
+	}
+}
+
+func Test_Watcher_MultipleEvents(t *testing.T) {
+	mock := &mockOnChange{}
+	watcher := testWatcher(t, mock.onChange)
+	sendTestWatcherEvent(watcher)
+	sendTestWatcherEvent(watcher)
+	sendTestWatcherEvent(watcher)
+	time.Sleep(watcher.eventDelay * 2)
+
+	if want, got := 1, mock.get(); want != got {
+		t.Errorf("want %v, got %v", want, got)
+	}
+}
+
+func Test_Watcher_EventsOverTime(t *testing.T) {
+	mock := &mockOnChange{}
+	watcher := testWatcher(t, mock.onChange)
+	sendTestWatcherEvent(watcher)
+	time.Sleep(watcher.eventDelay * 2)
+	sendTestWatcherEvent(watcher)
+	sendTestWatcherEvent(watcher)
+	time.Sleep(watcher.eventDelay * 2)
+	sendTestWatcherEvent(watcher)
+	sendTestWatcherEvent(watcher)
+	sendTestWatcherEvent(watcher)
+	time.Sleep(watcher.eventDelay * 2)
+
+	if want, got := 3, mock.get(); want != got {
+		t.Errorf("want %v, got %v", want, got)
+	}
+}