Method WatchKey

pkg/ring/kv/dynamodb/client.go:233–278

(ctx context.Context, key string, f func(any) bool)

}

func (c *Client) WatchKey(ctx context.Context, key string, f func(any) bool) {
	watchBackoffConfig := c.backoffConfig
	watchBackoffConfig.MaxRetries = 0
	bo := backoff.New(ctx, watchBackoffConfig)
	syncTimer := time.NewTimer(c.pullerSyncTime)
	defer syncTimer.Stop()

	for bo.Ongoing() {
		out, _, err := c.kv.Query(ctx, dynamodbKey{
			primaryKey: key,
		}, false)
		if err != nil {
			level.Error(c.logger).Log("msg", "error WatchKey", "key", key, "err", err)

			if bo.NumRetries() >= 10 {
				level.Error(c.logger).Log("msg", "failed to WatchKey after retries", "key", key, "err", err)
				if staleData := c.getStaleData(key); staleData != nil {
					if !f(staleData) {
						return
					}
				}
			}
			bo.Wait()
			continue
		}

		decoded, err := c.decodeMultikey(out)
		if err != nil {
			level.Error(c.logger).Log("msg", "error decoding key", "key", key, "err", err)
			continue
		}
		c.updateStaleData(key, decoded, time.Now().UTC())

		if !f(decoded) {
			return
		}

		bo.Reset()
		utiltimer.ResetTimer(syncTimer, c.pullerSyncTime)
		select {
		case <-ctx.Done():
			return
		case <-syncTimer.C:
		}
	}
}

func (c *Client) WatchPrefix(ctx context.Context, prefix string, f func(string, any) bool) {
	watchBackoffConfig := c.backoffConfig