/*
Copyright 2014 The Kubernetes Authors.
Copyright 2020 Authors of Arktos - file modified.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cache

import (
	"context"
	"errors"
	"fmt"
	"io"
	"k8s.io/client-go/apiserverupdate"
	"math/rand"
	"reflect"
	"strings"
	"sync"
	"time"

	"github.com/grafov/bcast"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/clock"
	"k8s.io/apimachinery/pkg/util/naming"
	utilnet "k8s.io/apimachinery/pkg/util/net"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/tools/pager"
	"k8s.io/klog"
	"k8s.io/utils/trace"
)

const defaultExpectedTypeName = "<unspecified>"

type filterBound struct {
	//ResetCh is the channel for incoming bounds changes
	resetCh *bcast.Member

	//sourceName is the app name that sends the reset message
	sourceName string

	//ownerKind is the owner's object kind
	ownerKind string

	//bounds is an array to save hashkey lower and upper bounds. The lower bound is
	// inclusive while the upper bound is exclusive
	bounds []int64
}

// Reflector watches a specified resource and causes all changes to be reflected in the given store.
type Reflector struct {
	// name identifies this reflector. By default it will be a file:line if possible.
	name string

	// The name of the type we expect to place in the store. The name
	// will be the stringification of expectedGVK if provided, and the
	// stringification of expectedType otherwise. It is for display
	// only, and should not be used for parsing or comparison.
	expectedTypeName string
	// The type of object we expect to place in the store.
	expectedType reflect.Type
	// The GVK of the object we expect to place in the store if unstructured.
	expectedGVK *schema.GroupVersionKind
	// The destination to sync up with the watch source
	store Store
	// listerWatcher is used to perform lists and watches.
	listerWatcher ListerWatcher

	// backoff manages backoff of ListWatch
	backoffManager wait.BackoffManager

	resyncPeriod time.Duration
	ShouldResync func() bool
	// clock allows tests to manipulate time
	clock clock.Clock
	// paginatedResult defines whether pagination should be forced for list calls.
	// It is set based on the result of the initial list call.
	paginatedResult bool
	// lastSyncResourceVersion is the resource version token last
	// observed when doing a sync with the underlying store
	// it is thread safe, but not synchronized with the underlying store
	lastSyncResourceVersion string
	// isLastSyncResourceVersionGone is true if the previous list or watch request with lastSyncResourceVersion
	// failed with an HTTP 410 (Gone) status code.
	isLastSyncResourceVersionGone bool
	// lastSyncResourceVersionMutex guards read/write access to lastSyncResourceVersion
	lastSyncResourceVersionMutex sync.RWMutex
	// WatchListPageSize is the requested chunk size of initial and resync watch lists.
	// If unset, for consistent reads (RV="") or reads that opt-into arbitrarily old data
	// (RV="0") it will default to pager.PageSize, for the rest (RV != "" && RV != "0")
	// it will turn off pagination to allow serving them from watch cache.
	// NOTE: It should be used carefully as paginated lists are always served directly from
	// etcd, which is significantly less efficient and may lead to serious performance and
	// scalability problems.
	WatchListPageSize int64

	// filterBounds are a list of list/watch filtering bounds
	filterBounds []filterBound

	// aggChan is the aggregate channel of reset channels
	aggChan chan interface{}

	// clientSetUpdateChan is the channel to get client set updates
	clientSetUpdateChan *bcast.Member

	// There are some watch that can only happen to certain api servers
	allowPartialWatch bool
}

var (
	// We try to spread the load on apiserver by setting timeouts for
	// watch requests - it is random in [minWatchTimeout, 2*minWatchTimeout].
	minWatchTimeout = 5 * time.Minute
)

// NewNamespaceKeyedIndexerAndReflector creates an Indexer and a Reflector
// The indexer is configured to key on namespace
func NewNamespaceKeyedIndexerAndReflector(lw ListerWatcher, expectedType interface{}, resyncPeriod time.Duration) (indexer Indexer, reflector *Reflector) {
	indexer = NewIndexer(MetaNamespaceKeyFunc, Indexers{NamespaceIndex: MetaNamespaceIndexFunc})
	reflector = NewReflector(lw, expectedType, indexer, resyncPeriod)
	return indexer, reflector
}

// NewReflector creates a new Reflector object which will keep the given store up to
// date with the server's contents for the given resource. Reflector promises to
// only put things in the store that have the type of expectedType, unless expectedType
// is nil. If resyncPeriod is non-zero, then lists will be executed after every
// resyncPeriod, so that you can use reflectors to periodically process everything as
// well as incrementally processing the things that change.
func NewReflector(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration) *Reflector {
	return NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod, false)
}

// NewReflectorWithReset creates a new Reflector object which will keep the given store up to
// date with the server's contents for the given resource. Reflector promises to
// only put things in the store that have the type of expectedType, unless expectedType
// is nil. If resyncPeriod is non-zero, then lists will be executed after every
// resyncPeriod, so that you can use reflectors to periodically process everything as
// well as incrementally processing the things that change. Also,  it introduces a  reset
// chan for any incoming bound changes
func NewReflectorWithReset(lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration, filterBounds []filterBound) *Reflector {
	r := NewNamedReflector(naming.GetNameFromCallsite(internalPackages...), lw, expectedType, store, resyncPeriod, false)
	r.filterBounds = filterBounds
	return r
}

// NewNamedReflector same as NewReflector, but with a specified name for logging
func NewNamedReflector(name string, lw ListerWatcher, expectedType interface{}, store Store, resyncPeriod time.Duration, allowPartialWatch bool) *Reflector {
	realClock := &clock.RealClock{}
	r := &Reflector{
		name:          name,
		listerWatcher: lw,
		store:         store,

		// We used to make the call every 1sec (1 QPS), the goal here is to achieve ~98% traffic reduction when
		// API server is not healthy. With these parameters, backoff will stop at [30,60) sec interval which is
		// 0.22 QPS. If we don't backoff for 2min, assume API server is healthy and we reset the backoff.
		backoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, realClock),
		resyncPeriod:   resyncPeriod,
		clock:          realClock,

		filterBounds:        make([]filterBound, 0),
		aggChan:             make(chan interface{}),
		clientSetUpdateChan: apiserverupdate.WatchClientSetUpdate(),
		allowPartialWatch:   allowPartialWatch,
		// We set lastSyncResourceVersion to "0", because it's the value which
		// we set as ResourceVersion to the first List() request.
	}
	r.setExpectedType(expectedType)
	return r
}

func (r *Reflector) setExpectedType(expectedType interface{}) {
	r.expectedType = reflect.TypeOf(expectedType)
	if r.expectedType == nil {
		r.expectedTypeName = defaultExpectedTypeName
		return
	}

	r.expectedTypeName = r.expectedType.String()

	if obj, ok := expectedType.(*unstructured.Unstructured); ok {
		// Use gvk to check that watch event objects are of the desired type.
		gvk := obj.GroupVersionKind()
		if gvk.Empty() {
			klog.V(4).Infof("Reflector from %s configured with expectedType of *unstructured.Unstructured with empty GroupVersionKind.", r.name)
			return
		}
		r.expectedGVK = &gvk
		r.expectedTypeName = gvk.String()
	}
}

// internalPackages are packages that ignored when creating a default reflector name. These packages are in the common
// call chains to NewReflector, so they'd be low entropy names for reflectors
var internalPackages = []string{"client-go/tools/cache/"}

// Run starts a watch and handles watch events. Will restart the watch if it is closed.
// Run will exit when stopCh is closed.
func (r *Reflector) Run(stopCh <-chan struct{}) {
	klog.V(2).Infof("Starting reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
	wait.BackoffUntil(func() {
		if len(r.filterBounds) == 0 {
			if err := r.ListAndWatch(stopCh); err != nil {
				utilruntime.HandleError(err)
			}
		} else {
			for i, fb := range r.filterBounds {
				go func(index int, fB filterBound) {
					for {
						select {
						case <-stopCh:
							return
						case signal, ok := <-fB.resetCh.Read:
							if !ok {
								klog.Errorf("reset channel closed. expectedType %v", r.expectedType)
								return
							} else {
								klog.V(4).Infof("Got reset channel message. expectedType %v, new bound [%+v]", r.expectedType, signal)
							}

							r.aggChan <- signal
						}
					}
				}(i, fb)
			}

			for {
				if err := r.ListAndWatch(stopCh); err != nil {
					if err == errorResetFilterBoundRequested || err == errorClientSetResetRequested {
						klog.V(4).Infof("Filter bounds reset message received, redo ListAndWatch with resetCh")
						continue
					}
					utilruntime.HandleError(err)
					break
				}
			}
		}
	}, r.backoffManager, true, stopCh)
	klog.V(2).Infof("Stopping reflector %s (%s) from %s", r.expectedTypeName, r.resyncPeriod, r.name)
}

var (
	// nothing will ever be sent down this channel
	neverExitWatch <-chan time.Time = make(chan time.Time)

	// Used to indicate that watching stopped because of a signal from the stop
	// channel passed in from a client of the reflector.
	errorStopRequested = errors.New("Stop requested")

	// Used to indicate that watching stopped because filtering bounds are updated
	errorResetFilterBoundRequested = errors.New("Reset requested")

	// Used to indicate that watching stopped because api server clients are updated
	errorClientSetResetRequested = errors.New("Clientset reset requested")
)

// resyncChan returns a channel which will receive something when a resync is
// required, and a cleanup function.
func (r *Reflector) resyncChan() (<-chan time.Time, func() bool) {
	if r.resyncPeriod == 0 {
		return neverExitWatch, func() bool { return false }
	}
	// The cleanup function is required: imagine the scenario where watches
	// always fail so we end up listing frequently. Then, if we don't
	// manually stop the timer, we could end up with many timers active
	// concurrently.
	t := r.clock.NewTimer(r.resyncPeriod)
	return t.C(), t.Stop
}

// ListAndWatch first lists all items and get the resource version at the moment of call,
// and then use the resource version to watch.
// It returns error if ListAndWatch didn't even try to initialize watch.
func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
	klog.V(3).Infof("ListAndWatch %v. filter bounds %+v. name %s. Watch page size %v. resync period %v",
		r.expectedTypeName, r.filterBounds, r.name, r.WatchListPageSize, r.resyncPeriod)
	var resourceVersion string

	options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}

	if len(r.filterBounds) > 0 {
		if r.hasInitBounds() {
			if !r.waitForBoundInit(stopCh) {
				return nil
			}
		} else {
			select {
			case <-stopCh:
				return nil
			default:
				klog.V(4).Infof("ListAndWatchWithReset default fall through. bounds %+v", r.filterBounds)
			}
		}

		// Pick up any bound changes
		options = appendFieldSelector(options, r.createHashkeyListOptions())
	}

	// LIST
	if err := func() error {
		initTrace := trace.New("Reflector " + r.name + " ListAndWatch")
		defer initTrace.LogIfLong(10 * time.Second)
		var list runtime.Object
		var paginatedResult bool
		var err error
		listCh := make(chan struct{}, 1)
		panicCh := make(chan interface{}, 1)
		go func() {
			defer func() {
				if r := recover(); r != nil {
					panicCh <- r
				}
			}()
			// Attempt to gather list in chunks, if supported by listerWatcher, if not, the first
			// list request will return the full response.
			pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
				return r.listerWatcher.List(opts)
			}))
			switch {
			case r.WatchListPageSize != 0:
				pager.PageSize = r.WatchListPageSize
			case r.paginatedResult:
				// We got a paginated result initially. Assume this resource and server honor
				// paging requests (i.e. watch cache is probably disabled) and leave the default
				// pager size set.
			case options.ResourceVersion != "" && options.ResourceVersion != "0":
				// User didn't explicitly request pagination.
				//
				// With ResourceVersion != "", we have a possibility to list from watch cache,
				// but we do that (for ResourceVersion != "0") only if Limit is unset.
				// To avoid thundering herd on etcd (e.g. on master upgrades), we explicitly
				// switch off pagination to force listing from watch cache (if enabled).
				// With the existing semantic of RV (result is at least as fresh as provided RV),
				// this is correct and doesn't lead to going back in time.
				//
				// We also don't turn off pagination for ResourceVersion="0", since watch cache
				// is ignoring Limit in that case anyway, and if watch cache is not enabled
				// we don't introduce regression.
				pager.PageSize = 0
			}

			list, paginatedResult, err = pager.List(context.Background(), options)
			if isExpiredError(err) {
				r.setIsLastSyncResourceVersionExpired(true)
				// Retry immediately if the resource version used to list is expired.
				// The pager already falls back to full list if paginated list calls fail due to an "Expired" error on
				// continuation pages, but the pager might not be enabled, or the full list might fail because the
				// resource version it is listing at is expired, so we need to fallback to resourceVersion="" in all
				// to recover and ensure the reflector makes forward progress.
				list, paginatedResult, err = pager.List(context.Background(), metav1.ListOptions{ResourceVersion: r.relistResourceVersion()})
			}
			close(listCh)
		}()
		select {
		case <-stopCh:
			return nil
		case r := <-panicCh:
			panic(r)
		case <-listCh:
		}
		if err != nil {
			return fmt.Errorf("%s: Failed to list %v: %v", r.name, r.expectedTypeName, err)
		}

		// We check if the list was paginated and if so set the paginatedResult based on that.
		// However, we want to do that only for the initial list (which is the only case
		// when we set ResourceVersion="0"). The reasoning behind it is that later, in some
		// situations we may force listing directly from etcd (by setting ResourceVersion="")
		// which will return paginated result, even if watch cache is enabled. However, in
		// that case, we still want to prefer sending requests to watch cache if possible.
		//
		// Paginated result returned for request with ResourceVersion="0" mean that watch
		// cache is disabled and there are a lot of objects of a given type. In such case,
		// there is no need to prefer listing from watch cache.
		if options.ResourceVersion == "0" && paginatedResult {
			r.paginatedResult = true
		}

		r.setIsLastSyncResourceVersionExpired(false) // list was successful
		initTrace.Step("Objects listed")
		listMetaInterface, err := meta.ListAccessor(list)
		if err != nil {
			return fmt.Errorf("%s: Unable to understand list result %#v: %v", r.name, list, err)
		}
		resourceVersion = listMetaInterface.GetResourceVersion()
		initTrace.Step("Resource version extracted")
		items, err := meta.ExtractList(list)
		if err != nil {
			return fmt.Errorf("%s: Unable to understand list result %#v (%v)", r.name, list, err)
		}
		initTrace.Step("Objects extracted")
		if err := r.syncWith(items, resourceVersion); err != nil {
			return fmt.Errorf("%s: Unable to sync list result: %v", r.name, err)
		}
		initTrace.Step("SyncWith done")
		r.setLastSyncResourceVersion(resourceVersion)
		initTrace.Step("Resource version updated")
		return nil
	}(); err != nil {
		return err
	}

	// RESYNC
	resyncerrc := make(chan error, 1)
	cancelCh := make(chan struct{})
	defer close(cancelCh)
	go func() {
		resyncCh, cleanup := r.resyncChan()
		defer func() {
			cleanup() // Call the last one written into cleanup
		}()
		for {
			select {
			case <-resyncCh:
			case <-stopCh:
				return
			case <-cancelCh:
				return
			}
			if r.ShouldResync == nil || r.ShouldResync() {
				klog.V(4).Infof("%s: forcing resync. type %v. resync period %v", r.name, r.expectedTypeName, r.resyncPeriod)
				if err := r.store.Resync(); err != nil {
					resyncerrc <- err
					return
				}
			}
			cleanup()
			resyncCh, cleanup = r.resyncChan()
		}
	}()

	// WATCH
	for {
		// give the stopCh a chance to stop the loop, even in case of continue statements further down on errors
		select {
		case <-stopCh:
			return nil
		default:
		}

		timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
		options = metav1.ListOptions{
			ResourceVersion: resourceVersion,
			// We want to avoid situations of hanging watchers. Stop any wachers that do not
			// receive any events within the timeout window.
			TimeoutSeconds: &timeoutSeconds,
			// To reduce load on kube-apiserver on watch restarts, you may enable watch bookmarks.
			// Reflector doesn't assume bookmarks are returned at all (if the server do not support
			// watch bookmarks, it will ignore this field).
			AllowWatchBookmarks: true,
			AllowPartialWatch:   r.allowPartialWatch,
		}

		if len(r.filterBounds) > 0 {
			options = appendFieldSelector(options, r.createHashkeyListOptions())
		}
		// start the clock before sending the request, since some proxies won't flush headers until after the first watch event is sent
		start := r.clock.Now()
		w, err := r.listerWatcher.Watch(options)
		if err != nil {
			switch {
			case isExpiredError(err):
				// Don't set LastSyncResourceVersionExpired - LIST call with ResourceVersion=RV already
				// has a semantic that it returns data at least as fresh as provided RV.
				// So first try to LIST with setting RV to resource version of last observed object.
				klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
			case err == io.EOF:
				// watch closed normally
			case err == io.ErrUnexpectedEOF:
				klog.V(1).Infof("%s: Watch for %v closed with unexpected EOF: %v", r.name, r.expectedTypeName, err)
			default:
				utilruntime.HandleError(fmt.Errorf("%s: Failed to watch %v: %v", r.name, r.expectedTypeName, err))
			}
			// If this is "connection refused" error, it means that most likely apiserver is not responsive.
			// It doesn't make sense to re-list all objects because most likely we will be able to restart
			// watch where we ended.
			// If that's the case wait and resend watch request.
			if utilnet.IsConnectionRefused(err) {
				time.Sleep(time.Second)
				continue
			}
			return nil
		}

		if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {
			if err == errorResetFilterBoundRequested || err == errorClientSetResetRequested {
				select {
				case cancelCh <- struct{}{}:
					klog.V(4).Infof("Sent message to Resync cancelCh.")
				default:
					klog.V(4).Infof("Resync cancelCh was closed.")
				}
				return err
			}
			if err != errorStopRequested {
				switch {
				case isExpiredError(err):
					// Don't set LastSyncResourceVersionExpired - LIST call with ResourceVersion=RV already
					// has a semantic that it returns data at least as fresh as provided RV.
					// So first try to LIST with setting RV to resource version of last observed object.
					klog.V(4).Infof("%s: watch of %v closed with: %v", r.name, r.expectedTypeName, err)
				default:
					klog.Warningf("%s: watch of %v ended with: %v", r.name, r.expectedTypeName, err)
				}
			}
			return nil
		}
	}
}

// syncWith replaces the store's items with the given list.
func (r *Reflector) syncWith(items []runtime.Object, resourceVersion string) error {
	found := make([]interface{}, 0, len(items))
	for _, item := range items {
		found = append(found, item)
	}
	return r.store.Replace(found, resourceVersion)
}

// watchHandler watches w and keeps *resourceVersion up to date.
func (r *Reflector) watchHandler(start time.Time, w watch.Interface, resourceVersion *string, errc chan error, stopCh <-chan struct{}) error {
	eventCount := 0

	// Stopping the watcher should be idempotent and if we return from this function there's no way
	// we're coming back in with the same watch interface.
	defer w.Stop()

loop:
	for {
		if len(r.filterBounds) > 0 {
			select {
			case <-r.clientSetUpdateChan.Read:
				klog.Infof("Got client set update message. Restarting ListAndWatch %v", r.expectedType)
				return errorClientSetResetRequested
			case <-stopCh:
				return errorStopRequested
			case signal, ok := <-r.aggChan:
				if !ok {
					klog.Error("resetCh channel closed")
					return errors.New("Reset channel closed")
				}
				r.setBounds(signal)

				return errorResetFilterBoundRequested
			case err := <-errc:
				return err
			case event, ok := <-w.ResultChan():
				if !ok {
					break loop
				}

				var err error
				err, eventCount = r.watchHandlerHelper(event, resourceVersion, eventCount)
				if err != nil {
					return err
				}
			}
		} else {
			select {
			case <-r.clientSetUpdateChan.Read:
				klog.Infof("Got client set update message. Restarting ListAndWatch %v", r.expectedType)
				return errorClientSetResetRequested
			case <-stopCh:
				return errorStopRequested
			case err := <-errc:
				return err
			case event, ok := <-w.ResultChan():
				if !ok {
					break loop
				}

				var err error
				err, eventCount = r.watchHandlerHelper(event, resourceVersion, eventCount)
				if err != nil {
					return err
				}
			}
		}
	}

	watchDuration := r.clock.Since(start)
	if watchDuration < 1*time.Second && eventCount == 0 {
		return fmt.Errorf("very short watch: %s: Unexpected watch close - watch lasted less than a second and no items received", r.name)
	}
	klog.V(4).Infof("%s: Watch close - %v total %v items received", r.name, r.expectedTypeName, eventCount)
	return nil
}

func (r *Reflector) watchHandlerHelper(event watch.Event, resourceVersion *string, eventCount int) (error, int) {
	if event.Type == watch.Error {
		return apierrs.FromObject(event.Object), eventCount
	}
	if r.expectedType != nil {
		if e, a := r.expectedType, reflect.TypeOf(event.Object); e != a {
			utilruntime.HandleError(fmt.Errorf("%s: expected type %v, but watch event object had type %v", r.name, e, a))
			return nil, eventCount
		}
	}
	if r.expectedGVK != nil {
		if e, a := *r.expectedGVK, event.Object.GetObjectKind().GroupVersionKind(); e != a {
			utilruntime.HandleError(fmt.Errorf("%s: expected gvk %v, but watch event object had gvk %v", r.name, e, a))
			return nil, eventCount
		}
	}
	meta, err := meta.Accessor(event.Object)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
		return nil, eventCount
	}
	newResourceVersion := meta.GetResourceVersion()
	switch event.Type {
	case watch.Added:
		err := r.store.Add(event.Object)
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("%s: unable to add watch event object (%#v) to store: %v", r.name, event.Object, err))
		}
	case watch.Modified:
		err := r.store.Update(event.Object)
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("%s: unable to update watch event object (%#v) to store: %v", r.name, event.Object, err))
		}
	case watch.Deleted:
		// TODO: Will any consumers need access to the "last known
		// state", which is passed in event.Object? If so, may need
		// to change this.
		err := r.store.Delete(event.Object)
		if err != nil {
			utilruntime.HandleError(fmt.Errorf("%s: unable to delete watch event object (%#v) from store: %v", r.name, event.Object, err))
		}
	case watch.Bookmark:
		// A `Bookmark` means watch has synced here, just update the resourceVersion
	default:
		utilruntime.HandleError(fmt.Errorf("%s: unable to understand watch event %#v", r.name, event))
	}
	*resourceVersion = newResourceVersion
	r.setLastSyncResourceVersion(newResourceVersion)
	eventCount++
	return nil, eventCount
}

// LastSyncResourceVersion is the resource version observed when last sync with the underlying store
// The value returned is not synchronized with access to the underlying store and is not thread-safe
func (r *Reflector) LastSyncResourceVersion() string {
	r.lastSyncResourceVersionMutex.RLock()
	defer r.lastSyncResourceVersionMutex.RUnlock()
	return r.lastSyncResourceVersion
}

func (r *Reflector) setLastSyncResourceVersion(v string) {
	r.lastSyncResourceVersionMutex.Lock()
	defer r.lastSyncResourceVersionMutex.Unlock()
	r.lastSyncResourceVersion = v
}

// relistResourceVersion is the resource version the reflector should list or relist from.
// relistResourceVersion determines the resource version the reflector should list or relist from.
// Returns either the lastSyncResourceVersion so that this reflector will relist with a resource
// versions no older than has already been observed in relist results or watch events, or, if the last relist resulted
// in an HTTP 410 (Gone) status code, returns "" so that the relist will use the latest resource version available in
// etcd via a quorum read.
func (r *Reflector) relistResourceVersion() string {
	r.lastSyncResourceVersionMutex.RLock()
	defer r.lastSyncResourceVersionMutex.RUnlock()

	if r.isLastSyncResourceVersionGone {
		// Since this reflector makes paginated list requests, and all paginated list requests skip the watch cache
		// if the lastSyncResourceVersion is expired, we set ResourceVersion="" and list again to re-establish reflector
		// to the latest available ResourceVersion, using a consistent read from etcd.
		return ""
	}
	if r.lastSyncResourceVersion == "" {
		// For performance reasons, initial list performed by reflector uses "0" as resource version to allow it to
		// be served from the watch cache if it is enabled.
		return "0"
	}
	return r.lastSyncResourceVersion
}

// setIsLastSyncResourceVersionExpired sets if the last list or watch request with lastSyncResourceVersion returned a
// expired error: HTTP 410 (Gone) Status Code.
func (r *Reflector) setIsLastSyncResourceVersionExpired(isExpired bool) {
	r.lastSyncResourceVersionMutex.Lock()
	defer r.lastSyncResourceVersionMutex.Unlock()
	r.isLastSyncResourceVersionGone = isExpired
}

func isExpiredError(err error) bool {
	// In Kubernetes 1.17 and earlier, the api server returns both apierrs.StatusReasonExpired and
	// apierrs.StatusReasonGone for HTTP 410 (Gone) status code responses. In 1.18 the kube server is more consistent
	// and always returns apierrs.StatusReasonExpired. For backward compatibility we can only remove the apierrs.IsGone
	// check when we fully drop support for Kubernetes 1.17 servers from reflectors.
	return apierrs.IsResourceExpired(err) || apierrs.IsGone(err)
}

func (r *Reflector) ListerWatcher() ListerWatcher {
	return r.listerWatcher
}

func appendFieldSelector(target metav1.ListOptions, fieldSelector metav1.ListOptions) metav1.ListOptions {
	if len(target.FieldSelector) == 0 {
		target.FieldSelector = fieldSelector.FieldSelector
	} else {
		target.FieldSelector = target.FieldSelector + "," + fieldSelector.FieldSelector
	}
	return target
}

func (r *Reflector) setBounds(signal interface{}) {
	newBounds, _ := signal.(FilterBound)
	klog.V(4).Infof("Setbound Begin. Old bounds [%+v]. New bounds [%+v]. expectedType %v", r.filterBounds, newBounds, r.expectedType)
	isBoundReset := false
	for i, filterBound := range r.filterBounds {
		if filterBound.sourceName == newBounds.OwnerName {
			r.filterBounds[i].bounds = []int64{newBounds.LowerBound, newBounds.UpperBound}
			isBoundReset = true
		}
	}
	if !isBoundReset {
		klog.Errorf("Cannot found matching filtering bounds. Old bounds [%+v]. New bounds [%+v]", r.filterBounds, newBounds)
	}
	klog.V(4).Infof("Setbound DONE. New bounds [%+v]. expectedType %v", r.filterBounds, r.expectedType)
}

func (r *Reflector) hasInitBounds() bool {
	for _, filterBound := range r.filterBounds {
		if len(filterBound.bounds) == 0 || filterBound.bounds[0] == -1 && filterBound.bounds[1] == -1 {
			return true
		}
	}
	return false
}

// return false if stopCh send message
func (r *Reflector) waitForBoundInit(stopCh <-chan struct{}) bool {
	for {
		select {
		case msg, ok := <-r.aggChan:
			if !ok {
				klog.Errorf("Cannot read from closed channel. expectedType %v", r.expectedType)
				return false
			}
			r.setBounds(msg)
		case <-stopCh:
			return false
		}

		if !r.hasInitBounds() {
			klog.V(4).Infof("Init bound succeeded. expectedType %v, Bounds: %+v", r.expectedType, r.filterBounds)
			return true
		} else {
			klog.V(4).Infof("Waiting for more init bound. expectedType %v, Bounds: %+v", r.expectedType, r.filterBounds)
		}
	}
}

// currently do not support all range
func (r *Reflector) createHashkeyListOptions() metav1.ListOptions {
	klog.V(3).Infof("createHashkeyListOptions filterBounds [%+v], expectedType %v", r.filterBounds, r.expectedType)
	listOptions := make([]string, 0)
	for _, filterBound := range r.filterBounds {
		if filterBound.bounds[0] == -1 && filterBound.bounds[1] == -1 {
			continue
		}
		operator := "gt"
		if filterBound.bounds[0] == 0 {
			operator = "gte"
		}
		var option string
		if filterBound.ownerKind != "" {
			option = fmt.Sprintf("metadata.ownerReferences.hashkey.%s=%s:%v,metadata.ownerReferences.hashkey.%s=lte:%+v", filterBound.ownerKind, operator, filterBound.bounds[0], filterBound.ownerKind, filterBound.bounds[1])
		} else {
			option = fmt.Sprintf("metadata.hashkey=%s:%v,metadata.hashkey=lte:%+v", operator, filterBound.bounds[0], filterBound.bounds[1])
		}

		listOptions = append(listOptions, option)
	}

	return metav1.ListOptions{
		FieldSelector: strings.Join(listOptions, ";"),
	}
}
