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 }