| 107 | } |
| 108 | |
| 109 | func (cfg *Config) Validate() error { |
| 110 | if cfg.ConsumeFromKafka { |
| 111 | if err := cfg.Ingest.Validate(); err != nil { |
| 112 | return err |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | if cfg.IngestConcurrency == 0 { |
| 117 | return errors.New("ingest concurrency must be greater than zero") |
| 118 | } |
| 119 | |
| 120 | if err := cfg.Processor.Validate(); err != nil { |
| 121 | return err |
| 122 | } |
| 123 | |
| 124 | if err := cfg.Storage.Validate(); err != nil { |
| 125 | return err |
| 126 | } |
| 127 | |
| 128 | if !slices.Contains(validCodecs, cfg.Codec) { |
| 129 | return fmt.Errorf("invalid codec: %s, valid choices are %s", cfg.Codec, validCodecs) |
| 130 | } |
| 131 | |
| 132 | switch cfg.RingMode { |
| 133 | case RingModePartition, RingModeGenerator: |
| 134 | default: |
| 135 | return fmt.Errorf("invalid ring mode: %s, valid values are %s and %s", cfg.RingMode, RingModePartition, RingModeGenerator) |
| 136 | } |
| 137 | |
| 138 | if cfg.ConsumeFromKafka && cfg.RingMode == RingModeGenerator && cfg.Ingest.Kafka.ConsumerGroup == "" { |
| 139 | return errors.New("ingest.kafka.consumer_group must be configured when metrics-generator ring mode is generator") |
| 140 | } |
| 141 | |
| 142 | switch cfg.LimiterType { |
| 143 | case LimiterTypeSeries, LimiterTypeEntity: |
| 144 | default: |
| 145 | return fmt.Errorf("invalid limiter type: %s, valid values are %s and %s", cfg.LimiterType, LimiterTypeSeries, LimiterTypeEntity) |
| 146 | } |
| 147 | |
| 148 | return nil |
| 149 | } |
| 150 | |
| 151 | type ProcessorConfig struct { |
| 152 | ServiceGraphs servicegraphs.Config `yaml:"service_graphs"` |