mirror of
https://github.com/prometheus/prometheus.git
synced 2025-03-05 20:59:13 -08:00
Move hasSynced interface and its implementations to *_test.go files.
Signed-off-by: Yecheng Fu <cofyc.jackson@gmail.com>
This commit is contained in:
parent
8ceb8f2ae8
commit
d73b0d3141
|
@ -256,27 +256,6 @@ func New(l log.Logger, conf *SDConfig) (*Discovery, error) {
|
||||||
|
|
||||||
const resyncPeriod = 10 * time.Minute
|
const resyncPeriod = 10 * time.Minute
|
||||||
|
|
||||||
type hasSynced interface {
|
|
||||||
// hasSynced returns true if all informers' store has synced.
|
|
||||||
// This is only used in testing to determine when the cache stores have synced.
|
|
||||||
hasSynced() bool
|
|
||||||
}
|
|
||||||
|
|
||||||
var _ hasSynced = &Discovery{}
|
|
||||||
|
|
||||||
func (d *Discovery) hasSynced() bool {
|
|
||||||
d.RLock()
|
|
||||||
defer d.RUnlock()
|
|
||||||
for _, discoverer := range d.discoverers {
|
|
||||||
if hasSynceddiscoverer, ok := discoverer.(hasSynced); ok {
|
|
||||||
if !hasSynceddiscoverer.hasSynced() {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
// Run implements the discoverer interface.
|
// Run implements the discoverer interface.
|
||||||
func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
|
func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
|
||||||
d.Lock()
|
d.Lock()
|
||||||
|
|
|
@ -119,11 +119,14 @@ func (d k8sDiscoveryTest) Run(t *testing.T) {
|
||||||
resChan := make(chan map[string]*targetgroup.Group)
|
resChan := make(chan map[string]*targetgroup.Group)
|
||||||
go readResultWithoutTimeout(t, ch, d.expectedMaxItems, time.Second, resChan)
|
go readResultWithoutTimeout(t, ch, d.expectedMaxItems, time.Second, resChan)
|
||||||
|
|
||||||
if dd, ok := d.discovery.(hasSynced); ok {
|
dd, ok := d.discovery.(hasSynced)
|
||||||
if !cache.WaitForCacheSync(ctx.Done(), dd.hasSynced) {
|
if !ok {
|
||||||
t.Errorf("discoverer failed to sync: %v", dd)
|
t.Errorf("discoverer does not implement hasSynced interface")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if !cache.WaitForCacheSync(ctx.Done(), dd.hasSynced) {
|
||||||
|
t.Errorf("discoverer failed to sync: %v", dd)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if d.afterStart != nil {
|
if d.afterStart != nil {
|
||||||
|
@ -184,3 +187,49 @@ func requireTargetGroups(t *testing.T, expected, res map[string]*targetgroup.Gro
|
||||||
|
|
||||||
require.JSONEq(t, string(b1), string(b2))
|
require.JSONEq(t, string(b1), string(b2))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type hasSynced interface {
|
||||||
|
// hasSynced returns true if all informers' store has synced.
|
||||||
|
// This is only used in testing to determine when the cache stores have synced.
|
||||||
|
hasSynced() bool
|
||||||
|
}
|
||||||
|
|
||||||
|
var _ hasSynced = &Discovery{}
|
||||||
|
var _ hasSynced = &Node{}
|
||||||
|
var _ hasSynced = &Endpoints{}
|
||||||
|
var _ hasSynced = &Ingress{}
|
||||||
|
var _ hasSynced = &Pod{}
|
||||||
|
var _ hasSynced = &Service{}
|
||||||
|
|
||||||
|
func (d *Discovery) hasSynced() bool {
|
||||||
|
d.RLock()
|
||||||
|
defer d.RUnlock()
|
||||||
|
for _, discoverer := range d.discoverers {
|
||||||
|
if hasSynceddiscoverer, ok := discoverer.(hasSynced); ok {
|
||||||
|
if !hasSynceddiscoverer.hasSynced() {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (n *Node) hasSynced() bool {
|
||||||
|
return n.informer.HasSynced()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (e *Endpoints) hasSynced() bool {
|
||||||
|
return e.endpointsInf.HasSynced() && e.serviceInf.HasSynced() && e.podInf.HasSynced()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (i *Ingress) hasSynced() bool {
|
||||||
|
return i.informer.HasSynced()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *Pod) hasSynced() bool {
|
||||||
|
return p.informer.HasSynced()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *Service) hasSynced() bool {
|
||||||
|
return s.informer.HasSynced()
|
||||||
|
}
|
||||||
|
|
|
@ -39,7 +39,6 @@ type Node struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ discoverer = &Node{}
|
var _ discoverer = &Node{}
|
||||||
var _ hasSynced = &Node{}
|
|
||||||
|
|
||||||
// NewNode returns a new node discovery.
|
// NewNode returns a new node discovery.
|
||||||
func NewNode(l log.Logger, inf cache.SharedInformer) *Node {
|
func NewNode(l log.Logger, inf cache.SharedInformer) *Node {
|
||||||
|
@ -73,10 +72,6 @@ func (e *Node) enqueue(obj interface{}) {
|
||||||
e.queue.Add(key)
|
e.queue.Add(key)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (n *Node) hasSynced() bool {
|
|
||||||
return n.informer.HasSynced()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Run implements the Discoverer interface.
|
// Run implements the Discoverer interface.
|
||||||
func (n *Node) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
|
func (n *Node) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
|
||||||
defer n.queue.ShutDown()
|
defer n.queue.ShutDown()
|
||||||
|
|
Loading…
Reference in a new issue