feat: heartbeat crd

Author: loks0n

.github/workflows/test.yml

@@ -149,4 +149,4 @@ jobs:
             echo "BETTERSTACK_TOKEN secret not set; failing"
             exit 1
           fi
-          go test -tags=e2e ./test/e2e -run TestBetterStackMonitorLifecycle -timeout 20m
+          go test -tags=e2e ./test/e2e -run TestBetterStackOperatorLifecycle -timeout 20m

config/crd/bases/monitoring.betterstack.io_betterstackheartbeats.yaml

@@ -17,6 +17,8 @@ spec:
     - name: v1alpha1
       served: true
       storage: true
+      subresources:
+        status: {}
       additionalPrinterColumns:
         - name: Name
           type: string

controllers/betterstackheartbeat_controller.go

@@ -2,8 +2,10 @@ package controllers
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
+	"strings"
 
 	"k8s.io/utils/ptr"
 
@@ -41,6 +43,11 @@ type BetterStackHeartbeatReconciler struct {
 	Clients    BetterStackHeartbeatClientFactory
 }
 
+const (
+	// ReasonHeartbeatQuotaExceeded marks a reconciliation failure caused by Better Stack heartbeat quota limits.
+	ReasonHeartbeatQuotaExceeded = "HeartbeatQuotaExceeded"
+)
+
 //+kubebuilder:rbac:groups=monitoring.betterstack.io,resources=betterstackheartbeats,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=monitoring.betterstack.io,resources=betterstackheartbeats/status,verbs=get;update;patch
 //+kubebuilder:rbac:groups=monitoring.betterstack.io,resources=betterstackheartbeats/finalizers,verbs=update
@@ -104,10 +111,18 @@ func (r *BetterStackHeartbeatReconciler) Reconcile(ctx context.Context, req ctrl
 
 	if err != nil {
 		logger.Error(err, "unable to reconcile Better Stack heartbeat")
+		syncReason := "SyncFailed"
+		syncMessage := err.Error()
+		readyMessage := "Heartbeat reconciliation failed"
+		if isHeartbeatQuotaExceeded(err) {
+			syncReason = ReasonHeartbeatQuotaExceeded
+			syncMessage = "Better Stack heartbeat quota reached"
+			readyMessage = "Better Stack heartbeat quota reached"
+		}
 		_ = r.patchStatus(ctx, heartbeat, func(status *monitoringv1alpha1.BetterStackHeartbeatStatus) {
 			now := metav1.Now()
-			status.SetCondition(conditions.New(monitoringv1alpha1.ConditionSync, metav1.ConditionFalse, "SyncFailed", err.Error(), &now))
-			status.SetCondition(conditions.New(monitoringv1alpha1.ConditionReady, metav1.ConditionFalse, "SyncFailed", "Heartbeat reconciliation failed", &now))
+			status.SetCondition(conditions.New(monitoringv1alpha1.ConditionSync, metav1.ConditionFalse, syncReason, syncMessage, &now))
+			status.SetCondition(conditions.New(monitoringv1alpha1.ConditionReady, metav1.ConditionFalse, syncReason, readyMessage, &now))
 		})
 		return ctrl.Result{RequeueAfter: requeueIntervalOnError}, nil
 	}
@@ -233,3 +248,14 @@ func (r *BetterStackHeartbeatReconciler) heartbeatService(baseURL, token string)
 	}
 	return factory.Heartbeat(baseURL, token, r.HTTPClient)
 }
+
+func isHeartbeatQuotaExceeded(err error) bool {
+	var apiErr *betterstack.APIError
+	if !errors.As(err, &apiErr) {
+		return false
+	}
+	if apiErr.StatusCode != http.StatusForbidden {
+		return false
+	}
+	return strings.Contains(strings.ToLower(apiErr.Message), "quota")
+}

controllers/betterstackheartbeat_controller_test.go

@@ -403,6 +403,66 @@ func TestHeartbeatReconcileHandlesCreateError(t *testing.T) {
 	assert.String(t, "sync reason", syncCond.Reason, "SyncFailed")
 }
 
+func TestHeartbeatReconcileHandlesQuotaExceeded(t *testing.T) {
+	scheme := controllertest.NewScheme(t)
+
+	heartbeat := &monitoringv1alpha1.BetterStackHeartbeat{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:       "example",
+			Namespace:  "default",
+			Generation: 4,
+			Finalizers: []string{monitoringv1alpha1.BetterStackHeartbeatFinalizer},
+		},
+		Spec: monitoringv1alpha1.BetterStackHeartbeatSpec{
+			Name:          "Example",
+			PeriodSeconds: 60,
+			BaseURL:       "https://api.test",
+			APITokenSecretRef: corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{Name: "api"},
+				Key:                  "token",
+			},
+		},
+	}
+
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "api", Namespace: "default"},
+		Data:       map[string][]byte{"token": []byte("abcd")},
+	}
+
+	service := &fakeHeartbeatService{
+		createFn: func(ctx context.Context, req betterstack.HeartbeatCreateRequest) (betterstack.Heartbeat, error) {
+			return betterstack.Heartbeat{}, &betterstack.APIError{StatusCode: http.StatusForbidden, Message: "Heartbeat quota reached. Please upgrade your account."}
+		},
+	}
+	factory := &fakeBetterStackHeartbeatClientFactory{heartbeat: service}
+
+	client := fake.NewClientBuilder().
+		WithScheme(scheme).
+		WithStatusSubresource(heartbeat).
+		WithObjects(heartbeat.DeepCopy(), secret.DeepCopy()).
+		Build()
+
+	r := &BetterStackHeartbeatReconciler{Client: client, Scheme: scheme, Clients: factory}
+
+	ctx := context.Background()
+	res, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: types.NamespacedName{Name: heartbeat.Name, Namespace: heartbeat.Namespace}})
+	assert.NoError(t, err, "reconcile")
+	assert.Equal(t, "requeueAfter", res.RequeueAfter, requeueIntervalOnError)
+
+	updated := &monitoringv1alpha1.BetterStackHeartbeat{}
+	assert.NoError(t, client.Get(ctx, types.NamespacedName{Name: heartbeat.Name, Namespace: heartbeat.Namespace}, updated), "fetch updated heartbeat")
+	syncCond := controllertest.FindCondition(updated.Status.Conditions, monitoringv1alpha1.ConditionSync)
+	assert.NotNil(t, "sync condition", syncCond)
+	assert.Equal(t, "sync status", syncCond.Status, metav1.ConditionFalse)
+	assert.String(t, "sync reason", syncCond.Reason, ReasonHeartbeatQuotaExceeded)
+	assert.String(t, "sync message", syncCond.Message, "Better Stack heartbeat quota reached")
+	readyCond := controllertest.FindCondition(updated.Status.Conditions, monitoringv1alpha1.ConditionReady)
+	assert.NotNil(t, "ready condition", readyCond)
+	assert.Equal(t, "ready status", readyCond.Status, metav1.ConditionFalse)
+	assert.String(t, "ready reason", readyCond.Reason, ReasonHeartbeatQuotaExceeded)
+	assert.String(t, "ready message", readyCond.Message, "Better Stack heartbeat quota reached")
+}
+
 func TestHeartbeatReconcileHandlesDeletion(t *testing.T) {
 	scheme := controllertest.NewScheme(t)
 

controllers/betterstackmonitor_controller_test.go

@@ -52,11 +52,13 @@ type fakeMonitorService struct {
 	updateFn func(ctx context.Context, id string, req betterstack.MonitorUpdateRequest) (betterstack.Monitor, error)
 	createFn func(ctx context.Context, req betterstack.MonitorCreateRequest) (betterstack.Monitor, error)
 	deleteFn func(ctx context.Context, id string) error
+	listFn   func(ctx context.Context) ([]betterstack.Monitor, error)
 
 	getCalls    int
 	updateCalls int
 	createCalls int
 	deleteCalls int
+	listCalls   int
 
 	lastUpdateReq betterstack.MonitorUpdateRequest
 	lastCreateReq betterstack.MonitorCreateRequest
@@ -96,6 +98,14 @@ func (s *fakeMonitorService) Delete(ctx context.Context, id string) error {
 	return nil
 }
 
+func (s *fakeMonitorService) List(ctx context.Context) ([]betterstack.Monitor, error) {
+	s.listCalls++
+	if s.listFn != nil {
+		return s.listFn(ctx)
+	}
+	return nil, nil
+}
+
 var _ betterstack.MonitorClient = (*fakeMonitorService)(nil)
 
 func TestReconcileAddsFinalizer(t *testing.T) {

go.mod

@@ -7,6 +7,7 @@ require (
 	k8s.io/apiextensions-apiserver v0.34.1
 	k8s.io/apimachinery v0.34.1
 	k8s.io/client-go v0.34.1
+	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397
 	sigs.k8s.io/controller-runtime v0.22.1
 )
 
@@ -60,7 +61,6 @@ require (
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b // indirect
-	k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 // indirect
 	sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect

helm/betterstack-operator/files/crds/monitoring.betterstack.io_betterstackheartbeats.yaml

@@ -17,6 +17,8 @@ spec:
     - name: v1alpha1
       served: true
       storage: true
+      subresources:
+        status: {}
       additionalPrinterColumns:
         - name: Name
           type: string

internal/testutil/assert/assert.go

@@ -130,6 +130,12 @@ func StringSlice(t testing.TB, field string, actual, expected []string) {
 	EqualSlice[string](t, field, actual, expected)
 }
 
+// Failf unconditionally fails the test with the formatted message.
+func Failf(t testing.TB, format string, args ...any) {
+	t.Helper()
+	t.Fatalf(format, args...)
+}
+
 func isNil(v any) bool {
 	if v == nil {
 		return true

pkg/betterstack/heartbeats.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 )
 
@@ -97,6 +98,13 @@ type heartbeatData struct {
 	Attributes HeartbeatAttributes `json:"attributes"`
 }
 
+type heartbeatListEnvelope struct {
+	Data  []heartbeatData `json:"data"`
+	Links struct {
+		Next string `json:"next"`
+	} `json:"links"`
+}
+
 // Create creates a heartbeat in Better Stack.
 func (s *HeartbeatService) Create(ctx context.Context, req HeartbeatCreateRequest) (Heartbeat, error) {
 	var respEnvelope heartbeatEnvelope
@@ -136,4 +144,34 @@ func (s *HeartbeatService) Delete(ctx context.Context, id string) error {
 	return err
 }
 
+// List returns the collection of heartbeats. Pagination is followed automatically.
+func (s *HeartbeatService) List(ctx context.Context) ([]Heartbeat, error) {
+	path := "/heartbeats"
+	var heartbeats []Heartbeat
+
+	for path != "" {
+		var envelope heartbeatListEnvelope
+		if err := s.client.do(ctx, http.MethodGet, path, nil, &envelope); err != nil {
+			return nil, err
+		}
+
+		for _, item := range envelope.Data {
+			heartbeats = append(heartbeats, Heartbeat{ID: item.ID, Attributes: item.Attributes})
+		}
+
+		next := strings.TrimSpace(envelope.Links.Next)
+		if next == "" {
+			break
+		}
+
+		// normalise next path relative to base URL when required
+		if strings.HasPrefix(next, s.client.baseURL) {
+			next = strings.TrimPrefix(next, s.client.baseURL)
+		}
+		path = next
+	}
+
+	return heartbeats, nil
+}
+
 var _ HeartbeatClient = (*HeartbeatService)(nil)

pkg/betterstack/monitors.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strings"
 	"time"
 )
 
@@ -15,6 +16,7 @@ type MonitorClient interface {
 	Get(ctx context.Context, id string) (Monitor, error)
 	Update(ctx context.Context, id string, req MonitorUpdateRequest) (Monitor, error)
 	Delete(ctx context.Context, id string) error
+	List(ctx context.Context) ([]Monitor, error)
 }
 
 // MonitorService provides monitor-specific Better Stack operations.
@@ -183,6 +185,13 @@ type monitorData struct {
 	Attributes MonitorAttributes `json:"attributes"`
 }
 
+type monitorListEnvelope struct {
+	Data  []monitorData `json:"data"`
+	Links struct {
+		Next string `json:"next"`
+	} `json:"links"`
+}
+
 // Create creates a monitor in Better Stack.
 func (s *MonitorService) Create(ctx context.Context, req MonitorCreateRequest) (Monitor, error) {
 	var respEnvelope monitorEnvelope
@@ -222,4 +231,32 @@ func (s *MonitorService) Delete(ctx context.Context, id string) error {
 	return err
 }
 
+// List returns all monitors, following pagination automatically.
+func (s *MonitorService) List(ctx context.Context) ([]Monitor, error) {
+	path := "/monitors"
+	var monitors []Monitor
+
+	for path != "" {
+		var envelope monitorListEnvelope
+		if err := s.client.do(ctx, http.MethodGet, path, nil, &envelope); err != nil {
+			return nil, err
+		}
+
+		for _, item := range envelope.Data {
+			monitors = append(monitors, Monitor{ID: item.ID, Attributes: item.Attributes})
+		}
+
+		next := strings.TrimSpace(envelope.Links.Next)
+		if next == "" {
+			break
+		}
+		if strings.HasPrefix(next, s.client.baseURL) {
+			next = strings.TrimPrefix(next, s.client.baseURL)
+		}
+		path = next
+	}
+
+	return monitors, nil
+}
+
 var _ MonitorClient = (*MonitorService)(nil)