feat(metrics-influxdb): add support for Influx 1.x, including use of an authorization token
This commit also adds batching support via `BatchedMetricsConsumer`.
Note `InfluxDb09Metrics` is now simply named `InfluxDbMetrics`.
Improved option parsing has been added.
Closes
MODIFY
gateway/engine/influxdb/src/main/java/io/apiman/gateway/engine/influxdb/InfluxDb09Driver.java
gateway/engine/influxdb/src/main/java/io/apiman/gateway/engine/influxdb/InfluxDb09Driver.java
*/
package io.apiman.gateway.engine.influxdb;
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.URLEncoder;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import io.apiman.gateway.engine.async.AsyncResultImpl;
import io.apiman.gateway.engine.async.IAsyncHandler;
-import io.apiman.gateway.engine.async.IAsyncResult;
-import io.apiman.gateway.engine.async.IAsyncResultHandler;
import io.apiman.gateway.engine.components.IHttpClientComponent;
import io.apiman.gateway.engine.components.http.HttpMethod;
import io.apiman.gateway.engine.components.http.IHttpClientRequest;
-import io.apiman.gateway.engine.components.http.IHttpClientResponse;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
/**
* A simple async HTTP impl of the influxdb driver. Contains only the subset of functionality we need.
*
- * @author Marc Savy <msavy@redhat.com>
+ * @author Marc Savy {@literal <marc@blackparrotlabs.io>}
*/
public class InfluxDb09Driver {
- private IHttpClientComponent httpClient;
-
- private StringBuilder writeUrl;
- private StringBuilder queryUrl;
-
- private String username;
- private String password;
- private String database;
- private String retentionPolicy;
- private String timePrecision;
-
- private static ObjectMapper objectMapper = new ObjectMapper();
+ private final IHttpClientComponent httpClient;
+ private final StringBuilder writeUrl;
+ private final String username;
+ private final String password;
+ private final String token;
+ private final String database;
+ private final String retentionPolicy;
+ private final String timePrecision;
@SuppressWarnings("nls")
- public InfluxDb09Driver(IHttpClientComponent httpClient, String endpoint, String username,
- String password, String database, String retentionPolicy, String timePrecision) {
+ public InfluxDb09Driver(IHttpClientComponent httpClient, InfluxDbMetricsOptionsParser options, String timePrecision) {
this.httpClient = httpClient;
- this.username = username;
- this.password = password;
- this.database = database;
- this.retentionPolicy = retentionPolicy;
+ this.username = options.getUsername();
+ this.password = options.getPassword();
+ this.token = options.getToken();
+ this.database = options.getDbName();
+ this.retentionPolicy = options.getRetentionPolicy();
this.timePrecision = timePrecision;
+ String endpoint = options.getInfluxEndpoint();
StringBuilder writeEndpoint = new StringBuilder();
@@ -71,22 +59,18 @@ public class InfluxDb09Driver {
// domain + port
writeEndpoint.append(endpoint);
- // Same basic structure, but with /query on end
- StringBuilder queryEndpoint = new StringBuilder().append(writeEndpoint).append("/query");
- this.queryUrl = buildParams(queryEndpoint, "SHOW DATABASES");
-
// Add user-name, password, etc
writeEndpoint.append("/write");
- this.writeUrl = buildParams(writeEndpoint, null);
+ this.writeUrl = buildParams(writeEndpoint);
}
/**
* Simple write to "/write". Must be valid Influx line format.
*
- * @param lineDocument document to write, as string
+ * @param payload document to write, as string (payload may be a batch).
* @param failureHandler handler in case of failure
*/
- public void write(String lineDocument,
+ public void write(String payload,
final IAsyncHandler<InfluxException> failureHandler) {
// Make request to influx
IHttpClientRequest request = httpClient.request(writeUrl.toString(), HttpMethod.POST,
@@ -98,75 +82,21 @@ public class InfluxDb09Driver {
});
// For some reason Java's URLEncoding doesn't seem to be parseable by influx?
//request.addHeader("Content-Type", "application/x-www-form-urlencoded");
- request.addHeader("Content-Type", "text/plain"); //$NON-NLS-1$ //$NON-NLS-2$
- request.write(lineDocument, StandardCharsets.UTF_8.name());
- request.end();
+ if (token != null) {
+ request.addHeader("Authorization", "Token " + token);
}
-
- /**
- * List all databases
- *
- * @param handler the result handler
- */
- @SuppressWarnings("nls")
- public void listDatabases(final IAsyncResultHandler<List<String>> handler) {
- IHttpClientRequest request = httpClient.request(queryUrl.toString(), HttpMethod.GET,
- result -> {
- try {
-
- if (result.isError() || result.getResult().getResponseCode() != 200) {
- handleError(result, handler);
- return;
- }
-
- List<String> results = new ArrayList<>();
-
- // {"results":
- JsonNode arrNode = objectMapper.readTree(result.getResult().getBody())
- .path("results").elements().next() // results: [ first-elem
- .path("series").elements().next(); // series: [ first-elem
- // values: [[db1], [db2], [...]] => db1, db2
- flattenArrays(arrNode.get("values"), results);
-
- // send results
- handler.handle(AsyncResultImpl.create(results));
-
- } catch (IOException e) {
- AsyncResultImpl.create(new RuntimeException(
- "Unable to parse Influx JSON response", e));
- }
- });
-
+ request.addHeader("Content-Type", "text/plain");
+ request.write(payload, StandardCharsets.UTF_8.name());
request.end();
}
- protected <T> void handleError(IAsyncResult<IHttpClientResponse> result, IAsyncResultHandler<T> handler) {
- if (result.isError()) {
- handler.handle(AsyncResultImpl.<T> create(result.getError()));
- } else if (result.getResult().getResponseCode() != 200) {
- handler.handle(AsyncResultImpl.<T> create(new InfluxException("Influx: " //$NON-NLS-1$
- + result.getResult().getResponseCode() + " " + result.getResult().getResponseMessage()))); //$NON-NLS-1$
- }
- }
-
- private void flattenArrays(JsonNode arrNode, List<String> results) {
- if (arrNode.isArray()) {
- for (JsonNode entry : arrNode) {
- flattenArrays(entry, results);
- }
- } else {
- results.add(arrNode.textValue());
- }
- }
-
@SuppressWarnings("nls")
- private StringBuilder buildParams(StringBuilder url, String query) {
+ private StringBuilder buildParams(StringBuilder url) {
addQueryParam(url, "db", database, "?");
addQueryParam(url, "u", username, "&");
addQueryParam(url, "p", password, "&");
addQueryParam(url, "rp", retentionPolicy, "&");
addQueryParam(url, "precision", timePrecision, "&");
- addQueryParam(url, "q", query, "&");
return url;
}

DELETE
gateway/engine/influxdb/src/main/java/io/apiman/gateway/engine/influxdb/InfluxDb09Metrics.java
-/*
- * Copyright 2015 JBoss Inc
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package io.apiman.gateway.engine.influxdb;
-
-import io.apiman.common.logging.ApimanLoggerFactory;
-import io.apiman.common.logging.IApimanLogger;
-import io.apiman.gateway.engine.IComponentRegistry;
-import io.apiman.gateway.engine.IMetrics;
-import io.apiman.gateway.engine.IRequiresInitialization;
-import io.apiman.gateway.engine.async.IAsyncResult;
-import io.apiman.gateway.engine.beans.exceptions.ConfigurationParseException;
-import io.apiman.gateway.engine.components.IHttpClientComponent;
-import io.apiman.gateway.engine.components.http.IHttpClientResponse;
-import io.apiman.gateway.engine.i18n.Messages;
-import io.apiman.gateway.engine.metrics.RequestMetric;
-
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.CountDownLatch;
-
-/**
- * InfluxDB 0.9.x metrics implementation
- *
- * @author Marc Savy <msavy@redhat.com>
- */
-@SuppressWarnings("nls")
-public class InfluxDb09Metrics implements IMetrics, IRequiresInitialization {
- private static final IApimanLogger LOGGER = ApimanLoggerFactory.getLogger(InfluxDb09Metrics.class);
- private static final String USER = "username";
- private static final String PWORD = "password";
- private static final String INFLUX_ENDPOINT = "endpoint";
- private static final String DATABASE = "database";
- private static final String RETENTION_POLICY = "retentionPolicy";
- private static final String SERIES_NAME = "measurement";
- private static final String TIMEPRECISION = "ms";
-
-
- private static final Map<String, String> DEFAULT_TAGS = new LinkedHashMap<>();
- static {
- DEFAULT_TAGS.put("generator", "apiman-gateway"); //$NON-NLS-2$
- }
-
- private final String dbName;
- private final String retentionPolicy;
- private final String seriesName;
- private final String influxEndpoint;
- private IHttpClientComponent httpClient;
-
- private InfluxDb09Driver driver;
- private final String username;
- private final String password;
-
- /**
- * Constructor.
- * @param config plugin configuration options
- */
- public InfluxDb09Metrics(Map<String, String> config) {
- this.influxEndpoint = getMandatoryString(config, INFLUX_ENDPOINT);
- this.dbName = getMandatoryString(config, DATABASE);
- this.retentionPolicy = getOptionalString(config, RETENTION_POLICY, null);
- this.seriesName = getMandatoryString(config, SERIES_NAME);
- this.username = getOptionalString(config, USER, null);
- this.password = getOptionalString(config, PWORD, null);
- }
-
- /**
- * @see io.apiman.gateway.engine.IRequiresInitialization#initialize()
- */
- @Override
- public void initialize() {
- driver = new InfluxDb09Driver(httpClient, influxEndpoint, username, password, dbName,
- retentionPolicy, TIMEPRECISION);
-
- if (!listDatabases().contains(dbName)) {
- throw new ConfigurationParseException(Messages.i18n.format(
- "InfluxDb09Metrics.databaseDoesNotExist", dbName));
- }
- }
-
-
- /**
- * @see io.apiman.gateway.engine.IMetrics#setComponentRegistry(io.apiman.gateway.engine.IComponentRegistry)
- */
- @Override
- public void setComponentRegistry(IComponentRegistry registry) {
- this.httpClient = registry.getComponent(IHttpClientComponent.class);
- }
-
- /**
- * @see io.apiman.gateway.engine.IMetrics#record(io.apiman.gateway.engine.metrics.RequestMetric)
- */
- @Override
- public void record(RequestMetric metric) {
- driver.write(buildRequest(metric),
- (InfluxException result) -> {
- if (result.isBadResponse()) {
- IHttpClientResponse response = result.getResponse();
- LOGGER.error(result,
- "Influx stats error. Code: {0} with message: {1}",
- response.getResponseCode(),
- response.getResponseMessage());
- } else {
- LOGGER.error(result.getMessage(), result);
- }
- });
- }
-
- protected String buildRequest(RequestMetric metric) {
- // TODO: calculate capacity more accurately
- StringBuilder sb = new StringBuilder(500);
-
- // Series name, followed by comma
- sb.append(seriesName).append(",");
-
- // Default tags, comma delimited
- for (Entry<String, String> entry : DEFAULT_TAGS.entrySet()) {
- write(entry.getKey(), entry.getValue(), sb);
- }
-
- // Metric tags, comma delimited, space at end.
- write("apiOrgId", quote(metric.getApiOrgId()), sb);
- write("apiId", quote(metric.getApiId()), sb);
- write("apiVersion", quote(metric.getApiVersion()), sb);
- write("planId", quote(metric.getPlanId()), sb);
- write("clientOrgId", quote(metric.getClientOrgId()), sb);
- write("clientId", quote(metric.getClientId()), sb);
- write("clientVersion", quote(metric.getClientVersion()), sb);
- write("contractId", quote(metric.getContractId()), sb);
- write("user", quote(metric.getUser()), sb);
-
- sb.deleteCharAt(sb.length()-1);
- sb.append(' ');
-
- // Data, comma delimited, space at end.
- write("requestStart", dateToLong(metric.getRequestStart()), sb);
- write("requestEnd", dateToLong(metric.getRequestEnd()), sb);
- write("apiStart", dateToLong(metric.getApiStart()), sb);
- write("apiEnd", dateToLong(metric.getApiEnd()), sb);
- write("url", quote(metric.getUrl()), sb);
- write("resource", quote(metric.getResource()), sb);
- write("method", quote(metric.getMethod()), sb);
- write("responseCode", Integer.toString(metric.getResponseCode()), sb);
- write("responseMessage", quote(metric.getResponseMessage()), sb);
- write("failureCode", Integer.toString(metric.getFailureCode()), sb);
- write("failureReason", quote(metric.getFailureReason()), sb);
- write("error", Boolean.toString(metric.isError()), sb);
- write("errorMessage", quote(metric.getErrorMessage()), sb);
-
- sb.deleteCharAt(sb.length()-1);
- sb.append(' ');
-
- // Timestamp in milliseconds. Newline would be needed after this point for batching.
- sb.append(System.currentTimeMillis());
-
- return sb.toString();
- }
-
- private void write(String tagname, String tagValue, StringBuilder sb) {
- if (tagValue == null)
- return;
-
- sb.append(tagname).append("=").append(tagValue).append(",");
- }
-
- private String quote(String item) {
- if (item == null)
- return null;
- return "\"" + item + "\"";
- }
-
- private String dateToLong(Date date) {
- return Long.toString(date.getTime());
- }
-
- private String getMandatoryString(Map<String, String> config, String keyname) {
- String value = config.get(keyname);
-
- if (value == null)
- throw new ConfigurationParseException(Messages.i18n.format(
- "InfluxDb09Metrics.mandatoryConfigMustBeSet", getClass().getCanonicalName(), keyname));
- return value;
- }
-
- private String getOptionalString(Map<String, String> config, String key, String dValue) {
- return config.getOrDefault(key, dValue);
- }
-
- private List<String> listDatabases() {
- final CountDownLatch endSignal = new CountDownLatch(1);
- final List<String> results = new ArrayList<>();
-
- driver.listDatabases((IAsyncResult<List<String>> result) -> {
- if(result.isSuccess()) {
- results.addAll(result.getResult());
- } else {
- throw new InfluxException(result.getError());
- }
- endSignal.countDown();
- });
-
- try {
- endSignal.await();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
-
- return results;
- }
-}

ADD
gateway/engine/influxdb/src/main/java/io/apiman/gateway/engine/influxdb/InfluxDbMetrics.java
+/*
+ * Copyright 2015 JBoss Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.apiman.gateway.engine.influxdb;
+
+import io.apiman.common.logging.ApimanLoggerFactory;
+import io.apiman.common.logging.IApimanLogger;
+import io.apiman.gateway.engine.IComponentRegistry;
+import io.apiman.gateway.engine.IMetrics;
+import io.apiman.gateway.engine.IRequiresInitialization;
+import io.apiman.gateway.engine.beans.ApiRequest;
+import io.apiman.gateway.engine.beans.ApiResponse;
+import io.apiman.gateway.engine.components.IHttpClientComponent;
+import io.apiman.gateway.engine.components.http.IHttpClientResponse;
+import io.apiman.gateway.engine.metrics.BatchedMetricsConsumer;
+import io.apiman.gateway.engine.metrics.RequestMetric;
+
+import java.util.Date;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+/**
+ * InfluxDB 1.x metrics implementation.
+ *
+ * @author Marc Savy {@literal <marc@blackparrotlabs.io>}
+ */
+@SuppressWarnings("nls")
+public class InfluxDbMetrics implements IMetrics, IRequiresInitialization {
+ private static final IApimanLogger LOGGER = ApimanLoggerFactory.getLogger(InfluxDbMetrics.class);
+ private static final String TIME_PRECISION = "ms";
+ private final Map<String, String> defaultTags = new LinkedHashMap<>();
+ private final InfluxDbMetricsOptionsParser options;
+ private InfluxDb09Driver driver;
+ private IHttpClientComponent httpClient;
+ private final BatchedMetricsConsumer<RequestMetric> metricsConsumer;
+
+ /**
+ * Constructor.
+ * @param config plugin configuration options
+ */
+ public InfluxDbMetrics(Map<String, String> config) {
+ options = new InfluxDbMetricsOptionsParser(config);
+ this.defaultTags.put("generator", options.getGeneratorName());
+
+ metricsConsumer = new BatchedMetricsConsumer<>("InfluxDB-Metrics-Consumer", options.getQueueCapacity(), options.getMaxBatchSize(),
+ this::handleMetricBatch);
+ }
+
+ @Override
+ public void initialize() {
+ LOGGER.info("Initialising InfluxDB metrics...");
+ driver = new InfluxDb09Driver(httpClient, options, TIME_PRECISION);
+ metricsConsumer.start();
+ }
+
+ @Override
+ public void setComponentRegistry(IComponentRegistry registry) {
+ this.httpClient = registry.getComponent(IHttpClientComponent.class);
+ }
+
+ @Override
+ public void record(RequestMetric metric, ApiRequest apiRequest, ApiResponse apiResponse) {
+ this.metricsConsumer.offer(metric);
+ }
+
+ private void handleMetricBatch(List<RequestMetric> batch) {
+ String encodedBatch = encodeBatch(batch);
+ driver.write(encodedBatch, (InfluxException result) -> {
+ if (result.isBadResponse()) {
+ IHttpClientResponse httpResponse = result.getResponse();
+ LOGGER.error(result,
+ "Influx stats error. Code: {0} with message: {1}",
+ httpResponse.getResponseCode(),
+ httpResponse.getResponseMessage());
+ } else {
+ LOGGER.error(result.getMessage(), result);
+ }
+ });
+ }
+
+ protected String encodeBatch(List<RequestMetric> batch) {
+ // TODO: calculate capacity more accurately
+ StringBuilder sb = new StringBuilder(512 * batch.size());
+ for (Iterator<RequestMetric> iterator = batch.iterator(); iterator.hasNext();) {
+ RequestMetric metric = iterator.next();
+
+ // Series name, followed by comma
+ sb.append(options.getSeriesName()).append(",");
+
+ // Default tags, comma delimited
+ for (Entry<String, String> entry : defaultTags.entrySet()) {
+ write(entry.getKey(), entry.getValue(), sb);
+ }
+
+ // Metric tags/metadata, comma delimited, space at end.
+ // Apparently performance is better if we sort this alphabetically:
+ // https://docs.influxdata.com/influxdb/v2.3/write-data/best-practices/optimize-writes/#sort-tags-by-key
+ write("apiId", quote(metric.getApiId()), sb);
+ write("apiOrgId", quote(metric.getApiOrgId()), sb);
+ write("apiVersion", quote(metric.getApiVersion()), sb);
+ write("clientId", quote(metric.getClientId()), sb);
+ write("clientOrgId", quote(metric.getClientOrgId()), sb);
+ write("clientVersion", quote(metric.getClientVersion()), sb);
+ write("contractId", quote(metric.getContractId()), sb);
+ write("planId", quote(metric.getPlanId()), sb);
+ write("user", quote(metric.getUser()), sb);
+
+ sb.deleteCharAt(sb.length() - 1);
+ sb.append(' ');
+
+ // Data/variables, comma delimited, space at end.
+ write("requestStart", dateToLong(metric.getRequestStart()), sb);
+ write("requestEnd", dateToLong(metric.getRequestEnd()), sb);
+ write("apiStart", dateToLong(metric.getApiStart()), sb);
+ write("apiEnd", dateToLong(metric.getApiEnd()), sb);
+ write("url", quote(metric.getUrl()), sb);
+ write("resource", quote(metric.getResource()), sb);
+ write("method", quote(metric.getMethod()), sb);
+ write("responseCode", Integer.toString(metric.getResponseCode()), sb);
+ write("responseMessage", quote(metric.getResponseMessage()), sb);
+ write("failureCode", Integer.toString(metric.getFailureCode()), sb);
+ write("failureReason", quote(metric.getFailureReason()), sb);
+ write("error", Boolean.toString(metric.isError()), sb);
+ write("errorMessage", quote(metric.getErrorMessage()), sb);
+
+ sb.deleteCharAt(sb.length() - 1);
+ sb.append(' ');
+
+ sb.append(System.currentTimeMillis());
+
+ if (iterator.hasNext()) {
+ sb.append("\n");
+ }
+ }
+
+ return sb.toString();
+ }
+
+ private void write(String name, String value, StringBuilder sb) {
+ if (value == null)
+ return;
+
+ sb.append(name).append("=").append(value).append(",");
+ }
+
+ private String quote(String item) {
+ if (item == null)
+ return null;
+ return "\"" + item + "\"";
+ }
+
+ private String dateToLong(Date date) {
+ return Long.toString(date.getTime());
+ }
+
+}

ADD
gateway/engine/influxdb/src/main/java/io/apiman/gateway/engine/influxdb/InfluxDbMetricsOptionsParser.java
+/*
+ * Copyright 2022. Black Parrot Labs Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.apiman.gateway.engine.influxdb;
+
+import io.apiman.common.config.options.GenericOptionsParser;
+import io.apiman.common.config.options.Predicates;
+
+import java.util.Map;
+import java.util.StringJoiner;
+
+/**
+ * @author Marc Savy {@literal <marc@blackparrotlabs.io>}
+ */
+public class InfluxDbMetricsOptionsParser extends GenericOptionsParser {
+ private static final String INFLUX_ENDPOINT = "endpoint";
+ private static final String USER = "username";
+ private static final String PWORD = "password";
+ private static final String DATABASE = "database";
+ private static final String RETENTION_POLICY = "retentionPolicy";
+ private static final String SERIES_NAME = "measurement";
+
+ private static final String GENERATOR_NAME = "generatorName";
+
+ private final String influxEndpoint;
+ private final String dbName;
+ private final String retentionPolicy;
+ private final String seriesName;
+ private final String username;
+ private final String password;
+ private final String token;
+ private final String generatorName;
+ private final int queueCapacity;
+ private final int maxBatchSize;
+
+ public InfluxDbMetricsOptionsParser(Map<String, String> options) {
+ super(options);
+ this.influxEndpoint = super.getRequiredString(keys(INFLUX_ENDPOINT), Predicates.noWhitespace(), Predicates.noWhitespaceMsg());
+ this.dbName = super.getRequiredString(keys(DATABASE), Predicates.noWhitespace(), Predicates.noWhitespaceMsg());
+ this.retentionPolicy = super.getString(keys(RETENTION_POLICY), null, Predicates.noWhitespace(), Predicates.noWhitespaceMsg());
+ this.seriesName = super.getRequiredString(keys(SERIES_NAME), Predicates.noWhitespace(), Predicates.noWhitespaceMsg());
+ this.username = super.getString(keys(USER), null, Predicates.anyOk(), "");
+ this.password = super.getString(keys(PWORD), null, Predicates.anyOk(), "");
+ this.token = super.getString(keys("token"), null, Predicates.noWhitespace(), Predicates.noWhitespaceMsg());
+ this.generatorName = super.getString(keys(GENERATOR_NAME), "apiman-gateway", Predicates.noWhitespace(), Predicates.noWhitespaceMsg());
+ this.queueCapacity = super.getInt(keys("queue.size"), 10_000, Predicates.greaterThanZeroInt(), Predicates.greaterThanZeroMsg());
+ this.maxBatchSize = super.getInt(keys("batch.size"), 5_000, Predicates.greaterThanZeroInt(), Predicates.greaterThanZeroMsg());
+ }
+
+ public String getInfluxEndpoint() {
+ return influxEndpoint;
+ }
+
+ public String getDbName() {
+ return dbName;
+ }
+
+ public String getRetentionPolicy() {
+ return retentionPolicy;
+ }
+
+ public String getSeriesName() {
+ return seriesName;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public String getToken() {
+ return token;
+ }
+
+ public String getGeneratorName() {
+ return generatorName;
+ }
+
+ public int getQueueCapacity() {
+ return queueCapacity;
+ }
+
+ public int getMaxBatchSize() {
+ return maxBatchSize;
+ }
+
+ @Override
+ public String toString() {
+ return new StringJoiner(", ", InfluxDbMetricsOptionsParser.class.getSimpleName() + "[", "]")
+ .add("influxEndpoint='" + influxEndpoint + "'")
+ .add("dbName='" + dbName + "'")
+ .add("retentionPolicy='" + retentionPolicy + "'")
+ .add("seriesName='" + seriesName + "'")
+ .add("username='" + username + "'")
+ .add("password='***'")
+ .add("token='***'")
+ .add("generatorName='" + generatorName + "'")
+ .add("queueCapacity='" + queueCapacity + "'")
+ .add("maxBatchSize='" + maxBatchSize + "'")
+ .toString();
+ }
+}

MODIFY
gateway/platforms/servlet/src/main/java/io/apiman/gateway/platforms/servlet/connectors/HttpConnectorFactory.java
gateway/platforms/servlet/src/main/java/io/apiman/gateway/platforms/servlet/connectors/HttpConnectorFactory.java
@@ -115,10 +115,9 @@ public class HttpConnectorFactory implements IConnectorFactory {
public IApiConnection connect(ApiRequest request,
IAsyncResultHandler<IApiConnectionResponse> handler) throws ConnectorException {
- HttpApiConnection connection = new HttpApiConnection(okClient, request, api,
+ return new HttpApiConnection(okClient, request, api,
requiredAuthType, getSslStrategy(requiredAuthType), hasDataPolicy,
connectorConfig, handler);
- return connection;
}
};
}

feat(metrics-influxdb): add support for Influx 1.x, including use of an authorization token

This commit also adds batching support via `BatchedMetricsConsumer`.

Note `InfluxDb09Metrics` is now simply named `InfluxDbMetrics`.

Improved option parsing has been added.

Closes #2124