@@ -18,12 +18,9 @@ import (
1818 "context"
1919 "fmt"
2020 "log/slog"
21- "os"
2221 "time"
2322
2423 "github.com/blang/semver/v4"
25- "github.com/go-kit/log"
26- "github.com/go-kit/log/level"
2724 "github.com/go-logr/logr"
2825 monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
2926 promv1alpha1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1alpha1"
@@ -53,6 +50,7 @@ const (
5350)
5451
5552func NewPrometheusCRWatcher (ctx context.Context , logger logr.Logger , cfg allocatorconfig.Config ) (* PrometheusCRWatcher , error ) {
53+ slogger := slog .New (logr .ToSlogHandler (logger ))
5654 var resourceSelector * prometheus.ResourceSelector
5755 mClient , err := monitoringclient .NewForConfig (cfg .ClusterConfig )
5856 if err != nil {
@@ -92,9 +90,7 @@ func NewPrometheusCRWatcher(ctx context.Context, logger logr.Logger, cfg allocat
9290 },
9391 }
9492
95- promOperatorLogger := level .NewFilter (log .NewLogfmtLogger (os .Stderr ), level .AllowWarn ())
96- promOperatorSlogLogger := slog .New (slog .NewTextHandler (os .Stderr , & slog.HandlerOptions {Level : slog .LevelWarn }))
97- generator , err := prometheus .NewConfigGenerator (promOperatorLogger , prom , true )
93+ generator , err := prometheus .NewConfigGenerator (slogger , prom , prometheus .WithEndpointSliceSupport ())
9894
9995 if err != nil {
10096 return nil , err
@@ -112,21 +108,21 @@ func NewPrometheusCRWatcher(ctx context.Context, logger logr.Logger, cfg allocat
112108 logger .Error (err , "Retrying namespace informer creation in promOperator CRD watcher" )
113109 return true
114110 }, func () error {
115- nsMonInf , err = getNamespaceInformer (ctx , map [string ]struct {}{v1 .NamespaceAll : {}}, promOperatorLogger , clientset , operatorMetrics )
111+ nsMonInf , err = getNamespaceInformer (ctx , map [string ]struct {}{v1 .NamespaceAll : {}}, slogger , clientset , operatorMetrics )
116112 return err
117113 })
118114 if getNamespaceInformerErr != nil {
119115 logger .Error (getNamespaceInformerErr , "Failed to create namespace informer in promOperator CRD watcher" )
120116 return nil , getNamespaceInformerErr
121117 }
122118
123- resourceSelector , err = prometheus .NewResourceSelector (promOperatorSlogLogger , prom , store , nsMonInf , operatorMetrics , eventRecorder )
119+ resourceSelector , err = prometheus .NewResourceSelector (slogger , prom , store , nsMonInf , operatorMetrics , eventRecorder )
124120 if err != nil {
125121 logger .Error (err , "Failed to create resource selector in promOperator CRD watcher" )
126122 }
127123
128124 return & PrometheusCRWatcher {
129- logger : logger ,
125+ logger : slogger ,
130126 kubeMonitoringClient : mClient ,
131127 k8sClient : clientset ,
132128 informers : monitoringInformers ,
@@ -145,7 +141,7 @@ func NewPrometheusCRWatcher(ctx context.Context, logger logr.Logger, cfg allocat
145141}
146142
147143type PrometheusCRWatcher struct {
148- logger logr .Logger
144+ logger * slog .Logger
149145 kubeMonitoringClient monitoringclient.Interface
150146 k8sClient kubernetes.Interface
151147 informers map [string ]* informers.ForResource
@@ -162,7 +158,7 @@ type PrometheusCRWatcher struct {
162158 store * assets.StoreBuilder
163159}
164160
165- func getNamespaceInformer (ctx context.Context , allowList map [string ]struct {}, promOperatorLogger log .Logger , clientset kubernetes.Interface , operatorMetrics * operator.Metrics ) (cache.SharedIndexInformer , error ) {
161+ func getNamespaceInformer (ctx context.Context , allowList map [string ]struct {}, promOperatorLogger * slog .Logger , clientset kubernetes.Interface , operatorMetrics * operator.Metrics ) (cache.SharedIndexInformer , error ) {
166162 kubernetesVersion , err := clientset .Discovery ().ServerVersion ()
167163 if err != nil {
168164 return nil , err
@@ -252,7 +248,7 @@ func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors ch
252248 } {
253249 sync , err := k8sutil .LabelSelectionHasChanged (old .Labels , cur .Labels , selector )
254250 if err != nil {
255- w .logger .Error (err , "Failed to check label selection between namespaces while handling namespace updates" , "selector" , name )
251+ w .logger .Error (err . Error () , "Failed to check label selection between namespaces while handling namespace updates" , "selector" , name )
256252 return
257253 }
258254
@@ -273,17 +269,20 @@ func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors ch
273269 for name , resource := range w .informers {
274270 resource .Start (w .stopChannel )
275271
276- if ok := cache .WaitForNamedCacheSync (name , w .stopChannel , resource .HasSynced ); ! ok {
277- success = false
272+ if ok := w .WaitForNamedCacheSync (name , resource .HasSynced ); ! ok {
273+ w .logger .Info ("skipping informer" , "informer" , name )
274+ continue
278275 }
279276
280277 // only send an event notification if there isn't one already
281278 resource .AddEventHandler (cache.ResourceEventHandlerFuncs {
282279 // these functions only write to the notification channel if it's empty to avoid blocking
283280 // if scrape config updates are being rate-limited
284281 AddFunc : func (obj interface {}) {
282+ w .logger .Info ("added" , "obj" , obj )
285283 select {
286284 case notifyEvents <- struct {}{}:
285+ w .logger .Info ("added" )
287286 default :
288287 }
289288 },
@@ -378,7 +377,7 @@ func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Confi
378377 "" ,
379378 nil ,
380379 nil ,
381- monitoringv1.TSDBSpec {},
380+ & monitoringv1.TSDBSpec {},
382381 nil ,
383382 nil ,
384383 serviceMonitorInstances ,
@@ -415,3 +414,41 @@ func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Confi
415414 return promCfg , nil
416415 }
417416}
417+
418+ // WaitForNamedCacheSync adds a timeout to the informer's wait for the cache to be ready.
419+ // If the PrometheusCRWatcher is unable to load an informer within 15 seconds, the method is
420+ // cancelled and returns false. A successful informer load will return true. This method also
421+ // will be cancelled if the target allocator's stopChannel is called before it returns.
422+ //
423+ // This method is inspired by the upstream prometheus-operator implementation, with a shorter timeout
424+ // and support for the PrometheusCRWatcher's stopChannel.
425+ // https://github.com/prometheus-operator/prometheus-operator/blob/293c16c854ce69d1da9fdc8f0705de2d67bfdbfa/pkg/operator/operator.go#L433
426+ func (w * PrometheusCRWatcher ) WaitForNamedCacheSync (controllerName string , inf cache.InformerSynced ) bool {
427+ ctx , cancel := context .WithTimeout (context .Background (), time .Second * 15 )
428+ t := time .NewTicker (time .Second * 5 )
429+ defer t .Stop ()
430+
431+ go func () {
432+ for {
433+ select {
434+ case <- t .C :
435+ w .logger .Debug ("cache sync not yet completed" )
436+ case <- ctx .Done ():
437+ return
438+ case <- w .stopChannel :
439+ w .logger .Warn ("stop received, shutting down cache syncing" )
440+ cancel ()
441+ return
442+ }
443+ }
444+ }()
445+
446+ ok := cache .WaitForNamedCacheSync (controllerName , ctx .Done (), inf )
447+ if ! ok {
448+ w .logger .Error ("failed to sync cache" )
449+ } else {
450+ w .logger .Debug ("successfully synced cache" )
451+ }
452+
453+ return ok
454+ }
0 commit comments