diff options
Diffstat (limited to 'configserver-client/src')
4 files changed, 156 insertions, 77 deletions
diff --git a/configserver-client/src/main/java/ai/vespa/hosted/client/AbstractConfigServerClient.java b/configserver-client/src/main/java/ai/vespa/hosted/client/AbstractConfigServerClient.java index d467402f79f..2e5e445d63a 100644 --- a/configserver-client/src/main/java/ai/vespa/hosted/client/AbstractConfigServerClient.java +++ b/configserver-client/src/main/java/ai/vespa/hosted/client/AbstractConfigServerClient.java @@ -9,11 +9,12 @@ import org.apache.hc.core5.http.ClassicHttpResponse; import org.apache.hc.core5.http.ContentType; import org.apache.hc.core5.http.HttpEntity; import org.apache.hc.core5.http.Method; +import org.apache.hc.core5.http.ParseException; +import org.apache.hc.core5.http.io.entity.EntityUtils; import org.apache.hc.core5.http.io.entity.HttpEntities; import org.apache.hc.core5.net.URIBuilder; import java.io.IOException; -import java.io.InputStream; import java.io.UncheckedIOException; import java.net.URI; import java.net.URISyntaxException; @@ -21,7 +22,6 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit; -import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; @@ -104,7 +104,7 @@ public abstract class AbstractConfigServerClient implements ConfigServerClient { } @Override - public RequestBuilder send(HostStrategy hosts, Method method) { + public ConfigServerClient.RequestBuilder send(HostStrategy hosts, Method method) { return new RequestBuilder(hosts, method); } @@ -114,10 +114,11 @@ public abstract class AbstractConfigServerClient implements ConfigServerClient { private final Method method; private final HostStrategy hosts; private final URIBuilder uriBuilder = new URIBuilder(); + private final List<String> pathSegments = new ArrayList<>(); private HttpEntity entity; private RequestConfig config = ConfigServerClient.defaultRequestConfig; - private BiConsumer<ClassicHttpResponse, ClassicHttpRequest> handler = ConfigServerClient::throwOnError; - private Consumer<IOException> catcher = ConfigServerClient::retryAll; + private ResponseVerifier verifier = ConfigServerClient.throwOnError; + private Consumer<IOException> catcher = ConfigServerClient.retryAll; private RequestBuilder(HostStrategy hosts, Method method) { if ( ! hosts.iterator().hasNext()) @@ -129,7 +130,7 @@ public abstract class AbstractConfigServerClient implements ConfigServerClient { @Override public RequestBuilder at(List<String> pathSegments) { - uriBuilder.setPathSegments(requireNonNull(pathSegments)); + this.pathSegments.addAll(pathSegments); return this; } @@ -145,12 +146,23 @@ public abstract class AbstractConfigServerClient implements ConfigServerClient { } @Override + public ConfigServerClient.RequestBuilder emptyParameters(List<String> keys) { + for (String key : keys) + uriBuilder.setParameter(key, null); + + return this; + } + + @Override public RequestBuilder parameters(List<String> pairs) { if (pairs.size() % 2 != 0) throw new IllegalArgumentException("Must supply parameter key/values in pairs"); - for (int i = 0; i < pairs.size(); ) - uriBuilder.setParameter(pairs.get(i++), pairs.get(i++)); + for (int i = 0; i < pairs.size(); ) { + String key = pairs.get(i++), value = pairs.get(i++); + if (value != null) + uriBuilder.setParameter(key, value); + } return this; } @@ -175,56 +187,54 @@ public abstract class AbstractConfigServerClient implements ConfigServerClient { } @Override - public RequestBuilder handling(BiConsumer<ClassicHttpResponse, ClassicHttpRequest> handler) { - this.handler = requireNonNull(handler); + public RequestBuilder throwing(ResponseVerifier verifier) { + this.verifier = requireNonNull(verifier); return this; } @Override - public <T> T read(Function<byte[], T> mapper) { - return mapIfSuccess(input -> { - try (input) { - return mapper.apply(input.readAllBytes()); + public String read() { + return handle((response, __) -> { + try (response) { + return response.getEntity() == null ? "" : EntityUtils.toString(response.getEntity()); } - catch (IOException e) { - throw new RetryException(e); + catch (ParseException e) { + throw new IllegalStateException(e); // This isn't actually thrown by apache >_< + } + }); + } + + @Override + public <T> T read(Function<byte[], T> mapper) { + return handle((response, __) -> { + try (response) { + return mapper.apply(response.getEntity() == null ? new byte[0] : EntityUtils.toByteArray(response.getEntity())); } }); } @Override public void discard() throws UncheckedIOException, ResponseException { - mapIfSuccess(input -> { - try (input) { + handle((response, __) -> { + try (response) { return null; } - catch (IOException e) { - throw new RetryException(e); - } }); } @Override - public InputStream stream() throws UncheckedIOException, ResponseException { - return mapIfSuccess(input -> input); + public HttpInputStream stream() throws UncheckedIOException, ResponseException { + return handle((response, __) -> new HttpInputStream(response)); } - /** Returns the mapped body, if successful, retrying any IOException. The caller must close the body stream. */ - private <T> T mapIfSuccess(Function<InputStream, T> mapper) { + @Override + public <T> T handle(ResponseHandler<T> handler) { + uriBuilder.setPathSegments(pathSegments); return execute(this, (response, request) -> { try { - handler.accept(response, request); // This throws on unacceptable responses. - - InputStream body = response.getEntity() != null ? response.getEntity().getContent() - : InputStream.nullInputStream(); - return mapper.apply(new ForwardingInputStream(body) { - @Override - public void close() throws IOException { - super.close(); - response.close(); - } - }); + verifier.verify(response, request); // This throws on unacceptable responses. + return handler.handle(response, request); } catch (IOException | RuntimeException | Error e) { try { @@ -247,6 +257,7 @@ public abstract class AbstractConfigServerClient implements ConfigServerClient { } + @SuppressWarnings("unchecked") private static <T extends Throwable> void sneakyThrow(Throwable t) throws T { throw (T) t; diff --git a/configserver-client/src/main/java/ai/vespa/hosted/client/ConfigServerClient.java b/configserver-client/src/main/java/ai/vespa/hosted/client/ConfigServerClient.java index c0f72378c1b..c92acd7cd0b 100644 --- a/configserver-client/src/main/java/ai/vespa/hosted/client/ConfigServerClient.java +++ b/configserver-client/src/main/java/ai/vespa/hosted/client/ConfigServerClient.java @@ -7,10 +7,10 @@ import org.apache.hc.core5.http.ClassicHttpResponse; import org.apache.hc.core5.http.HttpEntity; import org.apache.hc.core5.http.HttpStatus; import org.apache.hc.core5.http.Method; -import org.apache.hc.core5.http.ParseException; import org.apache.hc.core5.http.io.entity.EntityUtils; import org.apache.hc.core5.util.Timeout; +import java.io.Closeable; import java.io.IOException; import java.io.InputStream; import java.io.UncheckedIOException; @@ -20,18 +20,18 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; -import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.IntStream; +import static java.nio.charset.StandardCharsets.UTF_8; import static java.util.Objects.requireNonNull; import static java.util.stream.Collectors.toUnmodifiableList; /** * @author jonmv */ -public interface ConfigServerClient extends AutoCloseable { +public interface ConfigServerClient extends Closeable { RequestConfig defaultRequestConfig = RequestConfig.custom() .setConnectionRequestTimeout(Timeout.ofSeconds(5)) @@ -40,20 +40,12 @@ public interface ConfigServerClient extends AutoCloseable { .build(); /** Wraps with a {@link RetryException} and rethrows. */ - static void retryAll(IOException e) { + Consumer<IOException> retryAll = (e) -> { throw new RetryException(e); - } + }; /** Throws a a {@link RetryException} if {@code statusCode == 503}, or a {@link ResponseException} unless {@code 200 <= statusCode < 300}. */ - static void throwOnError(ClassicHttpResponse response, ClassicHttpRequest request) { - if (response.getCode() < HttpStatus.SC_OK || response.getCode() >= HttpStatus.SC_REDIRECTION) { - ResponseException e = ResponseException.of(response, request); - if (response.getCode() == HttpStatus.SC_SERVICE_UNAVAILABLE) - throw new RetryException(e); - - throw e; - } - } + ResponseVerifier throwOnError = new DefaultResponseVerifier() { }; /** Reads the response body, throwing an {@link UncheckedIOException} if this fails, or {@code null} if there is none. */ static byte[] getBytes(ClassicHttpResponse response) { @@ -71,10 +63,10 @@ public interface ConfigServerClient extends AutoCloseable { /** Builder for a request against a given set of hosts, using this config server client. */ interface RequestBuilder { - /** Sets the request path. */ + /** Appends to the request path. */ default RequestBuilder at(String... pathSegments) { return at(List.of(pathSegments)); } - /** Sets the request path. */ + /** Appends to the request path. */ RequestBuilder at(List<String> pathSegments); /** Sets the request body as UTF-8 application/json. */ @@ -83,12 +75,20 @@ public interface ConfigServerClient extends AutoCloseable { /** Sets the request body. */ RequestBuilder body(HttpEntity entity); - /** Sets the parameter key/values for the request. Number of arguments must be even. */ + /** Sets query parameters without a value, like {@code ?debug&recursive}. */ + default RequestBuilder emptyParameters(String... keys) { + return emptyParameters(Arrays.asList(keys)); + } + + /** Sets query parameters without a value, like {@code ?debug&recursive}. */ + RequestBuilder emptyParameters(List<String> keys); + + /** Sets the parameter key/values for the request. Number of arguments must be even. Null values are omitted. */ default RequestBuilder parameters(String... pairs) { return parameters(Arrays.asList(pairs)); } - /** Sets the parameter key/values for the request. Number of arguments must be even. */ + /** Sets the parameter key/values for the request. Number of arguments must be even. Null values are omitted. */ RequestBuilder parameters(List<String> pairs); /** Overrides the default socket read timeout of the request. {@code Duration.ZERO} gives infinite timeout. */ @@ -109,7 +109,10 @@ public interface ConfigServerClient extends AutoCloseable { * Sets the (error) response handler for this request. The default is {@link #throwOnError}. * When the handler returns normally, the response is treated as a success, and passed on to a response mapper. */ - RequestBuilder handling(BiConsumer<ClassicHttpResponse, ClassicHttpRequest> handler); + RequestBuilder throwing(ResponseVerifier handler); + + /** Reads the response as a {@link String}, or throws if unsuccessful. */ + String read(); /** Reads and maps the response, or throws if unsuccessful. */ <T> T read(Function<byte[], T> mapper); @@ -118,10 +121,88 @@ public interface ConfigServerClient extends AutoCloseable { void discard(); /** Returns the raw response input stream, or throws if unsuccessful. The caller must close the returned stream. */ - InputStream stream(); + HttpInputStream stream(); + + /** Uses the response and request, if successful, to generate a mapped response. */ + <T> T handle(ResponseHandler<T> handler); + + } + + + class HttpInputStream extends ForwardingInputStream { + + private final ClassicHttpResponse response; + + protected HttpInputStream(ClassicHttpResponse response) throws IOException { + super(response.getEntity() != null ? response.getEntity().getContent() + : InputStream.nullInputStream()); + this.response = response; + } + + public int statusCode() { return response.getCode(); } + + public String contentType() { return response.getEntity().getContentType(); } + + @Override + public void close() throws IOException { + super.close(); + response.close(); + } + + } + + + /** Reads a successful response and request to compute a result. */ + @FunctionalInterface + interface ResponseHandler<T> { + + /** Called with successful responses, as per {@link ResponseVerifier}. The caller must close the response. */ + T handle(ClassicHttpResponse response, ClassicHttpRequest request) throws IOException; } + + /** Verifies a response, throwing on error responses, possibly indicating retries. */ + @FunctionalInterface + interface ResponseVerifier { + + /** Whether this status code means the response is an error response. */ + default boolean isError(int statusCode) { + return statusCode < HttpStatus.SC_OK || HttpStatus.SC_REDIRECTION <= statusCode; + } + + /** Whether this status code means we should retry. Has no effect if this is not also an error. */ + default boolean shouldRetry(int statusCode) { + return statusCode == HttpStatus.SC_SERVICE_UNAVAILABLE; + } + + /** Verifies the given response, consuming it and throwing if it is an error; or leaving it otherwise. */ + default void verify(ClassicHttpResponse response, ClassicHttpRequest request) throws IOException { + if (isError(response.getCode())) { + try (response) { + byte[] body = response.getEntity() == null ? new byte[0] : EntityUtils.toByteArray(response.getEntity()); + RuntimeException exception = toException(response.getCode(), body, request); + throw shouldRetry(response.getCode()) ? new RetryException(exception) : exception; + } + } + } + + /** Throws the appropriate exception, for the given status code and body. */ + RuntimeException toException(int statusCode, byte[] body, ClassicHttpRequest request); + + } + + + interface DefaultResponseVerifier extends ResponseVerifier { + + @Override + default RuntimeException toException(int statusCode, byte[] body, ClassicHttpRequest request) { + return new ResponseException(request + " failed with status " + statusCode + " and body '" + new String(body, UTF_8) + "'"); + } + + } + + /** What host(s) to try for a request, in what order. A host may be specified multiple times, for retries. */ @FunctionalInterface interface HostStrategy extends Iterable<URI> { @@ -147,6 +228,7 @@ public interface ConfigServerClient extends AutoCloseable { } + /** Exception wrapper that signals retries should be attempted. */ final class RetryException extends RuntimeException { @@ -160,25 +242,12 @@ public interface ConfigServerClient extends AutoCloseable { } + /** An exception due to server error, a bad request, or similar, which resulted in a non-OK HTTP response. */ class ResponseException extends RuntimeException { - public ResponseException(String message, Throwable cause) { - super(message, cause); - } - - public static ResponseException of(ClassicHttpResponse response, ClassicHttpRequest request) { - String detail; - Throwable thrown = null; - try { - detail = request.getEntity() == null ? " and no body" - : " and body '" + EntityUtils.toString(request.getEntity()) + "'"; - } - catch (IOException | ParseException e) { - detail = ". Reading body failed"; - thrown = e; - } - return new ResponseException(request + " failed with status " + response.getCode() + detail, thrown); + public ResponseException(String message) { + super(message); } } diff --git a/configserver-client/src/main/java/ai/vespa/hosted/client/HttpConfigServerClient.java b/configserver-client/src/main/java/ai/vespa/hosted/client/HttpConfigServerClient.java index c5b07eceaf5..e00489f0c64 100644 --- a/configserver-client/src/main/java/ai/vespa/hosted/client/HttpConfigServerClient.java +++ b/configserver-client/src/main/java/ai/vespa/hosted/client/HttpConfigServerClient.java @@ -54,7 +54,6 @@ public class HttpConfigServerClient extends AbstractConfigServerClient { false) .disableAutomaticRetries() .setUserAgent(userAgent) - .setDefaultRequestConfig(defaultRequestConfig) .build(); } diff --git a/configserver-client/src/test/java/ai/vespa/hosted/client/HttpConfigServerClientTest.java b/configserver-client/src/test/java/ai/vespa/hosted/client/HttpConfigServerClientTest.java index 3b46501a4bf..b2f17f43f5c 100644 --- a/configserver-client/src/test/java/ai/vespa/hosted/client/HttpConfigServerClientTest.java +++ b/configserver-client/src/test/java/ai/vespa/hosted/client/HttpConfigServerClientTest.java @@ -70,14 +70,15 @@ class HttpConfigServerClientTest { server.resetRequests(); // Successful attempt returns. - server.stubFor(get("/root/boot")) + server.stubFor(get("/root/boot/toot")) .setResponse(okJson("{}").build()); assertEquals("{}", client.send(HostStrategy.repeating(URI.create("http://localhost:" + server.port()), 10), Method.GET) .at("root", "boot") + .at("toot") .read(String::new)); - server.verify(1, getRequestedFor(urlEqualTo("/root/boot"))); + server.verify(1, getRequestedFor(urlEqualTo("/root/boot/toot"))); server.verify(1, anyRequestedFor(anyUrl())); server.resetRequests(); @@ -88,10 +89,9 @@ class HttpConfigServerClientTest { () -> client.send(HostStrategy.repeating(URI.create("http://localhost:" + server.port()), 10), Method.GET) .read(String::new)); - assertEquals("GET / failed with status 409 and no body", thrown.getMessage()); + assertEquals("GET / failed with status 409 and body 'hi'", thrown.getMessage()); server.verify(1, getRequestedFor(urlEqualTo("/"))); server.verify(1, anyRequestedFor(anyUrl())); - } } |