reboot: add reboot_queue_entries metrics

Signed-off-by: Daichi Sakaue <daichi-sakaue@cybozu.co.jp>

Author: yokaze

docs/metrics.md

@@ -8,6 +8,7 @@ CKE exposes the following metrics with the Prometheus format at `/metrics` REST
 | leader                                | True (=1) if this server is the leader of CKE.                             | Gauge |         |
 | operation_phase                       | 1 if CKE is operating in the phase specified by the `phase` label.         | Gauge | `phase` |
 | operation_phase_timestamp_seconds     | The Unix timestamp when `operation_phase` was last updated.                | Gauge |         |
+| reboot_queue_entries                  | The number of reboot queue entries remaining.                              | Gauge |         |
 | sabakan_integration_successful        | True (=1) if sabakan-integration satisfies constraints.                    | Gauge |         |
 | sabakan_integration_timestamp_seconds | The Unix timestamp when `sabakan_integration_successful` was last updated. | Gauge |         |
 | sabakan_workers                       | The number of worker nodes for each role.                                  | Gauge | `role`  |

metrics/collector.go

@@ -54,6 +54,10 @@ func NewCollector(client *v3.Client) prometheus.Collector {
 				collectors:  []prometheus.Collector{operationPhase, operationPhaseTimestampSeconds},
 				isAvailable: isOperationPhaseAvailable,
 			},
+			"reboot": {
+				collectors:  []prometheus.Collector{rebootQueueEntries},
+				isAvailable: isRebootAvailable,
+			},
 			"sabakan_integration": {
 				collectors:  []prometheus.Collector{sabakanIntegrationSuccessful, sabakanIntegrationTimestampSeconds, sabakanWorkers, sabakanUnusedMachines},
 				isAvailable: isSabakanIntegrationAvailable,

metrics/metrics.go

@@ -33,6 +33,14 @@ var operationPhaseTimestampSeconds = prometheus.NewGauge(
 	},
 )
 
+var rebootQueueEntries = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: namespace,
+		Name:      "reboot_queue_entries",
+		Help:      "The number of reboot queue entries remaining.",
+	},
+)
+
 var sabakanIntegrationSuccessful = prometheus.NewGauge(
 	prometheus.GaugeOpts{
 		Namespace: namespace,

metrics/updater.go

@@ -39,6 +39,20 @@ func isOperationPhaseAvailable(_ context.Context, _ storage) (bool, error) {
 	return isLeader, nil
 }
 
+// UpdateReboot updates "reboot_queue_entries".
+func UpdateReboot(numEntries int) {
+	rebootQueueEntries.Set(float64(numEntries))
+}
+
+// DecrementReboot decrements "reboot_queue_entries".
+func DecrementReboot() {
+	rebootQueueEntries.Dec()
+}
+
+func isRebootAvailable(_ context.Context, _ storage) (bool, error) {
+	return isLeader, nil
+}
+
 // UpdateSabakanIntegration updates Sabakan integration metrics.
 func UpdateSabakanIntegration(isSuccessful bool, workersByRole map[string]int, unusedMachines int, ts time.Time) {
 	sabakanIntegrationTimestampSeconds.Set(float64(ts.Unix()))

metrics/updater_test.go

@@ -40,6 +40,12 @@ type updateOperationPhaseTestCase struct {
 	expected operationPhaseExpected
 }
 
+type updateRebootTestCase struct {
+	name     string
+	input    int
+	expected float64
+}
+
 type sabakanInput struct {
 	isLeader       bool
 	enabled        bool
@@ -64,6 +70,7 @@ type updateSabakanIntegrationTestCase struct {
 func TestMetricsUpdater(t *testing.T) {
 	t.Run("UpdateLeader", testUpdateLeader)
 	t.Run("UpdateOperationPhase", testUpdateOperationPhase)
+	t.Run("UpdateReboot", testUpdateReboot)
 	t.Run("UpdateSabakanIntegration", testUpdateSabakanIntegration)
 }
 
@@ -232,6 +239,62 @@ func testUpdateOperationPhase(t *testing.T) {
 	}
 }
 
+func testUpdateReboot(t *testing.T) {
+	testCases := []updateRebootTestCase{
+		{
+			name:     "zero",
+			input:    0,
+			expected: 0,
+		},
+		{
+			name:     "one",
+			input:    1,
+			expected: 1,
+		},
+		{
+			name:     "two",
+			input:    2,
+			expected: 2,
+		},
+	}
+	for _, tt := range testCases {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			defer ctx.Done()
+
+			collector, _ := newTestCollector()
+			handler := GetHandler(collector)
+
+			UpdateReboot(tt.input)
+
+			w := httptest.NewRecorder()
+			req := httptest.NewRequest("GET", "/metrics", nil)
+			handler.ServeHTTP(w, req)
+
+			metricsFamily, err := parseMetrics(w.Result())
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			metricsFound := false
+			for _, mf := range metricsFamily {
+				if *mf.Name != "cke_reboot_queue_entries" {
+					continue
+				}
+				for _, m := range mf.Metric {
+					metricsFound = true
+					if *m.Gauge.Value != tt.expected {
+						t.Errorf("value for cke_reboot_queue_entries is wrong.  expected: %f, actual: %f", tt.expected, *m.Gauge.Value)
+					}
+				}
+			}
+			if !metricsFound {
+				t.Errorf("metrics reboot_queue_entries was not found")
+			}
+		})
+	}
+}
+
 func testUpdateSabakanIntegration(t *testing.T) {
 	testCases := []updateSabakanIntegrationTestCase{
 		{

op/reboot_dequeue.go

@@ -5,6 +5,7 @@ import (
 	"strconv"
 
 	"github.com/cybozu-go/cke"
+	"github.com/cybozu-go/cke/metrics"
 )
 
 type rebootDequeueOp struct {
@@ -41,7 +42,12 @@ type rebootDequeueCommand struct {
 }
 
 func (c rebootDequeueCommand) Run(ctx context.Context, inf cke.Infrastructure, leaderKey string) error {
-	return inf.Storage().DeleteRebootsEntry(ctx, leaderKey, c.index)
+	err := inf.Storage().DeleteRebootsEntry(ctx, leaderKey, c.index)
+	if err != nil {
+		return err
+	}
+	metrics.DecrementReboot()
+	return nil
 }
 
 func (c rebootDequeueCommand) Command() cke.Command {

server/control.go

@@ -266,15 +266,15 @@ func (c Controller) runOnce(ctx context.Context, leaderKey string, tick <-chan t
 		return err
 	}
 
-	reboot, err := inf.Storage().GetRebootsFrontEntry(ctx)
-	switch err {
-	case nil:
-	case cke.ErrNotFound:
-		reboot = nil
-	default:
+	re, err := inf.Storage().GetRebootsEntries(ctx)
+	if err != nil {
 		return err
 	}
-
+	metrics.UpdateReboot(len(re))
+	var reboot *cke.RebootQueueEntry
+	if len(re) > 0 {
+		reboot = re[0]
+	}
 	ops, phase := DecideOps(cluster, status, constraints, rcs, reboot)
 
 	st := &cke.ServerStatus{

storage.go

@@ -794,14 +794,12 @@ func (s Storage) GetRebootsEntry(ctx context.Context, index int64) (*RebootQueue
 	return r, nil
 }
 
-func (s Storage) getRebootsEntries(ctx context.Context, count int64) ([]*RebootQueueEntry, error) {
+// GetRebootsEntries loads the entries from the reboot queue.
+func (s Storage) GetRebootsEntries(ctx context.Context) ([]*RebootQueueEntry, error) {
 	opts := []clientv3.OpOption{
 		clientv3.WithPrefix(),
 		clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend),
 	}
-	if count > 0 {
-		opts = append(opts, clientv3.WithLimit(count))
-	}
 	resp, err := s.Get(ctx, KeyRebootsPrefix, opts...)
 	if err != nil {
 		return nil, err
@@ -824,26 +822,6 @@ func (s Storage) getRebootsEntries(ctx context.Context, count int64) ([]*RebootQ
 	return reboots, nil
 }
 
-// GetRebootsEntries loads the entries from the reboot queue.
-func (s Storage) GetRebootsEntries(ctx context.Context) ([]*RebootQueueEntry, error) {
-	return s.getRebootsEntries(ctx, 0)
-}
-
-// GetRebootsFrontEntry loads the front entry from the reboot queue.
-// If the queue is empty, this returns ErrNotFound.
-func (s Storage) GetRebootsFrontEntry(ctx context.Context) (*RebootQueueEntry, error) {
-	reboots, err := s.getRebootsEntries(ctx, 1)
-	if err != nil {
-		return nil, err
-	}
-
-	if len(reboots) == 0 {
-		return nil, ErrNotFound
-	}
-
-	return reboots[0], nil
-}
-
 // DeleteRebootsEntry deletes the entry specified by the index from the reboot queue.
 func (s Storage) DeleteRebootsEntry(ctx context.Context, leaderKey string, index int64) error {
 	resp, err := s.Txn(ctx).

storage_test.go

@@ -653,11 +653,6 @@ func testStorageReboot(t *testing.T) {
 		t.Error("unexptected error:", err)
 	}
 
-	_, err = storage.GetRebootsFrontEntry(ctx)
-	if err != ErrNotFound {
-		t.Error("unexptected error:", err)
-	}
-
 	ents, err := storage.GetRebootsEntries(ctx)
 	if err != nil {
 		t.Fatal("GetRebootsEntries failed:", err)
@@ -698,14 +693,6 @@ func testStorageReboot(t *testing.T) {
 		t.Error("GetRebootsEntry returned unexpected result:", cmp.Diff(ent, entry2))
 	}
 
-	ent, err = storage.GetRebootsFrontEntry(ctx)
-	if err != nil {
-		t.Fatal("GetRebootsFrontEntry failed:", err)
-	}
-	if !cmp.Equal(ent, entry) {
-		t.Error("GetRebootsFrontEntry returned unexpected result:", cmp.Diff(ent, entry))
-	}
-
 	entries := []*RebootQueueEntry{entry, entry2}
 	ents, err = storage.GetRebootsEntries(ctx)
 	if err != nil {
@@ -725,7 +712,7 @@ func testStorageReboot(t *testing.T) {
 		t.Fatal("GetRebootsEntry failed:", err)
 	}
 	if !cmp.Equal(ent, entry) {
-		t.Error("GetRebootsFrontEntry returned unexpected result:", cmp.Diff(ent, entry))
+		t.Error("GetRebootsEntry returned unexpected result:", cmp.Diff(ent, entry))
 	}
 
 	err = storage.DeleteRebootsEntry(ctx, leaderKey, 0)