Add canary pinger for periodic shard ownership verification

Implement the client-side pinger that periodically pings random shard
owners to verify:
1. Executors can route to each other based on shard ownership
2. Shard ownership information is accurate
3. The shard distributor is functioning correctly

The pinger:
- Selects random shards at regular intervals (1s with 10% jitter)
- Sends ping requests to the executor owning each shard
- Validates that the receiving executor actually owns the shard
- Logs warnings when ownership is incorrect

Dependencies:
- Requires ShardDistributorExecutorCanaryAPI proto and client
- Will use SpectatorPeerChooser for routing (wired in later commit)

Signed-off-by: Jakob Haahr Taankvist <[email protected]>

Author: jakobht

service/sharddistributor/canary/pinger/canary_client_mock.go

@@ -0,0 +1,142 @@
+package pinger
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	"go.uber.org/fx"
+	"go.uber.org/yarpc"
+	"go.uber.org/zap"
+
+	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
+	"github.com/uber/cadence/common/backoff"
+	"github.com/uber/cadence/common/clock"
+	"github.com/uber/cadence/service/sharddistributor/client/spectatorclient"
+)
+
+//go:generate mockgen -package $GOPACKAGE -destination canary_client_mock.go github.com/uber/cadence/.gen/proto/sharddistributor/v1 ShardDistributorExecutorCanaryAPIYARPCClient
+
+const (
+	pingInterval    = 1 * time.Second
+	pingJitterCoeff = 0.1 // 10% jitter
+	pingTimeout     = 5 * time.Second
+)
+
+// Pinger periodically pings shard owners in the fixed namespace
+type Pinger struct {
+	logger       *zap.Logger
+	timeSource   clock.TimeSource
+	canaryClient sharddistributorv1.ShardDistributorExecutorCanaryAPIYARPCClient
+	namespace    string
+	numShards    int
+	ctx          context.Context
+	cancel       context.CancelFunc
+	wg           sync.WaitGroup
+}
+
+// Params are the parameters for creating a Pinger
+type Params struct {
+	fx.In
+
+	Logger       *zap.Logger
+	TimeSource   clock.TimeSource
+	CanaryClient sharddistributorv1.ShardDistributorExecutorCanaryAPIYARPCClient
+}
+
+// NewPinger creates a new Pinger for the fixed namespace
+func NewPinger(params Params, namespace string, numShards int) *Pinger {
+	return &Pinger{
+		logger:       params.Logger,
+		timeSource:   params.TimeSource,
+		canaryClient: params.CanaryClient,
+		namespace:    namespace,
+		numShards:    numShards,
+	}
+}
+
+// Start begins the periodic ping loop
+func (p *Pinger) Start(ctx context.Context) {
+	p.logger.Info("Starting canary pinger", zap.String("namespace", p.namespace), zap.Int("num_shards", p.numShards))
+	p.ctx, p.cancel = context.WithCancel(context.WithoutCancel(ctx))
+	p.wg.Add(1)
+	go p.pingLoop()
+}
+
+// Stop stops the ping loop
+func (p *Pinger) Stop() {
+	if p.cancel != nil {
+		p.cancel()
+	}
+	p.wg.Wait()
+}
+
+func (p *Pinger) pingLoop() {
+	defer p.wg.Done()
+
+	ticker := p.timeSource.NewTicker(backoff.JitDuration(pingInterval, pingJitterCoeff))
+	defer ticker.Stop()
+
+	p.logger.Info("Starting canary pinger",
+		zap.String("namespace", p.namespace),
+		zap.Int("num_shards", p.numShards))
+
+	for {
+		select {
+		case <-p.ctx.Done():
+			p.logger.Info("Pinger context done, stopping")
+			return
+		case <-ticker.Chan():
+			p.pingRandomShard()
+			ticker.Reset(backoff.JitDuration(pingInterval, pingJitterCoeff))
+		}
+	}
+}
+
+func (p *Pinger) pingRandomShard() {
+	// Pick a random shard number
+	shardNum := rand.Intn(p.numShards)
+	shardKey := fmt.Sprintf("%d", shardNum)
+
+	if err := p.pingShard(shardKey); err != nil {
+		p.logger.Error("Failed to ping shard",
+			zap.String("namespace", p.namespace),
+			zap.String("shard_key", shardKey),
+			zap.Error(err))
+	}
+}
+
+func (p *Pinger) pingShard(shardKey string) error {
+	// Create ping request
+	request := &sharddistributorv1.PingRequest{
+		ShardKey:  shardKey,
+		Namespace: p.namespace,
+	}
+
+	// Create context with deadline for the RPC call
+	ctx, cancel := context.WithTimeout(p.ctx, pingTimeout)
+	defer cancel()
+
+	response, err := p.canaryClient.Ping(ctx, request, yarpc.WithShardKey(shardKey), yarpc.WithHeader(spectatorclient.NamespaceHeader, p.namespace))
+	if err != nil {
+		return fmt.Errorf("ping rpc failed: %w", err)
+	}
+
+	// Verify response
+	if !response.GetOwnsShard() {
+		p.logger.Warn("Executor does not own shard",
+			zap.String("namespace", p.namespace),
+			zap.String("shard_key", shardKey),
+			zap.String("executor_id", response.GetExecutorId()))
+		return fmt.Errorf("executor %s does not own shard %s", response.GetExecutorId(), shardKey)
+	}
+
+	p.logger.Info("Successfully pinged shard owner",
+		zap.String("namespace", p.namespace),
+		zap.String("shard_key", shardKey),
+		zap.String("executor_id", response.GetExecutorId()))
+
+	return nil
+}

service/sharddistributor/canary/pinger/pinger.go

@@ -0,0 +1,115 @@
+package pinger
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"go.uber.org/mock/gomock"
+	"go.uber.org/zap"
+
+	sharddistributorv1 "github.com/uber/cadence/.gen/proto/sharddistributor/v1"
+	"github.com/uber/cadence/common/clock"
+)
+
+func TestPingerStartStop(t *testing.T) {
+	defer goleak.VerifyNone(t)
+
+	ctrl := gomock.NewController(t)
+	mockClient := NewMockShardDistributorExecutorCanaryAPIYARPCClient(ctrl)
+
+	pinger := NewPinger(Params{
+		Logger:       zap.NewNop(),
+		TimeSource:   clock.NewRealTimeSource(),
+		CanaryClient: mockClient,
+	}, "test-ns", 10)
+
+	pinger.Start(context.Background())
+	pinger.Stop()
+}
+
+func TestPingShard_Success(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockClient := NewMockShardDistributorExecutorCanaryAPIYARPCClient(ctrl)
+
+	pinger := NewPinger(Params{
+		Logger:       zap.NewNop(),
+		TimeSource:   clock.NewRealTimeSource(),
+		CanaryClient: mockClient,
+	}, "test-ns", 10)
+	pinger.ctx, pinger.cancel = context.WithCancel(context.Background())
+	defer pinger.cancel()
+
+	mockClient.EXPECT().
+		Ping(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(&sharddistributorv1.PingResponse{
+			OwnsShard:  true,
+			ExecutorId: "127.0.0.1:7953",
+		}, nil)
+
+	err := pinger.pingShard("5")
+	assert.NoError(t, err)
+}
+
+func TestPingShard_DoesNotOwnShard(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockClient := NewMockShardDistributorExecutorCanaryAPIYARPCClient(ctrl)
+
+	pinger := NewPinger(Params{
+		Logger:       zap.NewNop(),
+		TimeSource:   clock.NewRealTimeSource(),
+		CanaryClient: mockClient,
+	}, "test-ns", 10)
+	pinger.ctx, pinger.cancel = context.WithCancel(context.Background())
+	defer pinger.cancel()
+
+	mockClient.EXPECT().
+		Ping(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(&sharddistributorv1.PingResponse{
+			OwnsShard:  false,
+			ExecutorId: "127.0.0.1:7953",
+		}, nil)
+
+	err := pinger.pingShard("5")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "does not own shard")
+}
+
+func TestPingShard_RPCError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockClient := NewMockShardDistributorExecutorCanaryAPIYARPCClient(ctrl)
+
+	pinger := NewPinger(Params{
+		Logger:       zap.NewNop(),
+		TimeSource:   clock.NewRealTimeSource(),
+		CanaryClient: mockClient,
+	}, "test-ns", 10)
+	pinger.ctx, pinger.cancel = context.WithCancel(context.Background())
+	defer pinger.cancel()
+
+	mockClient.EXPECT().
+		Ping(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(nil, errors.New("network error"))
+
+	err := pinger.pingShard("5")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "ping rpc failed")
+}
+
+func TestNewPinger(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mockClient := NewMockShardDistributorExecutorCanaryAPIYARPCClient(ctrl)
+
+	pinger := NewPinger(Params{
+		Logger:       zap.NewNop(),
+		TimeSource:   clock.NewRealTimeSource(),
+		CanaryClient: mockClient,
+	}, "test-ns", 100)
+
+	require.NotNil(t, pinger)
+	assert.Equal(t, "test-ns", pinger.namespace)
+	assert.Equal(t, 100, pinger.numShards)
+}