ROSAENG-62134: Add rate-limit backoff guard for fleet mode HTTP 429

pkg/handlers/webhookrhobsreceiver.go

@@ -3,9 +3,11 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	stderrors "errors"
 	"fmt"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/prometheus/alertmanager/template"
@@ -41,6 +43,12 @@ var (
 	}
 
 	customIs409 = func(err error) bool { return errors.IsConflict(err) || errors.IsAlreadyExists(err) }
+
+	// rateLimitBackoffs tracks per-notification:cluster backoff timestamps after
+	// an OCM API 429 rate-limit response.  The key is "notification:clusterID"
+	// and the value is the time.Time when the 429 was received.
+	rateLimitBackoffs      sync.Map
+	rateLimitRetryInterval = 30 * time.Minute
 )
 
 type WebhookRHOBSReceiverHandler struct {
@@ -376,10 +384,30 @@ func (h *WebhookRHOBSReceiverHandler) processAlert(alert template.Alert, isCurre
 		return fmt.Errorf("unable to find ManagedFleetNotification %s", alert.Labels[AMLabelTemplateName])
 	}
 
+	// When an alert resolves, clear any rate-limit backoff entry so that
+	// if the alert fires again later, the send is not suppressed.
+	if !isCurrentlyFiring {
+		rateLimitKey := alert.Labels[AMLabelTemplateName] + ":" + alert.Labels[AMLabelAlertHCID]
+		rateLimitBackoffs.Delete(rateLimitKey)
+	}
+
 	if !fleetNotificationRetriever.fleetNotification.LimitedSupport && !isCurrentlyFiring {
 		return nil
 	}
 
+	// Skip firing alerts that are within the rate-limit backoff window.
+	if isCurrentlyFiring {
+		rateLimitKey := alert.Labels[AMLabelTemplateName] + ":" + alert.Labels[AMLabelAlertHCID]
+		if backoffTime, ok := rateLimitBackoffs.Load(rateLimitKey); ok {
+			if time.Since(backoffTime.(time.Time)) < rateLimitRetryInterval {
+				log.WithFields(log.Fields{
+					LogFieldNotificationName: alert.Labels[AMLabelTemplateName],
+				}).Warn("skipping alert due to OCM API rate-limit backoff")
+				return nil
+			}
+		}
+	}
+
 	var c *fleetNotificationContext
 	canSend := false
 	err = retryOnConflictOrAlreadyExists(retryConfig, func() error {
@@ -403,6 +431,7 @@ func (h *WebhookRHOBSReceiverHandler) processAlert(alert template.Alert, isCurre
 
 	if isCurrentlyFiring {
 		if canSend {
+			sendStartTime := time.Now()
 			err := c.sendNotification(h.ocm, alert)
 
 			var logService string
@@ -413,6 +442,14 @@ func (h *WebhookRHOBSReceiverHandler) processAlert(alert template.Alert, isCurre
 			}
 
 			if err != nil {
+				var rateLimitErr *ocm.RateLimitError
+				if stderrors.As(err, &rateLimitErr) {
+					rateLimitKey := alert.Labels[AMLabelTemplateName] + ":" + alert.Labels[AMLabelAlertHCID]
+					rateLimitBackoffs.Store(rateLimitKey, time.Now())
+					log.WithFields(log.Fields{
+						LogFieldNotificationName: fleetNotification.Name,
+					}).Warn("OCM API rate limit hit (HTTP 429), backing off for 30 minutes")
+				}
 				if fleetNotification.LimitedSupport { // Limited support case
 					metrics.IncrementFailedLimitedSupportSend(fleetNotification.Name)
 				} else { // Service log case
@@ -423,6 +460,17 @@ func (h *WebhookRHOBSReceiverHandler) processAlert(alert template.Alert, isCurre
 				return err
 			}
 
+			// Clear any rate-limit backoff for this notification:cluster pair
+			// after a successful send so normal operation resumes immediately.
+			// Only delete if the stored timestamp predates this send attempt,
+			// so a concurrent 429 that arrived while we were sending is preserved.
+			rateLimitKey := alert.Labels[AMLabelTemplateName] + ":" + alert.Labels[AMLabelAlertHCID]
+			if val, ok := rateLimitBackoffs.Load(rateLimitKey); ok {
+				if val.(time.Time).Before(sendStartTime) {
+					rateLimitBackoffs.Delete(rateLimitKey)
+				}
+			}
+
 			if fleetNotification.LimitedSupport { // Limited support case
 				metrics.IncrementLimitedSupportSentCount(fleetNotification.Name)
 			} else { // Service log case

pkg/handlers/webhookrhobsreceiver_test.go

@@ -664,3 +664,158 @@ func (f *FailingResponseWriter) Write([]byte) (int, error) {
 func (f *FailingResponseWriter) WriteHeader(statusCode int) {
 	f.statusCode = statusCode
 }
+
+func clearRateLimitBackoffs() {
+	rateLimitBackoffs.Range(func(k, v any) bool {
+		rateLimitBackoffs.Delete(k)
+		return true
+	})
+}
+
+var _ = Describe("Rate-limit backoff behavior", func() {
+	var (
+		mockCtrl                       *gomock.Controller
+		mockClient                     *clientmocks.MockClient
+		mockOCMClient                  *webhookreceivermock.MockOCMClient
+		testHandler                    *WebhookRHOBSReceiverHandler
+		testAlertFiring                template.Alert
+		mockStatusWriter               *clientmocks.MockStatusWriter
+		serviceLog                     *ocm.ServiceLog
+		managedFleetNotification       *ocmagentv1alpha1.ManagedFleetNotification
+		managedFleetNotificationRecord *ocmagentv1alpha1.ManagedFleetNotificationRecord
+	)
+
+	BeforeEach(func() {
+		clearRateLimitBackoffs()
+
+		mockCtrl = gomock.NewController(GinkgoT())
+		mockClient = clientmocks.NewMockClient(mockCtrl)
+		mockStatusWriter = clientmocks.NewMockStatusWriter(mockCtrl)
+		mockOCMClient = webhookreceivermock.NewMockOCMClient(mockCtrl)
+		testHandler = &WebhookRHOBSReceiverHandler{
+			c:   mockClient,
+			ocm: mockOCMClient,
+		}
+		testAlertFiring = testconst.NewTestAlert(false, true)
+
+		defaultMFN := testconst.NewManagedFleetNotification(false)
+		defaultMFN.Spec.FleetNotification.ResendWait = 1
+		managedFleetNotification = &defaultMFN
+
+		defaultRecord := testconst.NewManagedFleetNotificationRecordWithStatus()
+		managedFleetNotificationRecord = &defaultRecord
+
+		serviceLog = testconst.NewTestServiceLog(
+			ocm.ServiceLogActivePrefix+": "+testconst.ServiceLogSummary,
+			testconst.ServiceLogFleetDesc,
+			testconst.TestHostedClusterID,
+			testconst.TestNotification.Severity,
+			"",
+			testconst.TestNotification.References)
+
+		// Setup k8s mocks for ManagedFleetNotification and Record retrieval
+		mockClient.EXPECT().Get(gomock.Any(), client.ObjectKey{
+			Namespace: OCMAgentNamespaceName,
+			Name:      managedFleetNotification.ObjectMeta.Name,
+		}, gomock.Any()).DoAndReturn(
+			func(ctx context.Context, key client.ObjectKey, res *ocmagentv1alpha1.ManagedFleetNotification, opts ...client.GetOption) error {
+				*res = *managedFleetNotification
+				return nil
+			}).AnyTimes()
+
+		mockClient.EXPECT().Get(gomock.Any(), client.ObjectKey{
+			Namespace: OCMAgentNamespaceName,
+			Name:      managedFleetNotificationRecord.ObjectMeta.Name,
+		}, gomock.Any()).DoAndReturn(
+			func(ctx context.Context, key client.ObjectKey, res *ocmagentv1alpha1.ManagedFleetNotificationRecord, opts ...client.GetOption) error {
+				*res = *managedFleetNotificationRecord
+				return nil
+			}).AnyTimes()
+
+		mockClient.EXPECT().Status().Return(mockStatusWriter).AnyTimes()
+		mockStatusWriter.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	})
+
+	AfterEach(func() {
+		clearRateLimitBackoffs()
+		mockCtrl.Finish()
+	})
+
+	It("populates rateLimitBackoffs when SendServiceLog returns RateLimitError", func() {
+		rateLimitErr := &ocm.RateLimitError{Err: fmt.Errorf("rate limited")}
+		mockOCMClient.EXPECT().SendServiceLog(serviceLog).Return(rateLimitErr)
+
+		err := testHandler.processAlert(testAlertFiring, true)
+		Expect(err).To(HaveOccurred())
+
+		key := testconst.TestNotificationName + ":" + testconst.TestHostedClusterID
+		_, ok := rateLimitBackoffs.Load(key)
+		Expect(ok).To(BeTrue())
+	})
+
+	It("skips sending when within the rate-limit backoff window", func() {
+		key := testconst.TestNotificationName + ":" + testconst.TestHostedClusterID
+		rateLimitBackoffs.Store(key, time.Now())
+
+		// SendServiceLog should NOT be called because the backoff guard returns early
+		err := testHandler.processAlert(testAlertFiring, true)
+		Expect(err).ShouldNot(HaveOccurred())
+	})
+
+	It("proceeds normally after the backoff window expires", func() {
+		key := testconst.TestNotificationName + ":" + testconst.TestHostedClusterID
+		rateLimitBackoffs.Store(key, time.Now().Add(-rateLimitRetryInterval-time.Minute))
+
+		mockOCMClient.EXPECT().SendServiceLog(serviceLog).Return(nil)
+
+		err := testHandler.processAlert(testAlertFiring, true)
+		Expect(err).ShouldNot(HaveOccurred())
+	})
+
+	It("clears backoff entry on successful send", func() {
+		key := testconst.TestNotificationName + ":" + testconst.TestHostedClusterID
+		rateLimitBackoffs.Store(key, time.Now().Add(-rateLimitRetryInterval-time.Minute))
+
+		mockOCMClient.EXPECT().SendServiceLog(serviceLog).Return(nil)
+
+		err := testHandler.processAlert(testAlertFiring, true)
+		Expect(err).ShouldNot(HaveOccurred())
+
+		_, ok := rateLimitBackoffs.Load(key)
+		Expect(ok).To(BeFalse())
+	})
+
+	It("clears backoff entry when a resolved alert is received", func() {
+		key := testconst.TestNotificationName + ":" + testconst.TestHostedClusterID
+		rateLimitBackoffs.Store(key, time.Now())
+
+		testAlertResolved := testconst.NewTestAlert(true, true)
+		err := testHandler.processAlert(testAlertResolved, false)
+		Expect(err).ShouldNot(HaveOccurred())
+
+		_, ok := rateLimitBackoffs.Load(key)
+		Expect(ok).To(BeFalse())
+	})
+
+	It("does not clear a newer backoff stored during an in-flight send", func() {
+		key := testconst.TestNotificationName + ":" + testconst.TestHostedClusterID
+		// Seed an old, expired backoff so the request is not skipped.
+		rateLimitBackoffs.Store(key, time.Now().Add(-rateLimitRetryInterval-time.Minute))
+
+		// Simulate a concurrent 429: while SendServiceLog is running,
+		// another goroutine stores a fresh backoff timestamp.
+		mockOCMClient.EXPECT().SendServiceLog(serviceLog).DoAndReturn(
+			func(sl *ocm.ServiceLog) error {
+				rateLimitBackoffs.Store(key, time.Now())
+				return nil
+			},
+		)
+
+		err := testHandler.processAlert(testAlertFiring, true)
+		Expect(err).ShouldNot(HaveOccurred())
+
+		// The fresh backoff must survive the success-path cleanup.
+		_, ok := rateLimitBackoffs.Load(key)
+		Expect(ok).To(BeTrue())
+	})
+})

pkg/ocm/ocm.go

@@ -20,6 +20,19 @@ const (
 	ServiceLogResolvePrefix = "Issue Resolution"
 )
 
+// RateLimitError indicates that an OCM API call was rejected with HTTP 429.
+type RateLimitError struct {
+	Err error
+}
+
+func (e *RateLimitError) Error() string {
+	return e.Err.Error()
+}
+
+func (e *RateLimitError) Unwrap() error {
+	return e.Err
+}
+
 type ServiceLogBuilder struct {
 	wrappedBuilder *slv1.LogEntryBuilder
 	summary        string
@@ -242,10 +255,16 @@ func (o *ocmClientImpl) SendServiceLog(logEntry *slv1.LogEntry) error {
 	// Send the request to the OCM API.
 	response, err := request.Send()
 	if err != nil {
+		if response != nil && response.Status() == http.StatusTooManyRequests {
+			return &RateLimitError{Err: fmt.Errorf("can't post service log: rate limited (HTTP 429): %w", err)}
+		}
 		return fmt.Errorf("can't post service log: %v", err)
 	}
 
 	// Check the response status code.
+	if response.Status() == http.StatusTooManyRequests {
+		return &RateLimitError{Err: fmt.Errorf("can't post service log: rate limited (HTTP 429)")}
+	}
 	if response.Status() != http.StatusCreated {
 		// Extract error details from the response and return an appropriate error.
 		return fmt.Errorf("unexpected status: %d", response.Status())

pkg/ocm/ocm_test.go

@@ -1,6 +1,7 @@
 package ocm
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"testing"
@@ -312,6 +313,95 @@ var _ = Describe("OCM client Handler", func() {
 		})
 
 	})
+
+	Context("RateLimitError type", func() {
+		It("Error() returns the wrapped error message", func() {
+			inner := fmt.Errorf("some error")
+			rle := &RateLimitError{Err: inner}
+			Expect(rle.Error()).To(Equal("some error"))
+		})
+
+		It("Unwrap() returns the inner error", func() {
+			inner := fmt.Errorf("some error")
+			rle := &RateLimitError{Err: inner}
+			Expect(rle.Unwrap()).To(Equal(inner))
+		})
+
+		It("errors.As() detects RateLimitError from a wrapped error chain", func() {
+			inner := fmt.Errorf("sdk error")
+			rle := &RateLimitError{Err: inner}
+			wrapped := fmt.Errorf("outer context: %w", rle)
+
+			var target *RateLimitError
+			Expect(errors.As(wrapped, &target)).To(BeTrue())
+			Expect(target.Err.Error()).To(Equal("sdk error"))
+		})
+
+		It("errors.As() does not match a plain error as RateLimitError", func() {
+			plainErr := fmt.Errorf("just an error")
+			var target *RateLimitError
+			Expect(errors.As(plainErr, &target)).To(BeFalse())
+		})
+	})
+
+	Context("SendServiceLog returns RateLimitError on 429", func() {
+		It("should return a RateLimitError when the API responds with HTTP 429", func() {
+			mockServer.SetHandler(0, CombineHandlers(
+				VerifyRequest("POST", "/api/service_logs/v1/cluster_logs"),
+				RespondWith(
+					http.StatusTooManyRequests,
+					`{"kind": "Error", "id": "429", "href": "/api/service_logs/v1/errors/429", "code": "SERVICE-LOGS-429", "reason": "Rate limit exceeded"}`,
+					http.Header{"Content-Type": []string{"application/json"}},
+				),
+			))
+			// The OCM SDK may retry on certain errors, so add extra handlers
+			for i := 0; i < 5; i++ {
+				mockServer.AppendHandlers(CombineHandlers(
+					VerifyRequest("POST", "/api/service_logs/v1/cluster_logs"),
+					RespondWith(
+						http.StatusTooManyRequests,
+						`{"kind": "Error", "id": "429", "href": "/api/service_logs/v1/errors/429", "code": "SERVICE-LOGS-429", "reason": "Rate limit exceeded"}`,
+						http.Header{"Content-Type": []string{"application/json"}},
+					),
+				))
+			}
+
+			err := ocmClient.SendServiceLog(serviceLog)
+			Expect(err).To(HaveOccurred())
+
+			var rateLimitErr *RateLimitError
+			Expect(errors.As(err, &rateLimitErr)).To(BeTrue())
+			Expect(rateLimitErr.Error()).To(ContainSubstring("rate limited"))
+		})
+
+		It("should return a non-RateLimitError for other failure statuses", func() {
+			mockServer.SetHandler(0, CombineHandlers(
+				VerifyRequest("POST", "/api/service_logs/v1/cluster_logs"),
+				RespondWith(
+					http.StatusInternalServerError,
+					`{"kind": "Error", "id": "500", "href": "/api/service_logs/v1/errors/500", "code": "SERVICE-LOGS-500", "reason": "Internal server error"}`,
+					http.Header{"Content-Type": []string{"application/json"}},
+				),
+			))
+			for i := 0; i < 5; i++ {
+				mockServer.AppendHandlers(CombineHandlers(
+					VerifyRequest("POST", "/api/service_logs/v1/cluster_logs"),
+					RespondWith(
+						http.StatusInternalServerError,
+						`{"kind": "Error", "id": "500", "href": "/api/service_logs/v1/errors/500", "code": "SERVICE-LOGS-500", "reason": "Internal server error"}`,
+						http.Header{"Content-Type": []string{"application/json"}},
+					),
+				))
+			}
+
+			err := ocmClient.SendServiceLog(serviceLog)
+			Expect(err).To(HaveOccurred())
+
+			var rateLimitErr *RateLimitError
+			Expect(errors.As(err, &rateLimitErr)).To(BeFalse())
+		})
+	})
+
 	Context("Limit support", func() {
 		It("should not return an error on successful post", func() {
 			mockServer.SetHandler(0, CombineHandlers(