result = new ArrayList<>(ipAddresses.size() + hosts.size());
+ result.addAll(ipAddresses);
+ for (String host : hosts) {
+ result.addAll(actualDns.lookup(host));
+ }
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return "ConcatenatingDns(" + ipAddresses + "," + hosts + ")";
+ }
+ }
+
+ // ** Start code from Guava v20 **//
+ private static final int IPV4_PART_COUNT = 4;
+ private static final int IPV6_PART_COUNT = 8;
+
+ /**
+ * Returns the {@link InetAddress#getAddress()} having the given string representation or null if
+ * unable to parse.
+ *
+ * This deliberately avoids all nameservice lookups (e.g. no DNS).
+ *
+ *
This is the same as com.google.common.net.InetAddresses.ipStringToBytes(), except internally
+ * Splitter isn't used (as that would introduce more dependencies).
+ *
+ * @param ipString {@code String} containing an IPv4 or IPv6 string literal, e.g. {@code
+ * "192.168.0.1"} or {@code "2001:db8::1"}
+ */
+ @Nullable
+ static byte[] ipStringToBytes(String ipString) {
+ // PATCHED! adding null/empty escape
+ if (ipString == null || ipString.isEmpty()) return null;
+ // Make a first pass to categorize the characters in this string.
+ boolean hasColon = false;
+ boolean hasDot = false;
+ for (int i = 0; i < ipString.length(); i++) {
+ char c = ipString.charAt(i);
+ if (c == '.') {
+ hasDot = true;
+ } else if (c == ':') {
+ if (hasDot) {
+ return null; // Colons must not appear after dots.
+ }
+ hasColon = true;
+ } else if (Character.digit(c, 16) == -1) {
+ return null; // Everything else must be a decimal or hex digit.
+ }
+ }
+
+ // Now decide which address family to parse.
+ if (hasColon) {
+ if (hasDot) {
+ ipString = convertDottedQuadToHex(ipString);
+ if (ipString == null) {
+ return null;
+ }
+ }
+ return textToNumericFormatV6(ipString);
+ } else if (hasDot) {
+ return textToNumericFormatV4(ipString);
+ }
+ return null;
+ }
+
+ @Nullable
+ private static byte[] textToNumericFormatV4(String ipString) {
+ byte[] bytes = new byte[IPV4_PART_COUNT];
+ int i = 0;
+ try {
+ // PATCHED! for (String octet : IPV4_SPLITTER.split(ipString)) {
+ for (String octet : ipString.split("\\.", 5)) {
+ bytes[i++] = parseOctet(octet);
+ }
+ } catch (NumberFormatException ex) {
+ return null;
+ }
+
+ return i == IPV4_PART_COUNT ? bytes : null;
+ }
+
+ @Nullable
+ private static byte[] textToNumericFormatV6(String ipString) {
+ // An address can have [2..8] colons, and N colons make N+1 parts.
+ String[] parts = ipString.split(":", IPV6_PART_COUNT + 2);
+ if (parts.length < 3 || parts.length > IPV6_PART_COUNT + 1) {
+ return null;
+ }
+
+ // Disregarding the endpoints, find "::" with nothing in between.
+ // This indicates that a run of zeroes has been skipped.
+ int skipIndex = -1;
+ for (int i = 1; i < parts.length - 1; i++) {
+ if (parts[i].length() == 0) {
+ if (skipIndex >= 0) {
+ return null; // Can't have more than one ::
+ }
+ skipIndex = i;
+ }
+ }
+
+ int partsHi; // Number of parts to copy from above/before the "::"
+ int partsLo; // Number of parts to copy from below/after the "::"
+ if (skipIndex >= 0) {
+ // If we found a "::", then check if it also covers the endpoints.
+ partsHi = skipIndex;
+ partsLo = parts.length - skipIndex - 1;
+ if (parts[0].length() == 0 && --partsHi != 0) {
+ return null; // ^: requires ^::
+ }
+ if (parts[parts.length - 1].length() == 0 && --partsLo != 0) {
+ return null; // :$ requires ::$
+ }
+ } else {
+ // Otherwise, allocate the entire address to partsHi. The endpoints
+ // could still be empty, but parseHextet() will check for that.
+ partsHi = parts.length;
+ partsLo = 0;
+ }
+
+ // If we found a ::, then we must have skipped at least one part.
+ // Otherwise, we must have exactly the right number of parts.
+ int partsSkipped = IPV6_PART_COUNT - (partsHi + partsLo);
+ if (!(skipIndex >= 0 ? partsSkipped >= 1 : partsSkipped == 0)) {
+ return null;
+ }
+
+ // Now parse the hextets into a byte array.
+ ByteBuffer rawBytes = ByteBuffer.allocate(2 * IPV6_PART_COUNT);
+ try {
+ for (int i = 0; i < partsHi; i++) {
+ rawBytes.putShort(parseHextet(parts[i]));
+ }
+ for (int i = 0; i < partsSkipped; i++) {
+ rawBytes.putShort((short) 0);
+ }
+ for (int i = partsLo; i > 0; i--) {
+ rawBytes.putShort(parseHextet(parts[parts.length - i]));
+ }
+ } catch (NumberFormatException ex) {
+ return null;
+ }
+ return rawBytes.array();
+ }
+
+ @Nullable
+ private static String convertDottedQuadToHex(String ipString) {
+ int lastColon = ipString.lastIndexOf(':');
+ String initialPart = ipString.substring(0, lastColon + 1);
+ String dottedQuad = ipString.substring(lastColon + 1);
+ byte[] quad = textToNumericFormatV4(dottedQuad);
+ if (quad == null) {
+ return null;
+ }
+ String penultimate = Integer.toHexString(((quad[0] & 0xff) << 8) | (quad[1] & 0xff));
+ String ultimate = Integer.toHexString(((quad[2] & 0xff) << 8) | (quad[3] & 0xff));
+ return initialPart + penultimate + ":" + ultimate;
+ }
+
+ private static byte parseOctet(String ipPart) {
+ // Note: we already verified that this string contains only hex digits.
+ int octet = Integer.parseInt(ipPart);
+ // Disallow leading zeroes, because no clear standard exists on
+ // whether these should be interpreted as decimal or octal.
+ if (octet > 255 || (ipPart.startsWith("0") && ipPart.length() > 1)) {
+ throw new NumberFormatException();
+ }
+ return (byte) octet;
+ }
+
+ private static short parseHextet(String ipPart) {
+ // Note: we already verified that this string contains only hex digits.
+ int hextet = Integer.parseInt(ipPart, 16);
+ if (hextet > 0xffff) {
+ throw new NumberFormatException();
+ }
+ return (short) hextet;
+ }
+ // ** End code from Guava v20 **//
+}
diff --git a/storage-logzio/src/main/java/zipkin2/storage/logzio/StatusReporterFactory.java b/storage-logzio/src/main/java/zipkin2/storage/logzio/StatusReporterFactory.java
new file mode 100644
index 0000000..2d59899
--- /dev/null
+++ b/storage-logzio/src/main/java/zipkin2/storage/logzio/StatusReporterFactory.java
@@ -0,0 +1,38 @@
+package zipkin2.storage.logzio;
+
+import io.logz.sender.SenderStatusReporter;
+import org.slf4j.Logger;
+
+
+public class StatusReporterFactory {
+ public static SenderStatusReporter newSenderStatusReporter(final Logger logger) {
+ return new SenderStatusReporter() {
+
+ public void error(String s) {
+ logger.error(LogzioStorage.ZIPKIN_LOGZIO_STORAGE_MSG + s);
+ }
+
+ public void error(String s, Throwable throwable) {
+ logger.error(LogzioStorage.ZIPKIN_LOGZIO_STORAGE_MSG + s + " " + throwable.getMessage());
+ }
+
+ public void warning(String s) {
+ logger.warn(LogzioStorage.ZIPKIN_LOGZIO_STORAGE_MSG + s);
+ }
+
+ public void warning(String s, Throwable throwable) {
+ logger.warn(LogzioStorage.ZIPKIN_LOGZIO_STORAGE_MSG + s + " " + throwable.getMessage());
+ }
+
+ @Override
+ public void info(String s) {
+ logger.debug(LogzioStorage.ZIPKIN_LOGZIO_STORAGE_MSG + s);
+ }
+
+ @Override
+ public void info(String s, Throwable throwable) {
+ logger.debug(LogzioStorage.ZIPKIN_LOGZIO_STORAGE_MSG + s + " " + throwable.getMessage());
+ }
+ };
+ }
+}
diff --git a/storage-logzio/src/main/java/zipkin2/storage/logzio/client/Aggregation.java b/storage-logzio/src/main/java/zipkin2/storage/logzio/client/Aggregation.java
new file mode 100644
index 0000000..a146238
--- /dev/null
+++ b/storage-logzio/src/main/java/zipkin2/storage/logzio/client/Aggregation.java
@@ -0,0 +1,56 @@
+
+package zipkin2.storage.logzio.client;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class Aggregation {
+ transient final String field;
+ AggTerms terms;
+ Map min;
+ Map aggs;
+
+ Aggregation(String field) {
+ this.field = field;
+ }
+
+ public static Aggregation terms(String field, int size) {
+ Aggregation result = new Aggregation(field);
+ result.terms = new AggTerms(field, size);
+ return result;
+ }
+
+ public Aggregation orderBy(String subAgg, String direction) {
+ terms.order(subAgg, direction);
+ return this;
+ }
+
+ public static Aggregation min(String field) {
+ Aggregation result = new Aggregation(field);
+ result.min = Collections.singletonMap("field", field);
+ return result;
+ }
+
+ static class AggTerms {
+ AggTerms(String field, int size) {
+ this.field = field;
+ this.size = size;
+ }
+
+ final String field;
+ int size;
+ Map order;
+
+ AggTerms order(String agg, String direction) {
+ order = Collections.singletonMap(agg, direction);
+ return this;
+ }
+ }
+
+ public Aggregation addSubAggregation(Aggregation agg) {
+ if (aggs == null) aggs = new LinkedHashMap<>();
+ aggs.put(agg.field, agg);
+ return this;
+ }
+}
diff --git a/storage-logzio/src/main/java/zipkin2/storage/logzio/client/SearchCallFactory.java b/storage-logzio/src/main/java/zipkin2/storage/logzio/client/SearchCallFactory.java
new file mode 100644
index 0000000..ec5ce13
--- /dev/null
+++ b/storage-logzio/src/main/java/zipkin2/storage/logzio/client/SearchCallFactory.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2015-2018 The OpenZipkin Authors
+ *
+ * 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 zipkin2.storage.logzio.client;
+
+import com.squareup.moshi.JsonAdapter;
+import com.squareup.moshi.Moshi;
+import okhttp3.HttpUrl;
+import okhttp3.MediaType;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import zipkin2.elasticsearch.internal.client.HttpCall;
+import zipkin2.internal.Nullable;
+
+public class SearchCallFactory {
+ private static final MediaType APPLICATION_JSON = MediaType.parse("application/json");
+ public static final String API_TOKEN_HEADER = "X-API-TOKEN";
+
+ private final HttpCall.Factory http;
+ private final String apiToken;
+ private final JsonAdapter searchRequest =
+ new Moshi.Builder().build().adapter(SearchRequest.class);
+
+ public SearchCallFactory(HttpCall.Factory http, String apiToken) {
+ this.http = http;
+ this.apiToken = apiToken;
+ }
+
+ public HttpCall newCall(SearchRequest request, HttpCall.BodyConverter bodyConverter) {
+ Request httpRequest = new Request.Builder().url(lenientSearch(request.type))
+ .post(RequestBody.create(APPLICATION_JSON, searchRequest.toJson(request)))
+ .header("Content-Type", "application/json")
+ .header(API_TOKEN_HEADER, apiToken)
+ .tag(request.tag()).build();
+
+ return http.newCall(httpRequest, bodyConverter);
+ }
+
+ private HttpUrl lenientSearch(@Nullable String type) {
+ HttpUrl.Builder builder = http.baseUrl.newBuilder();
+ if (type != null) builder.addPathSegment(type);
+ return builder.build();
+ }
+
+}
diff --git a/storage-logzio/src/main/java/zipkin2/storage/logzio/client/SearchRequest.java b/storage-logzio/src/main/java/zipkin2/storage/logzio/client/SearchRequest.java
new file mode 100644
index 0000000..30e274f
--- /dev/null
+++ b/storage-logzio/src/main/java/zipkin2/storage/logzio/client/SearchRequest.java
@@ -0,0 +1,129 @@
+package zipkin2.storage.logzio.client;
+
+import zipkin2.internal.Nullable;
+import zipkin2.storage.logzio.ConsumerParams;
+
+import java.util.*;
+
+public final class SearchRequest {
+
+ public static SearchRequest create() {
+ return new SearchRequest(null);
+ }
+
+ public static SearchRequest create(String type) {
+ return new SearchRequest(type);
+ }
+
+ /**
+ * The maximum results returned in a query. This only affects non-aggregation requests.
+ *
+ * Not configurable as it implies adjustments to the index template (index.max_result_window)
+ *
+ *
See https://www.elastic.co/guide/en/elasticsearch/reference/current/search-request-from-size.html
+ */
+ private static final int MAX_RESULT_WINDOW = 10000; // the default logz.io allowed limit
+
+ @Nullable
+ transient final String type;
+
+ private Integer size = MAX_RESULT_WINDOW;
+ private Boolean _source;
+ private Object query;
+ private Map aggs;
+
+ SearchRequest(@Nullable String type) {
+
+ this.type = type;
+ }
+
+ public static class Filters extends ArrayList