| 144 | } |
| 145 | |
| 146 | func newQuerierWorkerWithProcessor(cfg Config, log log.Logger, processor processor, address string, servs []services.Service) (*querierWorker, error) { |
| 147 | f := &querierWorker{ |
| 148 | cfg: cfg, |
| 149 | log: log, |
| 150 | managers: map[string]*processorManager{}, |
| 151 | processor: processor, |
| 152 | } |
| 153 | |
| 154 | // Empty address is only used in tests, where individual targets are added manually. |
| 155 | if address != "" { |
| 156 | w, err := util.NewDNSWatcher(address, cfg.DNSLookupPeriod, f) |
| 157 | if err != nil { |
| 158 | return nil, err |
| 159 | } |
| 160 | |
| 161 | servs = append(servs, w) |
| 162 | } |
| 163 | |
| 164 | if len(servs) > 0 { |
| 165 | subservices, err := services.NewManager(servs...) |
| 166 | if err != nil { |
| 167 | return nil, errors.Wrap(err, "querier worker subservices") |
| 168 | } |
| 169 | |
| 170 | f.subservices = subservices |
| 171 | } |
| 172 | |
| 173 | f.BasicService = services.NewIdleService(f.starting, f.stopping) |
| 174 | return f, nil |
| 175 | } |
| 176 | |
| 177 | func (w *querierWorker) starting(ctx context.Context) error { |
| 178 | if w.subservices == nil { |