|
| 1 | +package frontend |
| 2 | + |
| 3 | +import ( |
| 4 | + "flag" |
| 5 | + "math/rand" |
| 6 | + "net/http" |
| 7 | + "sync" |
| 8 | + |
| 9 | + "github.com/go-kit/kit/log" |
| 10 | + "github.com/go-kit/kit/log/level" |
| 11 | + "github.com/weaveworks/common/httpgrpc" |
| 12 | + "github.com/weaveworks/common/httpgrpc/server" |
| 13 | + "github.com/weaveworks/common/user" |
| 14 | +) |
| 15 | + |
| 16 | +var ( |
| 17 | + errServerClosing = httpgrpc.Errorf(http.StatusTeapot, "server closing down") |
| 18 | + errTooManyRequest = httpgrpc.Errorf(http.StatusTooManyRequests, "too many outstanding requests") |
| 19 | + errCanceled = httpgrpc.Errorf(http.StatusInternalServerError, "context cancelled") |
| 20 | +) |
| 21 | + |
| 22 | +// Config for a Frontend. |
| 23 | +type Config struct { |
| 24 | + MaxOutstandingPerTenant int |
| 25 | + MaxRetries int |
| 26 | +} |
| 27 | + |
| 28 | +// RegisterFlags adds the flags required to config this to the given FlagSet. |
| 29 | +func (cfg *Config) RegisterFlags(f *flag.FlagSet) { |
| 30 | + f.IntVar(&cfg.MaxOutstandingPerTenant, "querier.max-outstanding-requests-per-tenant", 100, "") |
| 31 | + f.IntVar(&cfg.MaxRetries, "querier.max-retries-per-request", 5, "") |
| 32 | +} |
| 33 | + |
| 34 | +// Frontend queues HTTP requests, dispatches them to backends, and handles retries |
| 35 | +// for requests which failed. |
| 36 | +type Frontend struct { |
| 37 | + cfg Config |
| 38 | + log log.Logger |
| 39 | + |
| 40 | + mtx sync.Mutex |
| 41 | + cond *sync.Cond |
| 42 | + closed bool |
| 43 | + queues map[string]chan *request |
| 44 | +} |
| 45 | + |
| 46 | +type request struct { |
| 47 | + request *httpgrpc.HTTPRequest |
| 48 | + err chan error |
| 49 | + response chan *httpgrpc.HTTPResponse |
| 50 | +} |
| 51 | + |
| 52 | +// New creates a new frontend. |
| 53 | +func New(cfg Config, log log.Logger) (*Frontend, error) { |
| 54 | + f := &Frontend{ |
| 55 | + cfg: cfg, |
| 56 | + log: log, |
| 57 | + queues: map[string]chan *request{}, |
| 58 | + } |
| 59 | + f.cond = sync.NewCond(&f.mtx) |
| 60 | + return f, nil |
| 61 | +} |
| 62 | + |
| 63 | +// Close stops new requests and errors out any pending requests. |
| 64 | +func (f *Frontend) Close() { |
| 65 | + f.mtx.Lock() |
| 66 | + defer f.mtx.Unlock() |
| 67 | + |
| 68 | + f.closed = true |
| 69 | + f.cond.Broadcast() |
| 70 | + |
| 71 | + for _, queue := range f.queues { |
| 72 | + close(queue) |
| 73 | + for request := range queue { |
| 74 | + request.err <- errServerClosing |
| 75 | + } |
| 76 | + } |
| 77 | +} |
| 78 | + |
| 79 | +// ServeHTTP serves HTTP requests. |
| 80 | +func (f *Frontend) ServeHTTP(w http.ResponseWriter, r *http.Request) { |
| 81 | + if err := f.serveHTTP(w, r); err != nil { |
| 82 | + server.WriteError(w, err) |
| 83 | + } |
| 84 | +} |
| 85 | + |
| 86 | +func (f *Frontend) serveHTTP(w http.ResponseWriter, r *http.Request) error { |
| 87 | + ctx := r.Context() |
| 88 | + userID, err := user.ExtractOrgID(ctx) |
| 89 | + if err != nil { |
| 90 | + return err |
| 91 | + } |
| 92 | + |
| 93 | + req, err := server.HTTPRequest(r) |
| 94 | + if err != nil { |
| 95 | + return err |
| 96 | + } |
| 97 | + |
| 98 | + request := &request{ |
| 99 | + request: req, |
| 100 | + // Buffer of 1 to ensure response can be written even if client has gone away. |
| 101 | + err: make(chan error, 1), |
| 102 | + response: make(chan *httpgrpc.HTTPResponse, 1), |
| 103 | + } |
| 104 | + |
| 105 | + var lastErr error |
| 106 | + for retries := 0; retries < f.cfg.MaxRetries; retries++ { |
| 107 | + if err := f.queueRequest(userID, request); err != nil { |
| 108 | + return err |
| 109 | + } |
| 110 | + |
| 111 | + var resp *httpgrpc.HTTPResponse |
| 112 | + select { |
| 113 | + case <-ctx.Done(): |
| 114 | + // TODO propagate cancellation. |
| 115 | + //request.Cancel() |
| 116 | + return errCanceled |
| 117 | + |
| 118 | + case resp = <-request.response: |
| 119 | + case lastErr = <-request.err: |
| 120 | + level.Error(f.log).Log("msg", "error processing request", "try", retries, "err", lastErr) |
| 121 | + resp, _ = httpgrpc.HTTPResponseFromError(lastErr) |
| 122 | + } |
| 123 | + |
| 124 | + // Only fail is we get a valid HTTP non-500; otherwise retry. |
| 125 | + if resp != nil && resp.Code/100 != 5 { |
| 126 | + server.WriteResponse(w, resp) |
| 127 | + return nil |
| 128 | + } |
| 129 | + } |
| 130 | + |
| 131 | + return lastErr |
| 132 | +} |
| 133 | + |
| 134 | +// Process allows backends to pull requests from the frontend. |
| 135 | +func (f *Frontend) Process(server Frontend_ProcessServer) error { |
| 136 | + for { |
| 137 | + request := f.getNextRequest() |
| 138 | + if request == nil { |
| 139 | + // Occurs when server is shutting down. |
| 140 | + return nil |
| 141 | + } |
| 142 | + |
| 143 | + if err := server.Send(&ProcessRequest{ |
| 144 | + HttpRequest: request.request, |
| 145 | + }); err != nil { |
| 146 | + request.err <- err |
| 147 | + return err |
| 148 | + } |
| 149 | + |
| 150 | + response, err := server.Recv() |
| 151 | + if err != nil { |
| 152 | + request.err <- err |
| 153 | + return err |
| 154 | + } |
| 155 | + |
| 156 | + request.response <- response.HttpResponse |
| 157 | + } |
| 158 | +} |
| 159 | + |
| 160 | +func (f *Frontend) queueRequest(userID string, req *request) error { |
| 161 | + f.mtx.Lock() |
| 162 | + defer f.mtx.Unlock() |
| 163 | + |
| 164 | + if f.closed { |
| 165 | + return errServerClosing |
| 166 | + } |
| 167 | + |
| 168 | + queue, ok := f.queues[userID] |
| 169 | + if !ok { |
| 170 | + queue = make(chan *request, f.cfg.MaxOutstandingPerTenant) |
| 171 | + f.queues[userID] = queue |
| 172 | + } |
| 173 | + |
| 174 | + select { |
| 175 | + case queue <- req: |
| 176 | + f.cond.Signal() |
| 177 | + return nil |
| 178 | + default: |
| 179 | + return errTooManyRequest |
| 180 | + } |
| 181 | +} |
| 182 | + |
| 183 | +// getQueue picks a random queue and takes the next request off of it, so we |
| 184 | +// faily process users queries. Will block if there are no requests. |
| 185 | +func (f *Frontend) getNextRequest() *request { |
| 186 | + f.mtx.Lock() |
| 187 | + defer f.mtx.Unlock() |
| 188 | + |
| 189 | + for len(f.queues) == 0 && !f.closed { |
| 190 | + f.cond.Wait() |
| 191 | + } |
| 192 | + |
| 193 | + if f.closed { |
| 194 | + return nil |
| 195 | + } |
| 196 | + |
| 197 | + i, n := 0, rand.Intn(len(f.queues)) |
| 198 | + for userID, queue := range f.queues { |
| 199 | + if i < n { |
| 200 | + i++ |
| 201 | + continue |
| 202 | + } |
| 203 | + |
| 204 | + request := <-queue |
| 205 | + if len(queue) == 0 { |
| 206 | + delete(f.queues, userID) |
| 207 | + } |
| 208 | + return request |
| 209 | + } |
| 210 | + |
| 211 | + panic("should never happen") |
| 212 | +} |
0 commit comments