T-Gander/ProxmoxVEDHelperScripts
1
0
Fork 0
Code Issues Pull Requests Actions 5 Packages Projects Releases Wiki Activity

Unify telemetry storage and add repo filtering

Browse Source 
Refactor telemetry backend to store all telemetry in a single collection and add repo_source-based filtering.

Key changes:
- Added detect_repo_source() in misc/api.func to auto-detect/export REPO_SOURCE (ProxmoxVE/ProxmoxVED/external) when scripts are sourced.
- Consolidated PocketBase collections into a single default collection (_telemetry_data) across service, migration, and scripts; updated defaults in migrate.go, migration.go, migrate.sh and migration shell scripts.
- Simplified PBClient to use one targetColl and removed collection resolution logic; updated create/update/find/fetch functions to use targetColl.
- Introduced repo_source field (values: "ProxmoxVE", "ProxmoxVED", "external") on telemetry records and telemetry payloads; updated validation and logging.
- Added repo filtering to dashboard endpoints, FetchDashboardData and FetchRecordsPaginated, plus a repo selector in the dashboard UI; default filter is ProxmoxVE (production), with an "all" option.
- Adjusted API handlers and callers to pass repo filters and include repo_source when upserting telemetry.
- Misc: updated comments, error messages, and logging to reflect the new model; added telemetry-service.exe binary.

Purpose: simplify data model (single collection), make telemetry attributable to repository sources, and enable dashboard filtering by repo/source.
This commit is contained in:
CanbiZ (MickLesk)
2026-02-11 15:49:41 +01:00
parent 6529949f69
commit dc47c1c1c3
9 changed files with 165 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
misc/data/service.go
View File

@@ -24,11 +24,10 @@ type Config struct {
// PocketBase
PBBaseURL string
PBAuthCollection string // "_dev_telemetry_service"
PBAuthCollection string // "_telemetry_service"
PBIdentity string // email
PBPassword string
PBTargetColl string // "_dev_telemetry_data" (dev default)
PBLiveTargetColl string // "_live_telemetry_data" (production)
PBTargetColl string // "_telemetry_data"
// Limits
MaxBodyBytes int64
@@ -104,10 +103,10 @@ type TelemetryIn struct {
ErrorCategory string `json:"error_category,omitempty"` // "network", "storage", "dependency", "permission", "timeout", "unknown"
// Repository source for collection routing
RepoSource string `json:"repo_source,omitempty"` // "community-scripts/ProxmoxVE" or "community-scripts/ProxmoxVED"
RepoSource string `json:"repo_source,omitempty"` // "ProxmoxVE", "ProxmoxVED", or "external"
}
// TelemetryOut is sent to PocketBase (matches _dev_telemetry_data collection)
// TelemetryOut is sent to PocketBase (matches _telemetry_data collection)
type TelemetryOut struct {
RandomID string `json:"random_id"`
Type string `json:"type"`
@@ -133,6 +132,9 @@ type TelemetryOut struct {
RAMSpeed string `json:"ram_speed,omitempty"`
InstallDuration int `json:"install_duration,omitempty"`
ErrorCategory string `json:"error_category,omitempty"`
// Repository source: "ProxmoxVE", "ProxmoxVED", or "external"
RepoSource string `json:"repo_source,omitempty"`
}
// TelemetryStatusUpdate contains only fields needed for status updates
@@ -150,10 +152,11 @@ type TelemetryStatusUpdate struct {
RAMSpeed string `json:"ram_speed,omitempty"`
}
// Allowed values for 'repo_source' field — controls collection routing
// Allowed values for 'repo_source' field
var allowedRepoSource = map[string]bool{
"community-scripts/ProxmoxVE": true,
"community-scripts/ProxmoxVED": true,
"ProxmoxVE": true,
"ProxmoxVED": true,
"external": true,
}
type PBClient struct {
@@ -161,8 +164,7 @@ type PBClient struct {
authCollection string
identity string
password string
devColl string // "_dev_telemetry_data"
liveColl string // "_live_telemetry_data"
targetColl string // single collection for all telemetry data
mu sync.Mutex
token string
@@ -176,25 +178,13 @@ func NewPBClient(cfg Config) *PBClient {
authCollection: cfg.PBAuthCollection,
identity: cfg.PBIdentity,
password: cfg.PBPassword,
devColl: cfg.PBTargetColl,
liveColl: cfg.PBLiveTargetColl,
targetColl: cfg.PBTargetColl,
http: &http.Client{
Timeout: cfg.RequestTimeout,
},
}
}
// resolveCollection maps a repo_source value to the correct PocketBase collection.
// - "community-scripts/ProxmoxVE" → live collection
// - "community-scripts/ProxmoxVED" → dev collection
// - empty / unknown → dev collection (safe default)
func (p *PBClient) resolveCollection(repoSource string) string {
if repoSource == "community-scripts/ProxmoxVE" && p.liveColl != "" {
return p.liveColl
}
return p.devColl
}
func (p *PBClient) ensureAuth(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
@@ -246,8 +236,8 @@ func (p *PBClient) ensureAuth(ctx context.Context) error {
return nil
}
// FindRecordByRandomID searches for an existing record by random_id in the given collection
func (p *PBClient) FindRecordByRandomID(ctx context.Context, coll, randomID string) (string, error) {
// FindRecordByRandomID searches for an existing record by random_id
func (p *PBClient) FindRecordByRandomID(ctx context.Context, randomID string) (string, error) {
if err := p.ensureAuth(ctx); err != nil {
return "", err
}
@@ -256,7 +246,7 @@ func (p *PBClient) FindRecordByRandomID(ctx context.Context, coll, randomID stri
filter := fmt.Sprintf("random_id='%s'", randomID)
req, err := http.NewRequestWithContext(ctx, http.MethodGet,
fmt.Sprintf("%s/api/collections/%s/records?filter=%s&fields=id&perPage=1",
p.baseURL, coll, filter),
p.baseURL, p.targetColl, filter),
nil,
)
if err != nil {
@@ -290,14 +280,14 @@ func (p *PBClient) FindRecordByRandomID(ctx context.Context, coll, randomID stri
}
// UpdateTelemetryStatus updates only status, error, and exit_code of an existing record
func (p *PBClient) UpdateTelemetryStatus(ctx context.Context, coll, recordID string, update TelemetryStatusUpdate) error {
func (p *PBClient) UpdateTelemetryStatus(ctx context.Context, recordID string, update TelemetryStatusUpdate) error {
if err := p.ensureAuth(ctx); err != nil {
return err
}
b, _ := json.Marshal(update)
req, err := http.NewRequestWithContext(ctx, http.MethodPatch,
fmt.Sprintf("%s/api/collections/%s/records/%s", p.baseURL, coll, recordID),
fmt.Sprintf("%s/api/collections/%s/records/%s", p.baseURL, p.targetColl, recordID),
bytes.NewReader(b),
)
if err != nil {
@@ -319,8 +309,7 @@ func (p *PBClient) UpdateTelemetryStatus(ctx context.Context, coll, recordID str
}
// FetchRecordsPaginated retrieves records with pagination and optional filters.
// Uses devColl by default (dashboard shows dev data); for live data, use separate endpoint if needed.
func (p *PBClient) FetchRecordsPaginated(ctx context.Context, page, limit int, status, app, osType, sortField string) ([]TelemetryRecord, int, error) {
func (p *PBClient) FetchRecordsPaginated(ctx context.Context, page, limit int, status, app, osType, sortField, repoSource string) ([]TelemetryRecord, int, error) {
if err := p.ensureAuth(ctx); err != nil {
return nil, 0, err
}
@@ -336,6 +325,9 @@ func (p *PBClient) FetchRecordsPaginated(ctx context.Context, page, limit int, s
if osType != "" {
filters = append(filters, fmt.Sprintf("os_type='%s'", osType))
}
if repoSource != "" {
filters = append(filters, fmt.Sprintf("repo_source='%s'", repoSource))
}
filterStr := ""
if len(filters) > 0 {
@@ -361,7 +353,7 @@ func (p *PBClient) FetchRecordsPaginated(ctx context.Context, page, limit int, s
}
reqURL := fmt.Sprintf("%s/api/collections/%s/records?sort=%s&page=%d&perPage=%d%s",
p.baseURL, p.devColl, sort, page, limit, filterStr)
p.baseURL, p.targetColl, sort, page, limit, filterStr)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
if err != nil {
@@ -391,22 +383,18 @@ func (p *PBClient) FetchRecordsPaginated(ctx context.Context, page, limit int, s
}
// UpsertTelemetry handles both creation and updates intelligently.
// Routes to the correct PocketBase collection based on repoSource:
// - "community-scripts/ProxmoxVE" → _live_telemetry_data
// - "community-scripts/ProxmoxVED" → _dev_telemetry_data
// All records go to the same collection; repo_source is stored as a field.
//
// For status="installing": always creates a new record.
// For status!="installing": updates existing record (found by random_id).
func (p *PBClient) UpsertTelemetry(ctx context.Context, payload TelemetryOut, repoSource string) error {
coll := p.resolveCollection(repoSource)
func (p *PBClient) UpsertTelemetry(ctx context.Context, payload TelemetryOut) error {
// For "installing" status, always create new record
if payload.Status == "installing" {
return p.CreateTelemetry(ctx, coll, payload)
return p.CreateTelemetry(ctx, payload)
}
// For status updates (success/failed/unknown), find and update existing record
recordID, err := p.FindRecordByRandomID(ctx, coll, payload.RandomID)
recordID, err := p.FindRecordByRandomID(ctx, payload.RandomID)
if err != nil {
// Search failed, log and return error
return fmt.Errorf("cannot find record to update: %w", err)
@@ -415,7 +403,7 @@ func (p *PBClient) UpsertTelemetry(ctx context.Context, payload TelemetryOut, re
if recordID == "" {
// Record not found - this shouldn't happen normally
// Create a full record as fallback
return p.CreateTelemetry(ctx, coll, payload)
return p.CreateTelemetry(ctx, payload)
}
// Update only status, error, exit_code, and new metrics fields
@@ -432,17 +420,17 @@ func (p *PBClient) UpsertTelemetry(ctx context.Context, payload TelemetryOut, re
CPUModel: payload.CPUModel,
RAMSpeed: payload.RAMSpeed,
}
return p.UpdateTelemetryStatus(ctx, coll, recordID, update)
return p.UpdateTelemetryStatus(ctx, recordID, update)
}
func (p *PBClient) CreateTelemetry(ctx context.Context, coll string, payload TelemetryOut) error {
func (p *PBClient) CreateTelemetry(ctx context.Context, payload TelemetryOut) error {
if err := p.ensureAuth(ctx); err != nil {
return err
}
b, _ := json.Marshal(payload)
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
fmt.Sprintf("%s/api/collections/%s/records", p.baseURL, coll),
fmt.Sprintf("%s/api/collections/%s/records", p.baseURL, p.targetColl),
bytes.NewReader(b),
)
if err != nil {
@@ -730,9 +718,9 @@ func validate(in *TelemetryIn) error {
return errors.New("invalid install_duration (max 24h)")
}
// Validate repo_source: must be an allowed repository or empty
// Validate repo_source: must be a known value or empty
if in.RepoSource != "" && !allowedRepoSource[in.RepoSource] {
return errors.New("invalid repo_source (must be 'community-scripts/ProxmoxVE' or 'community-scripts/ProxmoxVED')")
return fmt.Errorf("rejected repo_source '%s' (must be 'ProxmoxVE', 'ProxmoxVED', or 'external')", in.RepoSource)
}
return nil
@@ -755,11 +743,10 @@ func main() {
TrustedProxiesCIDR: splitCSV(env("TRUSTED_PROXIES_CIDR", "")),
PBBaseURL: mustEnv("PB_URL"),
PBAuthCollection: env("PB_AUTH_COLLECTION", "_dev_telemetry_service"),
PBAuthCollection: env("PB_AUTH_COLLECTION", "_telemetry_service"),
PBIdentity: mustEnv("PB_IDENTITY"),
PBPassword: mustEnv("PB_PASSWORD"),
PBTargetColl: env("PB_TARGET_COLLECTION", "_dev_telemetry_data"),
PBLiveTargetColl: env("PB_LIVE_TARGET_COLLECTION", "_live_telemetry_data"),
PBTargetColl: env("PB_TARGET_COLLECTION", "_telemetry_data"),
MaxBodyBytes: envInt64("MAX_BODY_BYTES", 1024),
RateLimitRPM: envInt("RATE_LIMIT_RPM", 60),
@@ -870,7 +857,7 @@ func main() {
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel()
data, err := pb.FetchDashboardData(ctx, 1) // Last 24h only for metrics
data, err := pb.FetchDashboardData(ctx, 1, "ProxmoxVE") // Last 24h, production only for metrics
if err != nil {
http.Error(w, "failed to fetch metrics", http.StatusInternalServerError)
return
@@ -907,11 +894,21 @@ func main() {
}
}
// repo_source filter (default: ProxmoxVE)
repoSource := r.URL.Query().Get("repo")
if repoSource == "" {
repoSource = "ProxmoxVE"
}
// "all" means no filter
if repoSource == "all" {
repoSource = ""
}
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
// Try cache first
cacheKey := fmt.Sprintf("dashboard:%d", days)
cacheKey := fmt.Sprintf("dashboard:%d:%s", days, repoSource)
var data *DashboardData
if cfg.CacheEnabled && cache.Get(ctx, cacheKey, &data) {
w.Header().Set("Content-Type", "application/json")
@@ -920,7 +917,7 @@ func main() {
return
}
data, err := pb.FetchDashboardData(ctx, days)
data, err := pb.FetchDashboardData(ctx, days, repoSource)
if err != nil {
log.Printf("dashboard fetch failed: %v", err)
http.Error(w, "failed to fetch data", http.StatusInternalServerError)
@@ -945,6 +942,10 @@ func main() {
app := r.URL.Query().Get("app")
osType := r.URL.Query().Get("os")
sort := r.URL.Query().Get("sort")
repoSource := r.URL.Query().Get("repo")
if repoSource == "" {
repoSource = "ProxmoxVE" // Default filter: production data
}
if p := r.URL.Query().Get("page"); p != "" {
fmt.Sscanf(p, "%d", &page)
@@ -965,7 +966,7 @@ func main() {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
records, total, err := pb.FetchRecordsPaginated(ctx, page, limit, status, app, osType, sort)
records, total, err := pb.FetchRecordsPaginated(ctx, page, limit, status, app, osType, sort, repoSource)
if err != nil {
log.Printf("records fetch failed: %v", err)
http.Error(w, "failed to fetch records", http.StatusInternalServerError)
@@ -1052,6 +1053,9 @@ func main() {
return
}
if err := validate(&in); err != nil {
if cfg.EnableReqLogging {
log.Printf("telemetry rejected: %v", err)
}
http.Error(w, "invalid payload", http.StatusBadRequest)
return
}
@@ -1080,6 +1084,7 @@ func main() {
RAMSpeed: in.RAMSpeed,
InstallDuration: in.InstallDuration,
ErrorCategory: in.ErrorCategory,
RepoSource: in.RepoSource,
}
_ = computeHash(out) // For future deduplication
@@ -1087,8 +1092,8 @@ func main() {
defer cancel()
// Upsert: Creates new record if random_id doesn't exist, updates if it does
// Routes to correct collection based on repo_source
if err := pb.UpsertTelemetry(ctx, out, in.RepoSource); err != nil {
// repo_source is stored as a field on the record for filtering
if err := pb.UpsertTelemetry(ctx, out); err != nil {
// GDPR: don't log raw payload, don't log IPs; log only generic error
log.Printf("pocketbase write failed: %v", err)
http.Error(w, "upstream error", http.StatusBadGateway)
@@ -1096,7 +1101,7 @@ func main() {
}
if cfg.EnableReqLogging {
log.Printf("telemetry accepted nsapp=%s status=%s", out.NSAPP, out.Status)
log.Printf("telemetry accepted nsapp=%s status=%s repo=%s", out.NSAPP, out.Status, in.RepoSource)
}
w.WriteHeader(http.StatusAccepted)