Repository: beatlabs/harvester (go)

Issue:
Redis monitoring always setting the version to 0
## Is your feature request related to a problem? Please describe

When monitoring a Redis key, the version is always set to `0`. See [here](https://github.com/beatlabs/harvester/blob/master/monitor/redis/watcher.go#L80):

```go
// file: /monitor/redis/watcher.go#L80

changes = append(changes, change.New(config.SourceRedis, key, values[i].(string), 0))
```

Then, when the field's value is set, it will always trigger a debug log and send a change notification, see [here](https://github.com/beatlabs/harvester/blob/master/config/config.go#L102-L123):

```go
// file: /config/config.go#L102-L123

func (f *Field) Set(value string, version uint64) error {
	if version != 0 && version < f.version { // <--- version is 0, so it will not match
		log.Errorf("version %d is older than the field's %q (version %d)", version, f.name, f.version)
		return nil
	}

	if version != 0 && version == f.version { // <--- version is 0, so it will not match
		log.Debugf("version %d is the same as field's %q", version, f.name)
		return nil
	}

	prevValue := f.structField.String()

	if err := f.structField.SetString(value); err != nil {
		return err
	}

	f.version = version
	log.Debugf("field %q updated with value %q, version: %d", f.name, f, version) // <--- this one will always evaluate
	f.sendNotification(prevValue, value) // <--- this one as well
	return nil
}
```

2 issues:
1. There will be one log for each Redis key and at each monitoring tick, even though the value didn't change
2. It sends a notification to `chNotifications` for no reason

## Is this a bug? Please provide steps to reproduce, a failing test etc.

Yes.

To reproduce, monitor a Redis key, enable debug logs and tail the logs.

## Describe the solution

We need to increment the Redis version if there is a change.

Proposed solution:
1. When reading a value from Redis, generate a checksum
2. If the checksum is the same as the previously seen checksum for this key, discard the value (no changes)
3. If the checksum is different, it means that there's a change. In this case, get the latest known version for this key, increment it and trigger a notification.


Pull request description:
Fix Redis monitoring version
## Which problem is this PR solving?

Fixes #105: Redis monitoring always setting the version to 0

## Short description of the changes

Versioning:
* Hash and store the latest known value of each key
* Introduce a version counter, incremented every time a different value is found for a key
* Add Redis to docker-compose, to perform integration tests
* Unit and integration tests

Other changes:
* Do not send empty `[]*change.Change` if there are nos change when polling Redis

Patch:
diff --git a/docker-compose.yml b/docker-compose.yml
index 3492b25..250d5c1 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -17,3 +17,9 @@ services:
       - '8500:8500'
       - '8600:8600'
       - '8600:8600/udp'
+
+  redis:
+    container_name: harvester_redis_dev
+    image: redis:alpine
+    ports:
+      - '6379:6379'
\ No newline at end of file
diff --git a/monitor/redis/watcher.go b/monitor/redis/watcher.go
index abd2fa4..47e4925 100644
--- a/monitor/redis/watcher.go
+++ b/monitor/redis/watcher.go
@@ -3,6 +3,8 @@ package redis
 
 import (
 	"context"
+	"crypto/md5"
+	"encoding/hex"
 	"errors"
 	"time"
 
@@ -16,6 +18,8 @@ import (
 type Watcher struct {
 	client       redis.UniversalClient
 	keys         []string
+	versions     []uint64
+	hashes       []string
 	pollInterval time.Duration
 }
 
@@ -34,6 +38,8 @@ func New(client redis.UniversalClient, pollInterval time.Duration, keys []string
 	return &Watcher{
 		client:       client,
 		keys:         keys,
+		versions:     make([]uint64, len(keys)),
+		hashes:       make([]string, len(keys)),
 		pollInterval: pollInterval,
 	}, nil
 }
@@ -77,8 +83,26 @@ func (w *Watcher) getValues(ctx context.Context, ch chan<- []*change.Change) {
 			continue
 		}
 
-		changes = append(changes, change.New(config.SourceRedis, key, values[i].(string), 0))
+		value := values[i].(string)
+		hash := w.hash(value)
+		if hash == w.hashes[i] {
+			continue
+		}
+
+		w.versions[i]++
+		w.hashes[i] = hash
+
+		changes = append(changes, change.New(config.SourceRedis, key, value, w.versions[i]))
+	}
+
+	if len(changes) == 0 {
+		return
 	}
 
 	ch <- changes
 }
+
+func (w *Watcher) hash(value string) string {
+	hash := md5.Sum([]byte(value))
+	return hex.EncodeToString(hash[:])
+}


Test patch:
diff --git a/monitor/redis/watcher_integration_test.go b/monitor/redis/watcher_integration_test.go
index 080b31a..c204f32 100644
--- a/monitor/redis/watcher_integration_test.go
+++ b/monitor/redis/watcher_integration_test.go
@@ -1,3 +1,4 @@
+//go:build integration
 // +build integration
 
 package redis
@@ -8,6 +9,7 @@ import (
 	"time"
 
 	"github.com/beatlabs/harvester/change"
+	"github.com/beatlabs/harvester/config"
 	"github.com/go-redis/redis/v8"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -27,40 +29,57 @@ func TestWatch(t *testing.T) {
 		val3 = "value3"
 	)
 
-	set(t, client, key1, val1)
-	set(t, client, key2, val2)
-	set(t, client, key3, val3)
 	defer func() {
 		del(t, client, key1)
 		del(t, client, key2)
 		del(t, client, key3)
 	}()
 
-	ch := make(chan []*change.Change)
 	w, err := New(client, 10*time.Millisecond, []string{key1, key2, key3})
 	require.NoError(t, err)
 	require.NotNil(t, w)
+
 	ctx, cnl := context.WithCancel(context.Background())
 	defer cnl()
+
+	// Initial values, set even before watching - not all keys have a value
+	set(t, client, key1, val1)
+	set(t, client, key2, val1)
+
+	// Start watching
+	ch := make(chan []*change.Change, 100)
+	defer close(ch)
 	err = w.Watch(ctx, ch)
 	require.NoError(t, err)
 
-	for i := 0; i < 2; i++ {
-		cc := <-ch
-		for _, cng := range cc {
-			switch cng.Key() {
-			case key1:
-				assert.Equal(t, val1, cng.Value())
-			case key2:
-				assert.Equal(t, val2, cng.Value())
-			case key3:
-				assert.Equal(t, val3, cng.Value())
-			default:
-				assert.Fail(t, "key invalid", cng.Key())
-			}
-			assert.True(t, cng.Version() == 0)
-		}
+	// First values update
+	time.Sleep(1 * time.Second)
+	set(t, client, key1, val1) // Same value
+	set(t, client, key2, val2)
+	set(t, client, key3, val1) // First value for this key
+
+	// Second values update
+	time.Sleep(1 * time.Second)
+	set(t, client, key1, val1) // Same value
+	set(t, client, key2, val1) // Second value - same as the initial value
+	set(t, client, key3, val3)
+
+	time.Sleep(1 * time.Second)
+
+	found := transformChangesToSlices(ch)
+	expected := []*change.Change{
+		// Initial values
+		change.New(config.SourceRedis, key1, val1, 1),
+		change.New(config.SourceRedis, key2, val1, 1),
+		// First update
+		change.New(config.SourceRedis, key2, val2, 2),
+		change.New(config.SourceRedis, key3, val1, 1),
+		// Second update
+		change.New(config.SourceRedis, key2, val1, 3),
+		change.New(config.SourceRedis, key3, val3, 2),
 	}
+
+	assert.Equal(t, expected, found)
 }
 
 func set(t *testing.T, client redis.UniversalClient, key string, val string) {
@@ -74,3 +93,15 @@ func del(t *testing.T, client redis.UniversalClient, key string) {
 	require.NoError(t, err)
 	require.Equal(t, int64(1), delResult)
 }
+
+func transformChangesToSlices(ch chan []*change.Change) []*change.Change {
+	changes := make([]*change.Change, 0)
+	for {
+		select {
+		case cc := <-ch:
+			changes = append(changes, cc...)
+		default:
+			return changes
+		}
+	}
+}
diff --git a/monitor/redis/watcher_test.go b/monitor/redis/watcher_test.go
index 80c9778..7ce6cbe 100644
--- a/monitor/redis/watcher_test.go
+++ b/monitor/redis/watcher_test.go
@@ -2,10 +2,12 @@ package redis
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/beatlabs/harvester/change"
+	"github.com/beatlabs/harvester/config"
 	"github.com/go-redis/redis/v8"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -65,3 +67,84 @@ func TestWatcher_Watch(t *testing.T) {
 		})
 	}
 }
+
+func TestWatcher_Versioning(t *testing.T) {
+	client := (&clientStub{t: t}).
+		WithValues("val1.1", "val2.1", "val3.1"). // Initial values
+		WithValues("val1.1", "val2.2", "val3.2"). // Only keys 2 and 3 are updated
+		WithValues("val1.1", "val2.1", "val3.2")  // Only 2 is updated, to its previous value
+
+	expected := [][]*change.Change{
+		{
+			change.New(config.SourceRedis, "key1", "val1.1", 1),
+			change.New(config.SourceRedis, "key2", "val2.1", 1),
+			change.New(config.SourceRedis, "key3", "val3.1", 1),
+		},
+		{
+			change.New(config.SourceRedis, "key2", "val2.2", 2),
+			change.New(config.SourceRedis, "key3", "val3.2", 2),
+		},
+		{
+			change.New(config.SourceRedis, "key2", "val2.1", 3),
+		},
+	}
+
+	w, err := New(client, 1*time.Millisecond, []string{"key1", "key2", "key3"})
+	require.NoError(t, err)
+	assert.Equal(t, []uint64{0, 0, 0}, w.versions)
+	assert.Equal(t, []string{"", "", ""}, w.hashes)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	ch := make(chan []*change.Change, 10)
+	err = w.Watch(ctx, ch)
+	assert.NoError(t, err)
+
+	time.Sleep(100 * time.Millisecond)
+
+	cancel()
+
+	found := make([][]*change.Change, 0)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+
+	go func() {
+		for {
+			select {
+			case cc := <-ch:
+				if len(cc) == 0 {
+					break
+				}
+				found = append(found, cc)
+			default:
+				wg.Done()
+				return
+			}
+		}
+	}()
+	wg.Wait()
+
+	assert.Equal(t, expected, found)
+}
+
+type clientStub struct {
+	t *testing.T
+	*redis.Client
+
+	cmds []*redis.SliceCmd
+}
+
+func (c *clientStub) WithValues(values ...interface{}) *clientStub {
+	c.cmds = append(c.cmds, redis.NewSliceResult(values, nil))
+	return c
+}
+
+func (c *clientStub) MGet(_ context.Context, keys ...string) *redis.SliceCmd {
+	if len(c.cmds) == 0 {
+		return redis.NewSliceResult(make([]interface{}, len(keys)), nil)
+	}
+	shifted := c.cmds[0]
+	c.cmds = c.cmds[1:]
+	return shifted
+}