Commands wrapped in a Watch callback are not hooked if client is a cluster

[yansal]

Expected Behavior

Commands wrapped in a Watch callback should enter the client hook.

Current Behavior

Commands wrapped in a Watch callback do enter the client hook if the client has been created with the NewClient function, but not if the client is a cluster created with the NewClusterClient function

Steps to Reproduce

This program works:

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/go-redis/redis/v8"
)

func main() {
	if err := main1(); err != nil {
		log.Fatal(err)
	}
}

func main1() error {
	client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
	client.AddHook(hook{})
	ctx := context.Background()
	const key = "foo"
	if err := client.Watch(ctx, func(tx *redis.Tx) error {
		return tx.Set(ctx, key, "bar", 0).Err()
	}, key); err != nil {
		return err
	}
	return nil
}

type hook struct{}

func (h hook) BeforeProcess(ctx context.Context, cmder redis.Cmder) (context.Context, error) {
	return ctx, nil
}

func (h hook) AfterProcess(ctx context.Context, cmder redis.Cmder) error {
	fmt.Println(cmder)
	return nil
}

func (h hook) BeforeProcessPipeline(ctx context.Context, cmders []redis.Cmder) (context.Context, error) {
	return ctx, nil
}

func (h hook) AfterProcessPipeline(ctx context.Context, cmders []redis.Cmder) error { return nil }

It prints

watch foo: OK
set foo bar: OK
unwatch: OK

If we replace the client with a cluster client, then the program prints nothing.

This bug is present with the git sha 38caa12.

[github-actions]

This issue is marked stale. It will be closed in 30 days if it is not updated.

[yansal]

I do confirm that the issue still exists with the latest release of github.com/redis/go-redis.

The following program should print the following output but it doesn't.

watch foo: OK
set foo bar: OK
unwatch: OK

It works with a simple client, but it doesn't work with a cluster client.

module redis-watch

go 1.21.1

require github.com/redis/go-redis/v9 v9.2.1

require (
	github.com/cespare/xxhash/v2 v2.2.0 // indirect
	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
)

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/redis/go-redis/v9"
)

func main() {
	if err := main1(); err != nil {
		log.Fatal(err)
	}
}

func main1() error {
	client := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{"localhost:7000"}})
	client.AddHook(hook{})
	ctx := context.Background()
	const (
		key        = "foo"
		value      = "bar"
		expiration = 0
	)
	if err := client.Watch(ctx, func(tx *redis.Tx) error {
		return tx.Set(ctx, key, value, expiration).Err()
	}, key); err != nil {
		return err
	}
	return nil
}

type hook struct{}

func (h hook) DialHook(next redis.DialHook) redis.DialHook {
	return next
}
func (h hook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		err := next(ctx, cmd)
		if err != nil {
			return err
		}
		fmt.Println(cmd)
		return nil
	}
}
func (h hook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		err := next(ctx, cmds)
		if err != nil {
			return err
		}
		fmt.Println(cmds)
		return nil
	}
}

[trebor8x]

When using a ClusterClient one has to utilize OnNewNode, so that hooks are propagated to every new Client.
An example can be seen here:

go-redis/extra/redisotel/tracing.go

Line 35 in 0f0a284

rdb.OnNewNode(func(rdb *redis.Client) {

[ndyakov]

@yansal there are some changes on the hooks happening in #3320 , would you mind testing with the latest commit there?

[ndyakov]

closing for one month of inactivity.