feat: independent bandwidth throughput measurement (ADR-015, part of #73)

control-plane/internal/dashboard/agentendpoint.go

@@ -55,5 +55,64 @@ func (s *Server) agentEndpoint(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusNotFound, map[string]string{"error": "provider has no advertised agent endpoint"})
 		return
 	}
-	writeJSON(w, http.StatusOK, map[string]string{"agent_endpoint": endpoint, "public_key": hex.EncodeToString(publicKey)})
+	response := agentEndpointResponse{AgentEndpoint: endpoint, PublicKey: hex.EncodeToString(publicKey)}
+	if ingressMbps, egressMbps, ok := s.declaredBandwidth(ctx, providerID); ok {
+		response.BandwidthIngressMbps = ingressMbps
+		response.BandwidthEgressMbps = egressMbps
+	}
+	writeJSON(w, http.StatusOK, response)
+}
+
+// agentEndpointResponse is GET /api/v1/agent-endpoint/{provider_id}'s
+// response shape. The bandwidth fields are ADR-015 §5's "the provider's
+// own declared ResourceCapability.Bandwidth," added for the Network
+// Validator's MeasureBandwidth probe to score against -- omitempty (and
+// therefore simply absent, not zero-valued) when declaredBandwidth has
+// no fresh capability data for this provider, so a caller can tell
+// "nothing to compare against yet" apart from "provider declared 0
+// Mbps" (see networkvalidator.AgentEndpoint's doc comment for how the
+// caller-side type handles that same distinction).
+type agentEndpointResponse struct {
+	AgentEndpoint        string `json:"agent_endpoint"`
+	PublicKey            string `json:"public_key"`
+	BandwidthIngressMbps int32  `json:"bandwidth_ingress_mbps,omitempty"`
+	BandwidthEgressMbps  int32  `json:"bandwidth_egress_mbps,omitempty"`
+}
+
+// declaredBandwidth reads a provider's most recently heartbeated declared
+// bandwidth capacity (ResourceCapability.Bandwidth) from the same Redis
+// heartbeat cache the overview endpoint's per-provider capability display
+// already reads (openinfra:heartbeat:<provider_id>, decoded via the same
+// structHeartbeat this package already uses) -- ADR-015 §5 calls for
+// reading "the same live directory data the scheduler already uses," and
+// agentmanager's live provider directory (what the scheduler reads) is
+// itself backed by this identical heartbeat cache (see
+// internal/agentmanager/directory.go's heartbeatKeyPrefix). Deliberately
+// does not check the cache entry's TTL/freshness the way the overview
+// endpoint's liveness display does: a declared capacity is closer to a
+// slow-changing configuration value than a liveness signal, so serving a
+// slightly stale declared figure is an acceptable simplification here,
+// not a correctness bug -- ok is false only when there is no cached
+// capability data to read at all.
+func (s *Server) declaredBandwidth(ctx context.Context, providerID string) (ingressMbps, egressMbps int32, ok bool) {
+	if s.redis == nil {
+		return 0, 0, false
+	}
+	key := "openinfra:heartbeat:" + providerID
+	values, err := s.redis.HMGet(ctx, key, "payload").Result()
+	if err != nil || len(values) != 1 || values[0] == nil {
+		return 0, 0, false
+	}
+	payloadBytes, isString := values[0].(string)
+	if !isString {
+		return 0, 0, false
+	}
+	var payload structHeartbeat
+	if err := payload.Unmarshal([]byte(payloadBytes)); err != nil || payload.ProviderID != providerID {
+		return 0, 0, false
+	}
+	if payload.Capabilities == nil || payload.Capabilities.Bandwidth == nil {
+		return 0, 0, false
+	}
+	return payload.Capabilities.Bandwidth.IngressMbps, payload.Capabilities.Bandwidth.EgressMbps, true
 }

control-plane/internal/dashboard/agentendpoint_test.go

@@ -1,10 +1,16 @@
 package dashboard
 
 import (
+	"context"
 	"encoding/hex"
 	"encoding/json"
 	"net/http"
+	"os"
 	"testing"
+
+	sharedv1 "github.com/openinfra/network/protocol/generated/go/shared/v1"
+	"github.com/redis/go-redis/v9"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestAgentEndpointReturnsTheAdvertisedEndpointAndKey(t *testing.T) {
@@ -65,6 +71,105 @@ func TestAgentEndpointReturnsNotFoundWhenNoEndpointWasEverAdvertised(t *testing.
 	}
 }
 
+// Without a configured Redis client (this test's server, like most of
+// this package's, is built with a nil one -- see newAuthTestServer),
+// declaredBandwidth must degrade to "nothing to report" rather than
+// panicking on a nil-interface method call, and the response must simply
+// omit the bandwidth fields (they're omitempty), not zero-value them.
+func TestAgentEndpointOmitsBandwidthFieldsWithoutARedisClient(t *testing.T) {
+	ctx, server, pool := newAuthTestServer(t)
+	handler := server.Handler()
+
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO providers (provider_id, public_key, protocol_version, agent_version, capabilities, status, registered_at, agent_endpoint)
+		VALUES ('provider-no-redis', $1, '1', 'test', $2, 2, now(), 'https://agent.example.invalid:50052')`,
+		make([]byte, 32), []byte{},
+	); err != nil {
+		t.Fatal(err)
+	}
+
+	recorder := doJSON(t, handler, http.MethodGet, "/api/v1/agent-endpoint/provider-no-redis", nil)
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String())
+	}
+	var decoded map[string]any
+	if err := json.Unmarshal(recorder.Body.Bytes(), &decoded); err != nil {
+		t.Fatal(err)
+	}
+	if _, present := decoded["bandwidth_ingress_mbps"]; present {
+		t.Fatalf("expected no bandwidth_ingress_mbps field without a Redis client, got %v", decoded)
+	}
+	if _, present := decoded["bandwidth_egress_mbps"]; present {
+		t.Fatalf("expected no bandwidth_egress_mbps field without a Redis client, got %v", decoded)
+	}
+}
+
+// With a real Redis-backed heartbeat cache entry present (ADR-015 §5's
+// "the same live directory data the scheduler already uses"), the
+// agent-endpoint response must surface the provider's declared
+// ingress/egress bandwidth -- MeasureBandwidth's tolerance check
+// (control-plane/internal/networkvalidator) depends on this field
+// actually being populated, not just present-but-zero.
+func TestAgentEndpointIncludesDeclaredBandwidthFromTheHeartbeatCache(t *testing.T) {
+	redisURL := os.Getenv("OPENINFRA_TEST_REDIS_URL")
+	if redisURL == "" {
+		t.Skip("OPENINFRA_TEST_REDIS_URL is not set")
+	}
+	options, err := redis.ParseURL(redisURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	client := redis.NewClient(options)
+	t.Cleanup(func() { _ = client.Close() })
+
+	ctx, server, pool := newAuthTestServer(t)
+	server.redis = client
+	handler := server.Handler()
+
+	const providerID = "provider-with-declared-bandwidth"
+	if _, err := pool.Exec(ctx, `
+		INSERT INTO providers (provider_id, public_key, protocol_version, agent_version, capabilities, status, registered_at, agent_endpoint)
+		VALUES ($1, $2, '1', 'test', $3, 2, now(), 'https://agent.example.invalid:50052')`,
+		providerID, make([]byte, 32), []byte{},
+	); err != nil {
+		t.Fatal(err)
+	}
+
+	payload := &heartbeatPayload{
+		ProviderId: providerID,
+		Capabilities: &sharedv1.ResourceCapability{
+			Bandwidth: &sharedv1.Bandwidth{IngressMbps: 750, EgressMbps: 400},
+		},
+	}
+	encoded, err := proto.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	heartbeatKey := "openinfra:heartbeat:" + providerID
+	if err := client.HSet(ctx, heartbeatKey, "payload", encoded).Err(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = client.Del(context.Background(), heartbeatKey).Err() })
+
+	recorder := doJSON(t, handler, http.MethodGet, "/api/v1/agent-endpoint/"+providerID, nil)
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String())
+	}
+	var response struct {
+		BandwidthIngressMbps int32 `json:"bandwidth_ingress_mbps"`
+		BandwidthEgressMbps  int32 `json:"bandwidth_egress_mbps"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
+		t.Fatal(err)
+	}
+	if response.BandwidthIngressMbps != 750 {
+		t.Fatalf("bandwidth_ingress_mbps = %d, want 750", response.BandwidthIngressMbps)
+	}
+	if response.BandwidthEgressMbps != 400 {
+		t.Fatalf("bandwidth_egress_mbps = %d, want 400", response.BandwidthEgressMbps)
+	}
+}
+
 func TestAgentEndpointRejectsAnOversizedProviderID(t *testing.T) {
 	_, server, _ := newAuthTestServer(t)
 	handler := server.Handler()