mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 02:25:56 +00:00
Add PDM HTTP alert source
This commit is contained in:
rcourtman
@@ -6,10 +6,8 @@
|
||||
// into a known-offline state. Resources that come back online emit a lazy
|
||||
// resolve sentinel on the next observe trip -- there is no sweep goroutine.
|
||||
//
|
||||
// MVP: the real HTTP client is a follow-on; this lane ships only the
|
||||
// interface, the diff machinery, and a nil-source guard at the call site.
|
||||
// pdmAlertBridgeConfig is reserved for the future env-based credential
|
||||
// surface; the MVP leaves it zero-valued.
|
||||
// The real HTTP source is enabled only when the full PDM_API_URL,
|
||||
// PDM_API_TOKEN, and PDM_API_TOKEN_SECRET environment contract is present.
|
||||
package ai
|
||||
|
||||
import (
|
||||
@@ -30,7 +28,7 @@ const (
|
||||
|
||||
// pdmAlertSource is the interface the bridge uses to fetch resource
|
||||
// snapshots. Unit tests inject a deterministic fake; the real implementation
|
||||
// (future) makes an authenticated HTTP call to /api2/json/resources.
|
||||
// makes an authenticated HTTP call to /api2/extjs/resources/list.
|
||||
type pdmAlertSource interface {
|
||||
ResourceList(ctx context.Context) ([]pdmResource, error)
|
||||
}
|
||||
@@ -44,9 +42,12 @@ type pdmResource struct {
|
||||
Status string // "online", "offline", "running", "stopped", "failed", "unknown"
|
||||
}
|
||||
|
||||
// pdmAlertBridgeConfig is reserved for the forward-compatible credential
|
||||
// surface (PDM_API_URL, PDM_API_TOKEN). MVP leaves it zero-valued.
|
||||
type pdmAlertBridgeConfig struct{}
|
||||
type pdmAlertBridgeConfig struct {
|
||||
APIURL string
|
||||
APIToken string
|
||||
APITokenSecret string
|
||||
InsecureSkipVerify bool
|
||||
}
|
||||
|
||||
// pdmAlertBridge diffs PDM resource status across observe trips.
|
||||
type pdmAlertBridge struct {
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
package ai
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
|
||||
"github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil"
|
||||
)
|
||||
|
||||
const (
|
||||
pdmAPIURLEnv = "PDM_API_URL"
|
||||
pdmAPITokenEnv = "PDM_API_TOKEN"
|
||||
pdmAPITokenSecretEnv = "PDM_API_TOKEN_SECRET"
|
||||
pdmInsecureSkipVerifyEnv = "PDM_INSECURE_SKIP_VERIFY"
|
||||
pdmResourcesListPath = "api2/extjs/resources/list"
|
||||
pdmAPITokenAuthScheme = "PDMAPIToken"
|
||||
pdmHTTPClientRequestTimeout = 10 * time.Second
|
||||
)
|
||||
|
||||
type pdmHTTPClient struct {
|
||||
baseURL *url.URL
|
||||
httpClient *http.Client
|
||||
tokenID string
|
||||
tokenSecret string
|
||||
}
|
||||
|
||||
func loadPDMAlertBridgeConfigFromEnv() pdmAlertBridgeConfig {
|
||||
return pdmAlertBridgeConfig{
|
||||
APIURL: strings.TrimSpace(os.Getenv(pdmAPIURLEnv)),
|
||||
APIToken: strings.TrimSpace(os.Getenv(pdmAPITokenEnv)),
|
||||
APITokenSecret: strings.TrimSpace(os.Getenv(pdmAPITokenSecretEnv)),
|
||||
InsecureSkipVerify: parsePDMInsecureSkipVerify(os.Getenv(pdmInsecureSkipVerifyEnv)),
|
||||
}
|
||||
}
|
||||
|
||||
func parsePDMInsecureSkipVerify(raw string) bool {
|
||||
enabled, err := strconv.ParseBool(strings.TrimSpace(raw))
|
||||
return err == nil && enabled
|
||||
}
|
||||
|
||||
func (cfg pdmAlertBridgeConfig) enabled() bool {
|
||||
return cfg.APIURL != "" && cfg.APIToken != "" && cfg.APITokenSecret != ""
|
||||
}
|
||||
|
||||
func newPDMAlertSourceFromEnv() pdmAlertSource {
|
||||
client, err := newPDMHTTPClient(loadPDMAlertBridgeConfigFromEnv())
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
if client == nil {
|
||||
return nil
|
||||
}
|
||||
return client
|
||||
}
|
||||
|
||||
func newPDMHTTPClient(cfg pdmAlertBridgeConfig) (*pdmHTTPClient, error) {
|
||||
if !cfg.enabled() {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
baseURL, err := securityutil.NormalizeHTTPBaseURL(cfg.APIURL, "https")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid PDM API URL: %w", err)
|
||||
}
|
||||
|
||||
httpClient := tlsutil.CreateHTTPClientWithTimeout(!cfg.InsecureSkipVerify, "", pdmHTTPClientRequestTimeout)
|
||||
if httpClient == nil {
|
||||
return nil, fmt.Errorf("create PDM HTTP client")
|
||||
}
|
||||
|
||||
return &pdmHTTPClient{
|
||||
baseURL: baseURL,
|
||||
httpClient: httpClient,
|
||||
tokenID: cfg.APIToken,
|
||||
tokenSecret: cfg.APITokenSecret,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *pdmHTTPClient) ResourceList(ctx context.Context) ([]pdmResource, error) {
|
||||
if c == nil {
|
||||
return nil, fmt.Errorf("PDM HTTP client is nil")
|
||||
}
|
||||
|
||||
endpoint := securityutil.AppendURLPath(c.baseURL, pdmResourcesListPath)
|
||||
req, err := securityutil.NewValidatedRequestWithContext(ctx, http.MethodGet, endpoint, nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create PDM resources request: %w", err)
|
||||
}
|
||||
req.Header.Set("Accept", "application/json")
|
||||
req.Header.Set("Authorization", pdmAPITokenAuthScheme+" "+c.tokenID+":"+c.tokenSecret)
|
||||
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("fetch PDM resources: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
||||
return nil, fmt.Errorf("fetch PDM resources: unexpected status %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
var envelope pdmResourcesResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil {
|
||||
return nil, fmt.Errorf("decode PDM resources response: %w", err)
|
||||
}
|
||||
|
||||
return envelope.resources(), nil
|
||||
}
|
||||
|
||||
type pdmResourcesResponse struct {
|
||||
Data []pdmRemoteResources `json:"data"`
|
||||
}
|
||||
|
||||
type pdmRemoteResources struct {
|
||||
Remote string `json:"remote"`
|
||||
Error string `json:"error,omitempty"`
|
||||
Resources []pdmRawResource `json:"resources"`
|
||||
}
|
||||
|
||||
type pdmRawResource struct {
|
||||
Type string `json:"type"`
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Node string `json:"node"`
|
||||
Storage string `json:"storage"`
|
||||
Status string `json:"status"`
|
||||
Uptime uint64 `json:"uptime"`
|
||||
Maintenance *string `json:"maintenance"`
|
||||
}
|
||||
|
||||
func (r pdmResourcesResponse) resources() []pdmResource {
|
||||
var out []pdmResource
|
||||
for _, remote := range r.Data {
|
||||
remoteID := strings.TrimSpace(remote.Remote)
|
||||
if remoteID == "" {
|
||||
continue
|
||||
}
|
||||
for _, raw := range remote.Resources {
|
||||
resource, ok := raw.resource(remoteID)
|
||||
if ok {
|
||||
out = append(out, resource)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (r pdmRawResource) resource(remoteID string) (pdmResource, bool) {
|
||||
resourceType, ok := pdmBridgeResourceType(r.Type)
|
||||
if !ok {
|
||||
return pdmResource{}, false
|
||||
}
|
||||
|
||||
name := strings.TrimSpace(r.resourceName())
|
||||
if name == "" {
|
||||
return pdmResource{}, false
|
||||
}
|
||||
|
||||
status := r.resourceStatus()
|
||||
id := strings.TrimSpace(r.ID)
|
||||
if id == "" {
|
||||
id = remoteID + "/" + resourceType + "/" + name
|
||||
}
|
||||
|
||||
return pdmResource{
|
||||
ID: id,
|
||||
RemoteID: remoteID,
|
||||
Name: name,
|
||||
Type: resourceType,
|
||||
Status: status,
|
||||
}, true
|
||||
}
|
||||
|
||||
func (r pdmRawResource) resourceName() string {
|
||||
switch r.Type {
|
||||
case "pve-node":
|
||||
return r.Node
|
||||
case "pve-storage":
|
||||
return r.Storage
|
||||
default:
|
||||
return r.Name
|
||||
}
|
||||
}
|
||||
|
||||
func (r pdmRawResource) resourceStatus() string {
|
||||
switch r.Type {
|
||||
case "pbs-node":
|
||||
if r.Uptime > 0 {
|
||||
return "online"
|
||||
}
|
||||
return "offline"
|
||||
case "pbs-datastore":
|
||||
if r.Maintenance == nil {
|
||||
return "online"
|
||||
}
|
||||
return "unknown"
|
||||
default:
|
||||
status := strings.TrimSpace(strings.ToLower(r.Status))
|
||||
if status == "" {
|
||||
return "unknown"
|
||||
}
|
||||
return status
|
||||
}
|
||||
}
|
||||
|
||||
func pdmBridgeResourceType(resourceType string) (string, bool) {
|
||||
switch resourceType {
|
||||
case "pve-qemu":
|
||||
return "qemu", true
|
||||
case "pve-lxc":
|
||||
return "lxc", true
|
||||
case "pve-node", "pbs-node":
|
||||
return "node", true
|
||||
case "pve-storage", "pbs-datastore":
|
||||
return "storage", true
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package ai
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestPDMHTTPClientResourceListUsesCurrentAPIAuthAndGroupedResponse(t *testing.T) {
|
||||
const (
|
||||
tokenID = "root@pam!pulse"
|
||||
tokenSecret = "secret-value"
|
||||
)
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path != "/api2/extjs/resources/list" {
|
||||
t.Fatalf("path: want /api2/extjs/resources/list, got %q", r.URL.Path)
|
||||
}
|
||||
wantAuth := "PDMAPIToken " + tokenID + ":" + tokenSecret
|
||||
if got := r.Header.Get("Authorization"); got != wantAuth {
|
||||
t.Fatalf("Authorization: want %q, got %q", wantAuth, got)
|
||||
}
|
||||
if strings.Contains(r.Header.Get("Authorization"), "PVEAPIToken") {
|
||||
t.Fatal("Authorization used legacy PVE token scheme")
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, `{
|
||||
"data": [
|
||||
{
|
||||
"remote": "pve-a",
|
||||
"resources": [
|
||||
{"type": "pve-qemu", "id": "remote/pve-a/guest/101", "name": "vm-101", "status": "running"},
|
||||
{"type": "pve-lxc", "id": "remote/pve-a/guest/102", "name": "ct-102", "status": "stopped"},
|
||||
{"type": "pve-node", "id": "remote/pve-a/node/pve1", "node": "pve1", "status": "online"},
|
||||
{"type": "pve-storage", "id": "remote/pve-a/storage/local", "storage": "local", "status": "offline"},
|
||||
{"type": "pve-network", "id": "remote/pve-a/network/zone/vnet1", "name": "vnet1", "status": "available"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"remote": "pbs-a",
|
||||
"error": "cached remote warning",
|
||||
"resources": [
|
||||
{"type": "pbs-node", "id": "remote/pbs-a/node/pbs1", "name": "pbs1", "uptime": 12},
|
||||
{"type": "pbs-node", "id": "remote/pbs-a/node/pbs2", "name": "pbs2", "uptime": 0},
|
||||
{"type": "pbs-datastore", "id": "remote/pbs-a/datastore/fast", "name": "fast", "maintenance": null},
|
||||
{"type": "pbs-datastore", "id": "remote/pbs-a/datastore/archive", "name": "archive", "maintenance": "offline"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client, err := newPDMHTTPClient(pdmAlertBridgeConfig{
|
||||
APIURL: server.URL,
|
||||
APIToken: tokenID,
|
||||
APITokenSecret: tokenSecret,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPDMHTTPClient: %v", err)
|
||||
}
|
||||
|
||||
got, err := client.ResourceList(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("ResourceList: %v", err)
|
||||
}
|
||||
|
||||
want := []pdmResource{
|
||||
{ID: "remote/pve-a/guest/101", RemoteID: "pve-a", Name: "vm-101", Type: "qemu", Status: "running"},
|
||||
{ID: "remote/pve-a/guest/102", RemoteID: "pve-a", Name: "ct-102", Type: "lxc", Status: "stopped"},
|
||||
{ID: "remote/pve-a/node/pve1", RemoteID: "pve-a", Name: "pve1", Type: "node", Status: "online"},
|
||||
{ID: "remote/pve-a/storage/local", RemoteID: "pve-a", Name: "local", Type: "storage", Status: "offline"},
|
||||
{ID: "remote/pbs-a/node/pbs1", RemoteID: "pbs-a", Name: "pbs1", Type: "node", Status: "online"},
|
||||
{ID: "remote/pbs-a/node/pbs2", RemoteID: "pbs-a", Name: "pbs2", Type: "node", Status: "offline"},
|
||||
{ID: "remote/pbs-a/datastore/fast", RemoteID: "pbs-a", Name: "fast", Type: "storage", Status: "online"},
|
||||
{ID: "remote/pbs-a/datastore/archive", RemoteID: "pbs-a", Name: "archive", Type: "storage", Status: "unknown"},
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("resources:\nwant %#v\n got %#v", want, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDMHTTPClientNonSuccessErrorDoesNotExposeToken(t *testing.T) {
|
||||
const tokenSecret = "do-not-leak"
|
||||
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "failure", http.StatusInternalServerError)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client, err := newPDMHTTPClient(pdmAlertBridgeConfig{
|
||||
APIURL: server.URL,
|
||||
APIToken: "root@pam!pulse",
|
||||
APITokenSecret: tokenSecret,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPDMHTTPClient: %v", err)
|
||||
}
|
||||
|
||||
_, err = client.ResourceList(context.Background())
|
||||
if err == nil {
|
||||
t.Fatal("ResourceList should fail on non-2xx status")
|
||||
}
|
||||
errText := err.Error()
|
||||
if strings.Contains(errText, tokenSecret) || strings.Contains(errText, "Authorization") || strings.Contains(errText, "PDMAPIToken") {
|
||||
t.Fatalf("error exposes auth material: %q", errText)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDMAlertBridgeConfigEnvActivation(t *testing.T) {
|
||||
server := httptest.NewServer(http.NotFoundHandler())
|
||||
defer server.Close()
|
||||
|
||||
t.Setenv(pdmAPIURLEnv, "")
|
||||
t.Setenv(pdmAPITokenEnv, "root@pam!pulse")
|
||||
t.Setenv(pdmAPITokenSecretEnv, "secret")
|
||||
if got := newPDMAlertSourceFromEnv(); got != nil {
|
||||
t.Fatalf("incomplete env should not activate PDM source, got %T", got)
|
||||
}
|
||||
|
||||
t.Setenv(pdmAPIURLEnv, server.URL)
|
||||
t.Setenv(pdmAPITokenEnv, "root@pam!pulse")
|
||||
t.Setenv(pdmAPITokenSecretEnv, "secret")
|
||||
t.Setenv(pdmInsecureSkipVerifyEnv, "true")
|
||||
source := newPDMAlertSourceFromEnv()
|
||||
client, ok := source.(*pdmHTTPClient)
|
||||
if !ok {
|
||||
t.Fatalf("complete env should activate *pdmHTTPClient, got %T", source)
|
||||
}
|
||||
if client.httpClient.Timeout != pdmHTTPClientRequestTimeout {
|
||||
t.Fatalf("timeout: want %s, got %s", pdmHTTPClientRequestTimeout, client.httpClient.Timeout)
|
||||
}
|
||||
|
||||
patrol := NewPatrolService(nil, nil)
|
||||
if patrol.pdmAlertBridge == nil || patrol.pdmAlertBridge.source == nil {
|
||||
t.Fatal("NewPatrolService should wire PDM source when complete env is present")
|
||||
}
|
||||
}
|
||||
|
||||
func TestPDMHTTPClientTLSVerificationDefaultsOn(t *testing.T) {
|
||||
client, err := newPDMHTTPClient(pdmAlertBridgeConfig{
|
||||
APIURL: "https://pdm.example.test",
|
||||
APIToken: "root@pam!pulse",
|
||||
APITokenSecret: "secret",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPDMHTTPClient: %v", err)
|
||||
}
|
||||
transport, ok := client.httpClient.Transport.(*http.Transport)
|
||||
if !ok {
|
||||
t.Fatalf("transport: want *http.Transport, got %T", client.httpClient.Transport)
|
||||
}
|
||||
if transport.TLSClientConfig == nil {
|
||||
t.Fatal("TLSClientConfig should be set")
|
||||
}
|
||||
if transport.TLSClientConfig.InsecureSkipVerify {
|
||||
t.Fatal("TLS verification should be enabled by default")
|
||||
}
|
||||
if client.httpClient.Timeout != 10*time.Second {
|
||||
t.Fatalf("timeout: want 10s, got %s", client.httpClient.Timeout)
|
||||
}
|
||||
|
||||
insecureClient, err := newPDMHTTPClient(pdmAlertBridgeConfig{
|
||||
APIURL: "https://pdm.example.test",
|
||||
APIToken: "root@pam!pulse",
|
||||
APITokenSecret: "secret",
|
||||
InsecureSkipVerify: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("newPDMHTTPClient insecure: %v", err)
|
||||
}
|
||||
insecureTransport, ok := insecureClient.httpClient.Transport.(*http.Transport)
|
||||
if !ok {
|
||||
t.Fatalf("transport: want *http.Transport, got %T", insecureClient.httpClient.Transport)
|
||||
}
|
||||
if insecureTransport.TLSClientConfig == nil || !insecureTransport.TLSClientConfig.InsecureSkipVerify {
|
||||
t.Fatal("PDM_INSECURE_SKIP_VERIFY=true should opt out of default TLS verification")
|
||||
}
|
||||
}
|
||||
@@ -466,9 +466,8 @@ type PatrolService struct {
|
||||
updateSafetyWatcher *UpdateSafetyWatcher
|
||||
// PDM alert bridge -- polls the PDM resource list on each patrol cycle
|
||||
// and emits reliability findings for offline nodes and failed guests.
|
||||
// Nil source at MVP makes Observe a no-op; the real HTTP client is a
|
||||
// follow-on.
|
||||
pdmAlertBridge *pdmAlertBridge
|
||||
// Nil source makes Observe a no-op when PDM env configuration is absent.
|
||||
pdmAlertBridge *pdmAlertBridge
|
||||
// ReadState provides typed read-only views over resource state (VMs, nodes, hosts, etc.).
|
||||
// This is injected separately from stateProvider since stateProvider also contains
|
||||
// non-resource telemetry (alerts, backups, connection health) that isn't modeled as resources yet.
|
||||
@@ -692,6 +691,6 @@ func NewPatrolService(aiService *Service, stateProvider StateProvider) *PatrolSe
|
||||
p.stormThrottler = newFindingStormThrottler()
|
||||
p.findings.SetStormThrottler(p.stormThrottler)
|
||||
p.updateSafetyWatcher = newUpdateSafetyWatcher()
|
||||
p.pdmAlertBridge = newPDMAlertBridge(nil)
|
||||
p.pdmAlertBridge = newPDMAlertBridge(newPDMAlertSourceFromEnv())
|
||||
return p
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user