OTA-1813: Populate risks from alerts

Author: hongkailiu

pkg/clusterconditions/promql/alerts.go

@@ -0,0 +1,95 @@
+package promql
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/api"
+	prometheusv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"github.com/prometheus/common/config"
+
+	"k8s.io/klog/v2"
+
+	"github.com/openshift/cluster-version-operator/pkg/clusterconditions"
+)
+
+type Getter interface {
+	Get(ctx context.Context) (prometheusv1.AlertsResult, error)
+}
+
+func NewAlertGetter(promQLTarget clusterconditions.PromQLTarget) Getter {
+	p := NewPromQL(promQLTarget)
+	condition := p.Condition
+	v, ok := condition.(*PromQL)
+	if !ok {
+		panic("invalid condition type")
+	}
+	return &ocAlertGetter{promQL: v, expiration: 1 * time.Minute}
+}
+
+type ocAlertGetter struct {
+	promQL *PromQL
+
+	mutex       sync.Mutex
+	cached      prometheusv1.AlertsResult
+	expiration  time.Duration
+	lastRefresh time.Time
+}
+
+func (o *ocAlertGetter) Get(ctx context.Context) (prometheusv1.AlertsResult, error) {
+	if time.Now().After(o.lastRefresh.Add(o.expiration)) {
+		if err := o.refresh(ctx); err != nil {
+			klog.Errorf("Failed to refresh alerts, using stale cache instead: %v", err)
+		}
+	}
+	return o.cached, nil
+}
+
+func (o *ocAlertGetter) refresh(ctx context.Context) error {
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+
+	klog.Info("refresh alerts ...")
+	p := o.promQL
+	host, err := p.Host(ctx)
+	if err != nil {
+		return fmt.Errorf("failure determine thanos IP: %w", err)
+	}
+	p.url.Host = host
+	clientConfig := api.Config{Address: p.url.String()}
+
+	if roundTripper, err := config.NewRoundTripperFromConfig(p.HTTPClientConfig, "cluster-conditions"); err == nil {
+		clientConfig.RoundTripper = roundTripper
+	} else {
+		return fmt.Errorf("creating PromQL round-tripper: %w", err)
+	}
+
+	promqlClient, err := api.NewClient(clientConfig)
+	if err != nil {
+		return fmt.Errorf("creating PromQL client: %w", err)
+	}
+
+	client := &statusCodeNotImplementedForPostClient{
+		client: promqlClient,
+	}
+
+	v1api := prometheusv1.NewAPI(client)
+
+	queryContext := ctx
+	if p.QueryTimeout > 0 {
+		var cancel context.CancelFunc
+		queryContext, cancel = context.WithTimeout(ctx, p.QueryTimeout)
+		defer cancel()
+	}
+
+	r, err := v1api.Alerts(queryContext)
+	if err != nil {
+		return fmt.Errorf("failed to get alerts: %w", err)
+	}
+	o.cached = r
+	o.lastRefresh = time.Now()
+	klog.Infof("refreshed: %d alerts", len(o.cached.Alerts))
+	return nil
+}

pkg/cvo/availableupdates.go

@@ -15,6 +15,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
+	prometheusv1 "github.com/prometheus/client_golang/api/prometheus/v1"
 
 	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -145,6 +146,7 @@ func (optr *Operator) syncAvailableUpdates(ctx context.Context, config *configv1
 		optrAvailableUpdates.ShouldReconcileAcceptRisks = optr.shouldReconcileAcceptRisks
 		optrAvailableUpdates.AcceptRisks = acceptRisks
 		optrAvailableUpdates.ConditionRegistry = optr.conditionRegistry
+		optrAvailableUpdates.AlertGetter = optr.AlertGetter
 		optrAvailableUpdates.Condition = condition
 
 		responseFailed := (condition.Type == configv1.RetrievedUpdates &&
@@ -159,6 +161,11 @@ func (optr *Operator) syncAvailableUpdates(ctx context.Context, config *configv1
 		}
 	}
 
+	if optrAvailableUpdates.ShouldReconcileAcceptRisks() {
+		if err := optrAvailableUpdates.evaluateAlertRisks(ctx); err != nil {
+			klog.Errorf("Failed to evaluate alert conditions: %v", err)
+		}
+	}
 	optrAvailableUpdates.evaluateConditionalUpdates(ctx)
 
 	queueKey := optr.queueKey()
@@ -212,6 +219,11 @@ type availableUpdates struct {
 
 	// RiskConditions stores the condition for every risk (name, url, message, matchingRules).
 	RiskConditions map[string][]metav1.Condition
+
+	AlertGetter AlertGetter
+
+	// AlertRisks stores the condition for every alert that might have impact on cluster update
+	AlertRisks []configv1.ConditionalUpdateRisk
 }
 
 func (u *availableUpdates) RecentlyAttempted(interval time.Duration) bool {
@@ -317,6 +329,8 @@ func (optr *Operator) getAvailableUpdates() *availableUpdates {
 		ShouldReconcileAcceptRisks: optr.shouldReconcileAcceptRisks,
 		AcceptRisks:                optr.availableUpdates.AcceptRisks,
 		RiskConditions:             optr.availableUpdates.RiskConditions,
+		AlertRisks:                 optr.availableUpdates.AlertRisks,
+		AlertGetter:                optr.availableUpdates.AlertGetter,
 		LastAttempt:                optr.availableUpdates.LastAttempt,
 		LastSyncOrConfigChange:     optr.availableUpdates.LastSyncOrConfigChange,
 		Current:                    *optr.availableUpdates.Current.DeepCopy(),
@@ -513,6 +527,166 @@ func calculateAvailableUpdatesStatus(ctx context.Context, clusterID string, tran
 	}
 }
 
+type AlertGetter interface {
+	Get(ctx context.Context) (prometheusv1.AlertsResult, error)
+}
+
+func (u *availableUpdates) evaluateAlertRisks(ctx context.Context) error {
+	if u == nil || u.AlertGetter == nil {
+		u.AlertRisks = nil
+		return nil
+	}
+	alertsResult, err := u.AlertGetter.Get(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get alerts: %w", err)
+	}
+
+	u.AlertRisks = alertsToRisks(alertsResult.Alerts)
+	return nil
+}
+
+func alertsToRisks(alerts []prometheusv1.Alert) []configv1.ConditionalUpdateRisk {
+	klog.V(2).Infof("Found %d alerts", len(alerts))
+	risks := map[string]configv1.ConditionalUpdateRisk{}
+	for _, alert := range alerts {
+		var alertName string
+		if alertName = string(alert.Labels["alertname"]); alertName == "" {
+			continue
+		}
+		if alert.State == "pending" {
+			continue
+		}
+
+		var summary string
+		if summary = string(alert.Annotations["summary"]); summary == "" {
+			summary = alertName
+		}
+		if !strings.HasSuffix(summary, ".") {
+			summary += "."
+		}
+
+		var description string
+		alertMessage := string(alert.Annotations["message"])
+		alertDescription := string(alert.Annotations["description"])
+		switch {
+		case alertMessage != "" && alertDescription != "":
+			description += " The alert description is: " + alertDescription + " | " + alertMessage
+		case alertDescription != "":
+			description += " The alert description is: " + alertDescription
+		case alertMessage != "":
+			description += " The alert description is: " + alertMessage
+		default:
+			description += " The alert has no description."
+		}
+
+		var runbook string
+		if runbook = string(alert.Annotations["runbook"]); runbook == "" {
+			runbook = "<alert does not have a runbook_url annotation>"
+		}
+
+		details := fmt.Sprintf("%s%s %s", summary, description, runbook)
+
+		// TODO (hongkliu):
+		alertURL := "https://console.com/monitoring/alertrules/id"
+		alertPromQL := "todo-expression"
+
+		severity := string(alert.Labels["severity"])
+		if severity == "critical" {
+			if alertName == "PodDisruptionBudgetLimit" {
+				details = fmt.Sprintf("Namespace=%s, PodDisruptionBudget=%s. %s", alert.Labels["namespace"], alert.Labels["poddisruptionbudget"], details)
+			}
+			risks[alertName] = getRisk(risks, alertName, summary, alertURL, alertPromQL, metav1.Condition{
+				Type:               internal.ConditionalUpdateRiskConditionTypeApplies,
+				Status:             metav1.ConditionTrue,
+				Reason:             fmt.Sprintf("Alert:%s", alert.State),
+				Message:            fmt.Sprintf("%s alert %s %s, suggesting significant cluster issues worth investigating. %s", severity, alertName, alert.State, details),
+				LastTransitionTime: metav1.NewTime(alert.ActiveAt),
+			})
+			continue
+		}
+
+		if alertName == "PodDisruptionBudgetAtLimit" {
+			details = fmt.Sprintf("Namespace=%s, PodDisruptionBudget=%s. %s", alert.Labels["namespace"], alert.Labels["poddisruptionbudget"], details)
+			risks[alertName] = getRisk(risks, alertName, summary, alertURL, alertPromQL, metav1.Condition{
+				Type:               internal.ConditionalUpdateRiskConditionTypeApplies,
+				Status:             metav1.ConditionTrue,
+				Reason:             internal.AlertConditionReason(string(alert.State)),
+				Message:            fmt.Sprintf("%s alert %s %s, which might slow node drains. %s", severity, alertName, alert.State, details),
+				LastTransitionTime: metav1.NewTime(alert.ActiveAt),
+			})
+			continue
+		}
+
+		if alertName == "KubeContainerWaiting" {
+			risks[alertName] = getRisk(risks, alertName, summary, alertURL, alertPromQL, metav1.Condition{
+				Type:               internal.ConditionalUpdateRiskConditionTypeApplies,
+				Status:             metav1.ConditionTrue,
+				Reason:             internal.AlertConditionReason(string(alert.State)),
+				Message:            fmt.Sprintf("%s alert %s %s, which may slow workload redistribution during rolling node updates. %s", severity, alertName, alert.State, details),
+				LastTransitionTime: metav1.NewTime(alert.ActiveAt),
+			})
+			continue
+		}
+
+		if alertName == "KubeNodeNotReady" || alertName == "KubeNodeReadinessFlapping" || alertName == "KubeNodeUnreachable" {
+			risks[alertName] = getRisk(risks, alertName, summary, alertURL, alertPromQL, metav1.Condition{
+				Type:               internal.ConditionalUpdateRiskConditionTypeApplies,
+				Status:             metav1.ConditionTrue,
+				Reason:             internal.AlertConditionReason(string(alert.State)),
+				Message:            fmt.Sprintf("%s alert %s %s, which may slow workload redistribution during rolling node updates. %s", severity, alertName, alert.State, details),
+				LastTransitionTime: metav1.NewTime(alert.ActiveAt),
+			})
+			continue
+		}
+	}
+
+	klog.V(2).Infof("Got %d risks", len(risks))
+	if len(risks) == 0 {
+		return nil
+	}
+
+	var ret []configv1.ConditionalUpdateRisk
+	var keys []string
+	for k := range risks {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	for _, k := range keys {
+		ret = append(ret, risks[k])
+	}
+	return ret
+}
+
+func getRisk(risks map[string]configv1.ConditionalUpdateRisk, riskName, message, url, promQL string, condition metav1.Condition) configv1.ConditionalUpdateRisk {
+	risk, ok := risks[riskName]
+	if !ok {
+		return configv1.ConditionalUpdateRisk{
+			Name:    riskName,
+			Message: message,
+			URL:     url,
+			MatchingRules: []configv1.ClusterCondition{
+				{
+					Type: "PromQL",
+					PromQL: &configv1.PromQLClusterCondition{
+						PromQL: promQL,
+					},
+				},
+			},
+			Conditions: []metav1.Condition{condition},
+		}
+	}
+
+	if c := meta.FindStatusCondition(risk.Conditions, condition.Type); c != nil {
+		c.Message = fmt.Sprintf("%s; %s", c.Message, condition.Message)
+		if c.LastTransitionTime.After(condition.LastTransitionTime.Time) {
+			c.LastTransitionTime = condition.LastTransitionTime
+		}
+		meta.SetStatusCondition(&risk.Conditions, *c)
+	}
+
+	return risk
+}
+
 func (u *availableUpdates) evaluateConditionalUpdates(ctx context.Context) {
 	if u == nil {
 		return
@@ -522,7 +696,7 @@ func (u *availableUpdates) evaluateConditionalUpdates(ctx context.Context) {
 	risks := risksInOrder(riskVersions)
 	u.RiskConditions = loadRiskConditions(ctx, risks, riskVersions, u.ConditionRegistry)
 
-	if err := sanityCheck(u.ConditionalUpdates); err != nil {
+	if err := sanityCheck(u.ConditionalUpdates, u.AlertRisks); err != nil {
 		klog.Errorf("Sanity check failed which might impact risk evaluation: %v", err)
 	}
 	for i, conditionalUpdate := range u.ConditionalUpdates {
@@ -540,7 +714,7 @@ func (u *availableUpdates) evaluateConditionalUpdates(ctx context.Context) {
 	}
 }
 
-func sanityCheck(updates []configv1.ConditionalUpdate) error {
+func sanityCheck(updates []configv1.ConditionalUpdate, alertRisks []configv1.ConditionalUpdateRisk) error {
 	risks := map[string]configv1.ConditionalUpdateRisk{}
 	var errs []error
 	for _, update := range updates {
@@ -556,6 +730,11 @@ func sanityCheck(updates []configv1.ConditionalUpdate) error {
 			}
 		}
 	}
+	for _, alertRisk := range alertRisks {
+		if _, ok := risks[alertRisk.Name]; ok {
+			errs = append(errs, fmt.Errorf("found alert risk and conditional update risk share the name: %s", alertRisk.Name))
+		}
+	}
 	return utilerrors.NewAggregate(errs)
 }
 
@@ -600,6 +779,7 @@ const (
 	recommendedReasonAllExposedRisksAccepted = "AllExposedRisksAccepted"
 	recommendedReasonEvaluationFailed        = "EvaluationFailed"
 	recommendedReasonMultiple                = "MultipleReasons"
+	//recommendedReasonAlertFiring             = "AlertFiring"
 
 	// recommendedReasonExposed is used instead of the original name if it does
 	// not match the pattern for a valid k8s condition reason.
@@ -617,10 +797,13 @@ var reasonPattern = regexp.MustCompile(`^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
 func newRecommendedReason(now, want string) string {
 	switch {
 	case now == recommendedReasonRisksNotExposed ||
-		now == recommendedReasonAllExposedRisksAccepted ||
+		now == recommendedReasonAllExposedRisksAccepted && want != recommendedReasonRisksNotExposed ||
+		now == recommendedReasonEvaluationFailed && want == recommendedReasonExposed ||
 		now == want:
 		return want
-	case want == recommendedReasonRisksNotExposed:
+	case want == recommendedReasonRisksNotExposed ||
+		(now == recommendedReasonEvaluationFailed) && want == recommendedReasonAllExposedRisksAccepted ||
+		now == recommendedReasonExposed && (want == recommendedReasonAllExposedRisksAccepted || want == recommendedReasonEvaluationFailed):
 		return now
 	default:
 		return recommendedReasonMultiple
@@ -677,7 +860,6 @@ func evaluateConditionalUpdate(
 	if len(errorMessages) > 0 {
 		recommended.Message = strings.Join(errorMessages, "\n\n")
 	}
-
 	return recommended
 }