ITWorx-Pulse-Public/internal/metriccatalog/catalog.go

package metriccatalog

import (
	"bytes"
	"context"
	"crypto/sha256"
	"embed"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"regexp"
	"sort"
	"strings"

	"github.com/itworx/pulse/internal/datasource"
)

//go:embed seed.json
var seedFS embed.FS

const (
	SchemaVersion   = 1
	maxCatalogBytes = 4 << 20
)

var semanticNamePattern = regexp.MustCompile(`^[a-z][a-z0-9]*(\.[a-z][a-z0-9_]*)+$`)
var labelPattern = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
var placeholderPattern = regexp.MustCompile(`\{\{([a-zA-Z_][a-zA-Z0-9_]*)\}\}`)
var allowedPlaceholders = map[string]struct{}{"instance": {}, "window": {}, "container": {}, "device": {}, "pool": {}, "probe_id": {}}

// Definition is the validated wire-compatible semantic metric definition.
type Definition struct {
	SemanticName          string      `json:"semanticName"`
	Version               int         `json:"version"`
	Description           string      `json:"description"`
	Unit                  string      `json:"unit"`
	ValueKind             string      `json:"valueKind"`
	SourceKind            string      `json:"sourceKind"`
	QueryTemplate         string      `json:"queryTemplate"`
	RequiredCapabilities  []string    `json:"requiredCapabilities,omitempty"`
	AllowedLabels         []string    `json:"allowedLabels"`
	DefaultAggregation    string      `json:"defaultAggregation,omitempty"`
	CardinalityBudget     int         `json:"cardinalityBudget"`
	Limits                Limits      `json:"limits"`
	AllowedVisualizations []string    `json:"allowedVisualizations"`
	FreshnessSeconds      int         `json:"freshnessSeconds"`
	DefaultThresholds     []Threshold `json:"defaultThresholds,omitempty"`
}

type Limits struct {
	MaxRangeSeconds int `json:"maxRangeSeconds"`
	MaxSeries       int `json:"maxSeries"`
	MaxPoints       int `json:"maxPoints"`
	TimeoutSeconds  int `json:"timeoutSeconds"`
}

type Threshold struct {
	State           string  `json:"state"`
	Operator        string  `json:"operator"`
	Value           float64 `json:"value"`
	DurationSeconds int     `json:"durationSeconds,omitempty"`
}

type Catalog struct {
	SchemaVersion int          `json:"schemaVersion"`
	Metrics       []Definition `json:"metrics"`
}

type BindingState string

const (
	BindingSupported   BindingState = "supported"
	BindingUnsupported BindingState = "unsupported"
	BindingUnavailable BindingState = "unavailable"
)

type Binding struct {
	SemanticName            string       `json:"semanticName"`
	Version                 int          `json:"version"`
	State                   BindingState `json:"state"`
	MissingCapabilities     []string     `json:"missingCapabilities,omitempty"`
	UnavailableCapabilities []string     `json:"unavailableCapabilities,omitempty"`
	UnsupportedCapabilities []string     `json:"unsupportedCapabilities,omitempty"`
}

type Response struct {
	SchemaVersion int          `json:"schemaVersion"`
	Version       string       `json:"version"`
	Metrics       []Definition `json:"metrics"`
	Bindings      []Binding    `json:"bindings"`
}

type Registry struct {
	catalog      Catalog
	version      string
	capabilities datasource.CapabilitySet
}

func DefaultRegistry() (Registry, error) {
	data, err := seedFS.ReadFile("seed.json")
	if err != nil {
		return Registry{}, fmt.Errorf("read embedded metric seed: %w", err)
	}
	return Load(context.Background(), bytes.NewReader(data), nil)
}

func Load(ctx context.Context, reader io.Reader, capabilities datasource.CapabilitySet) (Registry, error) {
	if err := ctx.Err(); err != nil {
		return Registry{}, err
	}
	limited := io.LimitReader(reader, maxCatalogBytes+1)
	decoder := json.NewDecoder(limited)
	decoder.DisallowUnknownFields()
	var catalog Catalog
	if err := decoder.Decode(&catalog); err != nil {
		return Registry{}, fmt.Errorf("decode metric catalog: %w", err)
	}
	var extra any
	if err := decoder.Decode(&extra); err != io.EOF {
		return Registry{}, errors.New("metric catalog contains trailing data")
	}
	if err := ctx.Err(); err != nil {
		return Registry{}, err
	}
	return New(catalog, capabilities)
}

func New(catalog Catalog, capabilities datasource.CapabilitySet) (Registry, error) {
	if err := catalog.Validate(); err != nil {
		return Registry{}, err
	}
	canonical := struct {
		SchemaVersion int          `json:"schemaVersion"`
		Metrics       []Definition `json:"metrics"`
	}{catalog.SchemaVersion, append([]Definition(nil), catalog.Metrics...)}
	for i := range canonical.Metrics {
		canonical.Metrics[i].RequiredCapabilities = append([]string(nil), canonical.Metrics[i].RequiredCapabilities...)
		canonical.Metrics[i].AllowedLabels = append([]string(nil), canonical.Metrics[i].AllowedLabels...)
		canonical.Metrics[i].AllowedVisualizations = append([]string(nil), canonical.Metrics[i].AllowedVisualizations...)
	}
	sort.Slice(canonical.Metrics, func(i, j int) bool {
		if canonical.Metrics[i].SemanticName == canonical.Metrics[j].SemanticName {
			return canonical.Metrics[i].Version < canonical.Metrics[j].Version
		}
		return canonical.Metrics[i].SemanticName < canonical.Metrics[j].SemanticName
	})
	encoded, err := json.Marshal(canonical)
	if err != nil {
		return Registry{}, fmt.Errorf("canonicalize metric catalog: %w", err)
	}
	digest := sha256.Sum256(encoded)
	return Registry{catalog: Catalog{SchemaVersion: catalog.SchemaVersion, Metrics: canonical.Metrics}, version: hex.EncodeToString(digest[:]), capabilities: append(datasource.CapabilitySet(nil), capabilities...)}, nil
}

func (c Catalog) Validate() error {
	if c.SchemaVersion != SchemaVersion {
		return fmt.Errorf("unsupported metric catalog schema version %d", c.SchemaVersion)
	}
	if len(c.Metrics) == 0 || len(c.Metrics) > 500 {
		return errors.New("metric catalog must contain 1-500 metrics")
	}
	seen := make(map[string]struct{}, len(c.Metrics))
	for _, metric := range c.Metrics {
		if err := metric.Validate(); err != nil {
			return fmt.Errorf("metric %q: %w", metric.SemanticName, err)
		}
		key := fmt.Sprintf("%s@%d", metric.SemanticName, metric.Version)
		if _, ok := seen[key]; ok {
			return fmt.Errorf("duplicate metric %q", key)
		}
		seen[key] = struct{}{}
	}
	return nil
}

func (d Definition) Validate() error {
	if !semanticNamePattern.MatchString(d.SemanticName) || len(d.SemanticName) > 160 {
		return errors.New("semantic name is invalid")
	}
	if d.Version < 1 {
		return errors.New("version must be positive")
	}
	if strings.TrimSpace(d.Description) == "" || len(d.Description) > 1000 {
		return errors.New("description is required and bounded")
	}
	if len(d.Unit) > 40 {
		return errors.New("unit is too long")
	}
	if !oneOf(d.ValueKind, "gauge", "counter", "state", "histogram") || !oneOf(d.SourceKind, "prometheus", "inventory", "event", "derived") {
		return errors.New("value or source kind is invalid")
	}
	if strings.TrimSpace(d.QueryTemplate) == "" || len(d.QueryTemplate) > 4000 {
		return errors.New("query template is required and bounded")
	}
	if strings.Contains(d.QueryTemplate, "${") {
		return errors.New("query template contains unsupported interpolation")
	}
	for _, match := range placeholderPattern.FindAllStringSubmatch(d.QueryTemplate, -1) {
		if _, ok := allowedPlaceholders[match[1]]; !ok {
			return fmt.Errorf("query template placeholder %q is not allowed", match[1])
		}
	}
	if len(d.RequiredCapabilities) > 20 {
		return errors.New("too many required capabilities")
	}
	if err := uniqueBounded(d.RequiredCapabilities, 100, nil); err != nil {
		return fmt.Errorf("required capabilities: %w", err)
	}
	if len(d.AllowedLabels) > 30 {
		return errors.New("too many allowed labels")
	}
	if err := uniqueBounded(d.AllowedLabels, 0, labelPattern); err != nil {
		return fmt.Errorf("allowed labels: %w", err)
	}
	if d.DefaultAggregation != "" && !oneOf(d.DefaultAggregation, "none", "avg", "sum", "min", "max", "rate", "increase", "p50", "p95", "p99") {
		return errors.New("default aggregation is invalid")
	}
	if d.CardinalityBudget < 1 || d.CardinalityBudget > 10000 {
		return errors.New("cardinality budget is out of range")
	}
	if d.Limits.MaxRangeSeconds < 60 || d.Limits.MaxSeries < 1 || d.Limits.MaxSeries > 10000 || d.Limits.MaxPoints < 10 || d.Limits.MaxPoints > 1000000 || d.Limits.TimeoutSeconds < 1 || d.Limits.TimeoutSeconds > 120 {
		return errors.New("limits are out of range")
	}
	if len(d.AllowedVisualizations) == 0 || len(d.AllowedVisualizations) > 30 {
		return errors.New("at least one visualization is required")
	}
	if err := uniqueBounded(d.AllowedVisualizations, 0, nil); err != nil {
		return fmt.Errorf("visualizations: %w", err)
	}
	if d.FreshnessSeconds < 1 {
		return errors.New("freshness must be positive")
	}
	for _, threshold := range d.DefaultThresholds {
		if !oneOf(threshold.State, "attention", "degraded", "critical") || !oneOf(threshold.Operator, ">", ">=", "<", "<=", "==", "!=") || threshold.DurationSeconds < 0 {
			return errors.New("threshold is invalid")
		}
	}
	return nil
}

func uniqueBounded(values []string, maxLen int, pattern *regexp.Regexp) error {
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		if maxLen > 0 && len(value) > maxLen {
			return errors.New("value is too long")
		}
		if pattern != nil && !pattern.MatchString(value) {
			return fmt.Errorf("value %q is invalid", value)
		}
		if _, ok := seen[value]; ok {
			return fmt.Errorf("duplicate value %q", value)
		}
		seen[value] = struct{}{}
	}
	return nil
}
func oneOf(value string, allowed ...string) bool {
	for _, candidate := range allowed {
		if value == candidate {
			return true
		}
	}
	return false
}

func (r Registry) Find(semanticName string) (Definition, bool) {
	for _, metric := range r.catalog.Metrics {
		if metric.SemanticName == semanticName {
			return metric, true
		}
	}
	return Definition{}, false
}

func (r Registry) Version() string       { return r.version }
func (r Registry) Metrics() []Definition { return append([]Definition(nil), r.catalog.Metrics...) }
func (r Registry) Response() Response {
	return Response{SchemaVersion: r.catalog.SchemaVersion, Version: r.version, Metrics: r.Metrics(), Bindings: r.Bindings()}
}
func (r Registry) Bindings() []Binding {
	result := make([]Binding, 0, len(r.catalog.Metrics))
	for _, metric := range r.catalog.Metrics {
		binding := Binding{SemanticName: metric.SemanticName, Version: metric.Version, State: BindingSupported}
		for _, required := range metric.RequiredCapabilities {
			capability, ok := r.capabilities.Find(required)
			if !ok {
				binding.State = BindingUnsupported
				binding.MissingCapabilities = append(binding.MissingCapabilities, required)
				continue
			}
			switch capability.State {
			case datasource.CapabilityEnabled:
			case datasource.CapabilityUnsupported:
				binding.State = BindingUnsupported
				binding.UnsupportedCapabilities = append(binding.UnsupportedCapabilities, required)
			case datasource.CapabilityDisabled:
				if binding.State == BindingSupported {
					binding.State = BindingUnavailable
				}
				binding.UnavailableCapabilities = append(binding.UnavailableCapabilities, required)
			case datasource.CapabilityUnavailable:
				if binding.State == BindingSupported {
					binding.State = BindingUnavailable
				}
				binding.UnavailableCapabilities = append(binding.UnavailableCapabilities, required)
			default:
				if binding.State == BindingSupported {
					binding.State = BindingUnavailable
				}
				binding.UnavailableCapabilities = append(binding.UnavailableCapabilities, required)
			}
		}
		result = append(result, binding)
	}
	return result
}