device-health-oracle: add controller_success activation criterion

Add a criteria-based evaluation pattern to the device-health-oracle and
implement the first criterion: devices must have called the controller
at least once per minute over the burn-in period (verified via ClickHouse
controller_grpc_getconfig_success table).

- Introduce DeviceCriterion/LinkCriterion interfaces and stage-aware
  evaluators that enforce Pending → ReadyForLinks → ReadyForUsers
  progression for devices
- Add ControllerSuccessCriterion querying ClickHouse for controller call
  coverage over the burn-in window
- Optimize update logic to skip health writes when the value is already
  at the desired state
- ClickHouse connection is optional: when CLICKHOUSE_ADDR is not set,
  the oracle falls back to no-criteria behavior (backward compatible)

Author: nikw9944

controlplane/device-health-oracle/cmd/device-health-oracle/main.go

@@ -44,6 +44,7 @@ var (
 	slackWebhookURL         = flag.String("slack-webhook-url", "", "The Slack webhook URL to send alerts")
 	provisioningSlotCount   = flag.Uint64("provisioning-slot-count", defaultProvisioningSlotCount, "Burn-in slot count for new devices/links (~20 hours at 200000)")
 	drainedSlotCount        = flag.Uint64("drained-slot-count", defaultDrainedSlotCount, "Burn-in slot count for reactivated devices/links (~30 min at 5000)")
+	slotDurationMs          = flag.Uint64("slot-duration-ms", 400, "Duration of a ledger slot in milliseconds")
 	version                 = "dev"
 	commit                  = "none"
 	date                    = "unknown"
@@ -129,6 +130,48 @@ func main() {
 	serviceabilityExecutor := serviceability.NewExecutor(log, rpcClient, &signer, networkConfig.ServiceabilityProgramID)
 	telemetryClient := telemetry.New(log, rpcClient, nil, networkConfig.TelemetryProgramID)
 
+	// Initialize ClickHouse-dependent criteria.
+	var deviceCriteria []worker.DeviceCriterion
+	if chAddr := os.Getenv("CLICKHOUSE_ADDR"); chAddr != "" {
+		chDB := os.Getenv("CLICKHOUSE_DB")
+		if chDB == "" {
+			chDB = *env
+		}
+		chUser := os.Getenv("CLICKHOUSE_USER")
+		if chUser == "" {
+			chUser = "default"
+		}
+		chPass := os.Getenv("CLICKHOUSE_PASS")
+		chTLSDisabled := os.Getenv("CLICKHOUSE_TLS_DISABLED") == "true"
+
+		chClient, err := worker.NewClickHouseClient(chAddr, chDB, chUser, chPass, chTLSDisabled)
+		if err != nil {
+			log.Warn("ClickHouse connection failed, continuing without controller_success criterion", "addr", chAddr, "error", err)
+		} else {
+			log.Info("ClickHouse enabled", "addr", chAddr, "db", chDB, "user", chUser, "tls", !chTLSDisabled)
+			controllerSuccess := worker.NewControllerSuccessCriterion(
+				chClient,
+				*provisioningSlotCount,
+				*drainedSlotCount,
+				*slotDurationMs,
+				log,
+			)
+			deviceCriteria = append(deviceCriteria, controllerSuccess)
+		}
+	} else {
+		log.Info("ClickHouse disabled (CLICKHOUSE_ADDR not set), no controller_success criterion")
+	}
+
+	deviceEvaluator := &worker.DeviceHealthEvaluator{
+		ReadyForLinksCriteria: deviceCriteria,
+		ReadyForUsersCriteria: nil,
+		Log:                   log,
+	}
+	linkEvaluator := &worker.LinkHealthEvaluator{
+		ReadyForServiceCriteria: nil,
+		Log:                     log,
+	}
+
 	worker.MetricBuildInfo.WithLabelValues(version, commit, date).Set(1)
 	go func() {
 		listener, err := net.Listen("tcp", *metricsAddr)
@@ -155,6 +198,8 @@ func main() {
 		Env:                     *env,
 		ProvisioningSlotCount:   *provisioningSlotCount,
 		DrainedSlotCount:        *drainedSlotCount,
+		DeviceEvaluator:         deviceEvaluator,
+		LinkEvaluator:           linkEvaluator,
 	})
 	if err != nil {
 		log.Error("Failed to create worker", "error", err)

controlplane/device-health-oracle/internal/worker/clickhouse.go

@@ -0,0 +1,84 @@
+package worker
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+)
+
+// ControllerCallChecker queries ClickHouse for controller call records.
+type ControllerCallChecker interface {
+	ControllerCallCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (minutesWithCalls int64, err error)
+	Close() error
+}
+
+// ClickHouseClient wraps a ClickHouse connection for reading controller call data.
+type ClickHouseClient struct {
+	conn clickhouse.Conn
+	db   string
+}
+
+// NewClickHouseClient creates a new ClickHouse client connection.
+func NewClickHouseClient(addr, db, user, pass string, disableTLS bool) (*ClickHouseClient, error) {
+	addr = strings.TrimPrefix(addr, "https://")
+	addr = strings.TrimPrefix(addr, "http://")
+
+	opts := &clickhouse.Options{
+		Protocol: clickhouse.HTTP,
+		Addr:     []string{addr},
+		Auth: clickhouse.Auth{
+			Database: db,
+			Username: user,
+			Password: pass,
+		},
+		MaxOpenConns: 5,
+		DialTimeout:  30 * time.Second,
+	}
+	if !disableTLS {
+		opts.TLS = &tls.Config{}
+	}
+
+	conn, err := clickhouse.Open(opts)
+	if err != nil {
+		return nil, fmt.Errorf("clickhouse open: %w", err)
+	}
+	if err := conn.Ping(context.Background()); err != nil {
+		return nil, fmt.Errorf("clickhouse ping: %w", err)
+	}
+
+	return &ClickHouseClient{conn: conn, db: db}, nil
+}
+
+// ControllerCallCoverage returns the number of distinct minutes in [start, end] that have
+// at least one controller_grpc_getconfig_success record for the given device.
+func (c *ClickHouseClient) ControllerCallCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (int64, error) {
+	query := fmt.Sprintf(
+		`SELECT count(DISTINCT toStartOfMinute(timestamp)) AS minutes_with_calls
+		 FROM "%s".controller_grpc_getconfig_success
+		 WHERE device_pubkey = {pubkey:String}
+		   AND timestamp >= {start:DateTime64(3)}
+		   AND timestamp <= {end:DateTime64(3)}`,
+		c.db,
+	)
+
+	var minutesWithCalls int64
+	err := c.conn.QueryRow(ctx, query,
+		clickhouse.Named("pubkey", devicePubkey),
+		clickhouse.Named("start", start),
+		clickhouse.Named("end", end),
+	).Scan(&minutesWithCalls)
+	if err != nil {
+		return 0, fmt.Errorf("clickhouse query: %w", err)
+	}
+
+	return minutesWithCalls, nil
+}
+
+// Close closes the ClickHouse connection.
+func (c *ClickHouseClient) Close() error {
+	return c.conn.Close()
+}

controlplane/device-health-oracle/internal/worker/clickhouse_test.go

@@ -0,0 +1,146 @@
+package worker
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	chmodule "github.com/testcontainers/testcontainers-go/modules/clickhouse"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func setupClickHouseContainer(t *testing.T) (clickhouse.Conn, func()) {
+	t.Helper()
+	ctx := context.Background()
+
+	container, err := chmodule.Run(ctx,
+		"clickhouse/clickhouse-server:24.3",
+	)
+	require.NoError(t, err)
+
+	host, err := container.Host(ctx)
+	require.NoError(t, err)
+
+	port, err := container.MappedPort(ctx, "9000/tcp")
+	require.NoError(t, err)
+
+	conn, err := clickhouse.Open(&clickhouse.Options{
+		Addr: []string{fmt.Sprintf("%s:%s", host, port.Port())},
+		Auth: clickhouse.Auth{
+			Database: "default",
+			Username: "default",
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, conn.Ping(ctx))
+
+	// Create the table
+	err = conn.Exec(ctx, `
+		CREATE TABLE IF NOT EXISTS "default".controller_grpc_getconfig_success (
+			timestamp DateTime64(3),
+			device_pubkey LowCardinality(String)
+		) ENGINE = MergeTree
+		PARTITION BY toYYYYMM(timestamp)
+		ORDER BY (timestamp, device_pubkey)
+	`)
+	require.NoError(t, err)
+
+	cleanup := func() {
+		_ = conn.Close()
+		_ = container.Terminate(ctx)
+	}
+
+	return conn, cleanup
+}
+
+func TestClickHouseClient_ControllerCallCoverage(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	conn, cleanup := setupClickHouseContainer(t)
+	defer cleanup()
+
+	ctx := context.Background()
+	client := &ClickHouseClient{conn: conn, db: "default"}
+
+	devicePubkey := "TestDevice123"
+	now := time.Now().Truncate(time.Second)
+
+	// Insert records: one per minute for 10 minutes
+	for i := 0; i < 10; i++ {
+		ts := now.Add(-time.Duration(10-i) * time.Minute)
+		err := conn.Exec(ctx, fmt.Sprintf(
+			`INSERT INTO "default".controller_grpc_getconfig_success (timestamp, device_pubkey) VALUES (?, ?)`,
+		), ts, devicePubkey)
+		require.NoError(t, err)
+	}
+
+	t.Run("full coverage", func(t *testing.T) {
+		start := now.Add(-11 * time.Minute)
+		end := now
+		minutes, err := client.ControllerCallCoverage(ctx, devicePubkey, start, end)
+		require.NoError(t, err)
+		assert.Equal(t, int64(10), minutes)
+	})
+
+	t.Run("partial window", func(t *testing.T) {
+		start := now.Add(-5 * time.Minute)
+		end := now
+		minutes, err := client.ControllerCallCoverage(ctx, devicePubkey, start, end)
+		require.NoError(t, err)
+		assert.GreaterOrEqual(t, minutes, int64(4))
+		assert.LessOrEqual(t, minutes, int64(5))
+	})
+
+	t.Run("no data for different device", func(t *testing.T) {
+		start := now.Add(-11 * time.Minute)
+		end := now
+		minutes, err := client.ControllerCallCoverage(ctx, "OtherDevice456", start, end)
+		require.NoError(t, err)
+		assert.Equal(t, int64(0), minutes)
+	})
+
+	t.Run("empty time range", func(t *testing.T) {
+		start := now.Add(-1 * time.Hour)
+		end := now.Add(-50 * time.Minute)
+		minutes, err := client.ControllerCallCoverage(ctx, devicePubkey, start, end)
+		require.NoError(t, err)
+		assert.Equal(t, int64(0), minutes)
+	})
+}
+
+func TestClickHouseClient_ControllerCallCoverage_WithGaps(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+
+	conn, cleanup := setupClickHouseContainer(t)
+	defer cleanup()
+
+	ctx := context.Background()
+	client := &ClickHouseClient{conn: conn, db: "default"}
+
+	devicePubkey := "GappyDevice789"
+	now := time.Now().Truncate(time.Second)
+
+	// Insert records with gaps: minutes 0,1,2 then skip 3,4, then 5,6,7
+	gapMinutes := []int{10, 9, 8, 5, 4, 3}
+	for _, m := range gapMinutes {
+		ts := now.Add(-time.Duration(m) * time.Minute)
+		err := conn.Exec(ctx, fmt.Sprintf(
+			`INSERT INTO "default".controller_grpc_getconfig_success (timestamp, device_pubkey) VALUES (?, ?)`,
+		), ts, devicePubkey)
+		require.NoError(t, err)
+	}
+
+	start := now.Add(-11 * time.Minute)
+	end := now
+	minutes, err := client.ControllerCallCoverage(ctx, devicePubkey, start, end)
+	require.NoError(t, err)
+	assert.Equal(t, int64(6), minutes, "should count 6 distinct minutes (with 2-minute gap)")
+}

controlplane/device-health-oracle/internal/worker/config.go

@@ -47,6 +47,10 @@ type Config struct {
 	// DrainedSlotCount is used for reactivated devices/links (status = Drained, HardDrained, SoftDrained).
 	ProvisioningSlotCount uint64
 	DrainedSlotCount      uint64
+
+	// Health evaluators determine target health based on criteria.
+	DeviceEvaluator *DeviceHealthEvaluator
+	LinkEvaluator   *LinkHealthEvaluator
 }
 
 func (c *Config) Validate() error {
@@ -71,5 +75,11 @@ func (c *Config) Validate() error {
 	if c.Interval <= 0 {
 		return errors.New("interval must be greater than 0")
 	}
+	if c.DeviceEvaluator == nil {
+		return errors.New("device evaluator is required")
+	}
+	if c.LinkEvaluator == nil {
+		return errors.New("link evaluator is required")
+	}
 	return nil
 }