getPList() {
+ if (pBuilder_ == null) {
+ return java.util.Collections.unmodifiableList(p_);
+ } else {
+ return pBuilder_.getMessageList();
+ }
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public int getPCount() {
+ if (pBuilder_ == null) {
+ return p_.size();
+ } else {
+ return pBuilder_.getCount();
+ }
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint getP(int index) {
+ if (pBuilder_ == null) {
+ return p_.get(index);
+ } else {
+ return pBuilder_.getMessage(index);
+ }
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder setP(
+ int index, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint value) {
+ if (pBuilder_ == null) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensurePIsMutable();
+ p_.set(index, value);
+ onChanged();
+ } else {
+ pBuilder_.setMessage(index, value);
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder setP(
+ int index, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder builderForValue) {
+ if (pBuilder_ == null) {
+ ensurePIsMutable();
+ p_.set(index, builderForValue.build());
+ onChanged();
+ } else {
+ pBuilder_.setMessage(index, builderForValue.build());
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder addP(de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint value) {
+ if (pBuilder_ == null) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensurePIsMutable();
+ p_.add(value);
+ onChanged();
+ } else {
+ pBuilder_.addMessage(value);
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder addP(
+ int index, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint value) {
+ if (pBuilder_ == null) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ ensurePIsMutable();
+ p_.add(index, value);
+ onChanged();
+ } else {
+ pBuilder_.addMessage(index, value);
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder addP(
+ de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder builderForValue) {
+ if (pBuilder_ == null) {
+ ensurePIsMutable();
+ p_.add(builderForValue.build());
+ onChanged();
+ } else {
+ pBuilder_.addMessage(builderForValue.build());
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder addP(
+ int index, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder builderForValue) {
+ if (pBuilder_ == null) {
+ ensurePIsMutable();
+ p_.add(index, builderForValue.build());
+ onChanged();
+ } else {
+ pBuilder_.addMessage(index, builderForValue.build());
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder addAllP(
+ java.lang.Iterable values) {
+ if (pBuilder_ == null) {
+ ensurePIsMutable();
+ com.google.protobuf.AbstractMessageLite.Builder.addAll(
+ values, p_);
+ onChanged();
+ } else {
+ pBuilder_.addAllMessages(values);
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder clearP() {
+ if (pBuilder_ == null) {
+ p_ = java.util.Collections.emptyList();
+ bitField0_ = (bitField0_ & ~0x00000001);
+ onChanged();
+ } else {
+ pBuilder_.clear();
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public Builder removeP(int index) {
+ if (pBuilder_ == null) {
+ ensurePIsMutable();
+ p_.remove(index);
+ onChanged();
+ } else {
+ pBuilder_.remove(index);
+ }
+ return this;
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder getPBuilder(
+ int index) {
+ return getPFieldBuilder().getBuilder(index);
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPointOrBuilder getPOrBuilder(
+ int index) {
+ if (pBuilder_ == null) {
+ return p_.get(index); } else {
+ return pBuilder_.getMessageOrBuilder(index);
+ }
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public java.util.List
+ getPOrBuilderList() {
+ if (pBuilder_ != null) {
+ return pBuilder_.getMessageOrBuilderList();
+ } else {
+ return java.util.Collections.unmodifiableList(p_);
+ }
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder addPBuilder() {
+ return getPFieldBuilder().addBuilder(
+ de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.getDefaultInstance());
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder addPBuilder(
+ int index) {
+ return getPFieldBuilder().addBuilder(
+ index, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.getDefaultInstance());
+ }
+ /**
+ * repeated .StringPoint p = 1;
+ *
+ *
+ *The list of points
+ *
+ */
+ public java.util.List
+ getPBuilderList() {
+ return getPFieldBuilder().getBuilderList();
+ }
+ private com.google.protobuf.RepeatedFieldBuilder<
+ de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPointOrBuilder>
+ getPFieldBuilder() {
+ if (pBuilder_ == null) {
+ pBuilder_ = new com.google.protobuf.RepeatedFieldBuilder<
+ de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPoint.Builder, de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers.StringPointOrBuilder>(
+ p_,
+ ((bitField0_ & 0x00000001) == 0x00000001),
+ getParentForChildren(),
+ isClean());
+ p_ = null;
+ }
+ return pBuilder_;
+ }
+
+ // @@protoc_insertion_point(builder_scope:StringPoints)
+ }
+
+ static {
+ defaultInstance = new StringPoints(true);
+ defaultInstance.initFields();
+ }
+
+ // @@protoc_insertion_point(class_scope:StringPoints)
+ }
+
+ private static final com.google.protobuf.Descriptors.Descriptor
+ internal_static_StringPoint_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_StringPoint_fieldAccessorTable;
+ private static final com.google.protobuf.Descriptors.Descriptor
+ internal_static_StringPoints_descriptor;
+ private static
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable
+ internal_static_StringPoints_fieldAccessorTable;
+
+ public static com.google.protobuf.Descriptors.FileDescriptor
+ getDescriptor() {
+ return descriptor;
+ }
+ private static com.google.protobuf.Descriptors.FileDescriptor
+ descriptor;
+ static {
+ java.lang.String[] descriptorData = {
+ "\n\021StringPoint.proto\"#\n\013StringPoint\022\t\n\001t\030" +
+ "\001 \001(\003\022\t\n\001v\030\002 \002(\t\"\'\n\014StringPoints\022\027\n\001p\030\001 " +
+ "\003(\0132\014.StringPointBE\n*de.qaware.chronix.c" +
+ "onverter.serializer.genB\025StringProtocolB" +
+ "uffersH\001"
+ };
+ com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
+ new com.google.protobuf.Descriptors.FileDescriptor. InternalDescriptorAssigner() {
+ public com.google.protobuf.ExtensionRegistry assignDescriptors(
+ com.google.protobuf.Descriptors.FileDescriptor root) {
+ descriptor = root;
+ return null;
+ }
+ };
+ com.google.protobuf.Descriptors.FileDescriptor
+ .internalBuildGeneratedFileFrom(descriptorData,
+ new com.google.protobuf.Descriptors.FileDescriptor[] {
+ }, assigner);
+ internal_static_StringPoint_descriptor =
+ getDescriptor().getMessageTypes().get(0);
+ internal_static_StringPoint_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_StringPoint_descriptor,
+ new java.lang.String[] { "T", "V", });
+ internal_static_StringPoints_descriptor =
+ getDescriptor().getMessageTypes().get(1);
+ internal_static_StringPoints_fieldAccessorTable = new
+ com.google.protobuf.GeneratedMessage.FieldAccessorTable(
+ internal_static_StringPoints_descriptor,
+ new java.lang.String[] { "P", });
+ }
+
+ // @@protoc_insertion_point(outer_class_scope)
+}
diff --git a/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/StringTimeSeriesConverter.java b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/StringTimeSeriesConverter.java
new file mode 100644
index 0000000..6db23d7
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/StringTimeSeriesConverter.java
@@ -0,0 +1,114 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter;
+
+import de.qaware.chronix.converter.common.MetricTSSchema;
+import de.qaware.chronix.converter.serializer.JsonKassiopeiaStringSerializer;
+import de.qaware.chronix.converter.serializer.ProtoBufKassiopeiaStringSerializer;
+import de.qaware.chronix.timeseries.StringTimeSeries;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.Charset;
+
+/**
+ * The kassiopeia time series converter for the strace time series class
+ * based on de.qaware.chronix.converter.KassiopeiaSimpleConverter
+ *
+ * @author f.lautenschlager & m.jalowski
+ */
+public class StringTimeSeriesConverter implements TimeSeriesConverter {
+
+ public static final String DATA_AS_JSON_FIELD = "dataAsJson";
+ public static final String DATA_FUNCTION_VALUE = "function_value";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(KassiopeiaSimpleConverter.class);
+
+ @Override
+ public StringTimeSeries from(BinaryTimeSeries binaryTimeSeries, long queryStart, long queryEnd) {
+ LOGGER.debug("Converting {} to MetricTimeSeries starting at {} and ending at {}", binaryTimeSeries, queryStart, queryEnd);
+ //get the metric
+ String metric = binaryTimeSeries.get(MetricTSSchema.METRIC).toString();
+
+ //Third build a minimal time series
+ StringTimeSeries.Builder builder = new StringTimeSeries.Builder(metric);
+
+ //add all user defined attributes
+ binaryTimeSeries.getFields().forEach((field, value) -> {
+ if (MetricTSSchema.isUserDefined(field)) {
+ builder.attribute(field, value);
+ }
+ });
+
+ //Default serialization is protocol buffers.
+ if (binaryTimeSeries.getPoints().length > 0) {
+ fromProtocolBuffers(binaryTimeSeries, queryStart, queryEnd, builder);
+
+ } else if (binaryTimeSeries.getFields().containsKey(DATA_AS_JSON_FIELD)) {
+ //do it from json
+ fromJson(binaryTimeSeries, queryStart, queryEnd, builder);
+
+ } else if (binaryTimeSeries.get(DATA_FUNCTION_VALUE) != null) {
+ //we have a function (aggregation) result
+ String value = binaryTimeSeries.get(DATA_FUNCTION_VALUE).toString();
+ long meanDate = meanDate(binaryTimeSeries);
+ builder.point(meanDate, value);
+ }
+
+ return builder.build();
+
+ }
+
+ private void fromProtocolBuffers(BinaryTimeSeries binaryTimeSeries, long queryStart, long queryEnd, StringTimeSeries.Builder builder) {
+ ProtoBufKassiopeiaStringSerializer.from(binaryTimeSeries.getPoints(), binaryTimeSeries.getStart(), binaryTimeSeries.getEnd(), queryStart, queryEnd, builder);
+ }
+
+ private void fromJson(BinaryTimeSeries binaryTimeSeries, long queryStart, long queryEnd, StringTimeSeries.Builder builder) {
+ String jsonString = binaryTimeSeries.get(DATA_AS_JSON_FIELD).toString();
+ //Second deserialize
+ JsonKassiopeiaStringSerializer serializer = new JsonKassiopeiaStringSerializer();
+ serializer.fromJson(jsonString.getBytes(Charset.forName(JsonKassiopeiaStringSerializer.UTF_8)), queryStart, queryEnd, builder);
+ }
+
+ private long meanDate(BinaryTimeSeries binaryTimeSeries) {
+ long start = binaryTimeSeries.getStart();
+ long end = binaryTimeSeries.getEnd();
+
+ return start + ((end - start) / 2);
+ }
+
+ @Override
+ public BinaryTimeSeries to(StringTimeSeries timeSeries) {
+ LOGGER.debug("Converting {} to BinaryTimeSeries", timeSeries);
+ BinaryTimeSeries.Builder builder = new BinaryTimeSeries.Builder();
+
+ //serialize
+ byte[] compressedPoints = ProtoBufKassiopeiaStringSerializer.to(timeSeries.points().iterator());
+
+ //Add the minimum required fields
+ builder.start(timeSeries.getStart())
+ .end(timeSeries.getEnd())
+ .data(compressedPoints);
+
+ //Currently we only have a metric
+ builder.field(MetricTSSchema.METRIC, timeSeries.getMetric());
+
+ //Add a list of user defined attributes
+ timeSeries.attributes().forEach(builder::field);
+
+ return builder.build();
+ }
+}
diff --git a/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/StringTimeSeriesDefaults.java b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/StringTimeSeriesDefaults.java
new file mode 100644
index 0000000..064032c
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/StringTimeSeriesDefaults.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter;
+
+import de.qaware.chronix.timeseries.StringTimeSeries;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BinaryOperator;
+import java.util.function.Function;
+
+/**
+ * Holds default group by and reduce functions for the kassiopeia-strace time series
+ *
+ * @author f.lautenschlager & m.jalowski
+ */
+public class StringTimeSeriesDefaults {
+
+ private StringTimeSeriesDefaults() {
+ //avoid instances
+ }
+
+ /**
+ * Default group by function for the metric time series class.
+ * Groups time series on its metric name.
+ */
+ public static final Function GROUP_BY = StringTimeSeries::getMetric;
+
+ /**
+ * Default reduce function.
+ * Attributes in both collected and reduced are merged using set holding both values.
+ */
+ public static final BinaryOperator REDUCE = (collected, reduced) -> {
+ collected.addAll(reduced.getTimestampsAsArray(), reduced.getValuesAsArray());
+
+ //The collected attributes
+ Map attributesReference = collected.getAttributesReference();
+
+ //merge the attributes
+ //we iterate over the copy
+ for (HashMap.Entry entry : reduced.attributes().entrySet()) {
+ //Attribute to add
+ String attributeToAdd = entry.getKey();
+ Set set = new HashSet<>();
+
+ if (attributesReference.containsKey(attributeToAdd)) {
+ Object value = attributesReference.get(attributeToAdd);
+ set.add(value);
+ }
+ set.add(entry.getValue());
+ attributesReference.put(attributeToAdd, set);
+ }
+ return collected;
+ };
+}
diff --git a/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/serializer/JsonKassiopeiaStringSerializer.java b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/serializer/JsonKassiopeiaStringSerializer.java
new file mode 100644
index 0000000..d1bdf3f
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/serializer/JsonKassiopeiaStringSerializer.java
@@ -0,0 +1,126 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter.serializer;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonIOException;
+import com.google.gson.JsonSyntaxException;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonWriter;
+import de.qaware.chronix.timeseries.StringTimeSeries;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The json serializer for the kassiopeia strace
+ *time series
+ *
+ * @author f.lautenschlager & m.jalowski
+ */
+public class JsonKassiopeiaStringSerializer {
+
+ public static final String UTF_8 = "UTF-8";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JsonKassiopeiaSimpleSerializer.class);
+ private static final byte[] EMPTY_JSON = "[[],[]]".getBytes(Charset.forName(UTF_8));
+ private final Gson gson;
+
+ /**
+ * Constructs a new JsonKassiopeiaSimpleSerializer.
+ */
+ public JsonKassiopeiaStringSerializer() {
+ GsonBuilder gsonBuilder = new GsonBuilder();
+ gson = gsonBuilder.create();
+ }
+
+ /**
+ * Serializes the collection of metric data points to json
+ *
+ * @param timeSeries - the time series whose points should be serialized.
+ * @return a json serialized collection of metric data points
+ */
+ public byte[] toJson(StringTimeSeries timeSeries) {
+
+ if (!timeSeries.isEmpty()) {
+ try {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ JsonWriter writer = new JsonWriter(new OutputStreamWriter(baos, UTF_8));
+
+ List[] data = new List[]{new ArrayList<>(timeSeries.size()), new ArrayList<>(timeSeries.size())};
+
+ for (int i = 0; i < timeSeries.size(); i++) {
+ data[0].add(timeSeries.getTime(i));
+ data[1].add(timeSeries.getValue(i));
+ }
+ gson.toJson(data, List[].class, writer);
+ writer.close();
+ baos.flush();
+
+ return baos.toByteArray();
+ } catch (IOException e) {
+ LOGGER.error("Could not serialize data to json", e);
+
+ }
+ }
+ return EMPTY_JSON;
+ }
+
+ /**
+ * Deserialize the given json to a collection of metric data points
+ *
+ * @param json the json representation of collection holding metric data points
+ * @param queryStart the start of the query
+ * @param queryEnd the end of the query
+ * @param builder the builder for the time series
+ */
+ public void fromJson(byte[] json, final long queryStart, final long queryEnd, StringTimeSeries.Builder builder) {
+ if (queryStart <= 0 && queryEnd <= 0) {
+ return;
+ }
+
+ try {
+ ByteArrayInputStream bais = new ByteArrayInputStream(json);
+ JsonReader reader = new JsonReader(new InputStreamReader(bais, UTF_8));
+ List[] timestampsValues = gson.fromJson(reader, List[].class);
+ reader.close();
+
+ List times = (List) timestampsValues[0];
+ List values = (List) timestampsValues[1];
+
+
+ for (int i = 0; i < times.size(); i++) {
+ if (times.get(i) > queryEnd) {
+ break;
+ }
+
+ if (times.get(i) >= queryStart && times.get(i) <= queryEnd) {
+ builder.point(times.get(i).longValue(), values.get(i));
+ }
+ }
+
+ } catch (IOException | JsonSyntaxException | JsonIOException e) {
+ LOGGER.error("Could not deserialize json data. Returning empty lists.", e);
+ }
+
+ }
+
+}
diff --git a/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/serializer/ProtoBufKassiopeiaStringSerializer.java b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/serializer/ProtoBufKassiopeiaStringSerializer.java
new file mode 100644
index 0000000..c39b3fd
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/main/java/de/qaware/chronix/converter/serializer/ProtoBufKassiopeiaStringSerializer.java
@@ -0,0 +1,218 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter.serializer;
+
+import de.qaware.chronix.converter.common.Compression;
+import de.qaware.chronix.converter.serializer.gen.StringProtocolBuffers;
+import de.qaware.chronix.timeseries.StringTimeSeries;
+import de.qaware.chronix.timeseries.dt.LongList;
+import de.qaware.chronix.timeseries.dt.StringPoint;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * Class to easily convert the protocol buffer into Point
+ *
+ * @author f.lautenschlager & m.jalowski
+ */
+public class ProtoBufKassiopeiaStringSerializer {
+
+ /**
+ * Name of the system property to set the equals offset between the dates.
+ */
+ public static final String DATE_EQUALS_OFFSET_MS = "DATE_EQUALS_OFFSET_MS";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ProtoBufKassiopeiaSimpleSerializer.class);
+ private static final long ALMOST_EQUALS_OFFSET_MS = Long.parseLong(System.getProperty(DATE_EQUALS_OFFSET_MS, "10"));
+
+ /**
+ * Private constructor
+ */
+ private ProtoBufKassiopeiaStringSerializer() {
+ //utility class
+ }
+
+ /**
+ * Adds the points (compressed byte array) to the given builder
+ *
+ * @param compressedBytes the compressed bytes holding the data points
+ * @param timeSeriesStart the start of the time series
+ * @param timeSeriesEnd the end of the time series
+ * @param builder the time series builder
+ */
+ public static void from(final byte[] compressedBytes, long timeSeriesStart, long timeSeriesEnd, StringTimeSeries.Builder builder) {
+ from(compressedBytes, timeSeriesStart, timeSeriesEnd, timeSeriesStart, timeSeriesEnd, builder);
+ }
+
+
+ /**
+ * Adds the points (compressed byte array) to the given builder
+ *
+ * @param compressedBytes the compressed bytes holding the data points
+ * @param timeSeriesStart the start of the time series
+ * @param timeSeriesEnd the end of the time series
+ * @param from including points from
+ * @param to including points to
+ * @param builder the time series builder
+ */
+ public static void from(final byte[] compressedBytes, long timeSeriesStart, long timeSeriesEnd, long from, long to, StringTimeSeries.Builder builder) {
+ if (from == -1 || to == -1) {
+ throw new IllegalArgumentException("FROM or TO have to be >= 0");
+ }
+
+ //if to is left of the time series, we have no points to return
+ if (to < timeSeriesStart) {
+ return;
+ }
+ //if from is greater to, we have nothing to return
+ if (from > to) {
+ return;
+ }
+
+ //if from is right of the time series we have nothing to return
+ if (from > timeSeriesEnd) {
+ return;
+ }
+
+ try {
+ InputStream decompressedPointStream = Compression.decompressToStream(compressedBytes);
+ StringProtocolBuffers.StringPoints protocolBufferPoints = StringProtocolBuffers.StringPoints.parseFrom(decompressedPointStream);
+
+ long lastOffset = ALMOST_EQUALS_OFFSET_MS;
+ long calculatedPointDate = timeSeriesStart;
+
+ List pList = protocolBufferPoints.getPList();
+
+ int size = pList.size();
+ StringProtocolBuffers.StringPoint[] points = pList.toArray(new StringProtocolBuffers.StringPoint[0]);
+
+ long[] timestamps = new long[pList.size()];
+ String[] values = new String[pList.size()];
+
+ int lastPointIndex = 0;
+
+ for (int i = 0; i < size; i++) {
+ StringProtocolBuffers.StringPoint p = points[i];
+
+ if (i > 0) {
+ long offset = p.getT();
+ if (offset != 0) {
+ lastOffset = offset;
+ }
+ calculatedPointDate += lastOffset;
+ }
+
+ //only add the point if it is within the date
+ if (calculatedPointDate >= from && calculatedPointDate <= to) {
+ timestamps[i] = calculatedPointDate;
+ values[i] = p.getV();
+ lastPointIndex++;
+ }
+ }
+ ArrayList valueList = new ArrayList<>(lastPointIndex);
+ valueList.addAll(Arrays.asList(values));
+ builder.points(new LongList(timestamps, lastPointIndex), valueList);
+
+ } catch (IOException e) {
+ LOGGER.info("Could not decode protocol buffers points");
+ }
+
+ }
+
+ /**
+ * Converts the given iterator of our point class to protocol buffers and compresses (gzip) it.
+ *
+ * @param metricDataPoints - the list with points
+ * @return a protocol buffer points object
+ */
+ public static byte[] to(Iterator metricDataPoints) {
+ long previousDate = 0;
+ long previousOffset = 0;
+
+ int timesSinceLastOffset = 0;
+ long lastStoredDate = 0;
+
+ StringProtocolBuffers.StringPoint.Builder builder = StringProtocolBuffers.StringPoint.newBuilder();
+ StringProtocolBuffers.StringPoints.Builder points = StringProtocolBuffers.StringPoints.newBuilder();
+
+
+ while (metricDataPoints.hasNext()) {
+
+ StringPoint p = metricDataPoints.next();
+
+ if (p == null) {
+ LOGGER.debug("Skipping 'null' point.");
+ continue;
+ }
+
+ long offset;
+ if (previousDate == 0) {
+ offset = 0;
+ // set lastStoredDate to the value of the first timestamp
+ lastStoredDate = p.getTimestamp();
+ } else {
+ offset = p.getTimestamp() - previousDate;
+ }
+
+ //Semantic Compression
+ if (ALMOST_EQUALS_OFFSET_MS == -1) {
+ builder.clearT()
+ .setV(p.getValue());
+ points.addP(builder.build());
+ } else {
+ if (almostEquals(previousOffset, offset) && noDrift(p.getTimestamp(), lastStoredDate, timesSinceLastOffset)) {
+ builder.clearT()
+ .setV(p.getValue());
+ points.addP(builder.build());
+ timesSinceLastOffset += 1;
+
+ } else {
+ builder.setT(offset)
+ .setV(p.getValue())
+ .build();
+ points.addP(builder.build());
+ //reset the offset counter
+ timesSinceLastOffset = 1;
+ lastStoredDate = p.getTimestamp();
+ }
+ //set current as former previous date
+ previousOffset = offset;
+ previousDate = p.getTimestamp();
+ }
+ }
+
+ return Compression.compress(points.build().toByteArray());
+ }
+
+ private static boolean noDrift(long timestamp, long lastStoredDate, int timesSinceLastOffset) {
+ long calculatedMaxOffset = ALMOST_EQUALS_OFFSET_MS * timesSinceLastOffset;
+ long drift = lastStoredDate + calculatedMaxOffset - timestamp;
+
+ return (drift <= (ALMOST_EQUALS_OFFSET_MS / 2));
+ }
+
+ private static boolean almostEquals(long previousOffset, long offset) {
+ double diff = Math.abs(offset - previousOffset);
+ return (diff <= ALMOST_EQUALS_OFFSET_MS);
+ }
+}
diff --git a/chronix-kassiopeia-converter-string/src/main/proto/StringPoint.proto b/chronix-kassiopeia-converter-string/src/main/proto/StringPoint.proto
new file mode 100644
index 0000000..8070306
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/main/proto/StringPoint.proto
@@ -0,0 +1,34 @@
+/*
+ * Copyright (C) 2015 QAware GmbH
+ *
+ * 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.
+ */
+
+option java_package = "de.qaware.chronix.converter.serializer.gen";
+
+option java_outer_classname = "StringProtocolBuffers";
+
+option optimize_for = SPEED;
+
+//Our point
+message StringPoint {
+ //The date as int64 (java long)
+ optional int64 t = 1;
+ required string v = 2;
+}
+
+//The data of a time series is a list of points
+message StringPoints {
+ //The list of points
+ repeated StringPoint p = 1;
+}
\ No newline at end of file
diff --git a/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/StringTimeSeriesConverterTest.groovy b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/StringTimeSeriesConverterTest.groovy
new file mode 100644
index 0000000..4d84b4d
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/StringTimeSeriesConverterTest.groovy
@@ -0,0 +1,115 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter
+
+import de.qaware.chronix.timeseries.StringTimeSeries
+import spock.lang.Specification
+
+import java.time.Instant
+
+/**
+ * Unit test for the kassiopeia simple converter
+ * @author f.lautenschlager & m.jalowski
+ */
+class StringTimeSeriesConverterTest extends Specification {
+
+ def "test to and from compressed data"() {
+ given:
+ def ts = new StringTimeSeries.Builder("\\Load\\avg").attribute("MyField", "4711")
+ def start = Instant.now()
+
+ 100.times {
+ ts.point(start.plusSeconds(it).toEpochMilli(), it * 2 + " as string")
+ }
+
+ def converter = new StringTimeSeriesConverter();
+
+ when:
+ def binaryTimeSeries = converter.to(ts.build())
+ def tsReconverted = converter.from(binaryTimeSeries, start.toEpochMilli(), start.plusSeconds(20).toEpochMilli())
+
+ then:
+ tsReconverted.metric == "\\Load\\avg"
+ tsReconverted.size() == 21
+ tsReconverted.getValue(1) == "2 as string"
+ tsReconverted.attribute("MyField") == "4711"
+
+ }
+
+ def "test to and from aggregated value"() {
+ given:
+ def converter = new StringTimeSeriesConverter();
+
+ def binTs = new BinaryTimeSeries.Builder()
+ .field("function_value", "4711")
+ .field("metric", "\\Load\\avg")
+ .start(0)
+ .end(10)
+
+ when:
+ def tsReconverted = converter.from(binTs.build(), 0, 100)
+
+ then:
+ tsReconverted.metric == "\\Load\\avg"
+ tsReconverted.size() == 1
+ tsReconverted.getValue(0) == "4711"
+ tsReconverted.start == 5
+ tsReconverted.end == 5
+ }
+
+ def "test to and from json data"() {
+ given:
+ def converter = new StringTimeSeriesConverter();
+
+ def binTs = new BinaryTimeSeries.Builder()
+ .field("function_value", "4711")
+ .field("metric", "\\Load\\avg")
+ .field("dataAsJson", "[[0,1,2,3],['string0','string1','string2','string3']]")
+ .start(0)
+ .end(10)
+
+ when:
+ def tsReconverted = converter.from(binTs.build(), 0, 100)
+
+ then:
+ tsReconverted.metric == "\\Load\\avg"
+ tsReconverted.size() == 4
+ tsReconverted.getValue(3) == 'string3'
+ tsReconverted.start == 0
+ tsReconverted.end == 3
+ }
+
+ def "test to and from json data with encoding exception"() {
+ given:
+ def converter = new StringTimeSeriesConverter();
+
+ def binTs = new BinaryTimeSeries.Builder()
+ .field("function_value", "4711")
+ .field("metric", "\\Load\\avg")
+ .field("dataAsJson", new String("[[0,1,2,3],['string0','string1','string2','string3']]".getBytes("IBM420")))
+ .start(0)
+ .end(10)
+
+ when:
+ def tsReconverted = converter.from(binTs.build(), 0, 100)
+
+ then:
+ tsReconverted.metric == "\\Load\\avg"
+ tsReconverted.size() == 0
+ tsReconverted.start == 0
+ tsReconverted.end == 0
+ }
+}
diff --git a/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/StringTimeSeriesDefaultsTest.groovy b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/StringTimeSeriesDefaultsTest.groovy
new file mode 100644
index 0000000..4349c28
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/StringTimeSeriesDefaultsTest.groovy
@@ -0,0 +1,84 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter
+
+import de.qaware.chronix.timeseries.StringTimeSeries
+import de.qaware.chronix.timeseries.dt.StringPoint
+import spock.lang.Specification
+
+import java.util.stream.Collectors
+
+/**
+ * Unit test for the reduce and group by defaults
+ * @author f.lautenschlager
+ */
+class StringTimeSeriesDefaultsTest extends Specification {
+
+ def "test default group by"() {
+ given:
+ def straceTimeSeries = new StringTimeSeries.Builder("metric").build();
+
+ when:
+ def groupBy = StringTimeSeriesDefaults.GROUP_BY.apply(straceTimeSeries)
+
+ then:
+ groupBy == "metric"
+ }
+
+ def "test default reduce"() {
+
+ given:
+ def ts1 = new StringTimeSeries.Builder("metric")
+ .attribute("existsInBoth", 1)
+ .attribute("onlyInTs1", "value only in ts1")
+ .attribute("simpleList", ["one", "two", "three"])
+ .point(1, "2")
+ .point(3, "6")
+ .point(5, "10")
+ .build();
+ def ts2 = new StringTimeSeries.Builder("metric")
+ .attribute("existsInBoth", 2)
+ .attribute("onlyInTs2", "ts1 would never see me")
+ .attribute("simpleList", ["four", "five", "six"])
+ .attribute("threads", [1, 2, 3])
+ .point(2, "4")
+ .point(4, "8")
+ .point(6, "12")
+ .build();
+ when:
+ def merged = StringTimeSeriesDefaults.REDUCE.apply(ts1, ts2)
+
+ then:
+ merged.size() == 6
+ merged.sort()
+ def list = merged.points().collect(Collectors.toList())
+ list.containsAll([new StringPoint(0, 1, "2"), new StringPoint(1, 2, "4"), new StringPoint(2, 3, "6")
+ , new StringPoint(3, 4, "8"), new StringPoint(4, 5, "10"), new StringPoint(5, 6, "12")])
+ merged.attributes().size() == 5
+ merged.attribute("existsInBoth") as Set == [1, 2] as Set
+ merged.attribute("onlyInTs1") as String == "value only in ts1"
+ merged.attribute("onlyInTs2") as Set == ["ts1 would never see me"] as Set
+ merged.attribute("threads") as Set == [[1, 2, 3]] as Set
+ merged.attribute("simpleList") as Set == [["one", "two", "three"], ["four", "five", "six"]] as Set
+ }
+
+ def "test private constructor"() {
+ when:
+ StringTimeSeriesDefaults.newInstance()
+ then:
+ noExceptionThrown()
+ }
+}
diff --git a/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/serializer/JsonKassiopeiaStringSerializerTest.groovy b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/serializer/JsonKassiopeiaStringSerializerTest.groovy
new file mode 100644
index 0000000..40996eb
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/serializer/JsonKassiopeiaStringSerializerTest.groovy
@@ -0,0 +1,119 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter.serializer
+
+import de.qaware.chronix.timeseries.StringTimeSeries
+import de.qaware.chronix.timeseries.dt.LongList
+import spock.lang.Specification
+import spock.lang.Unroll
+
+import java.nio.charset.Charset
+
+/**
+ * Unit test for the kassiopeia serializer
+ * @author f.lautenschlager & m.jalowski
+ */
+class JsonKassiopeiaStringSerializerTest extends Specification {
+
+ def "test serialize to and deserialize from json"() {
+ given:
+ def times = longList([0, 1, 2])
+ def values = ["string0", "string1", "string2"]
+ def ts = new StringTimeSeries.Builder("test").points(times, values).build()
+
+ def serializer = new JsonKassiopeiaStringSerializer()
+ def start = 0l
+ def end = 2l
+
+ when:
+ def json = serializer.toJson(ts)
+
+ then:
+ def builder = new StringTimeSeries.Builder("test")
+ serializer.fromJson(json, start, end, builder)
+ def recoverdTs = builder.build();
+
+
+ times.size() == 3
+ recoverdTs.size() == 3
+
+ recoverdTs.getTime(0) == 0l
+ recoverdTs.getTime(1) == 1l
+ recoverdTs.getTime(2) == 2l
+
+ recoverdTs.getValue(0) == "string0"
+ recoverdTs.getValue(1) == "string1"
+ recoverdTs.getValue(2) == "string2"
+ }
+
+ @Unroll
+ def "test serialize and deserialize from json with filter #start and #end expecting #size elements"() {
+ given:
+ def times = longList([0, 1, 2, 3, 4, 5])
+ def values = ["string0", "string1", "string2", "string3", "string4", "string5"]
+ def ts = new StringTimeSeries.Builder("test").points(times, values).build()
+
+ def serializer = new JsonKassiopeiaStringSerializer()
+
+ when:
+ def builder = new StringTimeSeries.Builder("test")
+
+ def json = serializer.toJson(ts)
+ serializer.fromJson(json, start, end, builder)
+ then:
+ builder.build().size() == size
+
+ where:
+ start << [0, 1, 1, 0]
+ end << [0, 1, 3, 6]
+ size << [0, 1, 3, 6]
+ }
+
+ def "test serialize to json with empty timestamps / values"() {
+ given:
+ def serializer = new JsonKassiopeiaStringSerializer()
+ def ts = new StringTimeSeries.Builder("test").points(times, values).build()
+
+ when:
+ def json = serializer.toJson(ts)
+
+ then:
+ new String(json) == "[[],[]]"
+
+ where:
+ times << [null, longList([0l, 1l, 2l]) as LongList]
+ values << [["string0", "string1", "string2"], null]
+ }
+
+ def longList(ArrayList longs) {
+ def times = new LongList()
+ longs.each { l -> times.add(l) }
+
+ times
+ }
+
+ def "test deserialize from empty json "() {
+ given:
+ def serializer = new JsonKassiopeiaStringSerializer()
+ def builder = new StringTimeSeries.Builder("test")
+
+ when:
+ def result = serializer.fromJson("[[],[]]".getBytes(Charset.forName("UTF-8")), 1, 2000, builder)
+ def ts = builder.build()
+ then:
+ ts.size() == 0
+ }
+}
diff --git a/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/serializer/ProtoBufKassiopeiaStringSerializerTest.groovy b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/serializer/ProtoBufKassiopeiaStringSerializerTest.groovy
new file mode 100644
index 0000000..ff0c17e
--- /dev/null
+++ b/chronix-kassiopeia-converter-string/src/test/groovy/de/qaware/chronix/converter/serializer/ProtoBufKassiopeiaStringSerializerTest.groovy
@@ -0,0 +1,214 @@
+/*
+ * Copyright (C) 2016 QAware GmbH
+ *
+ * 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 de.qaware.chronix.converter.serializer
+
+import de.qaware.chronix.timeseries.StringTimeSeries
+import de.qaware.chronix.timeseries.dt.StringPoint
+import spock.lang.Shared
+import spock.lang.Specification
+
+import java.time.Instant
+import java.util.stream.Collectors
+
+/**
+ * Unit test for the protocol buffers serializer
+ * @author f.lautenschlager & m.jalowski
+ */
+class ProtoBufKassiopeiaStringSerializerTest extends Specification {
+
+ def "test from without range query"() {
+ given:
+ def points = []
+ 100.times {
+ points.add(new StringPoint(it, it + 1, it * 100 + " as string"))
+ }
+ def compressedProtoPoints = ProtoBufKassiopeiaStringSerializer.to(points.iterator())
+
+ when:
+ def builder = new StringTimeSeries.Builder("metric");
+ ProtoBufKassiopeiaStringSerializer.from(compressedProtoPoints, 0, points.size(), 0, points.size(), builder)
+ def ts = builder.build();
+ then:
+ 100.times {
+ ts.getValue(it) == it * 100 + " as string"
+ ts.getTime(it) == it
+ }
+ }
+
+ @Shared
+ def start = Instant.now()
+ @Shared
+ def end = start.plusSeconds(100 * 100)
+
+ def "test from with range query"() {
+ given:
+ def points = []
+
+ 100.times {
+ points.add(new StringPoint(it, start.plusSeconds(it).toEpochMilli(), it * 100 + " as string"))
+ }
+ def compressedProtoPoints = ProtoBufKassiopeiaStringSerializer.to(points.iterator())
+ def builder = new StringTimeSeries.Builder("metric");
+
+ when:
+ ProtoBufKassiopeiaStringSerializer.from(compressedProtoPoints, start.toEpochMilli(), end.toEpochMilli(), from, to, builder)
+ def ts = builder.build();
+
+ then:
+ List list = ts.points().collect(Collectors.toList())
+ list.size() == size
+ if (size == 21) {
+ list.get(0).timestamp == 1456394850774
+ list.get(0).value == 5000 + " as string"
+
+ list.get(20).timestamp == 1456394870774
+ list.get(20).value == 7000 + " as string"
+ }
+
+ where:
+ from << [end.toEpochMilli() + 2, 0, start.toEpochMilli() + 4, start.plusSeconds(50).toEpochMilli()]
+ to << [end.toEpochMilli() + 3, 0, start.toEpochMilli() + 2, start.plusSeconds(70).toEpochMilli()]
+ size << [0, 0, 0, 21]
+ }
+
+ def "test convert to protocol buffers points"() {
+ given:
+ def points = []
+ 100.times {
+ points.add(new StringPoint(it, it + 15, it * 100 + " as string"))
+ }
+ //Points that are null are ignored
+ points.add(null)
+ def builder = new StringTimeSeries.Builder("");
+ when:
+ def compressedPoints = ProtoBufKassiopeiaStringSerializer.to(points.iterator())
+ ProtoBufKassiopeiaStringSerializer.from(compressedPoints, 0, 114, builder)
+
+ then:
+ builder.build().size() == 100
+ }
+
+ def "test iterator with invalid arguments"() {
+ when:
+ ProtoBufKassiopeiaStringSerializer.from(null, 0, 0, from, to, new StringTimeSeries.Builder(""))
+ then:
+ thrown IllegalArgumentException
+ where:
+ from << [-1, 0, -1]
+ to << [0, -1, -1]
+
+ }
+
+
+ def "test private constructor"() {
+ when:
+ ProtoBufKassiopeiaStringSerializer.newInstance()
+ then:
+ noExceptionThrown()
+ }
+
+
+ def "test date-delta-compaction"() {
+ given:
+ def points = []
+ points.add(new StringPoint(0, 10, "string0"))
+ points.add(new StringPoint(1, 20, "string1"))
+ points.add(new StringPoint(2, 30, "string2"))
+ points.add(new StringPoint(3, 39, "string3"))
+ points.add(new StringPoint(4, 48, "string4"))
+ points.add(new StringPoint(5, 57, "string5"))
+ points.add(new StringPoint(6, 66, "string6"))
+ points.add(new StringPoint(7, 75, "string7"))
+ points.add(new StringPoint(8, 84, "string8"))
+ points.add(new StringPoint(9, 93, "string9"))
+ points.add(new StringPoint(10, 102, "string10"))
+ points.add(new StringPoint(11, 111, "string11"))
+ points.add(new StringPoint(12, 120, "string12"))
+ points.add(new StringPoint(13, 129, "string13"))
+ points.add(new StringPoint(14, 138, "string14"))
+
+ def builder = new StringTimeSeries.Builder("metric");
+
+ when:
+ def compressedProtoPoints = ProtoBufKassiopeiaStringSerializer.to(points.iterator())
+ ProtoBufKassiopeiaStringSerializer.from(compressedProtoPoints, 10l, 1036l, builder)
+ def ts = builder.build()
+ def listPoints = ts.points().collect(Collectors.toList()) as List
+
+ then: //diff to origin
+ listPoints.get(0).timestamp == 10//0
+ listPoints.get(1).timestamp == 20//0
+ listPoints.get(2).timestamp == 30//0
+ listPoints.get(3).timestamp == 40//1
+ listPoints.get(4).timestamp == 50//2
+ listPoints.get(5).timestamp == 60//3
+ listPoints.get(6).timestamp == 70//4
+ listPoints.get(7).timestamp == 80//5
+ listPoints.get(8).timestamp == 89//5
+ listPoints.get(9).timestamp == 98//5
+ listPoints.get(10).timestamp == 107//5
+ listPoints.get(11).timestamp == 116//5
+ listPoints.get(12).timestamp == 125//5
+ listPoints.get(13).timestamp == 134//5
+ listPoints.get(14).timestamp == 143//5
+ }
+
+
+ def "test date-delta-compaction with different values"() {
+ given:
+ def points = []
+ points.add(new StringPoint(0, 1462892410, "string0"))
+ points.add(new StringPoint(1, 1462892420, "string1"))
+ points.add(new StringPoint(2, 1462892430, "string2"))
+ points.add(new StringPoint(3, 1462892439, "string3"))
+ points.add(new StringPoint(4, 1462892448, "string4"))
+ points.add(new StringPoint(5, 1462892457, "string5"))
+ points.add(new StringPoint(6, 1462892466, "string6"))
+ points.add(new StringPoint(7, 1462892475, "string7"))
+ points.add(new StringPoint(8, 1462892484, "string8"))
+ points.add(new StringPoint(9, 1462892493, "string9"))
+ points.add(new StringPoint(10, 1462892502, "string10"))
+ points.add(new StringPoint(11, 1462892511, "string11"))
+ points.add(new StringPoint(12, 1462892520, "string12"))
+ points.add(new StringPoint(13, 1462892529, "string13"))
+ points.add(new StringPoint(14, 1462892538, "string14"))
+
+ def builder = new StringTimeSeries.Builder("metric1");
+
+ when:
+ def compressedProtoPoints = ProtoBufKassiopeiaStringSerializer.to(points.iterator())
+ ProtoBufKassiopeiaStringSerializer.from(compressedProtoPoints, 1462892410l, 1462892543l, builder)
+ def ts = builder.build()
+ def listPoints = ts.points().collect(Collectors.toList()) as List
+
+ then: //diff to origin
+ listPoints.get(0).timestamp == 1462892410//0
+ listPoints.get(1).timestamp == 1462892420//0
+ listPoints.get(2).timestamp == 1462892430//0
+ listPoints.get(3).timestamp == 1462892440//1
+ listPoints.get(4).timestamp == 1462892450//2
+ listPoints.get(5).timestamp == 1462892460//3
+ listPoints.get(6).timestamp == 1462892470//4
+ listPoints.get(7).timestamp == 1462892480//5
+ listPoints.get(8).timestamp == 1462892489//5
+ listPoints.get(9).timestamp == 1462892498//5
+ listPoints.get(10).timestamp == 1462892507//5
+ listPoints.get(11).timestamp == 1462892516//5
+ listPoints.get(12).timestamp == 1462892525//5
+ listPoints.get(13).timestamp == 1462892534//5
+ listPoints.get(14).timestamp == 1462892543//5
+ }
+}