repo_description: Prometheus Operator creates/configures/manages Prometheus clusters atop Kubernetes
repo_topics: hacktoberfest kubernetes monitoring prometheus
code: // Copyright The prometheus-operator Authors
  //
  // 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 operator
  
  import (
  	"context"
  	"fmt"
  	"log/slog"
  	"slices"
  	"sync"
  	"time"
  
  	"github.com/prometheus/client_golang/prometheus"
  	appsv1 "k8s.io/api/apps/v1"
  	corev1 "k8s.io/api/core/v1"
  	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  	"k8s.io/apimachinery/pkg/runtime"
  	"k8s.io/apimachinery/pkg/runtime/schema"
  	"k8s.io/apimachinery/pkg/watch"
  	"k8s.io/client-go/kubernetes"
  	"k8s.io/client-go/tools/cache"
  	"k8s.io/client-go/tools/events"
  
  	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
  	"github.com/prometheus-operator/prometheus-operator/pkg/client/versioned/scheme"
  )
  
  const (
  	// InvalidConfigurationEvent is the  type used for events reporting invalid
  	// configuration resources.
  	InvalidConfigurationEvent = "InvalidConfiguration"
  )
  
  var (
  	syncsDesc = prometheus.NewDesc(
  		"prometheus_operator_syncs",
  		"Number of objects per sync status (ok/failed)",
  		[]string{"status"},
  		nil,
  	)
  	resourcesDesc = prometheus.NewDesc(
  		"prometheus_operator_managed_resources",
  		"Number of resources managed by the operator's controller per state (selected/rejected)",
  		[]string{"resource", "state"},
  		nil,
  	)
  )
  
  type ReconciliationStatus struct {
  	err     error
  	reason  string
  	message string
  }
  
  func (rs ReconciliationStatus) Reason() string {
  	if rs.Ok() {
  		return rs.reason
  	}
  
  	return "ReconciliationFailed"
  }
  
  func (rs ReconciliationStatus) Message() string {
  	if rs.Ok() {
  		return rs.message
  	}
  
  	return rs.err.Error()
  }
  
  func (rs ReconciliationStatus) Ok() bool {
  	return rs.err == nil
  }
  
  // ReconciliationTracker tracks reconciliation status per object.
  //
  // It only uses their `<namespace>/<name>` key to identify objects.
  //
  // The zero ReconciliationTracker is ready to use.
  type ReconciliationTracker struct {
  	once sync.Once
  
  	// mtx protects all fields below.
  	mtx            sync.RWMutex
  	statusByObject map[string]ReconciliationStatus
  	refTracker     map[string]ReferenceTracker
  }
  
  // ReferenceTracker returns true if it has a reference to the object.
  type ReferenceTracker interface {
  	Has(runtime.Object) bool
  }
  
  func (rt *ReconciliationTracker) init() {
  	rt.once.Do(func() {
  		rt.statusByObject = map[string]ReconciliationStatus{}
  		rt.refTracker = map[string]ReferenceTracker{}
  	})
  }
  
  // HasRefTo returns true if the object identified by key has a direct or
  // indirect reference to obj (secret or configmap).
  func (rt *ReconciliationTracker) HasRefTo(key string, obj runtime.Object) bool {
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	refTracker, found := rt.refTracker[key]
  	if !found {
  		return false
  	}
  
  	return refTracker.Has(obj)
  }
  
  // UpdateReferenceTracker updates the reference tracker for the object identified by key.
  func (rt *ReconciliationTracker) UpdateReferenceTracker(key string, refTracker ReferenceTracker) {
  	rt.init()
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	rt.refTracker[key] = refTracker
  }
  
  // ResetStatus resets the reconciliation status for the object identified by key.
  func (rt *ReconciliationTracker) ResetStatus(key string) {
  	rt.init()
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	rt.statusByObject[key] = ReconciliationStatus{}
  }
  
  // SetStatus updates the last reconciliation status for the object identified by key.
  func (rt *ReconciliationTracker) SetStatus(key string, err error) {
  	rt.init()
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	rs := rt.statusByObject[key]
  	rs.err = err
  	rt.statusByObject[key] = rs
  }
  
  // SetReasonAndMessage updates the reason and message for the object identified by key.
  // The reason and message are only used when the reconciliation returned no error.
  func (rt *ReconciliationTracker) SetReasonAndMessage(key string, reason, message string) {
  	rt.init()
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	rs := rt.statusByObject[key]
  	rs.reason = reason
  	rs.message = message
  	rt.statusByObject[key] = rs
  }
  
  // GetStatus returns the last reconciliation status for the given object.
  // The second value indicates whether the object is known or not.
  func (rt *ReconciliationTracker) getStatus(k string) (ReconciliationStatus, bool) {
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	s, found := rt.statusByObject[k]
  	if !found {
  		return ReconciliationStatus{}, false
  	}
  
  	return s, true
  }
  
  // GetCondition returns a monitoringv1.Condition for the last-known
  // reconciliation status of the given object.
  func (rt *ReconciliationTracker) GetCondition(k string, gen int64) monitoringv1.Condition {
  	condition := monitoringv1.Condition{
  		Type:   monitoringv1.Reconciled,
  		Status: monitoringv1.ConditionTrue,
  		LastTransitionTime: metav1.Time{
  			Time: time.Now().UTC(),
  		},
  		ObservedGeneration: gen,
  	}
  
  	reconciliationStatus, found := rt.getStatus(k)
  	if !found {
  		condition.Status = monitoringv1.ConditionUnknown
  		condition.Reason = "NotFound"
  		condition.Message = fmt.Sprintf("object %q not found", k)
  	} else {
  		if !reconciliationStatus.Ok() {
  			condition.Status = monitoringv1.ConditionFalse
  		}
  		condition.Reason = reconciliationStatus.Reason()
  		condition.Message = reconciliationStatus.Message()
  	}
  
  	return condition
  }
  
  // ForgetObject removes the given object from the tracker.
  // It should be called when the controller detects that the object has been deleted.
  func (rt *ReconciliationTracker) ForgetObject(key string) {
  	rt.mtx.Lock()
  	defer rt.mtx.Unlock()
  
  	if rt.statusByObject == nil {
  		return
  	}
  
  	delete(rt.statusByObject, key)
  	delete(rt.refTracker, key)
  }
  
  // Describe implements the prometheus.Collector interface.
  func (rt *ReconciliationTracker) Describe(ch chan<- *prometheus.Desc) {
  	ch <- syncsDesc
  }
  
  // Collect implements the prometheus.Collector interface.
  func (rt *ReconciliationTracker) Collect(ch chan<- prometheus.Metric) {
  	rt.mtx.RLock()
  	defer rt.mtx.RUnlock()
  
  	var ok, failed float64
  	for _, st := range rt.statusByObject {
  		if st.Ok() {
  			ok++
  		} else {
  			failed++
  		}
  	}
  
  	ch <- prometheus.MustNewConstMetric(
  		syncsDesc,
  		prometheus.GaugeValue,
  		ok,
  		"ok",
  	)
  	ch <- prometheus.MustNewConstMetric(
  		syncsDesc,
  		prometheus.GaugeValue,
  		failed,
  		"failed",
  	)
  }
  
  // Metrics represents metrics associated to an operator.
  type Metrics struct {
  	reg prometheus.Registerer
  
  	listCounter            prometheus.Counter
  	listFailedCounter      prometheus.Counter
  	watchCounter           prometheus.Counter
  	watchFailedCounter     prometheus.Counter
  	stsDeleteCreateCounter prometheus.Counter
  	// triggerByCounter is a set of counters keeping track of the amount
  	// of times Prometheus Operator was triggered to reconcile its created
  	// objects. It is split in the dimensions of Kubernetes objects and
  	// corresponding actions (add, delete, update).
  	triggerByCounter *prometheus.CounterVec
  	ready            prometheus.Gauge
  
  	// mtx protects all fields below.
  	mtx       sync.RWMutex
  	resources map[resourceKey]map[string]int
  }
  
  type resourceKey struct {
  	resource string
  	state    resourceState
  }
  
  // NewMetrics initializes operator metrics and registers them with the given registerer.
  func NewMetrics(r prometheus.Registerer) *Metrics {
  	m := Metrics{
  		reg: r,
  		triggerByCounter: prometheus.NewCounterVec(prometheus.CounterOpts{
  			Name: "prometheus_operator_triggered_total",
  			Help: "Number of times a Kubernetes object add, delete or update event" +
  				" triggered the Prometheus Operator to reconcile an object",
  		}, []string{"triggered_by", "action"}),
  		stsDeleteCreateCounter: prometheus.NewCounter(prometheus.CounterOpts{
  			Name: "prometheus_operator_reconcile_sts_delete_create_total",
  			Help: "Number of times that reconciling a statefulset required deleting and re-creating it",
  		}),
  		listCounter: prometheus.NewCounter(prometheus.CounterOpts{
  			Name: "prometheus_operator_list_operations_total",
  			Help: "Total number of list operations",
  		}),
  		listFailedCounter: prometheus.NewCounter(prometheus.CounterOpts{
  			Name: "prometheus_operator_list_operations_failed_total",
  			Help: "Total number of list operations that failed",
  		}),
  		watchCounter: prometheus.NewCounter(prometheus.CounterOpts{
  			Name: "prometheus_operator_watch_operations_total",
  			Help: "Total number of watch operations",
  		}),
  		watchFailedCounter: prometheus.NewCounter(prometheus.CounterOpts{
  			Name: "prometheus_operator_watch_operations_failed_total",
  			Help: "Total number of watch operations that failed",
  		}),
  		ready: prometheus.NewGauge(prometheus.GaugeOpts{
  			Name: "prometheus_operator_ready",
  			Help: "1 when the controller is ready to reconcile resources, 0 otherwise",
  		}),
  
  		resources: make(map[resourceKey]map[string]int),
  	}
  
  	m.reg.MustRegister(
  		m.triggerByCounter,
  		m.stsDeleteCreateCounter,
  		m.listCounter,
  		m.listFailedCounter,
  		m.watchCounter,
  		m.watchFailedCounter,
  		m.ready,
  		&m,
  	)
  
  	return &m
  }
  
  // EventRecorderFactory returns an function to create EventRecorder objects.
  type EventRecorderFactory func(client kubernetes.Interface, component string) NewEventRecorderFunc
  
  // NewEventRecorderFunc returns an EventRecorder which will automatically inject the given runtime.Object as related.
  type NewEventRecorderFunc func(related runtime.Object) *EventRecorder
  
  // EventRecorder records events which are related to the associated Object.
  type EventRecorder struct {
  	er      events.EventRecorder
  	related runtime.Object
  }
  
  func NewFakeRecorder(bufferSize int, related runtime.Object) *EventRecorder {
  	return &EventRecorder{
  		related: related,
  		er:      events.NewFakeRecorder(bufferSize),
  	}
  }
  
  // Eventf records a Kubernetes event.
  func (er *EventRecorder) Eventf(regarding runtime.Object, eventtype, reason, action, note string, args ...any) {
  	er.er.Eventf(
  		regarding,
  		er.related,
  		eventtype,
  		reason,
  		action,
  		note,
  		args...,
  	)
  }
  
  func NewEventRecorderFactory(emitEvents bool) EventRecorderFactory {
  	return func(client kubernetes.Interface, component string) NewEventRecorderFunc {
  		eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: client.EventsV1()})
  		eventBroadcaster.StartStructuredLogging(0)
  
  		if emitEvents {
  			_ = eventBroadcaster.StartRecordingToSinkWithContext(context.Background())
  		}
  
  		eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, component)
  
  		return func(related runtime.Object) *EventRecorder {
  			return &EventRecorder{
  				er:      eventRecorder,
  				related: related,
  			}
  		}
  	}
  }
  
  // StsDeleteCreateCounter returns a counter to track statefulset's recreations.
  func (m *Metrics) StsDeleteCreateCounter() prometheus.Counter {
  	return m.stsDeleteCreateCounter
  }
  
  type HandlerEvent string
  
  const (
  	AddEvent    = HandlerEvent("add")
  	DeleteEvent = HandlerEvent("delete")
  	UpdateEvent = HandlerEvent("update")
  )
  
  // TriggerByCounter returns a counter to track operator actions by resource type and action (add/delete/update).
  func (m *Metrics) TriggerByCounter(triggeredBy string, action HandlerEvent) prometheus.Counter {
  	return m.triggerByCounter.With(prometheus.Labels{"triggered_by": triggeredBy, "action": string(action)})
  }
  
  const (
  	selected int = iota
  	rejected
  )
  
  type resourceState int
  
  func (r resourceState) String() string {
  	switch int(r) {
  	case selected:
  		return "selected"
  	case rejected:
  		return "rejected"
  	}
  	return ""
  }
  
  // SetSelectedResources sets the number of resources that the controller selected for the given object's key.
  func (m *Metrics) SetSelectedResources(objKey, resource string, v int) {
  	m.setResources(objKey, resourceKey{resource: resource, state: resourceState(selected)}, v)
  }
  
  // SetRejectedResources sets the number of resources that the controller rejected for the given object's key.
  func (m *Metrics) SetRejectedResources(objKey, resource string, v int) {
  	m.setResources(objKey, resourceKey{resource: resource, state: resourceState(rejected)}, v)
  }
  
  func (m *Metrics) setResources(objKey string, resKey resourceKey, v int) {
  	m.mtx.Lock()
  	defer m.mtx.Unlock()
  
  	if _, found := m.resources[resKey]; !found {
  		m.resources[resourceKey{resource: resKey.resource, state: resourceState(selected)}] = make(map[string]int)
  		m.resources[resourceKey{resource: resKey.resource, state: resourceState(rejected)}] = make(map[string]int)
  	}
  
  	m.resources[resKey][objKey] = v
  }
  
  // Ready returns a gauge to track whether the controller is ready or not.
  func (m *Metrics) Ready() prometheus.Gauge {
  	return m.ready
  }
  
  // MustRegister registers metrics with the Metrics registerer.
  func (m *Metrics) MustRegister(metrics ...prometheus.Collector) {
  	m.reg.MustRegister(metrics...)
  }
  
  // Describe implements the prometheus.Collector interface.
  func (m *Metrics) Describe(ch chan<- *prometheus.Desc) {
  	ch <- resourcesDesc
  }
  
  // Collect implements the prometheus.Collector interface.
  func (m *Metrics) Collect(ch chan<- prometheus.Metric) {
  	m.mtx.RLock()
  	defer m.mtx.RUnlock()
  
  	for rKey := range m.resources {
  		var total int
  		for _, v := range m.resources[rKey] {
  			total += v
  		}
  		ch <- prometheus.MustNewConstMetric(
  			resourcesDesc,
  			prometheus.GaugeValue,
  			float64(total),
  			rKey.resource,
  			rKey.state.String(),
  		)
  	}
  }
  
  type instrumentedListerWatcher struct {
  	next        cache.ListerWatcher
  	listTotal   prometheus.Counter
  	listFailed  prometheus.Counter
  	watchTotal  prometheus.Counter
  	watchFailed prometheus.Counter
  }
  
  // NewInstrumentedListerWatcher returns a cache.ListerWatcher with instrumentation.
  func (m *Metrics) NewInstrumentedListerWatcher(lw cache.ListerWatcher) cache.ListerWatcher {
  	return &instrumentedListerWatcher{
  		next:        lw,
  		listTotal:   m.listCounter,
  		listFailed:  m.listFailedCounter,
  		watchTotal:  m.watchCounter,
  		watchFailed: m.watchFailedCounter,
  	}
  }
  
  // List implements the cache.ListerWatcher interface.
  func (i *instrumentedListerWatcher) List(options metav1.ListOptions) (runtime.Object, error) {
  	i.listTotal.Inc()
  	//nolint:staticcheck // Ignore SA1019 the function is deprecated.
  	ret, err := i.next.List(options)
  	if err != nil {
  		i.listFailed.Inc()
  	}
  	return ret, err
  }
  
  // IsWatchListSemanticsUnSupported calls the wrapped ListerWatcher if it
  // implements the function. Otherwise it returns false.
  // It is only required for the unpriviliged namespace ListerWatcher which
  // doesn't support the new ListWatch semantics.
  func (i *instrumentedListerWatcher) IsWatchListSemanticsUnSupported() bool {
  	type unSupportedWatchListSemantics interface {
  		IsWatchListSemanticsUnSupported() bool
  	}
  
  	lw, ok := i.next.(unSupportedWatchListSemantics)
  	if !ok {
  		return false
  	}
  	return lw.IsWatchListSemanticsUnSupported()
  }
  
  // Watch implements the cache.ListerWatcher interface.
  func (i *instrumentedListerWatcher) Watch(options metav1.ListOptions) (watch.Interface, error) {
  	i.watchTotal.Inc()
  	//nolint:staticcheck // Ignore SA1019 the function is deprecated.
  	ret, err := i.next.Watch(options)
  	if err != nil {
  		i.watchFailed.Inc()
  	}
  	return ret, err
  }
  
  // SanitizeSTS removes values for APIVersion and Kind from the VolumeClaimTemplates.
  // This prevents update failures due to these fields changing when applied.
  // See https://github.com/kubernetes/kubernetes/issues/87583
  func SanitizeSTS(sts *appsv1.StatefulSet) {
  	for i := range sts.Spec.VolumeClaimTemplates {
  		sts.Spec.VolumeClaimTemplates[i].APIVersion = ""
  		sts.Spec.VolumeClaimTemplates[i].Kind = ""
  	}
  }
  
  // WaitForNamedCacheSync synchronizes the informer's cache and will log a
  // warning every minute if the operation hasn't completed yet, until it reaches
  // a timeout of 10 minutes.
  // Under normal circumstances, the cache sync should be fast. If it takes more
  // than 1 minute, it means that something is stuck and the message will
  // indicate to the admin which informer is the culprit.
  // See https://github.com/prometheus-operator/prometheus-operator/issues/3347.
  func WaitForNamedCacheSync(ctx context.Context, controllerName string, logger *slog.Logger, inf cache.SharedIndexInformer) bool {
  	ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
  	defer cancel()
  
  	t := time.NewTicker(time.Minute)
  	defer t.Stop()
  
  	go func() {
  		for {
  			select {
  			case <-t.C:
  				logger.Warn("cache sync not yet completed")
  			case <-ctx.Done():
  				return
  			}
  		}
  	}()
  
  	ok := cache.WaitForNamedCacheSync(controllerName, ctx.Done(), inf.HasSynced)
  	if !ok {
  		logger.Error("failed to sync cache")
  	} else {
  		logger.Debug("successfully synced cache")
  	}
  
  	return ok
  }
  
  // ConfigMapGVK returns the GroupVersionKind representing ConfigMap objects.
  func ConfigMapGVK() schema.GroupVersionKind {
  	return corev1.SchemeGroupVersion.WithKind("ConfigMap")
  }
  
  // SecretGVK returns the GroupVersionKind representing Secret objects.
  func SecretGVK() schema.GroupVersionKind {
  	return corev1.SchemeGroupVersion.WithKind("Secret")
  }
  
  // SelectNamespacesFromCache returns the selected namespaces from the informer's cache.
  func SelectNamespacesFromCache(obj metav1.Object, sel *metav1.LabelSelector, nsInfs cache.SharedIndexInformer) ([]string, error) {
  	// If the selector is nil, return the object's namespace.
  	if sel == nil {
  		return []string{obj.GetNamespace()}, nil
  	}
  
  	labelSelector, err := metav1.LabelSelectorAsSelector(sel)
  	if err != nil {
  		return nil, fmt.Errorf("failed to convert namespace label selector to selector: %w", err)
  	}
  
  	var ns []string
  	err = cache.ListAll(nsInfs.GetStore(), labelSelector, func(obj any) {
  		ns = append(ns, obj.(*corev1.Namespace).Name)
  	})
  	if err != nil {
  		return nil, fmt.Errorf("failed to list namespaces: %w", err)
  	}
  	slices.Sort(ns)
  
  	return ns, nil
  }
  