(instance instance, err error)
| 149 | } |
| 150 | |
| 151 | func (b *batchTracker) record(instance instance, err error) { |
| 152 | // If we reach the required number of successful puts on this sample, then decrement the |
| 153 | // number of pending samples by one. |
| 154 | // |
| 155 | // The use of atomic increments here is needed as: |
| 156 | // * rpcsPending and rpcsPending guarantees only a single sendSamples goroutine will write to either channel |
| 157 | // * succeeded, failed4xx, failed5xx and remaining guarantees that the "return decision" is made atomically |
| 158 | // avoiding race condition |
| 159 | sampleTrackers := instance.itemTrackers |
| 160 | for i := range sampleTrackers { |
| 161 | if err != nil { |
| 162 | // Track the number of errors by error family, and if it exceeds maxFailures |
| 163 | // shortcut the waiting rpc. |
| 164 | errCount := instance.itemTrackers[i].recordError(err) |
| 165 | // We should return an error if we reach the maxFailure (quorum) on a given error family OR |
| 166 | // we dont have any remaining ingesters to try |
| 167 | // Ex: 2xx, 4xx, 5xx -> return 4xx |
| 168 | // Ex: 2xx, 5xx, 4xx -> return 4xx |
| 169 | // Ex: 4xx, 4xx, _ -> return 4xx |
| 170 | // Ex: 5xx, _, 5xx -> return 5xx |
| 171 | if errCount > int32(sampleTrackers[i].maxFailures) { |
| 172 | if b.rpcsFailed.Inc() == 1 { |
| 173 | b.err <- httpgrpcutil.WrapHTTPGrpcError( |
| 174 | sampleTrackers[i].getError(), "maxFailure (quorum) on a given error family, addr=%s state=%s zone=%s", |
| 175 | instance.desc.Addr, instance.desc.State, instance.desc.Zone) |
| 176 | } |
| 177 | continue |
| 178 | } |
| 179 | if sampleTrackers[i].remaining.Dec() == 0 { |
| 180 | if b.rpcsFailed.Inc() == 1 { |
| 181 | b.err <- httpgrpcutil.WrapHTTPGrpcError( |
| 182 | sampleTrackers[i].getError(), "not enough remaining instances to try, addr=%s state=%s zone=%s", |
| 183 | instance.desc.Addr, instance.desc.State, instance.desc.Zone) |
| 184 | } |
| 185 | continue |
| 186 | } |
| 187 | } else { |
| 188 | // If we successfully push all samples to min success instances, |
| 189 | // wake up the waiting rpc so it can return early. |
| 190 | if sampleTrackers[i].succeeded.Inc() >= int32(sampleTrackers[i].minSuccess) { |
| 191 | if b.rpcsPending.Dec() == 0 { |
| 192 | b.done <- struct{}{} |
| 193 | } |
| 194 | continue |
| 195 | } |
| 196 | |
| 197 | // If we succeeded to call this particular ingester but we dont have any remaining ingesters to try |
| 198 | // and we did not succeeded calling `minSuccess` ingesters we need to return the last error |
| 199 | // Ex: 4xx, 5xx, 2xx |
| 200 | if sampleTrackers[i].remaining.Dec() == 0 { |
| 201 | if b.rpcsFailed.Inc() == 1 { |
| 202 | b.err <- httpgrpcutil.WrapHTTPGrpcError( |
| 203 | sampleTrackers[i].getError(), "not enough remaining instances to try, addr=%s state=%s zone=%s", |
| 204 | instance.desc.Addr, instance.desc.State, instance.desc.Zone) |
| 205 | } |
| 206 | } |
| 207 | } |
| 208 | } |
no test coverage detected