| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | package pipeline |
| 4 | |
| 5 | import ( |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "strings" |
| 10 | |
| 11 | "github.com/netdata/netdata/go/plugins/plugin/agent/discovery/sd/model" |
| 12 | "github.com/netdata/netdata/go/plugins/plugin/agent/internal/naming" |
| 13 | "github.com/netdata/netdata/go/plugins/plugin/framework/confgroup" |
| 14 | ) |
| 15 | |
| 16 | type Config struct { |
| 17 | Source string `yaml:"-" json:"-"` |
| 18 | ConfigDefaults confgroup.Registry `yaml:"-" json:"-"` |
| 19 | |
| 20 | Disabled bool `yaml:"disabled,omitempty" json:"disabled,omitempty"` |
| 21 | Name string `yaml:"name" json:"name"` |
| 22 | |
| 23 | // Canonical format: discoverer: { <type>: <config> } |
| 24 | Discoverer DiscovererPayload `yaml:"discoverer,omitempty" json:"discoverer"` |
| 25 | |
| 26 | // New single-step format for service rules: |
| 27 | Services []ServiceRuleConfig `yaml:"services,omitempty" json:"services,omitempty"` |
| 28 | |
| 29 | // Legacy formats (converted during unmarshal, excluded from JSON): |
| 30 | LegacyDiscover []LegacyDiscoveryConfig `yaml:"discover,omitempty" json:"-"` |
| 31 | LegacyClassify []ClassifyRuleConfig `yaml:"classify,omitempty" json:"-"` |
| 32 | LegacyCompose []ComposeRuleConfig `yaml:"compose,omitempty" json:"-"` |
| 33 | } |
| 34 | |
| 35 | // DiscovererPayload stores discoverer type/config internally in a generic form |
| 36 | // while preserving canonical external shape discoverer: { <type>: <config> }. |
| 37 | type DiscovererPayload struct { |
| 38 | Kind string `yaml:"-" json:"-"` |
| 39 | Config json.RawMessage `yaml:"-" json:"-"` |
| 40 | } |
| 41 | |
| 42 | func (d DiscovererPayload) TypeName() string { |
| 43 | return strings.TrimSpace(d.Kind) |
| 44 | } |
| 45 | |
| 46 | func (d DiscovererPayload) Type() string { |
| 47 | return d.TypeName() |
| 48 | } |
| 49 | |
| 50 | func (d DiscovererPayload) Empty() bool { |
| 51 | return d.TypeName() == "" |
| 52 | } |
| 53 | |
| 54 | func (d *DiscovererPayload) UnmarshalJSON(data []byte) error { |
| 55 | var m map[string]json.RawMessage |
| 56 | if err := json.Unmarshal(data, &m); err != nil { |
| 57 | return err |
| 58 | } |
| 59 | |
| 60 | if len(m) == 0 { |
| 61 | *d = DiscovererPayload{} |
| 62 | return nil |
| 63 | } |
| 64 | if len(m) > 1 { |
| 65 | return errors.New("multiple discoverers configured, only one is allowed") |
| 66 | } |
| 67 | |
| 68 | for typ, cfg := range m { |
| 69 | *d = DiscovererPayload{Kind: typ, Config: cloneRaw(cfg)} |
| 70 | return nil |
| 71 | } |
| 72 | |
| 73 | *d = DiscovererPayload{} |
| 74 | return nil |
| 75 | } |
| 76 | |
| 77 | func (d DiscovererPayload) MarshalJSON() ([]byte, error) { |
| 78 | if d.Empty() { |
| 79 | return []byte("{}"), nil |
| 80 | } |
| 81 | |
| 82 | cfg := cloneRaw(d.Config) |
| 83 | if len(cfg) == 0 { |
| 84 | cfg = []byte("{}") |
| 85 | } |
| 86 | |
| 87 | return json.Marshal(map[string]json.RawMessage{d.TypeName(): cfg}) |
| 88 | } |
| 89 | |
| 90 | func (d *DiscovererPayload) UnmarshalYAML(unmarshal func(any) error) error { |
| 91 | var raw any |
| 92 | if err := unmarshal(&raw); err != nil { |
| 93 | return err |
| 94 | } |
| 95 | if raw == nil { |
| 96 | *d = DiscovererPayload{} |
| 97 | return nil |
| 98 | } |
| 99 | |
| 100 | norm, err := normalizeYAMLValue(raw) |
| 101 | if err != nil { |
| 102 | return err |
| 103 | } |
| 104 | |
| 105 | m, ok := norm.(map[string]any) |
| 106 | if !ok { |
| 107 | return fmt.Errorf("invalid discoverer format: expected map, got %T", norm) |
| 108 | } |
| 109 | |
| 110 | if len(m) == 0 { |
| 111 | *d = DiscovererPayload{} |
| 112 | return nil |
| 113 | } |
| 114 | if len(m) > 1 { |
| 115 | return errors.New("multiple discoverers configured, only one is allowed") |
| 116 | } |
| 117 | |
| 118 | for typ, cfg := range m { |
| 119 | bs, err := json.Marshal(cfg) |
| 120 | if err != nil { |
| 121 | return fmt.Errorf("marshal discoverer %q config: %w", typ, err) |
| 122 | } |
| 123 | *d = DiscovererPayload{Kind: typ, Config: bs} |
| 124 | return nil |
| 125 | } |
| 126 | |
| 127 | *d = DiscovererPayload{} |
| 128 | return nil |
| 129 | } |
| 130 | |
| 131 | func (d DiscovererPayload) MarshalYAML() (any, error) { |
| 132 | if d.Empty() { |
| 133 | return map[string]any{}, nil |
| 134 | } |
| 135 | |
| 136 | cfg := any(map[string]any{}) |
| 137 | if len(d.Config) != 0 { |
| 138 | if err := json.Unmarshal(d.Config, &cfg); err != nil { |
| 139 | return nil, fmt.Errorf("unmarshal discoverer %q config json: %w", d.TypeName(), err) |
| 140 | } |
| 141 | } |
| 142 | |
| 143 | return map[string]any{d.TypeName(): cfg}, nil |
| 144 | } |
| 145 | |
| 146 | func cloneRaw(raw json.RawMessage) json.RawMessage { |
| 147 | if raw == nil { |
| 148 | return nil |
| 149 | } |
| 150 | out := make([]byte, len(raw)) |
| 151 | copy(out, raw) |
| 152 | return out |
| 153 | } |
| 154 | |
| 155 | func normalizeYAMLValue(v any) (any, error) { |
| 156 | switch vv := v.(type) { |
| 157 | case map[any]any: |
| 158 | m := make(map[string]any, len(vv)) |
| 159 | for k, iv := range vv { |
| 160 | ks, ok := k.(string) |
| 161 | if !ok { |
| 162 | return nil, fmt.Errorf("yaml map key must be string, got %T", k) |
| 163 | } |
| 164 | norm, err := normalizeYAMLValue(iv) |
| 165 | if err != nil { |
| 166 | return nil, err |
| 167 | } |
| 168 | m[ks] = norm |
| 169 | } |
| 170 | return m, nil |
| 171 | case map[string]any: |
| 172 | m := make(map[string]any, len(vv)) |
| 173 | for k, iv := range vv { |
| 174 | norm, err := normalizeYAMLValue(iv) |
| 175 | if err != nil { |
| 176 | return nil, err |
| 177 | } |
| 178 | m[k] = norm |
| 179 | } |
| 180 | return m, nil |
| 181 | case []any: |
| 182 | arr := make([]any, 0, len(vv)) |
| 183 | for _, iv := range vv { |
| 184 | norm, err := normalizeYAMLValue(iv) |
| 185 | if err != nil { |
| 186 | return nil, err |
| 187 | } |
| 188 | arr = append(arr, norm) |
| 189 | } |
| 190 | return arr, nil |
| 191 | default: |
| 192 | return v, nil |
| 193 | } |
| 194 | } |
| 195 | |
| 196 | // CleanName returns the name sanitized for use in dyncfg IDs. |
| 197 | // Sanitizes for safe use in IDs and paths. |
| 198 | func (c Config) CleanName() string { |
| 199 | return naming.Sanitize(c.Name) |
| 200 | } |
| 201 | |
| 202 | // UnmarshalYAML implements yaml.Unmarshaler. |
| 203 | // It converts legacy formats to the canonical format: |
| 204 | // - discover[] -> discoverer{} |
| 205 | // - classify/compose -> services[] |
| 206 | func (c *Config) UnmarshalYAML(unmarshal func(any) error) error { |
| 207 | type plain Config // avoid recursion |
| 208 | if err := unmarshal((*plain)(c)); err != nil { |
| 209 | return err |
| 210 | } |
| 211 | |
| 212 | // Convert legacy discover[] to new discoverer{} format |
| 213 | if len(c.LegacyDiscover) > 0 && c.Discoverer.Empty() { |
| 214 | if err := c.convertLegacyDiscover(); err != nil { |
| 215 | return fmt.Errorf("failed to convert legacy discover config: %w", err) |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // Convert legacy classify/compose to canonical services format |
| 220 | if len(c.Services) == 0 && (len(c.LegacyClassify) > 0 || len(c.LegacyCompose) > 0) { |
| 221 | services, err := ConvertOldToServices(c.LegacyClassify, c.LegacyCompose) |
| 222 | if err != nil { |
| 223 | return fmt.Errorf("failed to convert legacy config: %w", err) |
| 224 | } |
| 225 | c.Services = services |
| 226 | } |
| 227 | |
| 228 | // Clear legacy fields - config is now in canonical form |
| 229 | c.LegacyDiscover = nil |
| 230 | c.LegacyClassify = nil |
| 231 | c.LegacyCompose = nil |
| 232 | |
| 233 | return nil |
| 234 | } |
| 235 | |
| 236 | // convertLegacyDiscover converts legacy discover[] array to new discoverer{} struct. |
| 237 | // For non-k8s discoverers, first value wins; for k8s, arrays are merged. |
| 238 | func (c *Config) convertLegacyDiscover() error { |
| 239 | var converted bool |
| 240 | |
| 241 | for i, d := range c.LegacyDiscover { |
| 242 | typ := strings.TrimSpace(d.Discoverer) |
| 243 | if typ == "" { |
| 244 | continue |
| 245 | } |
| 246 | |
| 247 | rawCfg, ok := d.Config[typ] |
| 248 | if !ok { |
| 249 | return fmt.Errorf("legacy discover[%d]: missing config for discoverer %q", i, typ) |
| 250 | } |
| 251 | |
| 252 | norm, err := normalizeYAMLValue(rawCfg) |
| 253 | if err != nil { |
| 254 | return fmt.Errorf("legacy discover[%d]: normalize %q config: %w", i, typ, err) |
| 255 | } |
| 256 | |
| 257 | cfgJSON, err := json.Marshal(norm) |
| 258 | if err != nil { |
| 259 | return fmt.Errorf("legacy discover[%d]: marshal %q config: %w", i, typ, err) |
| 260 | } |
| 261 | |
| 262 | if c.Discoverer.Empty() { |
| 263 | c.Discoverer = DiscovererPayload{Kind: typ, Config: cfgJSON} |
| 264 | converted = true |
| 265 | continue |
| 266 | } |
| 267 | |
| 268 | if c.Discoverer.Type() != typ || typ != "k8s" { |
| 269 | continue |
| 270 | } |
| 271 | |
| 272 | merged, err := mergeJSONArrays(c.Discoverer.Config, cfgJSON) |
| 273 | if err != nil { |
| 274 | return fmt.Errorf("legacy discover[%d]: merge %q configs: %w", i, typ, err) |
| 275 | } |
| 276 | c.Discoverer.Config = merged |
| 277 | converted = true |
| 278 | } |
| 279 | |
| 280 | if !converted { |
| 281 | return errors.New("legacy discover[] did not provide a usable discoverer config") |
| 282 | } |
| 283 | |
| 284 | return nil |
| 285 | } |
| 286 | |
| 287 | func mergeJSONArrays(aRaw, bRaw json.RawMessage) (json.RawMessage, error) { |
| 288 | var a []any |
| 289 | if len(aRaw) != 0 { |
| 290 | if err := json.Unmarshal(aRaw, &a); err != nil { |
| 291 | return nil, err |
| 292 | } |
| 293 | } |
| 294 | |
| 295 | var b []any |
| 296 | if len(bRaw) != 0 { |
| 297 | if err := json.Unmarshal(bRaw, &b); err != nil { |
| 298 | return nil, err |
| 299 | } |
| 300 | } |
| 301 | |
| 302 | out := append(a, b...) |
| 303 | bs, err := json.Marshal(out) |
| 304 | if err != nil { |
| 305 | return nil, err |
| 306 | } |
| 307 | return bs, nil |
| 308 | } |
| 309 | |
| 310 | func NewDiscovererPayload(typ string, cfg any) (DiscovererPayload, error) { |
| 311 | bs, err := json.Marshal(cfg) |
| 312 | if err != nil { |
| 313 | return DiscovererPayload{}, err |
| 314 | } |
| 315 | return DiscovererPayload{Kind: typ, Config: bs}, nil |
| 316 | } |
| 317 | |
| 318 | // MarshalYAML implements yaml.Marshaler. |
| 319 | // It only marshals the canonical format, not legacy fields. |
| 320 | func (c Config) MarshalYAML() (any, error) { |
| 321 | type output struct { |
| 322 | Disabled bool `yaml:"disabled,omitempty"` |
| 323 | Name string `yaml:"name,omitempty"` |
| 324 | Discoverer DiscovererPayload `yaml:"discoverer,omitempty"` |
| 325 | Services []ServiceRuleConfig `yaml:"services,omitempty"` |
| 326 | } |
| 327 | return output{ |
| 328 | Disabled: c.Disabled, |
| 329 | Name: c.Name, |
| 330 | Discoverer: c.Discoverer, |
| 331 | Services: c.Services, |
| 332 | }, nil |
| 333 | } |
| 334 | |
| 335 | // LegacyDiscoveryConfig is the old discover[] array item format. |
| 336 | // Kept for backwards compatibility during unmarshal. |
| 337 | type LegacyDiscoveryConfig struct { |
| 338 | Discoverer string `yaml:"discoverer"` |
| 339 | Config map[string]any `yaml:",inline"` |
| 340 | } |
| 341 | |
| 342 | type ServiceRuleConfig struct { |
| 343 | ID string `yaml:"id" json:"id"` // mandatory (for logging/diagnostics) |
| 344 | Match string `yaml:"match" json:"match"` // mandatory |
| 345 | ConfigTemplate string `yaml:"config_template,omitempty" json:"config_template,omitempty"` // optional (drop if empty) |
| 346 | } |
| 347 | |
| 348 | type ClassifyRuleConfig struct { |
| 349 | Name string `yaml:"name"` |
| 350 | Selector string `yaml:"selector"` // mandatory |
| 351 | Tags string `yaml:"tags"` // mandatory |
| 352 | Match []struct { |
| 353 | Tags string `yaml:"tags"` // mandatory |
| 354 | Expr string `yaml:"expr"` // mandatory |
| 355 | } `yaml:"match"` // mandatory, at least 1 |
| 356 | } |
| 357 | |
| 358 | type ComposeRuleConfig struct { |
| 359 | Name string `yaml:"name"` // optional |
| 360 | Selector string `yaml:"selector"` // mandatory |
| 361 | Config []struct { |
| 362 | Selector string `yaml:"selector"` // mandatory |
| 363 | Template string `yaml:"template"` // mandatory |
| 364 | } `yaml:"config"` // mandatory, at least 1 |
| 365 | } |
| 366 | |
| 367 | // ValidateConfig validates a pipeline configuration. |
| 368 | // Exported for use by dyncfg validation. |
| 369 | func ValidateConfig(cfg Config) error { |
| 370 | if cfg.Name == "" { |
| 371 | return errors.New("'name' not set") |
| 372 | } |
| 373 | if cfg.Discoverer.Empty() { |
| 374 | return errors.New("no discoverer configured") |
| 375 | } |
| 376 | if err := validateServicesConfig(cfg.Services); err != nil { |
| 377 | return fmt.Errorf("services rules: %v", err) |
| 378 | } |
| 379 | return nil |
| 380 | } |
| 381 | |
| 382 | func validateServicesConfig(rules []ServiceRuleConfig) error { |
| 383 | if len(rules) == 0 { |
| 384 | return errors.New("empty config, need at least 1 service rule") |
| 385 | } |
| 386 | for i, r := range rules { |
| 387 | i++ |
| 388 | if r.ID == "" { |
| 389 | return fmt.Errorf("'service[%d]->id' not set", i) |
| 390 | } |
| 391 | if r.Match == "" { |
| 392 | return fmt.Errorf("'service[%s][%d]->match' not set", r.ID, i) |
| 393 | } |
| 394 | // config_template is optional |
| 395 | } |
| 396 | return nil |
| 397 | } |
| 398 | |
| 399 | func ConvertOldToServices(cls []ClassifyRuleConfig, cmp []ComposeRuleConfig) ([]ServiceRuleConfig, error) { |
| 400 | var out []ServiceRuleConfig |
| 401 | |
| 402 | // Build quick lookups for tag -> list of match exprs that add this tag. |
| 403 | tagToExprs := map[string][]string{} |
| 404 | for _, r := range cls { |
| 405 | for _, m := range r.Match { |
| 406 | // split tags line into tokens: |
| 407 | tags, _ := model.ParseTags(m.Tags) // reuse existing parser if accessible |
| 408 | for tag := range tags { |
| 409 | if strings.HasPrefix(tag, "-") { // ignore deletions |
| 410 | continue |
| 411 | } |
| 412 | tagToExprs[tag] = append(tagToExprs[tag], m.Expr) |
| 413 | } |
| 414 | } |
| 415 | // also include rule-level tags |
| 416 | rtags, _ := model.ParseTags(r.Tags) |
| 417 | for tag := range rtags { |
| 418 | if strings.HasPrefix(tag, "-") { |
| 419 | continue |
| 420 | } |
| 421 | // no expr here; this is too generic to build a service rule from. |
| 422 | } |
| 423 | } |
| 424 | |
| 425 | // For each compose rule config entry, create services for its selector tags. |
| 426 | for _, r := range cmp { |
| 427 | for _, c := range r.Config { |
| 428 | sel := strings.TrimSpace(c.Selector) |
| 429 | exprs := tagToExprs[sel] |
| 430 | for i, expr := range exprs { |
| 431 | id := sel |
| 432 | if i > 0 { |
| 433 | id = fmt.Sprintf("%s_%d", sel, i+1) |
| 434 | } |
| 435 | out = append(out, ServiceRuleConfig{ |
| 436 | ID: id, |
| 437 | Match: expr, |
| 438 | ConfigTemplate: c.Template, |
| 439 | }) |
| 440 | } |
| 441 | } |
| 442 | } |
| 443 | |
| 444 | return out, nil |
| 445 | } |