Skip to content

Commit 7c3c957

Browse files
nikw9944nikw9944
authored
device-health-oracle: add controller_success device activation criterion (#3503)
Resolves: #3493 ## Summary of Changes - Add `controller_success` device activation criterion that queries ClickHouse to verify devices have called the controller at least once per minute over the burn-in period before advancing health - Introduce criteria-based evaluation pattern with stage-aware health progression (`Pending → ReadyForLinks → ReadyForUsers`) to support the many criteria planned in [RFC-12](https://github.com/malbeclabs/doublezero/blob/main/rfcs/rfc12-network-provisioning.md) - Skip onchain health writes when the value is already at the desired state - ClickHouse connection is optional via `CLICKHOUSE_ADDR` env var — backward compatible when not set ## Diff Breakdown | Category | Files | Lines (+/-) | Net | |--------------|-------|-------------|------| | Core logic | 4 | +332 / -4 | +328 | | Scaffolding | 3 | +72 / -2 | +70 | | Tests | 3 | +437 / -0 | +437 | | Docs | 2 | +4 / -2 | +2 | ~330 lines of core logic implementing the criteria pattern and ClickHouse integration, well-covered by ~440 lines of tests. <details> <summary>Key files (click to expand)</summary> - [`criteria.go`](https://github.com/malbeclabs/doublezero/pull/3503/files#diff-4fd40f530c225ddc7f12656d12ff3edacdf21e045027f1c1444c6e3a3bfca64c) — DeviceCriterion/LinkCriterion interfaces, BurnInTimes context helpers, and stage-aware evaluators - [`clickhouse.go`](https://github.com/malbeclabs/doublezero/pull/3503/files#diff-4e66562452cebb316aec46805cd7345d34a82fbc3d17cec0b2509bc7895f5ec6) — ClickHouse client with ControllerCallCoverage query and database name validation - [`controller_success.go`](https://github.com/malbeclabs/doublezero/pull/3503/files#diff-79edd4570aac4cdcdfb6d26e555134aeca31259c4984f3cfd04b6bf445bcd525) — ControllerSuccessCriterion querying ClickHouse for per-minute call coverage - [`worker.go`](https://github.com/malbeclabs/doublezero/pull/3503/files#diff-cb4d126aa894fbc1f5b6dec64090e91f14bee1fa90b19535ada50dc4afd9ca02) — Evaluator integration, GetBlockTime burn-in resolution, skip-update optimization </details> ## Testing Verification - Evaluator enforces stage progression — devices at `Pending` advance to `ReadyForLinks` (not directly to `ReadyForUsers`), and failing criteria block advancement - ClickHouse client tested via mocked `driver.Conn`, verifying query construction, database name quoting, and error propagation - Burn-in slot count correctly selects provisioning (200K) vs drained (5K) based on `status.IsDrained()` - Zero-length burn-in window (new environment) short-circuits to pass without querying ClickHouse
1 parent d510e3c commit 7c3c957

12 files changed

Lines changed: 856 additions & 8 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,13 @@
33
All notable changes to this project will be documented in this file.
44

55
## Unreleased
6-
6+
77
### Breaking
88

99
### Changes
1010

11+
- Device Health Oracle
12+
- Add `controller_success` activation criterion to device-health-oracle to verify devices have consistent controller call coverage over a configurable burn-in period by querying ClickHouse
1113
- Telemetry
1214
- Add `GET /device-link/agent-versions` endpoint to data-api and `agent-versions` subcommand to data-cli, exposing per-device telemetry agent version and commit from onchain `DeviceLatencySamplesHeader`
1315
- Smartcontract

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

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -129,6 +129,43 @@ func main() {
129129
serviceabilityExecutor := serviceability.NewExecutor(log, rpcClient, &signer, networkConfig.ServiceabilityProgramID)
130130
telemetryClient := telemetry.New(log, rpcClient, nil, networkConfig.TelemetryProgramID)
131131

132+
// Initialize ClickHouse-dependent criteria.
133+
var deviceCriteria []worker.DeviceCriterion
134+
if chAddr := os.Getenv("CLICKHOUSE_ADDR"); chAddr != "" {
135+
chDB := os.Getenv("CLICKHOUSE_DB")
136+
if chDB == "" {
137+
chDB = "default"
138+
}
139+
chUser := os.Getenv("CLICKHOUSE_USER")
140+
if chUser == "" {
141+
chUser = "default"
142+
}
143+
chPass := os.Getenv("CLICKHOUSE_PASS")
144+
chTLSDisabled := os.Getenv("CLICKHOUSE_TLS_DISABLED") == "true"
145+
146+
chClient, err := worker.NewClickHouseClient(chAddr, chDB, chUser, chPass, chTLSDisabled)
147+
if err != nil {
148+
log.Warn("ClickHouse connection failed, continuing without controller_success criterion", "addr", chAddr, "error", err)
149+
} else {
150+
defer chClient.Close()
151+
log.Info("ClickHouse enabled", "addr", chAddr, "db", chDB, "user", chUser, "tls", !chTLSDisabled)
152+
controllerSuccess := worker.NewControllerSuccessCriterion(chClient, log)
153+
deviceCriteria = append(deviceCriteria, controllerSuccess)
154+
}
155+
} else {
156+
log.Error("ClickHouse disabled (CLICKHOUSE_ADDR not set), no controller_success criterion")
157+
}
158+
159+
deviceEvaluator := &worker.DeviceHealthEvaluator{
160+
ReadyForLinksCriteria: deviceCriteria,
161+
ReadyForUsersCriteria: nil,
162+
Log: log,
163+
}
164+
linkEvaluator := &worker.LinkHealthEvaluator{
165+
ReadyForServiceCriteria: nil,
166+
Log: log,
167+
}
168+
132169
worker.MetricBuildInfo.WithLabelValues(version, commit, date).Set(1)
133170
go func() {
134171
listener, err := net.Listen("tcp", *metricsAddr)
@@ -155,6 +192,8 @@ func main() {
155192
Env: *env,
156193
ProvisioningSlotCount: *provisioningSlotCount,
157194
DrainedSlotCount: *drainedSlotCount,
195+
DeviceEvaluator: deviceEvaluator,
196+
LinkEvaluator: linkEvaluator,
158197
})
159198
if err != nil {
160199
log.Error("Failed to create worker", "error", err)

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

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
package worker
2+
3+
import (
4+
"context"
5+
"crypto/tls"
6+
"fmt"
7+
"regexp"
8+
"strings"
9+
"time"
10+
11+
"github.com/ClickHouse/clickhouse-go/v2"
12+
)
13+
14+
var validDBName = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
15+
16+
// ControllerCallChecker queries ClickHouse for controller call records.
17+
type ControllerCallChecker interface {
18+
ControllerCallCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (minutesWithCalls int64, err error)
19+
Close() error
20+
}
21+
22+
// ClickHouseClient wraps a ClickHouse connection for reading controller call data.
23+
type ClickHouseClient struct {
24+
conn clickhouse.Conn
25+
db string
26+
}
27+
28+
func NewClickHouseClient(addr, db, user, pass string, disableTLS bool) (*ClickHouseClient, error) {
29+
if !validDBName.MatchString(db) {
30+
return nil, fmt.Errorf("invalid clickhouse database name: %q", db)
31+
}
32+
33+
addr = strings.TrimPrefix(addr, "https://")
34+
addr = strings.TrimPrefix(addr, "http://")
35+
36+
opts := &clickhouse.Options{
37+
Protocol: clickhouse.HTTP,
38+
Addr: []string{addr},
39+
Auth: clickhouse.Auth{
40+
Database: db,
41+
Username: user,
42+
Password: pass,
43+
},
44+
MaxOpenConns: 5,
45+
DialTimeout: 30 * time.Second,
46+
}
47+
if !disableTLS {
48+
opts.TLS = &tls.Config{}
49+
}
50+
51+
conn, err := clickhouse.Open(opts)
52+
if err != nil {
53+
return nil, fmt.Errorf("clickhouse open: %w", err)
54+
}
55+
if err := conn.Ping(context.Background()); err != nil {
56+
return nil, fmt.Errorf("clickhouse ping: %w", err)
57+
}
58+
59+
return &ClickHouseClient{conn: conn, db: db}, nil
60+
}
61+
62+
// ControllerCallCoverage returns the number of distinct minutes in [start, end] that have
63+
// at least one controller_grpc_getconfig_success record for the given device.
64+
func (c *ClickHouseClient) ControllerCallCoverage(ctx context.Context, devicePubkey string, start, end time.Time) (int64, error) {
65+
query := fmt.Sprintf(
66+
`SELECT count(DISTINCT toStartOfMinute(timestamp)) AS minutes_with_calls
67+
FROM "%s".controller_grpc_getconfig_success
68+
WHERE device_pubkey = ?
69+
AND timestamp >= ?
70+
AND timestamp <= ?`,
71+
c.db,
72+
)
73+
74+
var minutesWithCalls uint64
75+
err := c.conn.QueryRow(ctx, query, devicePubkey, start, end).Scan(&minutesWithCalls)
76+
if err != nil {
77+
return 0, fmt.Errorf("clickhouse query: %w", err)
78+
}
79+
80+
return int64(minutesWithCalls), nil
81+
}
82+
83+
func (c *ClickHouseClient) Close() error {
84+
return c.conn.Close()
85+
}

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

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
package worker
2+
3+
import (
4+
"context"
5+
"errors"
6+
"testing"
7+
"time"
8+
9+
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
10+
"github.com/stretchr/testify/assert"
11+
"github.com/stretchr/testify/require"
12+
)
13+
14+
// mockRow implements driver.Row for testing.
15+
type mockRow struct {
16+
scanFunc func(dest ...any) error
17+
}
18+
19+
func (r *mockRow) Err() error { return nil }
20+
func (r *mockRow) Scan(dest ...any) error { return r.scanFunc(dest...) }
21+
func (r *mockRow) ScanStruct(_ any) error { return nil }
22+
23+
// mockConn implements the subset of driver.Conn used by ClickHouseClient.
24+
type mockConn struct {
25+
driver.Conn
26+
queryRowFunc func(ctx context.Context, query string, args ...any) driver.Row
27+
}
28+
29+
func (c *mockConn) QueryRow(ctx context.Context, query string, args ...any) driver.Row {
30+
return c.queryRowFunc(ctx, query, args...)
31+
}
32+
33+
func TestControllerCallCoverage_ReturnsCount(t *testing.T) {
34+
conn := &mockConn{
35+
queryRowFunc: func(_ context.Context, query string, args ...any) driver.Row {
36+
assert.Contains(t, query, `"testdb".controller_grpc_getconfig_success`)
37+
assert.Len(t, args, 3)
38+
assert.Equal(t, "device123", args[0])
39+
return &mockRow{
40+
scanFunc: func(dest ...any) error {
41+
p := dest[0].(*uint64)
42+
*p = 42
43+
return nil
44+
},
45+
}
46+
},
47+
}
48+
49+
client := &ClickHouseClient{conn: conn, db: "testdb"}
50+
start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
51+
end := start.Add(1 * time.Hour)
52+
53+
minutes, err := client.ControllerCallCoverage(context.Background(), "device123", start, end)
54+
require.NoError(t, err)
55+
assert.Equal(t, int64(42), minutes)
56+
}
57+
58+
func TestControllerCallCoverage_QueryError(t *testing.T) {
59+
conn := &mockConn{
60+
queryRowFunc: func(_ context.Context, _ string, _ ...any) driver.Row {
61+
return &mockRow{
62+
scanFunc: func(_ ...any) error {
63+
return errors.New("connection reset")
64+
},
65+
}
66+
},
67+
}
68+
69+
client := &ClickHouseClient{conn: conn, db: "testdb"}
70+
start := time.Now().Add(-1 * time.Hour)
71+
end := time.Now()
72+
73+
_, err := client.ControllerCallCoverage(context.Background(), "device123", start, end)
74+
assert.ErrorContains(t, err, "connection reset")
75+
}
76+
77+
func TestControllerCallCoverage_QuotesDatabaseName(t *testing.T) {
78+
// Verify that database names with hyphens (mainnet-beta) are quoted.
79+
conn := &mockConn{
80+
queryRowFunc: func(_ context.Context, query string, _ ...any) driver.Row {
81+
assert.Contains(t, query, `"mainnet-beta".controller_grpc_getconfig_success`)
82+
return &mockRow{
83+
scanFunc: func(dest ...any) error {
84+
p := dest[0].(*uint64)
85+
*p = 0
86+
return nil
87+
},
88+
}
89+
},
90+
}
91+
92+
client := &ClickHouseClient{conn: conn, db: "mainnet-beta"}
93+
start := time.Now().Add(-1 * time.Hour)
94+
end := time.Now()
95+
96+
minutes, err := client.ControllerCallCoverage(context.Background(), "device123", start, end)
97+
require.NoError(t, err)
98+
assert.Equal(t, int64(0), minutes)
99+
}
100+
101+
func TestNewClickHouseClient_StripsScheme(t *testing.T) {
102+
tests := []struct {
103+
name string
104+
addr string
105+
}{
106+
{"plain host:port", "localhost:8123"},
107+
{"https prefix", "https://clickhouse.example.com:8443"},
108+
{"http prefix", "http://localhost:8123"},
109+
}
110+
111+
for _, tt := range tests {
112+
t.Run(tt.name, func(t *testing.T) {
113+
_, err := NewClickHouseClient(tt.addr, "default", "default", "", true)
114+
// Connection will fail (no server) — verify no panic and an error is returned.
115+
assert.Error(t, err)
116+
})
117+
}
118+
}

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,10 @@ type Config struct {
4747
// DrainedSlotCount is used for reactivated devices/links (status = Drained, HardDrained, SoftDrained).
4848
ProvisioningSlotCount uint64
4949
DrainedSlotCount uint64
50+
51+
// Health evaluators determine target health based on criteria.
52+
DeviceEvaluator *DeviceHealthEvaluator
53+
LinkEvaluator *LinkHealthEvaluator
5054
}
5155

5256
func (c *Config) Validate() error {
@@ -71,5 +75,11 @@ func (c *Config) Validate() error {
7175
if c.Interval <= 0 {
7276
return errors.New("interval must be greater than 0")
7377
}
78+
if c.DeviceEvaluator == nil {
79+
return errors.New("device evaluator is required")
80+
}
81+
if c.LinkEvaluator == nil {
82+
return errors.New("link evaluator is required")
83+
}
7484
return nil
7585
}

‎controlplane/device-health-oracle/internal/worker/controller_success.go‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
package worker
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"log/slog"
7+
8+
"github.com/gagliardetto/solana-go"
9+
"github.com/malbeclabs/doublezero/smartcontract/sdk/go/serviceability"
10+
)
11+
12+
// ControllerSuccessCriterion checks that a device has called the controller
13+
// at least once per minute over the burn-in period by querying the ClickHouse
14+
// controller_grpc_getconfig_success table.
15+
//
16+
// The burn-in start times are resolved from ledger slot numbers via GetBlockTime
17+
// and passed through the context (see BurnInTimes / ContextWithBurnInTimes).
18+
type ControllerSuccessCriterion struct {
19+
checker ControllerCallChecker
20+
log *slog.Logger
21+
}
22+
23+
func NewControllerSuccessCriterion(checker ControllerCallChecker, log *slog.Logger) *ControllerSuccessCriterion {
24+
return &ControllerSuccessCriterion{
25+
checker: checker,
26+
log: log,
27+
}
28+
}
29+
30+
func (c *ControllerSuccessCriterion) Name() string {
31+
return "controller_success"
32+
}
33+
34+
func (c *ControllerSuccessCriterion) Check(ctx context.Context, device serviceability.Device) (bool, string) {
35+
start, now, expectedMinutes, ok := DeviceBurnIn(ctx, device.Status)
36+
if !ok {
37+
return false, "burn-in times not available in context"
38+
}
39+
if expectedMinutes == 0 {
40+
return true, ""
41+
}
42+
43+
pubkey := solana.PublicKeyFromBytes(device.PubKey[:]).String()
44+
minutesWithCalls, err := c.checker.ControllerCallCoverage(ctx, pubkey, start, now)
45+
if err != nil {
46+
c.log.Error("Failed to query controller call coverage",
47+
"device", pubkey, "code", device.Code, "error", err)
48+
return false, fmt.Sprintf("clickhouse query failed: %v", err)
49+
}
50+
51+
c.log.Debug("Controller call coverage",
52+
"device", pubkey, "code", device.Code,
53+
"minutesWithCalls", minutesWithCalls,
54+
"expectedMinutes", expectedMinutes,
55+
"start", start)
56+
57+
if minutesWithCalls < expectedMinutes {
58+
return false, fmt.Sprintf("controller calls cover %d/%d minutes", minutesWithCalls, expectedMinutes)
59+
}
60+
61+
return true, ""
62+
}

0 commit comments

Comments
 (0)