(address string)
| 199 | } |
| 200 | |
| 201 | func (w *querierWorker) AddressAdded(address string) { |
| 202 | ctx := w.ServiceContext() |
| 203 | if ctx == nil || ctx.Err() != nil { |
| 204 | return |
| 205 | } |
| 206 | |
| 207 | w.mu.Lock() |
| 208 | defer w.mu.Unlock() |
| 209 | |
| 210 | if m := w.managers[address]; m != nil { |
| 211 | return |
| 212 | } |
| 213 | |
| 214 | level.Info(w.log).Log("msg", "adding connection", "addr", address) |
| 215 | conn, err := w.connect(address) |
| 216 | if err != nil { |
| 217 | level.Error(w.log).Log("msg", "error connecting", "addr", address, "err", err) |
| 218 | return |
| 219 | } |
| 220 | |
| 221 | w.managers[address] = newProcessorManager(ctx, w.processor, conn, address) |
| 222 | // Called with lock. |
| 223 | w.resetConcurrency() |
| 224 | } |
| 225 | |
| 226 | func (w *querierWorker) AddressRemoved(address string) { |
| 227 | level.Info(w.log).Log("msg", "removing connection", "addr", address) |
nothing calls this directly
no test coverage detected