Files
pulse/internal/api/resourceapi/resources.go
T
rcourtman 6dea7101a7 Split operator-linked resources when a merge is reported
POST /api/resources/{id}/report-merge excluded only candidate IDs that
SourceTargets derives from the merged resource's type. For a source an
operator link folded in, that candidate names neither side of the link:
an agent linked into a VM got a vm- ID, and an agent linked with a node
or Docker host keyed by its machine ID got a source-specific one. No
link row matched, so the pair stayed merged while the API answered
"Merge reported" and the drawer toasted "Resource split applied". Two
linked agents share their only source and were refused outright as
"Resource is not merged".

applyManualLinks now records each link it applies on the merged
resource: the link row's own pair and the sources the folded side
brought, once per pair, carrying earlier folds so a chain keeps every
pair. The record rides in-memory clones, so the resources API's
registry, seeded from the monitor's already-linked listing, has it
although it never holds the folded row. Report-merge excludes those
pairs for the reported sources, and the store's one-decision-per-pair
rule deletes the link, as unlink does.

A link also joins only the pair its row names now. Canonical-ID
succession moves primary_id even when the endpoint re-key collides
with the successor's own row, and honouring that primary merged the
successor through a row that no unlink or report-merge of the merged
pair removes.
2026-10-08 01:06:07 +01:00

2939 lines
90 KiB
Go

package resourceapi
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/rcourtman/pulse-go-rewrite/internal/actionlifecycle"
"github.com/rcourtman/pulse-go-rewrite/internal/api/apicontext"
"github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/storagehealth"
unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
"github.com/rcourtman/pulse-go-rewrite/pkg/auth"
"github.com/rs/zerolog/log"
"golang.org/x/sync/singleflight"
)
// QueryService provides HTTP handlers for the unified resource API.
type QueryService struct {
cfg *config.Config
// storeLifecycle prevents eviction from overtaking in-flight constructors.
storeLifecycle sync.RWMutex
storeOpening singleflight.Group
storeMu sync.Mutex
// openStore is an optional constructor seam, set only before serving requests.
openStore func(string, string) (unified.ResourceStore, error)
stores map[string]unified.ResourceStore
cacheMu sync.Mutex
registryCache map[string]registryCacheEntry
supplementalMu sync.RWMutex
supplementalRecords map[unified.DataSource]SupplementalRecordsProvider
stateProvider SnapshotProvider
tenantStateProvider TenantStateProvider
actionAvailability actionlifecycle.AvailabilityChecker
discoveryReadiness ResourceDiscoveryReadinessProvider
}
// SnapshotProvider supplies the legacy monitoring snapshot used while callers
// migrate to the canonical unified-resource snapshot contract.
type SnapshotProvider = models.SnapshotProvider
// TenantStateProvider is the tenant-scoped resource read contract.
type TenantStateProvider interface {
UnifiedReadStateForTenant(orgID string) unified.ReadState
UnifiedResourceSnapshotForTenant(orgID string) ([]unified.Resource, time.Time)
}
// TenantResourceHealthAlertProvider is the optional tenant-safe alert slice
// used to decorate resource responses with canonical fleet posture.
type TenantResourceHealthAlertProvider interface {
ResourceHealthAlertsForTenant(orgID string) []unified.ResourceHealthAlert
}
// UnifiedResourceSnapshotProvider supplies a canonical unified-resource seed
// and freshness marker for the default tenant.
type UnifiedResourceSnapshotProvider interface {
UnifiedResourceSnapshot() ([]unified.Resource, time.Time)
}
// ResourceDiscoveryReadinessProvider projects service-discovery state onto a
// unified resource without coupling the resource API to discovery storage.
type ResourceDiscoveryReadinessProvider interface {
DiscoveryReadinessForResource(resource unified.Resource, now time.Time) unified.ResourceDiscoveryReadiness
}
type registrySeed struct {
snapshot models.StateSnapshot
resources []unified.Resource
lastUpdate time.Time
unifiedSource bool
}
// SupplementalRecordsProvider provides out-of-band ingest records for a specific source.
type SupplementalRecordsProvider interface {
GetCurrentRecords() []unified.IngestRecord
}
// TenantSupplementalRecordsProvider is an optional interface for providers that can scope
// supplemental records to a specific organization. This prevents cross-tenant leakage when
// Pulse runs in multi-tenant mode.
//
// Providers that do not implement this interface will be treated as "global"/legacy and
// their records will be ingested into every tenant registry.
type TenantSupplementalRecordsProvider interface {
GetCurrentRecordsForOrg(orgID string) []unified.IngestRecord
}
// SupplementalSnapshotSourceOwner is an optional interface for providers that
// own source-native resource ingestion and want matching legacy snapshot slices
// suppressed during registry construction.
type SupplementalSnapshotSourceOwner interface {
SnapshotOwnedSources() []unified.DataSource
}
// TenantSupplementalSnapshotSourceOwner is the tenant-aware variant of
// SupplementalSnapshotSourceOwner.
type TenantSupplementalSnapshotSourceOwner interface {
SnapshotOwnedSourcesForOrg(orgID string) []unified.DataSource
}
// NewQueryService creates a unified resource query and storage service.
func NewQueryService(cfg *config.Config) *QueryService {
return &QueryService{
cfg: cfg,
stores: make(map[string]unified.ResourceStore),
registryCache: make(map[string]registryCacheEntry),
supplementalRecords: make(map[unified.DataSource]SupplementalRecordsProvider),
}
}
// SetStateProvider sets the state provider for on-demand population.
func (h *QueryService) SetStateProvider(provider SnapshotProvider) {
h.stateProvider = provider
}
// SetTenantStateProvider sets the tenant-aware provider.
func (h *QueryService) SetTenantStateProvider(provider TenantStateProvider) {
h.tenantStateProvider = provider
}
// SetActionAvailabilityChecker configures the read-only projection of action
// readiness onto resource responses. Mutation lifecycle ownership remains in
// the API composition layer.
func (h *QueryService) SetActionAvailabilityChecker(checker actionlifecycle.AvailabilityChecker) {
h.actionAvailability = checker
}
// SetDiscoveryReadinessProvider wires the canonical discovery-readiness
// projection onto resource responses and Assistant context packs.
func (h *QueryService) SetDiscoveryReadinessProvider(provider ResourceDiscoveryReadinessProvider) {
h.discoveryReadiness = provider
}
// SetSupplementalRecordsProvider configures additional records for a source.
func (h *QueryService) SetSupplementalRecordsProvider(source unified.DataSource, provider SupplementalRecordsProvider) {
h.supplementalMu.Lock()
if h.supplementalRecords == nil {
h.supplementalRecords = make(map[unified.DataSource]SupplementalRecordsProvider)
}
if provider == nil {
delete(h.supplementalRecords, source)
} else {
h.supplementalRecords[source] = provider
}
h.supplementalMu.Unlock()
// Provider changes alter ingestion inputs, so clear all cached registries.
h.cacheMu.Lock()
h.registryCache = make(map[string]registryCacheEntry)
h.cacheMu.Unlock()
}
// HandleListResources handles GET /api/resources.
func (h *QueryService) HandleListResources(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
sharedResources, registry, err := h.sharedPresentationResources(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
// Flat copy so the per-request decorations below stay top-level writes
// on request-owned elements while nested data remains shared.
allResources := flatCopyResources(sharedResources)
allResources = unified.AttachResourceHealth(allResources, h.resourceHealthAlerts(orgID), time.Now().UTC())
h.applyActionAvailability(r.Context(), allResources)
resources := allResources
if unsupported := unsupportedResourceTypeFilterTokens(r.URL.Query().Get("type")); len(unsupported) > 0 {
http.Error(w, "unsupported type filter token(s): "+strings.Join(unsupported, ", "), http.StatusBadRequest)
return
}
filters := parseListFilters(r)
facetFilters := filters
// Type facets drive evidence-gated workflow navigation. They must honor the
// rest of the request scope (especially source and excludeSource) while
// ignoring the active route's type selection, otherwise a storage-only
// query could never prove that an Images or Networks workflow also exists.
facetFilters.types = nil
facetResources := applyFilters(allResources, facetFilters)
resources = applyFilters(resources, filters)
applySorting(resources, filters.sortField, filters.sortOrder)
paged, meta := paginate(resources, filters.page, filters.limit)
attachDiscoveryTargets(paged)
h.attachDiscoveryReadinesses(paged, time.Now().UTC())
attachMetricsTargets(paged, registry)
paged = unified.RefreshCanonicalMetadataSlice(paged)
pruneResourcesForListResponse(paged)
// Build aggregations from the same presentation resource set returned by
// the list, so top-level host coalescing cannot drift between counts and rows.
stats := computeResourceContractStats(allResources)
stats.PolicyPosture = resourcePolicyPostureAggregation(allResources)
applyResourceContractTypes(paged)
response := EmptyResourcesResponse()
response.Data = paged
response.Meta = meta
response.Aggregations = stats
response.Facets = buildResourceListFacets(facetResources)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response.NormalizeCollections())
}
func (h *QueryService) resourceHealthAlerts(orgID string) []unified.ResourceHealthAlert {
if provider, ok := h.tenantStateProvider.(TenantResourceHealthAlertProvider); ok {
return provider.ResourceHealthAlertsForTenant(orgID)
}
if orgID != "" && orgID != "default" {
return nil
}
if h.stateProvider == nil {
return nil
}
snapshot := h.stateProvider.ReadSnapshot()
out := make([]unified.ResourceHealthAlert, 0, len(snapshot.ActiveAlerts))
for _, alert := range snapshot.ActiveAlerts {
out = append(out, unified.ResourceHealthAlert{
ResourceID: alert.ResourceID,
Level: alert.Level,
Type: alert.Type,
})
}
return out
}
// HandleStorageSummary handles GET /api/resources/storage-summary.
func (h *QueryService) HandleStorageSummary(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
resources, _, err := h.sharedRawResources(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
filters := parseListFilters(r)
storageSubjects := make([]unified.Resource, 0, len(resources))
for _, resource := range resources {
if !isStorageSummaryResource(resource) {
continue
}
storageSubjects = append(storageSubjects, resource)
}
storageSubjects = applyFilters(storageSubjects, filters)
response := buildStorageSummaryResponse(storageSubjects)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response)
}
// HandleStorageIncidents handles GET /api/resources/storage-incidents.
func (h *QueryService) HandleStorageIncidents(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
resources, _, err := h.sharedRawResources(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
filters := parseListFilters(r)
incidentSubjects := make([]unified.Resource, 0, len(resources))
for _, resource := range resources {
if !isStorageSummaryResource(resource) || resource.IncidentCount == 0 {
continue
}
incidentSubjects = append(incidentSubjects, resource)
}
incidentSubjects = applyFilters(incidentSubjects, filters)
response := buildStorageIncidentsResponse(incidentSubjects)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response)
}
// pruneResourcesForListResponse removes heavy, platform-specific fields that would bloat
// the list response. Detail drawers can fetch full payloads via GET /api/resources/{id}.
func pruneResourcesForListResponse(resources []unified.Resource) {
for i := range resources {
pruneResourceForListResponse(&resources[i])
}
}
func pruneResourceForListResponse(resource *unified.Resource) {
if resource == nil {
return
}
// PMG domain stats can be very large; keep summary-only in list.
// Clone before clearing: the PMG struct is reachable through the shared
// per-generation resource cache, and writing through the pointer would
// corrupt it for every other request.
if resource.PMG != nil {
pruned := *resource.PMG
pruned.RelayDomains = nil
pruned.DomainStats = nil
pruned.DomainStatsAsOf = time.Time{}
resource.PMG = &pruned
}
}
// HandleGetResource handles GET /api/resources/{id}.
func (h *QueryService) HandleGetResource(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
registry, err := h.buildRegistry(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
resourceID := strings.TrimPrefix(r.URL.Path, "/api/resources/")
resourceID = strings.TrimSuffix(resourceID, "/")
resourceID = unified.CanonicalResourceID(resourceID)
if resourceID == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
resource, ok := presentationResourceByID(registry, resourceID)
if !ok {
http.Error(w, "Resource not found", http.StatusNotFound)
return
}
resourceCopy := *resource
resourceCopies := []unified.Resource{resourceCopy}
h.applyActionAvailability(r.Context(), resourceCopies)
resourceCopy = resourceCopies[0]
attachDiscoveryTarget(&resourceCopy)
h.attachDiscoveryReadiness(&resourceCopy, time.Now().UTC())
attachMetricsTarget(&resourceCopy, registry)
unified.RefreshCanonicalMetadata(&resourceCopy)
resourceCopy.Type = resourceContractType(resourceCopy)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resourceCopy)
}
func (h *QueryService) applyActionAvailability(ctx context.Context, resources []unified.Resource) {
checker := h.actionAvailability
if checker == nil {
return
}
for i := range resources {
if len(resources[i].Capabilities) == 0 {
continue
}
filtered := make([]unified.ResourceCapability, 0, len(resources[i].Capabilities))
readinesses := make([]unified.ResourceActionReadiness, 0, len(resources[i].Capabilities))
for _, capability := range resources[i].Capabilities {
req := unified.ActionRequest{
RequestID: "resource-capability-availability",
ResourceID: resources[i].ID,
CapabilityName: capability.Name,
Params: map[string]any{},
Reason: "Projecting currently executable resource capabilities.",
RequestedBy: "system:resource-api",
}
readiness := checker.CheckActionAvailable(ctx, req, resources[i])
if readiness.Name != "" && !readiness.Available {
readinesses = append(readinesses, readiness)
continue
}
filtered = append(filtered, capability)
}
if len(filtered) == 0 {
resources[i].Capabilities = nil
} else {
resources[i].Capabilities = filtered
}
if len(readinesses) > 0 {
resources[i].ActionReadiness = readinesses
}
}
}
func presentationResourcesFromRegistry(registry *unified.ResourceRegistry) []unified.Resource {
if registry == nil {
return nil
}
return registry.ListForPresentation()
}
func presentationResourceByID(registry *unified.ResourceRegistry, resourceID string) (*unified.Resource, bool) {
resource, _, ok := presentationResourceByReference(registry, resourceID)
return resource, ok
}
// PresentationResourceByID resolves a resource through the canonical
// presentation identity map used by list and detail handlers.
func PresentationResourceByID(registry *unified.ResourceRegistry, resourceID string) (*unified.Resource, bool) {
return presentationResourceByID(registry, resourceID)
}
func presentationResourceByReference(registry *unified.ResourceRegistry, ref string) (*unified.Resource, string, bool) {
ref = unified.CanonicalResourceID(ref)
if ref == "" || registry == nil {
return nil, "", false
}
rawResource, resolvedID, ok := registry.GetByReference(ref)
if !ok {
return nil, "", false
}
for _, resource := range presentationResourcesFromRegistry(registry) {
if unified.CanonicalResourceID(resource.ID) == resolvedID {
resourceCopy := resource
return &resourceCopy, resolvedID, true
}
}
return rawResource, resolvedID, true
}
// PresentationResourceByReference resolves canonical IDs and accepted aliases
// through the same presentation identity map as the resource API.
func PresentationResourceByReference(registry *unified.ResourceRegistry, ref string) (*unified.Resource, string, bool) {
return presentationResourceByReference(registry, ref)
}
type resourceFacetCountsResponse = unified.ResourceFacetCounts
type resourceFacetBundleResponse struct {
ResourceID string `json:"resourceId"`
Capabilities []unified.ResourceCapability `json:"capabilities,omitempty"`
Relationships []unified.ResourceRelationship `json:"relationships,omitempty"`
RecentChanges []unified.ResourceChange `json:"recentChanges"`
Counts resourceFacetCountsResponse `json:"counts"`
}
type resourceTimelineQueryOptions struct {
since time.Time
limit int
filters unified.ResourceChangeFilters
}
func parseResourceTimelineQuery(r *http.Request, defaultLimit int, includeRelated bool) (resourceTimelineQueryOptions, string) {
query := resourceTimelineQueryOptions{
limit: defaultLimit,
}
if raw := strings.TrimSpace(r.URL.Query().Get("since")); raw != "" {
parsed, parseErr := time.Parse(time.RFC3339, raw)
if parseErr != nil {
return query, "Invalid since value"
}
query.since = parsed.UTC()
}
if raw := strings.TrimSpace(r.URL.Query().Get("limit")); raw != "" {
parsed, parseErr := strconv.Atoi(raw)
if parseErr != nil || parsed <= 0 {
return query, "Invalid limit value"
}
query.limit = parsed
}
filters, err := unified.ParseResourceChangeFilters(r.URL.Query()["kind"], r.URL.Query()["sourceType"], r.URL.Query()["sourceAdapter"])
if err != nil {
return query, err.Error()
}
if includeRelated {
filters.IncludeRelated = true
}
query.filters = filters
return query, ""
}
// HandleResourceRoutes dispatches nested resource routes.
func (h *QueryService) HandleResourceRoutes(w http.ResponseWriter, r *http.Request) {
if strings.TrimSuffix(r.URL.Path, "/") == "/api/resources/timeline" {
h.HandleListResourceTimeline(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/facets") {
h.HandleGetResourceFacets(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/timeline") {
h.HandleGetResourceTimeline(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/children") {
h.HandleGetChildren(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/metrics") {
h.HandleGetMetrics(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/link") {
h.HandleLink(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/unlink") {
h.HandleUnlink(w, r)
return
}
if strings.HasSuffix(r.URL.Path, "/report-merge") {
h.HandleReportMerge(w, r)
return
}
h.HandleGetResource(w, r)
}
// HandleGetResourceFacets handles GET /api/resources/{id}/facets.
func (h *QueryService) HandleGetResourceFacets(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
registry, err := h.buildRegistry(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
store, err := h.getStore(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
resourceID := strings.TrimPrefix(r.URL.Path, "/api/resources/")
resourceID = strings.TrimSuffix(resourceID, "/facets")
resourceID = strings.TrimSuffix(resourceID, "/")
resourceID = unified.CanonicalResourceID(resourceID)
if resourceID == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
resource, ok := registry.Get(resourceID)
if !ok {
http.Error(w, "Resource not found", http.StatusNotFound)
return
}
timelineQuery, parseError := parseResourceTimelineQuery(r, 25, true)
if parseError != "" {
http.Error(w, parseError, http.StatusBadRequest)
return
}
since := timelineQuery.since
limit := timelineQuery.limit
filters := timelineQuery.filters
recentChanges, err := store.GetRecentChangesFiltered(resourceID, since, limit, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
changeCount, err := store.CountRecentChangesFiltered(resourceID, since, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
changeKindCounts, err := store.CountRecentChangesByKindFiltered(resourceID, since, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
sourceTypeCounts, err := store.CountRecentChangesBySourceTypeFiltered(resourceID, since, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
sourceAdapterCounts, err := store.CountRecentChangesBySourceAdapterFiltered(resourceID, since, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resourceFacetBundleResponse{
ResourceID: resourceID,
Capabilities: append([]unified.ResourceCapability(nil), resource.Capabilities...),
Relationships: unified.ResourceRelationshipsWithCanonicalParent(*resource),
RecentChanges: recentChanges,
Counts: resourceFacetCountsResponse{
RecentChanges: changeCount,
RecentChangeKinds: changeKindCounts,
RecentChangeSourceTypes: sourceTypeCounts,
RecentChangeSourceAdapters: sourceAdapterCounts,
},
})
}
// HandleGetChildren handles GET /api/resources/{id}/children.
func (h *QueryService) HandleGetChildren(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
registry, err := h.buildRegistry(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
path := strings.TrimPrefix(r.URL.Path, "/api/resources/")
path = strings.TrimSuffix(path, "/children")
path = strings.TrimSuffix(path, "/")
path = unified.CanonicalResourceID(path)
if path == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
children := registry.GetChildren(path)
attachDiscoveryTargets(children)
h.attachDiscoveryReadinesses(children, time.Now().UTC())
children = unified.RefreshCanonicalMetadataSlice(children)
applyResourceContractTypes(children)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"data": children,
"count": len(children),
})
}
// HandleGetMetrics handles GET /api/resources/{id}/metrics.
func (h *QueryService) HandleGetMetrics(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
registry, err := h.buildRegistry(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
path := strings.TrimPrefix(r.URL.Path, "/api/resources/")
path = strings.TrimSuffix(path, "/metrics")
path = strings.TrimSuffix(path, "/")
path = unified.CanonicalResourceID(path)
if path == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
resource, ok := registry.Get(path)
if !ok {
http.Error(w, "Resource not found", http.StatusNotFound)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(resource.Metrics)
}
// HandleListResourceTimeline handles GET /api/resources/timeline.
func (h *QueryService) HandleListResourceTimeline(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
store, err := h.getStore(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
resourceID := ""
timelineQuery, parseError := parseResourceTimelineQuery(r, 100, false)
if parseError != "" {
http.Error(w, parseError, http.StatusBadRequest)
return
}
since := timelineQuery.since
limit := timelineQuery.limit
filters := timelineQuery.filters
changes, err := store.GetRecentChangesFiltered(resourceID, since, limit, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
changeCount, err := store.CountRecentChangesFiltered(resourceID, since, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{
"resourceId": resourceID,
"recentChanges": changes,
"count": changeCount,
})
}
// HandleGetResourceTimeline handles GET /api/resources/{id}/timeline.
func (h *QueryService) HandleGetResourceTimeline(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
store, err := h.getStore(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
resourceID := strings.TrimPrefix(r.URL.Path, "/api/resources/")
resourceID = strings.TrimSuffix(resourceID, "/timeline")
resourceID = strings.TrimSuffix(resourceID, "/")
resourceID = unified.CanonicalResourceID(resourceID)
if resourceID == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
timelineQuery, parseError := parseResourceTimelineQuery(r, 100, true)
if parseError != "" {
http.Error(w, parseError, http.StatusBadRequest)
return
}
since := timelineQuery.since
limit := timelineQuery.limit
filters := timelineQuery.filters
changes, err := store.GetRecentChangesFiltered(resourceID, since, limit, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
changeCount, err := store.CountRecentChangesFiltered(resourceID, since, filters)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{
"resourceId": resourceID,
"recentChanges": changes,
"count": changeCount,
})
}
// HandleStats handles GET /api/resources/stats.
func (h *QueryService) HandleStats(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
allResources, _, err := h.sharedPresentationResources(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
// Both aggregations are read-only over the shared list.
stats := computeResourceContractStats(allResources)
stats.PolicyPosture = resourcePolicyPostureAggregation(allResources)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(stats)
}
// HandleLink handles POST /api/resources/{id}/link.
func (h *QueryService) HandleLink(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
store, err := h.getStore(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
path := strings.TrimPrefix(r.URL.Path, "/api/resources/")
path = strings.TrimSuffix(path, "/link")
path = strings.TrimSuffix(path, "/")
path = unified.CanonicalResourceID(path)
if path == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
var payload struct {
TargetID string `json:"targetId"`
Reason string `json:"reason"`
}
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, "Invalid JSON", http.StatusBadRequest)
return
}
if payload.TargetID == "" {
http.Error(w, "targetId required", http.StatusBadRequest)
return
}
payload.TargetID = unified.CanonicalResourceID(payload.TargetID)
link := unified.ResourceLink{
ResourceA: path,
ResourceB: payload.TargetID,
PrimaryID: path,
Reason: payload.Reason,
CreatedBy: getUserID(r),
CreatedAt: time.Now().UTC(),
}
if err := store.AddLink(link); err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
h.invalidateCache(orgID)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{
"status": "ok",
"message": "Resources linked",
})
}
// HandleUnlink handles POST /api/resources/{id}/unlink.
func (h *QueryService) HandleUnlink(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
store, err := h.getStore(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
path := strings.TrimPrefix(r.URL.Path, "/api/resources/")
path = strings.TrimSuffix(path, "/unlink")
path = strings.TrimSuffix(path, "/")
path = unified.CanonicalResourceID(path)
if path == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
var payload struct {
TargetID string `json:"targetId"`
Reason string `json:"reason"`
}
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil {
http.Error(w, "Invalid JSON", http.StatusBadRequest)
return
}
if payload.TargetID == "" {
http.Error(w, "targetId required", http.StatusBadRequest)
return
}
payload.TargetID = unified.CanonicalResourceID(payload.TargetID)
exclusion := unified.ResourceExclusion{
ResourceA: path,
ResourceB: payload.TargetID,
Reason: payload.Reason,
CreatedBy: getUserID(r),
CreatedAt: time.Now().UTC(),
}
if err := store.AddExclusion(exclusion); err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
h.invalidateCache(orgID)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{
"status": "ok",
"message": "Resources unlinked",
})
}
// HandleReportMerge handles POST /api/resources/{id}/report-merge.
func (h *QueryService) HandleReportMerge(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
orgID := apicontext.OrgID(r.Context())
store, err := h.getStore(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
path := strings.TrimPrefix(r.URL.Path, "/api/resources/")
path = strings.TrimSuffix(path, "/report-merge")
path = strings.TrimSuffix(path, "/")
path = unified.CanonicalResourceID(path)
if path == "" {
http.Error(w, "Resource ID required", http.StatusBadRequest)
return
}
var payload struct {
Sources []string `json:"sources"`
Notes string `json:"notes"`
}
if err := json.NewDecoder(r.Body).Decode(&payload); err != nil && !errors.Is(err, io.EOF) {
http.Error(w, "Invalid JSON", http.StatusBadRequest)
return
}
registry, err := h.buildRegistry(orgID)
if err != nil {
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
resource, ok := registry.Get(path)
if !ok {
http.Error(w, "Resource not found", http.StatusNotFound)
return
}
// Two resources an operator linked can share every source (two agents),
// so an applied link marks a merge too.
linkFolds := registry.ManualLinkFolds(path)
if len(resource.Sources) < 2 && len(linkFolds) == 0 {
http.Error(w, "Resource is not merged", http.StatusBadRequest)
return
}
sourceTargets := registry.SourceTargets(path)
if len(sourceTargets) == 0 && len(linkFolds) == 0 {
http.Error(w, "No source targets found", http.StatusBadRequest)
return
}
filteredSources := make(map[string]struct{})
for _, source := range payload.Sources {
filteredSources[strings.ToLower(strings.TrimSpace(source))] = struct{}{}
}
selected := func(sources ...unified.DataSource) bool {
if len(filteredSources) == 0 {
return true
}
for _, source := range sources {
if _, ok := filteredSources[strings.ToLower(string(source))]; ok {
return true
}
}
return false
}
reason := strings.TrimSpace(payload.Notes)
if reason == "" {
reason = "reported_incorrect_merge"
}
// An operator link that brought a reported source in is undone by
// excluding the link's own pair, which replaces the link in the store.
// Identity merges are kept apart by excluding the candidate ID each
// source's record takes when no match is allowed; for a linked source that
// candidate names neither side of the link.
type exclusionPair struct{ a, b string }
pairs := make([]exclusionPair, 0, len(linkFolds)+len(sourceTargets))
for _, fold := range linkFolds {
if selected(fold.Sources...) {
pairs = append(pairs, exclusionPair{fold.HolderID, fold.FoldedID})
}
}
for _, target := range sourceTargets {
if !selected(target.Source) || target.CandidateID == "" || target.CandidateID == path {
continue
}
pairs = append(pairs, exclusionPair{path, target.CandidateID})
}
exclusionsAdded := 0
seen := make(map[string]struct{})
for _, pair := range pairs {
key := pair.a + "|" + pair.b
if pair.b < pair.a {
key = pair.b + "|" + pair.a
}
if _, ok := seen[key]; ok {
continue
}
seen[key] = struct{}{}
exclusion := unified.ResourceExclusion{
ResourceA: pair.a,
ResourceB: pair.b,
Reason: reason,
CreatedBy: getUserID(r),
CreatedAt: time.Now().UTC(),
}
if err := store.AddExclusion(exclusion); err != nil {
log.Error().
Err(err).
Str("orgID", orgID).
Str("resourceID", path).
Str("resourceA", pair.a).
Str("resourceB", pair.b).
Msg("Failed to add resource merge exclusion")
http.Error(w, sanitizeError(err, "Internal server error"), http.StatusInternalServerError)
return
}
exclusionsAdded += 1
}
if exclusionsAdded == 0 {
http.Error(w, "No exclusions created", http.StatusBadRequest)
return
}
log.Info().
Str("orgID", orgID).
Str("resourceID", path).
Int("exclusionsAdded", exclusionsAdded).
Str("userID", getUserID(r)).
Strs("sources", payload.Sources).
Msg("Reported resource merge issue")
h.invalidateCache(orgID)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]any{
"status": "ok",
"message": "Merge reported",
"exclusions": exclusionsAdded,
})
}
// buildRegistry constructs a registry for the current tenant.
func (h *QueryService) buildRegistry(orgID string) (*unified.ResourceRegistry, error) {
store, err := h.getStore(orgID)
if err != nil {
return nil, err
}
key := cacheKey(orgID)
seed, err := h.registrySeed(orgID)
if err != nil {
return nil, err
}
h.cacheMu.Lock()
entry, ok := h.registryCache[key]
if ok && entry.registry != nil && !seed.lastUpdate.IsZero() && entry.lastUpdate.Equal(seed.lastUpdate) {
h.cacheMu.Unlock()
return entry.registry, nil
}
h.cacheMu.Unlock()
h.supplementalMu.RLock()
supplementalProviders := make(map[unified.DataSource]SupplementalRecordsProvider, len(h.supplementalRecords))
for source, provider := range h.supplementalRecords {
supplementalProviders[source] = provider
}
h.supplementalMu.RUnlock()
registry := unified.NewRegistry(store)
ownedSources := supplementalSnapshotOwnedSources(supplementalProviders, orgID)
supplementalSources := sortedSupplementalSources(supplementalProviders)
if seed.unifiedSource {
registry.IngestResources(seed.resources)
seedSources := unifiedSeedSources(seed.resources)
for _, source := range supplementalSources {
provider := supplementalProviders[source]
// Availability records are safe authoritative replacements and
// must replay even when an older unified seed contains standalone
// endpoints. Pre-fix seeds could otherwise hide attached targets
// while falsely looking source-complete.
if provider == nil ||
!sourceOwnedBySupplementalProvider(source, ownedSources) ||
(source != unified.SourceAvailability && unifiedSeedIncludesSource(seedSources, source)) {
continue
}
records := supplementalRecordsForOrg(provider, orgID)
if len(records) == 0 {
continue
}
registry.IngestRecords(source, records)
}
} else {
registry.IngestSnapshot(unified.SnapshotWithoutSources(seed.snapshot, ownedSources))
for _, source := range supplementalSources {
provider := supplementalProviders[source]
if provider == nil {
continue
}
records := supplementalRecordsForOrg(provider, orgID)
if len(records) == 0 {
continue
}
registry.IngestRecords(source, records)
}
}
h.cacheMu.Lock()
if h.registryCache == nil {
h.registryCache = make(map[string]registryCacheEntry)
}
h.registryCache[key] = registryCacheEntry{registry: registry, lastUpdate: seed.lastUpdate}
h.cacheMu.Unlock()
return registry, nil
}
// BuildRegistry returns the canonical tenant resource registry. It is exposed
// for API composition services that coordinate resource mutations and action
// lifecycle decisions against the same durable store.
func (h *QueryService) BuildRegistry(orgID string) (*unified.ResourceRegistry, error) {
return h.buildRegistry(orgID)
}
func sortedSupplementalSources(
providers map[unified.DataSource]SupplementalRecordsProvider,
) []unified.DataSource {
sources := make([]unified.DataSource, 0, len(providers))
for source := range providers {
sources = append(sources, source)
}
sort.Slice(sources, func(i, j int) bool {
if sources[i] == unified.SourceAvailability {
return false
}
if sources[j] == unified.SourceAvailability {
return true
}
return string(sources[i]) < string(sources[j])
})
return sources
}
func (h *QueryService) registrySeed(orgID string) (registrySeed, error) {
seed := registrySeed{}
if orgID != "" && orgID != "default" {
if h.tenantStateProvider == nil {
return seed, errors.New("tenant state provider unavailable")
}
resources, lastUpdate := h.tenantStateProvider.UnifiedResourceSnapshotForTenant(orgID)
seed.resources = resources
seed.lastUpdate = lastUpdate
seed.unifiedSource = true
return seed, nil
}
if provider, ok := any(h.stateProvider).(UnifiedResourceSnapshotProvider); ok {
resources, lastUpdate := provider.UnifiedResourceSnapshot()
if len(resources) > 0 || !lastUpdate.IsZero() {
seed.resources = resources
seed.lastUpdate = lastUpdate
seed.unifiedSource = true
return seed, nil
}
}
if h.stateProvider != nil {
seed.snapshot = h.stateProvider.ReadSnapshot()
seed.lastUpdate = seed.snapshot.LastUpdate
}
return seed, nil
}
func supplementalSnapshotOwnedSources(providers map[unified.DataSource]SupplementalRecordsProvider, orgID string) []unified.DataSource {
if len(providers) == 0 {
return nil
}
owned := make(map[string]unified.DataSource)
for _, provider := range providers {
if provider == nil {
continue
}
var sources []unified.DataSource
if tenantOwner, ok := any(provider).(TenantSupplementalSnapshotSourceOwner); ok {
sources = tenantOwner.SnapshotOwnedSourcesForOrg(orgID)
} else if owner, ok := any(provider).(SupplementalSnapshotSourceOwner); ok {
sources = owner.SnapshotOwnedSources()
}
for _, source := range sources {
key := strings.ToLower(strings.TrimSpace(string(source)))
if key == "" {
continue
}
owned[key] = unified.DataSource(key)
}
}
if len(owned) == 0 {
return nil
}
keys := make([]string, 0, len(owned))
for key := range owned {
keys = append(keys, key)
}
sort.Strings(keys)
out := make([]unified.DataSource, 0, len(keys))
for _, key := range keys {
out = append(out, owned[key])
}
return out
}
func supplementalRecordsForOrg(provider SupplementalRecordsProvider, orgID string) []unified.IngestRecord {
if provider == nil {
return nil
}
var records []unified.IngestRecord
if tenantProvider, ok := any(provider).(TenantSupplementalRecordsProvider); ok {
records = tenantProvider.GetCurrentRecordsForOrg(orgID)
} else {
records = provider.GetCurrentRecords()
}
if len(records) == 0 {
return nil
}
out := make([]unified.IngestRecord, len(records))
copy(out, records)
return out
}
func unifiedSeedSources(resources []unified.Resource) map[unified.DataSource]struct{} {
if len(resources) == 0 {
return nil
}
sources := make(map[unified.DataSource]struct{})
for _, resource := range resources {
availabilityOwned := unified.CanonicalResourceType(resource.Type) == unified.ResourceTypeNetworkEndpoint
for _, source := range resource.Sources {
if normalized := normalizeDataSourceAlias(source); normalized != "" {
if normalized == unified.SourceAvailability && !availabilityOwned {
continue
}
sources[normalized] = struct{}{}
}
}
for source := range resource.SourceStatus {
if normalized := normalizeDataSourceAlias(source); normalized != "" {
if normalized == unified.SourceAvailability && !availabilityOwned {
continue
}
sources[normalized] = struct{}{}
}
}
switch {
case resource.TrueNAS != nil:
sources[unified.SourceTrueNAS] = struct{}{}
case resource.VMware != nil:
sources[unified.SourceVMware] = struct{}{}
case availabilityOwned &&
len(unified.AvailabilityChecksForResource(resource)) > 0:
sources[unified.SourceAvailability] = struct{}{}
}
}
return sources
}
// UnifiedSeedSources reports source-owned records present in a canonical seed.
func UnifiedSeedSources(resources []unified.Resource) map[unified.DataSource]struct{} {
return unifiedSeedSources(resources)
}
func unifiedSeedIncludesSource(seedSources map[unified.DataSource]struct{}, source unified.DataSource) bool {
if len(seedSources) == 0 {
return false
}
_, ok := seedSources[normalizeDataSourceAlias(source)]
return ok
}
func sourceOwnedBySupplementalProvider(source unified.DataSource, ownedSources []unified.DataSource) bool {
normalized := normalizeDataSourceAlias(source)
if normalized == "" {
return false
}
for _, ownedSource := range ownedSources {
if normalizeDataSourceAlias(ownedSource) == normalized {
return true
}
}
return false
}
func normalizeDataSourceAlias(source unified.DataSource) unified.DataSource {
switch strings.ToLower(strings.TrimSpace(string(source))) {
case "", "all":
return ""
case "k8s":
return unified.SourceK8s
case "vmware-vsphere":
return unified.SourceVMware
default:
return unified.DataSource(strings.ToLower(strings.TrimSpace(string(source))))
}
}
// NormalizeDataSourceAlias canonicalizes legacy source identifiers accepted at
// the router and query boundaries.
func NormalizeDataSourceAlias(source unified.DataSource) unified.DataSource {
return normalizeDataSourceAlias(source)
}
func (h *QueryService) getStore(orgID string) (unified.ResourceStore, error) {
h.storeLifecycle.RLock()
defer h.storeLifecycle.RUnlock()
key := cacheKey(orgID)
lookup := func() (unified.ResourceStore, bool) {
h.storeMu.Lock()
defer h.storeMu.Unlock()
store, ok := h.stores[key]
return store, ok
}
if store, ok := lookup(); ok {
return store, nil
}
// Deduplicate only this tenant. An unrelated tenant's cold constructor must
// not hold the cache mutex or delay already-open stores.
value, err, _ := h.storeOpening.Do(key, func() (any, error) {
if store, ok := lookup(); ok {
return store, nil
}
dataDir := ""
if h.cfg != nil {
dataDir = h.cfg.DataPath
}
open := h.openStore
if open == nil {
open = func(dir, org string) (unified.ResourceStore, error) {
return unified.NewSQLiteResourceStore(dir, org)
}
}
store, err := open(dataDir, key)
if err != nil {
return nil, err
}
h.storeMu.Lock()
if h.stores == nil {
h.stores = make(map[string]unified.ResourceStore)
}
h.stores[key] = store
h.storeMu.Unlock()
return store, nil
})
if err != nil {
return nil, err
}
return value.(unified.ResourceStore), nil
}
// InvalidateCache clears a tenant's resource registry and presentation cache
// after an external mutation coordinated by another API domain.
func (h *QueryService) InvalidateCache(orgID string) {
h.invalidateCache(orgID)
}
// Store returns the tenant-scoped durable resource store shared by resource
// queries and API mutation services.
func (h *QueryService) Store(orgID string) (unified.ResourceStore, error) {
return h.getStore(orgID)
}
// CloseTenantStore closes and evicts one org's cached resource store.
//
// getStore opens a SQLite handle per org and caches it for the process
// lifetime. Without this, offboarding a tenant left its handle open: the
// -wal and -shm files stay on disk, the file descriptors are never returned,
// and the tenant directory cannot be fully removed.
func (h *QueryService) CloseTenantStore(orgID string) error {
if h == nil {
return nil
}
key := cacheKey(orgID)
h.storeLifecycle.Lock()
h.storeMu.Lock()
store, ok := h.stores[key]
if ok {
delete(h.stores, key)
}
h.storeMu.Unlock()
h.storeLifecycle.Unlock()
h.invalidateCache(orgID)
if !ok || store == nil {
return nil
}
return store.Close()
}
// CloseStores closes and evicts every cached resource store. Used on shutdown
// so SQLite handles are released rather than left to process exit.
func (h *QueryService) CloseStores() error {
if h == nil {
return nil
}
h.storeLifecycle.Lock()
h.storeMu.Lock()
stores := make(map[string]unified.ResourceStore, len(h.stores))
for key, store := range h.stores {
stores[key] = store
}
h.stores = make(map[string]unified.ResourceStore)
h.storeMu.Unlock()
h.storeLifecycle.Unlock()
var errs []error
for key, store := range stores {
if store == nil {
continue
}
if err := store.Close(); err != nil {
errs = append(errs, fmt.Errorf("close resource store for %s: %w", key, err))
}
h.invalidateCache(key)
}
return errors.Join(errs...)
}
func (h *QueryService) invalidateCache(orgID string) {
key := cacheKey(orgID)
h.cacheMu.Lock()
delete(h.registryCache, key)
h.cacheMu.Unlock()
}
func cacheKey(orgID string) string {
key := strings.TrimSpace(orgID)
if key == "" {
key = "default"
}
return key
}
// ResourcesResponse represents the list response for the unified resources API.
type ResourcesResponse struct {
Data []unified.Resource `json:"data"`
Meta ResourcesMeta `json:"meta"`
Aggregations unified.ResourceStats `json:"aggregations"`
Facets ResourceListFacets `json:"facets"`
}
// ResourceListFacets describes the complete resource scope selected by the
// non-type filters on a list request. The paged rows remain type-filtered;
// these compact counts let clients evidence-gate adjacent workflow routes
// without hydrating every row from every resource type.
type ResourceListFacets struct {
ByType map[unified.ResourceType]int `json:"byType"`
IncidentCount int `json:"incidentCount"`
}
func buildResourceListFacets(resources []unified.Resource) ResourceListFacets {
facets := ResourceListFacets{
ByType: make(map[unified.ResourceType]int, 8),
}
for _, resource := range resources {
facets.ByType[resourceContractType(resource)]++
facets.IncidentCount += resource.IncidentCount
}
return facets
}
func EmptyResourcesResponse() ResourcesResponse {
return ResourcesResponse{}.NormalizeCollections()
}
func (r ResourcesResponse) NormalizeCollections() ResourcesResponse {
if r.Data == nil {
r.Data = []unified.Resource{}
}
if r.Aggregations.PolicyPosture == nil {
r.Aggregations.PolicyPosture = unified.EmptyResourcePolicyPostureSummary()
} else {
r.Aggregations.PolicyPosture = r.Aggregations.PolicyPosture.NormalizeCollections()
}
if r.Aggregations.ByType == nil {
r.Aggregations.ByType = map[unified.ResourceType]int{}
}
if r.Aggregations.ByStatus == nil {
r.Aggregations.ByStatus = map[unified.ResourceStatus]int{}
}
if r.Aggregations.BySource == nil {
r.Aggregations.BySource = map[unified.DataSource]int{}
}
if r.Facets.ByType == nil {
r.Facets.ByType = map[unified.ResourceType]int{}
}
return r
}
// ResourcesMeta represents pagination metadata.
type ResourcesMeta struct {
Page int `json:"page"`
Limit int `json:"limit"`
Total int `json:"total"`
TotalPages int `json:"totalPages"`
}
type StorageSummaryResponse struct {
GeneratedAt time.Time `json:"generatedAt"`
TotalResources int `json:"totalResources"`
RiskyResources int `json:"riskyResources"`
CriticalResources int `json:"criticalResources"`
WarningResources int `json:"warningResources"`
ProtectionReducedCount int `json:"protectionReducedCount"`
RebuildInProgressCount int `json:"rebuildInProgressCount"`
DependentResourceCount int `json:"dependentResourceCount"`
ProtectedWorkloadCount int `json:"protectedWorkloadCount"`
AffectedDatastoreCount int `json:"affectedDatastoreCount"`
ByPlatform map[string]int `json:"byPlatform"`
ByResourceType map[string]int `json:"byResourceType"`
ByIncidentCategory map[string]int `json:"byIncidentCategory"`
TopIncidents []StorageSummaryIncident `json:"topIncidents"`
}
func EmptyStorageSummaryResponse() StorageSummaryResponse {
return StorageSummaryResponse{}.NormalizeCollections()
}
func (r StorageSummaryResponse) NormalizeCollections() StorageSummaryResponse {
if r.ByPlatform == nil {
r.ByPlatform = map[string]int{}
}
if r.ByResourceType == nil {
r.ByResourceType = map[string]int{}
}
if r.ByIncidentCategory == nil {
r.ByIncidentCategory = map[string]int{}
}
if r.TopIncidents == nil {
r.TopIncidents = []StorageSummaryIncident{}
}
return r
}
type StorageSummaryIncident struct {
ResourceID string `json:"resourceId"`
ResourceType string `json:"resourceType"`
Name string `json:"name"`
ParentName string `json:"parentName,omitempty"`
Platform string `json:"platform,omitempty"`
Topology string `json:"topology,omitempty"`
Status string `json:"status"`
IncidentCount int `json:"incidentCount"`
IncidentCategory string `json:"incidentCategory,omitempty"`
IncidentLabel string `json:"incidentLabel,omitempty"`
IncidentSeverity string `json:"incidentSeverity,omitempty"`
IncidentPriority int `json:"incidentPriority,omitempty"`
IncidentSummary string `json:"incidentSummary,omitempty"`
IncidentImpactSummary string `json:"incidentImpactSummary,omitempty"`
IncidentUrgency string `json:"incidentUrgency,omitempty"`
IncidentAction string `json:"incidentAction,omitempty"`
ProtectionReduced bool `json:"protectionReduced,omitempty"`
RebuildInProgress bool `json:"rebuildInProgress,omitempty"`
ConsumerCount int `json:"consumerCount,omitempty"`
ProtectedWorkloads int `json:"protectedWorkloads,omitempty"`
AffectedDatastores int `json:"affectedDatastores,omitempty"`
}
type StorageIncidentsResponse struct {
GeneratedAt time.Time `json:"generatedAt"`
TotalResources int `json:"totalResources"`
CriticalResources int `json:"criticalResources"`
WarningResources int `json:"warningResources"`
ByCategory map[string]int `json:"byCategory"`
ByUrgency map[string]int `json:"byUrgency"`
Sections []StorageIncidentSection `json:"sections"`
}
func EmptyStorageIncidentsResponse() StorageIncidentsResponse {
return StorageIncidentsResponse{}.NormalizeCollections()
}
func (r StorageIncidentsResponse) NormalizeCollections() StorageIncidentsResponse {
if r.ByCategory == nil {
r.ByCategory = map[string]int{}
}
if r.ByUrgency == nil {
r.ByUrgency = map[string]int{}
}
if r.Sections == nil {
r.Sections = []StorageIncidentSection{}
}
for i := range r.Sections {
if r.Sections[i].Resources == nil {
r.Sections[i].Resources = []StorageSummaryIncident{}
}
}
return r
}
type StorageIncidentSection struct {
Category string `json:"category"`
Label string `json:"label"`
ResourceCount int `json:"resourceCount"`
CriticalResources int `json:"criticalResources"`
WarningResources int `json:"warningResources"`
PrimaryUrgency string `json:"primaryUrgency,omitempty"`
Resources []StorageSummaryIncident `json:"resources"`
}
type registryCacheEntry struct {
registry *unified.ResourceRegistry
lastUpdate time.Time
// rawList and presentation are lazily built once per registry generation
// and SHARED between requests. Both the slices and the nested data of
// their elements are read-only: take a flat copy (flatCopyResources)
// before any top-level field writes, and never write through nested
// pointers or into nested maps/slices of shared elements.
rawList []unified.Resource
presentation []unified.Resource
}
// flatCopyResources returns a top-level copy of a shared resource list.
// Element values are copies, so top-level field writes and in-place
// reordering are safe; nested data stays shared with the cache, so writes
// through nested pointers remain forbidden (clone the nested struct first,
// as pruneResourceForListResponse does for PMG).
func flatCopyResources(shared []unified.Resource) []unified.Resource {
out := make([]unified.Resource, len(shared))
copy(out, shared)
return out
}
// sharedPresentationResources returns the org's cached presentation-shape
// resource list for the current registry generation, building it at most
// once per generation instead of deep-cloning the registry on every request.
func (h *QueryService) sharedPresentationResources(orgID string) ([]unified.Resource, *unified.ResourceRegistry, error) {
registry, err := h.buildRegistry(orgID)
if err != nil {
return nil, nil, err
}
key := cacheKey(orgID)
h.cacheMu.Lock()
if entry, ok := h.registryCache[key]; ok && entry.registry == registry && entry.presentation != nil {
list := entry.presentation
h.cacheMu.Unlock()
return list, registry, nil
}
h.cacheMu.Unlock()
list := presentationResourcesFromRegistry(registry)
h.cacheMu.Lock()
if entry, ok := h.registryCache[key]; ok && entry.registry == registry {
entry.presentation = list
h.registryCache[key] = entry
}
h.cacheMu.Unlock()
return list, registry, nil
}
// sharedRawResources is sharedPresentationResources for the raw (uncoalesced)
// registry list.
func (h *QueryService) sharedRawResources(orgID string) ([]unified.Resource, *unified.ResourceRegistry, error) {
registry, err := h.buildRegistry(orgID)
if err != nil {
return nil, nil, err
}
key := cacheKey(orgID)
h.cacheMu.Lock()
if entry, ok := h.registryCache[key]; ok && entry.registry == registry && entry.rawList != nil {
list := entry.rawList
h.cacheMu.Unlock()
return list, registry, nil
}
h.cacheMu.Unlock()
list := registry.List()
h.cacheMu.Lock()
if entry, ok := h.registryCache[key]; ok && entry.registry == registry {
entry.rawList = list
h.registryCache[key] = entry
}
h.cacheMu.Unlock()
return list, registry, nil
}
func buildStorageSummaryResponse(resources []unified.Resource) StorageSummaryResponse {
response := EmptyStorageSummaryResponse()
response.GeneratedAt = time.Now().UTC()
incidentCandidates := make([]unified.Resource, 0, len(resources))
for _, resource := range resources {
response.TotalResources++
platform := storageSummaryPlatform(resource)
if platform == "" {
platform = "unknown"
}
response.ByPlatform[platform]++
response.ByResourceType[string(resourceContractType(resource))]++
if storageSummaryProtectionReduced(resource) {
response.ProtectionReducedCount++
}
if storageSummaryRebuildInProgress(resource) {
response.RebuildInProgressCount++
}
response.DependentResourceCount += storageSummaryConsumerCount(resource)
response.ProtectedWorkloadCount += storageSummaryProtectedWorkloadCount(resource)
response.AffectedDatastoreCount += storageSummaryAffectedDatastoreCount(resource)
if resource.IncidentCount > 0 {
response.RiskyResources++
switch resource.IncidentSeverity {
case storagehealth.RiskCritical:
response.CriticalResources++
default:
response.WarningResources++
}
category := strings.TrimSpace(resource.IncidentCategory)
if category != "" {
response.ByIncidentCategory[category]++
}
incidentCandidates = append(incidentCandidates, resource)
}
}
sort.Slice(incidentCandidates, func(i, j int) bool {
return storageIncidentLess(incidentCandidates[i], incidentCandidates[j])
})
for i, resource := range incidentCandidates {
if i == 10 {
break
}
response.TopIncidents = append(response.TopIncidents, buildStorageSummaryIncident(resource))
}
return response.NormalizeCollections()
}
func buildStorageIncidentsResponse(resources []unified.Resource) StorageIncidentsResponse {
response := EmptyStorageIncidentsResponse()
response.GeneratedAt = time.Now().UTC()
if len(resources) == 0 {
return response.NormalizeCollections()
}
incidentResources := cloneStorageIncidentSubjects(resources)
sort.Slice(incidentResources, func(i, j int) bool {
return storageIncidentLess(incidentResources[i], incidentResources[j])
})
sectionIndex := make(map[string]int)
for _, resource := range incidentResources {
response.TotalResources++
switch resource.IncidentSeverity {
case storagehealth.RiskCritical:
response.CriticalResources++
default:
response.WarningResources++
}
category := strings.TrimSpace(resource.IncidentCategory)
if category == "" {
category = unified.IncidentCategoryHealth
}
response.ByCategory[category]++
urgency := strings.TrimSpace(resource.IncidentUrgency)
if urgency == "" {
urgency = unified.IncidentUrgencyPlan
}
response.ByUrgency[urgency]++
idx, ok := sectionIndex[category]
if !ok {
response.Sections = append(response.Sections, StorageIncidentSection{
Category: category,
Label: storageIncidentSectionLabel(category),
Resources: make([]StorageSummaryIncident, 0, 8),
})
idx = len(response.Sections) - 1
sectionIndex[category] = idx
}
section := &response.Sections[idx]
section.ResourceCount++
switch resource.IncidentSeverity {
case storagehealth.RiskCritical:
section.CriticalResources++
default:
section.WarningResources++
}
if storageIncidentUrgencyRank(urgency) > storageIncidentUrgencyRank(section.PrimaryUrgency) {
section.PrimaryUrgency = urgency
}
section.Resources = append(section.Resources, buildStorageSummaryIncident(resource))
}
sort.SliceStable(response.Sections, func(i, j int) bool {
left := response.Sections[i]
right := response.Sections[j]
if storageIncidentCategoryRank(left.Category) != storageIncidentCategoryRank(right.Category) {
return storageIncidentCategoryRank(left.Category) < storageIncidentCategoryRank(right.Category)
}
if left.CriticalResources != right.CriticalResources {
return left.CriticalResources > right.CriticalResources
}
if left.ResourceCount != right.ResourceCount {
return left.ResourceCount > right.ResourceCount
}
return left.Label < right.Label
})
return response.NormalizeCollections()
}
func buildStorageSummaryIncident(resource unified.Resource) StorageSummaryIncident {
return StorageSummaryIncident{
ResourceID: resource.ID,
ResourceType: string(resourceContractType(resource)),
Name: resource.Name,
ParentName: resource.ParentName,
Platform: storageSummaryPlatform(resource),
Topology: storageSummaryTopology(resource),
Status: string(resource.Status),
IncidentCount: resource.IncidentCount,
IncidentCategory: resource.IncidentCategory,
IncidentLabel: resource.IncidentLabel,
IncidentSeverity: string(resource.IncidentSeverity),
IncidentPriority: resource.IncidentPriority,
IncidentSummary: resource.IncidentSummary,
IncidentImpactSummary: resource.IncidentImpactSummary,
IncidentUrgency: resource.IncidentUrgency,
IncidentAction: resource.IncidentAction,
ProtectionReduced: storageSummaryProtectionReduced(resource),
RebuildInProgress: storageSummaryRebuildInProgress(resource),
ConsumerCount: storageSummaryConsumerCount(resource),
ProtectedWorkloads: storageSummaryProtectedWorkloadCount(resource),
AffectedDatastores: storageSummaryAffectedDatastoreCount(resource),
}
}
func cloneStorageIncidentSubjects(resources []unified.Resource) []unified.Resource {
cloned := make([]unified.Resource, len(resources))
copy(cloned, resources)
return cloned
}
func storageIncidentLess(left, right unified.Resource) bool {
if left.IncidentPriority != right.IncidentPriority {
return left.IncidentPriority > right.IncidentPriority
}
if left.IncidentSeverity != right.IncidentSeverity {
return left.IncidentSeverity > right.IncidentSeverity
}
if left.Name != right.Name {
return left.Name < right.Name
}
return left.ID < right.ID
}
func storageIncidentSectionLabel(category string) string {
switch strings.TrimSpace(category) {
case unified.IncidentCategoryRecoverability:
return "Backup & Recoverability"
case unified.IncidentCategoryProtection:
return "Protection & Redundancy"
case unified.IncidentCategoryRebuild:
return "Rebuild & Recovery"
case unified.IncidentCategoryCapacity:
return "Capacity Pressure"
case unified.IncidentCategoryDiskHealth:
return "Disk Health"
case unified.IncidentCategoryAvailability:
return "Availability"
default:
return "Storage Health"
}
}
func storageIncidentCategoryRank(category string) int {
switch strings.TrimSpace(category) {
case unified.IncidentCategoryRecoverability:
return 1
case unified.IncidentCategoryProtection:
return 2
case unified.IncidentCategoryRebuild:
return 3
case unified.IncidentCategoryCapacity:
return 4
case unified.IncidentCategoryDiskHealth:
return 5
case unified.IncidentCategoryAvailability:
return 6
default:
return 7
}
}
func storageIncidentUrgencyRank(urgency string) int {
switch strings.TrimSpace(urgency) {
case unified.IncidentUrgencyNow:
return 4
case unified.IncidentUrgencyToday:
return 3
case unified.IncidentUrgencyPlan:
return 2
case unified.IncidentUrgencyMonitor:
return 1
default:
return 0
}
}
func isStorageSummaryResource(resource unified.Resource) bool {
switch unified.CanonicalResourceType(resource.Type) {
case unified.ResourceTypeStorage, unified.ResourceTypePhysicalDisk, unified.ResourceTypePBS:
return true
case unified.ResourceTypeAgent:
if resource.TrueNAS != nil {
return true
}
if resource.Agent != nil && (resource.Agent.Unraid != nil || resource.Agent.StorageRisk != nil) {
return true
}
}
return false
}
func storageSummaryPlatform(resource unified.Resource) string {
if resource.Storage != nil && strings.TrimSpace(resource.Storage.Platform) != "" {
return strings.TrimSpace(resource.Storage.Platform)
}
if resource.TrueNAS != nil {
return "truenas"
}
if resource.PBS != nil {
return "pbs"
}
if resource.Agent != nil && resource.Agent.Unraid != nil {
return "unraid"
}
for _, source := range resource.Sources {
if trimmed := strings.TrimSpace(string(source)); trimmed != "" {
return trimmed
}
}
return ""
}
func storageSummaryTopology(resource unified.Resource) string {
if resource.Storage != nil && strings.TrimSpace(resource.Storage.Topology) != "" {
return strings.TrimSpace(resource.Storage.Topology)
}
switch unified.CanonicalResourceType(resource.Type) {
case unified.ResourceTypePhysicalDisk:
return "disk"
case unified.ResourceTypePBS:
return "backup-server"
case unified.ResourceTypeAgent:
if resource.Agent != nil && resource.Agent.Unraid != nil {
return "array-host"
}
if resource.TrueNAS != nil {
return "nas"
}
}
return string(resourceContractType(resource))
}
func storageSummaryProtectionReduced(resource unified.Resource) bool {
if resource.Storage != nil && resource.Storage.ProtectionReduced {
return true
}
if resource.Agent != nil && resource.Agent.ProtectionReduced {
return true
}
if resource.TrueNAS != nil && resource.TrueNAS.ProtectionReduced {
return true
}
return false
}
func storageSummaryRebuildInProgress(resource unified.Resource) bool {
if resource.Storage != nil && resource.Storage.RebuildInProgress {
return true
}
if resource.Agent != nil && resource.Agent.RebuildInProgress {
return true
}
if resource.TrueNAS != nil && resource.TrueNAS.RebuildInProgress {
return true
}
return false
}
func storageSummaryConsumerCount(resource unified.Resource) int {
if resource.Storage != nil {
return resource.Storage.ConsumerCount
}
return 0
}
func storageSummaryProtectedWorkloadCount(resource unified.Resource) int {
if resource.PBS != nil {
return resource.PBS.ProtectedWorkloadCount
}
return 0
}
func storageSummaryAffectedDatastoreCount(resource unified.Resource) int {
if resource.PBS != nil {
return resource.PBS.AffectedDatastoreCount
}
return 0
}
// Filtering helpers.
type listFilters struct {
types map[unified.ResourceType]struct{}
sources map[unified.DataSource]struct{}
excludedSources map[unified.DataSource]struct{}
statuses map[unified.ResourceStatus]struct{}
parent string
cluster string
namespace string
query string
tags map[string]struct{}
page int
limit int
sortField string
sortOrder string
}
func parseListFilters(r *http.Request) listFilters {
filters := listFilters{
types: parseResourceTypes(r.URL.Query().Get("type")),
sources: parseSources(r.URL.Query().Get("source")),
excludedSources: parseSources(r.URL.Query().Get("excludeSource")),
statuses: parseStatuses(r.URL.Query().Get("status")),
parent: strings.TrimSpace(r.URL.Query().Get("parent")),
cluster: strings.TrimSpace(r.URL.Query().Get("cluster")),
namespace: strings.TrimSpace(r.URL.Query().Get("namespace")),
query: strings.TrimSpace(strings.ToLower(r.URL.Query().Get("q"))),
tags: parseTags(r.URL.Query().Get("tags")),
page: parseIntDefault(r.URL.Query().Get("page"), 1),
limit: parseIntDefault(r.URL.Query().Get("limit"), 50),
sortField: strings.TrimSpace(r.URL.Query().Get("sort")),
sortOrder: strings.TrimSpace(strings.ToLower(r.URL.Query().Get("order"))),
}
if filters.page < 1 {
filters.page = 1
}
if filters.limit < 1 {
filters.limit = 50
}
if filters.limit > 100 {
filters.limit = 100
}
if filters.sortField == "" {
filters.sortField = "name"
}
if filters.sortOrder != "desc" {
filters.sortOrder = "asc"
}
return filters
}
func applyFilters(resources []unified.Resource, filters listFilters) []unified.Resource {
out := make([]unified.Resource, 0, len(resources))
for _, r := range resources {
if len(filters.types) > 0 {
if _, ok := filters.types[resourceContractType(r)]; !ok {
continue
}
}
if len(filters.sources) > 0 {
matched := false
for _, source := range r.Sources {
if _, ok := filters.sources[source]; ok {
matched = true
break
}
}
if !matched {
continue
}
}
if len(filters.excludedSources) > 0 {
excluded := false
for _, source := range r.Sources {
if _, ok := filters.excludedSources[source]; ok {
excluded = true
break
}
}
if excluded {
continue
}
}
if len(filters.statuses) > 0 {
if _, ok := filters.statuses[r.Status]; !ok {
continue
}
}
if filters.parent != "" {
if r.ParentID == nil || *r.ParentID != filters.parent {
continue
}
}
if filters.cluster != "" {
cluster := strings.ToLower(filters.cluster)
identityCluster := strings.ToLower(r.Identity.ClusterName)
proxmoxCluster := ""
if r.Proxmox != nil {
proxmoxCluster = strings.ToLower(r.Proxmox.ClusterName)
}
dockerSwarmClusterName := ""
dockerSwarmClusterID := ""
if r.Docker != nil && r.Docker.Swarm != nil {
dockerSwarmClusterName = strings.ToLower(strings.TrimSpace(r.Docker.Swarm.ClusterName))
dockerSwarmClusterID = strings.ToLower(strings.TrimSpace(r.Docker.Swarm.ClusterID))
}
if identityCluster != cluster && proxmoxCluster != cluster && dockerSwarmClusterName != cluster && dockerSwarmClusterID != cluster {
continue
}
}
if filters.namespace != "" {
if r.Kubernetes == nil {
continue
}
want := strings.ToLower(filters.namespace)
got := strings.ToLower(strings.TrimSpace(r.Kubernetes.Namespace))
if got != want {
continue
}
}
if filters.query != "" {
name := strings.ToLower(r.Name)
if !strings.Contains(name, filters.query) {
continue
}
}
if len(filters.tags) > 0 {
matched := false
for _, tag := range r.Tags {
if _, ok := filters.tags[strings.ToLower(tag)]; ok {
matched = true
break
}
}
if !matched {
continue
}
}
out = append(out, r)
}
return out
}
func applySorting(resources []unified.Resource, field, order string) {
sort.SliceStable(resources, func(i, j int) bool {
a := resources[i]
b := resources[j]
comparison := 0
switch field {
case "status":
comparison = strings.Compare(string(a.Status), string(b.Status))
case "type":
comparison = strings.Compare(string(a.Type), string(b.Type))
case "lastSeen":
if a.LastSeen.Before(b.LastSeen) {
comparison = -1
} else if b.LastSeen.Before(a.LastSeen) {
comparison = 1
}
default:
comparison = unified.CompareResourcesByCanonicalName(a, b)
}
if comparison == 0 {
comparison = unified.CompareResourcesByCanonicalName(a, b)
}
if order == "desc" {
return comparison > 0
}
return comparison < 0
})
}
func paginate(resources []unified.Resource, page, limit int) ([]unified.Resource, ResourcesMeta) {
total := len(resources)
start := (page - 1) * limit
if start > total {
start = total
}
end := start + limit
if end > total {
end = total
}
paged := resources[start:end]
totalPages := total / limit
if total%limit != 0 {
totalPages++
}
meta := ResourcesMeta{
Page: page,
Limit: limit,
Total: total,
TotalPages: totalPages,
}
return paged, meta
}
func parseResourceTypes(raw string) map[unified.ResourceType]struct{} {
result := make(map[unified.ResourceType]struct{})
for _, part := range splitCSV(raw) {
for _, resourceType := range resourceTypeFilterAdapter(part) {
result[resourceType] = struct{}{}
}
}
return result
}
// ParseResourceTypes parses the public resource type filter grammar.
func ParseResourceTypes(raw string) map[unified.ResourceType]struct{} {
return parseResourceTypes(raw)
}
func unsupportedResourceTypeFilterTokens(raw string) []string {
if strings.TrimSpace(raw) == "" {
return nil
}
var unsupported []string
for _, part := range splitCSV(raw) {
if !isSupportedResourceTypeFilterToken(part) {
unsupported = append(unsupported, part)
}
}
return unsupported
}
// UnsupportedResourceTypeFilterTokens returns rejected public filter tokens.
func UnsupportedResourceTypeFilterTokens(raw string) []string {
return unsupportedResourceTypeFilterTokens(raw)
}
func isSupportedResourceTypeFilterToken(token string) bool {
return len(resourceTypeFilterAdapter(token)) > 0
}
func resourceTypeFilterAdapter(token string) []unified.ResourceType {
switch token {
case "agent", "agents", "node", "nodes":
return []unified.ResourceType{unified.ResourceTypeAgent}
case "docker-host":
return []unified.ResourceType{"docker-host"}
case "docker-image", "docker-images":
return []unified.ResourceType{unified.ResourceTypeDockerImage}
case "docker-volume", "docker-volumes":
return []unified.ResourceType{unified.ResourceTypeDockerVolume}
case "docker-network", "docker-networks":
return []unified.ResourceType{unified.ResourceTypeDockerNetwork}
case "docker-task", "docker-tasks":
return []unified.ResourceType{unified.ResourceTypeDockerTask}
case "docker-swarm-node", "docker-swarm-nodes", "docker-node", "docker-nodes", "swarm-node", "swarm-nodes":
return []unified.ResourceType{unified.ResourceTypeDockerSwarmNode}
case "docker-secret", "docker-secrets", "swarm-secret", "swarm-secrets":
return []unified.ResourceType{unified.ResourceTypeDockerSecret}
case "docker-config", "docker-configs", "swarm-config", "swarm-configs":
return []unified.ResourceType{unified.ResourceTypeDockerConfig}
case "vm", "vms":
return []unified.ResourceType{unified.ResourceTypeVM}
case "system-container", "system-containers", "oci-container":
return []unified.ResourceType{unified.ResourceTypeSystemContainer}
case "app-container", "app-containers":
return []unified.ResourceType{unified.ResourceTypeAppContainer}
case "docker-service", "service", "services":
return []unified.ResourceType{unified.ResourceTypeDockerService}
case "pod", "pods":
return []unified.ResourceType{unified.ResourceTypePod}
case "k8s-cluster", "k8s-clusters":
return []unified.ResourceType{unified.ResourceTypeK8sCluster}
case "k8s-node", "k8s-nodes":
return []unified.ResourceType{unified.ResourceTypeK8sNode}
case "k8s-deployment", "k8s-deployments":
return []unified.ResourceType{unified.ResourceTypeK8sDeployment}
case "k8s-replicaset", "k8s-replicasets":
return []unified.ResourceType{unified.ResourceTypeK8sReplicaSet}
case "k8s-namespace", "k8s-namespaces":
return []unified.ResourceType{unified.ResourceTypeK8sNamespace}
case "k8s-service", "k8s-services":
return []unified.ResourceType{unified.ResourceTypeK8sService}
case "k8s-statefulset", "k8s-statefulsets":
return []unified.ResourceType{unified.ResourceTypeK8sStatefulSet}
case "k8s-daemonset", "k8s-daemonsets":
return []unified.ResourceType{unified.ResourceTypeK8sDaemonSet}
case "k8s-job", "k8s-jobs":
return []unified.ResourceType{unified.ResourceTypeK8sJob}
case "k8s-cronjob", "k8s-cronjobs":
return []unified.ResourceType{unified.ResourceTypeK8sCronJob}
case "k8s-ingress", "k8s-ingresses":
return []unified.ResourceType{unified.ResourceTypeK8sIngress}
case "k8s-endpoint-slice", "k8s-endpoint-slices":
return []unified.ResourceType{unified.ResourceTypeK8sEndpointSlice}
case "k8s-network-policy", "k8s-network-policies":
return []unified.ResourceType{unified.ResourceTypeK8sNetworkPolicy}
case "k8s-persistent-volume", "k8s-persistent-volumes":
return []unified.ResourceType{unified.ResourceTypeK8sPV}
case "k8s-persistent-volume-claim", "k8s-persistent-volume-claims":
return []unified.ResourceType{unified.ResourceTypeK8sPVC}
case "k8s-storage-class", "k8s-storage-classes":
return []unified.ResourceType{unified.ResourceTypeK8sStorageClass}
case "k8s-configmap", "k8s-configmaps":
return []unified.ResourceType{unified.ResourceTypeK8sConfigMap}
case "k8s-secret", "k8s-secrets":
return []unified.ResourceType{unified.ResourceTypeK8sSecret}
case "k8s-serviceaccount", "k8s-serviceaccounts":
return []unified.ResourceType{unified.ResourceTypeK8sServiceAccount}
case "k8s-role", "k8s-roles":
return []unified.ResourceType{unified.ResourceTypeK8sRole}
case "k8s-cluster-role", "k8s-cluster-roles":
return []unified.ResourceType{unified.ResourceTypeK8sClusterRole}
case "k8s-role-binding", "k8s-role-bindings":
return []unified.ResourceType{unified.ResourceTypeK8sRoleBinding}
case "k8s-cluster-role-binding", "k8s-cluster-role-bindings":
return []unified.ResourceType{unified.ResourceTypeK8sClusterRoleBinding}
case "k8s-resource-quota", "k8s-resource-quotas":
return []unified.ResourceType{unified.ResourceTypeK8sResourceQuota}
case "k8s-limit-range", "k8s-limit-ranges":
return []unified.ResourceType{unified.ResourceTypeK8sLimitRange}
case "k8s-pod-disruption-budget", "k8s-pod-disruption-budgets":
return []unified.ResourceType{unified.ResourceTypeK8sPDB}
case "k8s-horizontal-pod-autoscaler", "k8s-horizontal-pod-autoscalers", "k8s-hpa", "k8s-hpas":
return []unified.ResourceType{unified.ResourceTypeK8sHPA}
case "k8s-event", "k8s-events":
return []unified.ResourceType{unified.ResourceTypeK8sEvent}
case "storage":
return []unified.ResourceType{unified.ResourceTypeStorage}
case "network", "networks":
return []unified.ResourceType{unified.ResourceTypeNetwork}
case "pbs":
return []unified.ResourceType{unified.ResourceTypePBS}
case "pmg":
return []unified.ResourceType{unified.ResourceTypePMG}
case "ceph", "pool":
return []unified.ResourceType{unified.ResourceTypeCeph}
case "physical_disk", "physical-disk", "physicaldisk", "disk":
return []unified.ResourceType{unified.ResourceTypePhysicalDisk}
case "network-share", "network-shares", "share", "shares":
return []unified.ResourceType{unified.ResourceTypeNetworkShare}
case "network-endpoint", "network-endpoints", "endpoint", "endpoints", "availability":
return []unified.ResourceType{unified.ResourceTypeNetworkEndpoint}
default:
return nil
}
}
func parseSources(raw string) map[unified.DataSource]struct{} {
result := make(map[unified.DataSource]struct{})
for _, part := range splitCSV(raw) {
switch part {
case "proxmox":
result[unified.SourceProxmox] = struct{}{}
case "agent":
result[unified.SourceAgent] = struct{}{}
case "docker":
result[unified.SourceDocker] = struct{}{}
case "pbs":
result[unified.SourcePBS] = struct{}{}
case "pmg":
result[unified.SourcePMG] = struct{}{}
case "kubernetes":
result[unified.SourceK8s] = struct{}{}
case "truenas":
result[unified.SourceTrueNAS] = struct{}{}
case "vmware", "vmware-vsphere":
result[unified.SourceVMware] = struct{}{}
case "availability":
result[unified.SourceAvailability] = struct{}{}
}
}
return result
}
func parseStatuses(raw string) map[unified.ResourceStatus]struct{} {
result := make(map[unified.ResourceStatus]struct{})
for _, part := range splitCSV(raw) {
switch part {
case "online":
result[unified.StatusOnline] = struct{}{}
case "offline":
result[unified.StatusOffline] = struct{}{}
case "warning":
result[unified.StatusWarning] = struct{}{}
case "unknown":
result[unified.StatusUnknown] = struct{}{}
}
}
return result
}
func parseTags(raw string) map[string]struct{} {
result := make(map[string]struct{})
for _, part := range splitCSV(raw) {
if part == "" {
continue
}
result[strings.ToLower(part)] = struct{}{}
}
return result
}
func attachDiscoveryTargets(resources []unified.Resource) {
for i := range resources {
attachDiscoveryTarget(&resources[i])
}
}
func attachDiscoveryTarget(resource *unified.Resource) {
if resource == nil {
return
}
if target := buildDiscoveryTarget(*resource); target != nil {
resource.DiscoveryTarget = target
}
}
func (h *QueryService) attachDiscoveryReadinesses(resources []unified.Resource, now time.Time) {
if h == nil || h.discoveryReadiness == nil {
return
}
for i := range resources {
h.attachDiscoveryReadiness(&resources[i], now)
}
}
func (h *QueryService) attachDiscoveryReadiness(resource *unified.Resource, now time.Time) {
if h == nil || h.discoveryReadiness == nil || resource == nil {
return
}
readiness := h.discoveryReadiness.DiscoveryReadinessForResource(*resource, now)
if readiness.State == "" {
return
}
resource.DiscoveryReadiness = &readiness
}
// AttachDiscoveryReadiness projects current discovery state onto a resource
// used by another API domain.
func (h *QueryService) AttachDiscoveryReadiness(resource *unified.Resource, now time.Time) {
h.attachDiscoveryReadiness(resource, now)
}
func attachMetricsTargets(resources []unified.Resource, registry *unified.ResourceRegistry) {
for i := range resources {
attachMetricsTarget(&resources[i], registry)
}
}
func attachMetricsTarget(resource *unified.Resource, registry *unified.ResourceRegistry) {
if resource == nil || registry == nil {
return
}
resource.MetricsTarget = registry.MetricsTarget(resource.ID)
}
// resourceContractType maps internal unified resource shapes onto the canonical
// REST resource-type contract without exposing legacy aliases.
func resourceContractType(r unified.Resource) unified.ResourceType {
return unified.ContractResourceType(r)
}
// ContractType maps an internal resource shape onto the public API contract.
func ContractType(resource unified.Resource) unified.ResourceType {
return resourceContractType(resource)
}
// AttachDiscoveryTarget projects the canonical discovery target metadata.
func AttachDiscoveryTarget(resource *unified.Resource) {
attachDiscoveryTarget(resource)
}
// applyResourceContractTypes rewrites Type fields on the response slice so the
// REST API exposes the canonical resource-type contract.
func applyResourceContractTypes(resources []unified.Resource) {
for i := range resources {
resources[i].Type = resourceContractType(resources[i])
}
}
// computeResourceContractByType builds the ByType aggregation using the
// canonical REST resource-type contract. Does not mutate the input slice.
func computeResourceContractByType(resources []unified.Resource) map[unified.ResourceType]int {
m := make(map[unified.ResourceType]int, 8)
for _, r := range resources {
m[resourceContractType(r)]++
}
return m
}
func computeResourceContractStats(resources []unified.Resource) unified.ResourceStats {
stats := unified.ResourceStats{
Total: len(resources),
ByType: make(map[unified.ResourceType]int, 8),
ByStatus: make(map[unified.ResourceStatus]int, 8),
BySource: make(map[unified.DataSource]int, 8),
}
for _, resource := range resources {
stats.ByType[resourceContractType(resource)]++
stats.ByStatus[resource.Status]++
for _, source := range resource.Sources {
stats.BySource[source]++
}
}
admission := unified.BuildPlatformAdmission(resources)
stats.PlatformAdmission = &admission
return stats
}
func resourcePolicyPostureAggregation(resources []unified.Resource) *unified.ResourcePolicyPostureSummary {
canonicalResources := unified.RefreshCanonicalMetadataSlice(resources)
return unified.ResourcePolicyPostureContract(unified.SummarizePolicyPosture(canonicalResources))
}
func buildDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
switch unified.CanonicalResourceType(resource.Type) {
case unified.ResourceTypeAgent:
return hostDiscoveryTarget(resource)
case unified.ResourceTypeVM:
return proxmoxGuestDiscoveryTarget(resource, "vm")
case unified.ResourceTypeSystemContainer:
return proxmoxGuestDiscoveryTarget(resource, "system-container")
case unified.ResourceTypeAppContainer:
return dockerContainerDiscoveryTarget(resource)
case unified.ResourceTypePBS:
return hostDiscoveryTarget(resource)
case unified.ResourceTypePMG:
return hostDiscoveryTarget(resource)
case unified.ResourceTypeCeph:
return cephDiscoveryTarget(resource)
case unified.ResourceTypeK8sCluster:
return kubernetesDiscoveryTarget(resource)
case unified.ResourceTypeK8sNode:
return kubernetesDiscoveryTarget(resource)
case unified.ResourceTypePod:
return kubernetesDiscoveryTarget(resource)
case unified.ResourceTypeK8sDeployment:
return kubernetesDiscoveryTarget(resource)
case unified.ResourceTypePhysicalDisk:
return physicalDiskDiscoveryTarget(resource)
default:
return nil
}
}
func cephDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
if resource.Ceph == nil {
return nil
}
hostID := firstNonEmptyTrimmed(
resource.Ceph.FSID,
resource.Name,
resource.ID,
)
if hostID == "" {
return nil
}
return &unified.DiscoveryTarget{
ResourceType: "ceph",
AgentID: hostID,
ResourceID: resource.ID,
Hostname: resource.Name,
}
}
// CephDiscoveryTarget constructs the canonical discovery target for a Ceph resource.
func CephDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
return cephDiscoveryTarget(resource)
}
func hostDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
linkedAgentID := ""
agentBacked := hasSource(resource.Sources, unified.SourceAgent) || resource.Agent != nil
if agentBacked {
linkedAgentID = proxmoxLinkedAgentID(resource.Proxmox)
}
if linkedAgentID != "" {
agentBacked = true
}
hostID := firstNonEmptyTrimmed(
agentID(resource.Agent),
linkedAgentID,
dockerAgentID(resource.Docker),
proxmoxNodeName(resource.Proxmox),
agentHostname(resource.Agent),
dockerHostname(resource.Docker),
pbsHostname(resource.PBS),
pmgHostname(resource.PMG),
firstResourceHostname(resource),
resource.Name,
resource.ID,
)
if hostID == "" {
return nil
}
return &unified.DiscoveryTarget{
ResourceType: "agent",
AgentID: hostID,
ResourceID: hostID,
Hostname: firstNonEmptyTrimmed(
agentHostname(resource.Agent),
proxmoxNodeName(resource.Proxmox),
dockerHostname(resource.Docker),
pbsHostname(resource.PBS),
pmgHostname(resource.PMG),
firstResourceHostname(resource),
resource.Name,
hostID,
),
}
}
func dockerContainerDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
if resource.Docker == nil {
return nil
}
hostID := strings.TrimSpace(resource.Docker.AgentID)
if hostID == "" {
return nil
}
resourceID := firstNonEmptyTrimmed(
resource.Name,
resource.Docker.ContainerID,
resource.ID,
)
if resourceID == "" {
return nil
}
return &unified.DiscoveryTarget{
ResourceType: "app-container",
AgentID: hostID,
ResourceID: resourceID,
Hostname: firstNonEmptyTrimmed(
resource.Docker.Hostname,
firstResourceHostname(resource),
resource.Name,
),
}
}
func proxmoxGuestDiscoveryTarget(resource unified.Resource, resourceType string) *unified.DiscoveryTarget {
if resource.Proxmox == nil || resource.Proxmox.NodeName == "" || resource.Proxmox.VMID == 0 {
return nil
}
resourceID := strconv.Itoa(resource.Proxmox.VMID)
hostID := proxmoxLinkedAgentID(resource.Proxmox)
if hostID == "" || resourceID == "" {
return nil
}
return &unified.DiscoveryTarget{
ResourceType: string(resourceType),
AgentID: hostID,
ResourceID: resourceID,
Hostname: firstNonEmptyTrimmed(
firstResourceHostname(resource),
resource.Name,
resourceID,
),
}
}
func kubernetesDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
if resource.Kubernetes == nil {
return nil
}
resourceType := unified.CanonicalResourceType(resource.Type)
hostID := firstNonEmptyTrimmed(
resource.Kubernetes.AgentID,
resource.Kubernetes.ClusterID,
resource.Kubernetes.ClusterName,
)
if hostID == "" {
return nil
}
resourceID := ""
switch resourceType {
case unified.ResourceTypeK8sCluster:
resourceID = firstNonEmptyTrimmed(
resource.Kubernetes.ClusterID,
resource.Kubernetes.ClusterName,
resource.Name,
)
case unified.ResourceTypeK8sNode:
resourceID = firstNonEmptyTrimmed(
resource.Kubernetes.NodeUID,
resource.Kubernetes.NodeName,
resource.Name,
)
case unified.ResourceTypeK8sDeployment:
resourceID = firstNonEmptyTrimmed(
resource.Kubernetes.DeploymentUID,
kubernetesNamespacedName(resource.Kubernetes.Namespace, resource.Name),
resource.Name,
)
case unified.ResourceTypePod:
resourceID = firstNonEmptyTrimmed(
resource.Kubernetes.PodUID,
kubernetesNamespacedName(resource.Kubernetes.Namespace, resource.Name),
resource.Name,
)
default:
return nil
}
if resourceID == "" {
return nil
}
return &unified.DiscoveryTarget{
ResourceType: string(resourceType),
AgentID: hostID,
ResourceID: resourceID,
Hostname: firstNonEmptyTrimmed(
resource.Kubernetes.ClusterName,
resource.Kubernetes.Context,
resource.Name,
),
}
}
func physicalDiskDiscoveryTarget(resource unified.Resource) *unified.DiscoveryTarget {
if resource.PhysicalDisk == nil {
return nil
}
hostID := ""
if resource.Proxmox != nil {
hostID = resource.Proxmox.NodeName
}
if hostID == "" {
hostID = firstNonEmptyTrimmed(firstResourceHostname(resource), resource.Name)
}
if hostID == "" {
return nil
}
return &unified.DiscoveryTarget{
ResourceType: "disk",
AgentID: hostID,
ResourceID: resource.ID,
Hostname: hostID,
}
}
func kubernetesNamespacedName(namespace, name string) string {
ns := strings.TrimSpace(namespace)
n := strings.TrimSpace(name)
if ns == "" {
return n
}
if n == "" {
return ns
}
return ns + "/" + n
}
func firstResourceHostname(resource unified.Resource) string {
for _, hostname := range resource.Identity.Hostnames {
trimmed := strings.TrimSpace(hostname)
if trimmed != "" {
return trimmed
}
}
return ""
}
func firstNonEmptyTrimmed(values ...string) string {
for _, value := range values {
trimmed := strings.TrimSpace(value)
if trimmed != "" {
return trimmed
}
}
return ""
}
func hasSource(sources []unified.DataSource, target unified.DataSource) bool {
for _, source := range sources {
if source == target {
return true
}
}
return false
}
func agentID(agent *unified.AgentData) string {
if agent == nil {
return ""
}
return agent.AgentID
}
func dockerAgentID(docker *unified.DockerData) string {
if docker == nil {
return ""
}
return docker.AgentID
}
func agentHostname(agent *unified.AgentData) string {
if agent == nil {
return ""
}
return agent.Hostname
}
func proxmoxLinkedAgentID(proxmox *unified.ProxmoxData) string {
if proxmox == nil {
return ""
}
return proxmox.LinkedAgentID
}
func proxmoxNodeName(proxmox *unified.ProxmoxData) string {
if proxmox == nil {
return ""
}
return proxmox.NodeName
}
func dockerHostname(docker *unified.DockerData) string {
if docker == nil {
return ""
}
return docker.Hostname
}
func pbsHostname(pbs *unified.PBSData) string {
if pbs == nil {
return ""
}
return pbs.Hostname
}
func pmgHostname(pmg *unified.PMGData) string {
if pmg == nil {
return ""
}
return pmg.Hostname
}
func splitCSV(raw string) []string {
raw = strings.NewReplacer("%2c", ",", "%2C", ",").Replace(raw)
parts := strings.Split(raw, ",")
out := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(strings.ToLower(part))
if part == "" {
continue
}
out = append(out, part)
}
return out
}
func parseIntDefault(raw string, def int) int {
if raw == "" {
return def
}
val, err := strconv.Atoi(raw)
if err != nil {
return def
}
return val
}
// getUserID attempts to resolve the user ID for auditing.
func getUserID(r *http.Request) string {
return auth.GetUser(r.Context())
}
func sanitizeError(err error, genericMessage string) string {
if err != nil {
log.Error().Err(err).Msg(genericMessage)
}
return genericMessage
}