mirror of
https://github.com/tiennm99/tjahzi.git
synced 2026-10-14 10:19:01 +00:00
Merge branch 'master' into 5-in-house-protobuf
This commit is contained in:
commit
3f7b2cad24
14 files changed
+513
-39
No files matched your search
+10
-6
@@ -1,7 +1,7 @@
|
||||
version: 2.1
|
||||
|
||||
orbs:
|
||||
circleci-maven-release-orb: sonatype-nexus-community/circleci-maven-release-orb@0.0.15
|
||||
circleci-maven-release-orb: sonatype-nexus-community/circleci-maven-release-orb@0.0.16
|
||||
|
||||
jobs:
|
||||
build-and-test:
|
||||
@@ -46,17 +46,21 @@ executors:
|
||||
machine:
|
||||
image: ubuntu-2004:202008-01
|
||||
|
||||
parameters:
|
||||
run_workflow_release:
|
||||
default: false
|
||||
type: boolean
|
||||
|
||||
workflows:
|
||||
version: 2
|
||||
main:
|
||||
jobs:
|
||||
- build-and-test
|
||||
release:
|
||||
when: << pipeline.parameters.run_workflow_release >>
|
||||
jobs:
|
||||
- build-and-test
|
||||
- circleci-maven-release-orb/run-maven-release:
|
||||
filters:
|
||||
branches:
|
||||
ignore: /.*/
|
||||
tags:
|
||||
only: /.*/
|
||||
requires:
|
||||
- build-and-test
|
||||
executor: new-java-maven-executor
|
||||
@@ -1,12 +1,11 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>tjahzi</artifactId>
|
||||
<version>0.7-SNAPSHOT</version>
|
||||
<version>0.8.2-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<name>tjahzi-log4j2-appender</name>
|
||||
|
||||
+124
@@ -0,0 +1,124 @@
|
||||
package pl.tkowalcz.tjahzi.log4j2;
|
||||
|
||||
import io.restassured.RestAssured;
|
||||
import io.restassured.http.ContentType;
|
||||
import io.restassured.parsing.Parser;
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import org.awaitility.Awaitility;
|
||||
import org.awaitility.Durations;
|
||||
import org.hamcrest.BaseMatcher;
|
||||
import org.hamcrest.Description;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.testcontainers.containers.BindMode;
|
||||
import org.testcontainers.containers.GenericContainer;
|
||||
import org.testcontainers.containers.wait.strategy.Wait;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static io.restassured.RestAssured.given;
|
||||
import static org.hamcrest.CoreMatchers.equalTo;
|
||||
import static org.hamcrest.CoreMatchers.hasItems;
|
||||
|
||||
@Testcontainers
|
||||
class LokiAppenderLargeBatchesTestTest {
|
||||
|
||||
@Container
|
||||
public GenericContainer loki = new GenericContainer("grafana/loki:latest")
|
||||
.withCommand("-config.file=/etc/loki-config.yaml")
|
||||
.withClasspathResourceMapping("loki-config.yaml",
|
||||
"/etc/loki-config.yaml",
|
||||
BindMode.READ_ONLY)
|
||||
.waitingFor(
|
||||
Wait.forHttp("/ready")
|
||||
.forPort(3100)
|
||||
)
|
||||
.withExposedPorts(3100);
|
||||
|
||||
@Test
|
||||
void shouldSendData() throws URISyntaxException {
|
||||
// Given
|
||||
System.setProperty("loki.host", loki.getHost());
|
||||
System.setProperty("loki.port", loki.getFirstMappedPort().toString());
|
||||
|
||||
URI uri = getClass()
|
||||
.getClassLoader()
|
||||
.getResource("appender-test-large-batches.xml")
|
||||
.toURI();
|
||||
|
||||
((org.apache.logging.log4j.core.LoggerContext) LogManager.getContext(false))
|
||||
.setConfigLocation(uri);
|
||||
|
||||
String expectedLogLine = "Cupcake ipsum dolor sit amet cake wafer. " +
|
||||
"Souffle jelly beans biscuit topping. " +
|
||||
"Danish bonbon gummies powder caramels. " +
|
||||
"Danish jelly beans sweet roll topping jelly beans oat cake toffee. " +
|
||||
"Chocolate cake sesame snaps brownie biscuit cheesecake. " +
|
||||
"Ice cream dessert sweet donut marshmallow. " +
|
||||
"Muffin bear claw cookie jelly-o sugar plum jelly beans apple pie fruitcake cookie. " +
|
||||
"Tootsie roll carrot cake pastry jujubes jelly beans chupa chups. " +
|
||||
"Souffle cake muffin liquorice tart souffle pie sesame snaps.";
|
||||
|
||||
long expectedTimestamp = System.currentTimeMillis();
|
||||
Logger logger = LogManager.getLogger(LokiAppenderLargeBatchesTestTest.class);
|
||||
|
||||
// When
|
||||
for (int i = 0; i < 1000; i++) {
|
||||
logger.info(i + " " + expectedLogLine);
|
||||
}
|
||||
|
||||
// Then
|
||||
RestAssured.port = loki.getFirstMappedPort();
|
||||
RestAssured.baseURI = "http://" + loki.getHost();
|
||||
RestAssured.registerParser("text/plain", Parser.JSON);
|
||||
|
||||
Awaitility
|
||||
.await()
|
||||
.atMost(Durations.ONE_MINUTE)
|
||||
.pollInterval(Durations.ONE_SECOND)
|
||||
.ignoreExceptions()
|
||||
.untilAsserted(() -> {
|
||||
given()
|
||||
.contentType(ContentType.URLENC)
|
||||
.urlEncodingEnabled(false)
|
||||
.formParam("&start=" + expectedTimestamp + "&limit=1000&query=%7Bserver%3D%22127.0.0.1%22%7D")
|
||||
.when()
|
||||
.get("/loki/api/v1/query_range")
|
||||
.then()
|
||||
.log()
|
||||
.all()
|
||||
.statusCode(200)
|
||||
.body("status", equalTo("success"))
|
||||
.body("data.result.size()", equalTo(1))
|
||||
.body("data.result[0].stream.server", equalTo("127.0.0.1"))
|
||||
.body("data.result[0].values.size()", equalTo(1000))
|
||||
.body("data.result[0].values", hasItems(new BaseMatcher<>() {
|
||||
@Override
|
||||
public boolean matches(Object o) {
|
||||
List<Object> list = (List<Object>) o;
|
||||
if (list.size() != 2) {
|
||||
return false;
|
||||
}
|
||||
|
||||
long actualTimestamp = Long.parseLong(list.get(0).toString());
|
||||
String actualLogLine = list.get(1).toString();
|
||||
|
||||
return actualLogLine.contains(expectedLogLine)
|
||||
&& (expectedTimestamp - actualTimestamp) < TimeUnit.MINUTES.toMillis(1);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void describeTo(Description description) {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
));
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<configuration status="OFF" shutdownHook="disable" packages="pl.tkowalcz.tjahzi.log4j2">
|
||||
<Loggers>
|
||||
<Root level="INFO">
|
||||
<AppenderRef ref="Loki"/>
|
||||
</Root>
|
||||
</Loggers>
|
||||
|
||||
<appenders>
|
||||
<Loki name="Loki" bufferSizeMegabytes="2" batchWait="10000">
|
||||
<host>${sys:loki.host}</host>
|
||||
<port>${sys:loki.port}</port>
|
||||
|
||||
<ThresholdFilter level="ALL"/>
|
||||
<PatternLayout>
|
||||
<Pattern>%X{tid} [%t] %d{MM-dd HH:mm:ss.SSS} %5p %c{1} - %m%n%exception{full}</Pattern>
|
||||
</PatternLayout>
|
||||
|
||||
<Header name="server" value="127.0.0.1"/>
|
||||
<Label name="server" value="127.0.0.1"/>
|
||||
</Loki>
|
||||
</appenders>
|
||||
</configuration>
|
||||
@@ -38,6 +38,7 @@ limits_config:
|
||||
enforce_metric_name: false
|
||||
reject_old_samples: true
|
||||
reject_old_samples_max_age: 168h
|
||||
max_entries_limit_per_query: 10000
|
||||
|
||||
chunk_store_config:
|
||||
max_look_back_period: 0s
|
||||
|
||||
@@ -5,7 +5,7 @@
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>tjahzi</artifactId>
|
||||
<packaging>pom</packaging>
|
||||
<version>0.7-SNAPSHOT</version>
|
||||
<version>0.8.2-SNAPSHOT</version>
|
||||
|
||||
<name>Tjahzi</name>
|
||||
<description>Tjahzi is a low overhead logging client for Loki</description>
|
||||
@@ -20,7 +20,7 @@
|
||||
<scm>
|
||||
<developerConnection>scm:git:git@github.com:tkowalcz/tjahzi.git</developerConnection>
|
||||
<url>https://github.com/tkowalcz/tjahzi</url>
|
||||
<tag>HEAD</tag>
|
||||
<tag>tjahzi-0.7</tag>
|
||||
</scm>
|
||||
|
||||
<distributionManagement>
|
||||
@@ -56,6 +56,7 @@
|
||||
<compiler.source.level>9</compiler.source.level>
|
||||
<compiler.target.level>9</compiler.target.level>
|
||||
|
||||
<!--suppress UnresolvedMavenProperty -->
|
||||
<gpg.passphrase>${env.GPG_PASSPHRASE}</gpg.passphrase>
|
||||
</properties>
|
||||
|
||||
@@ -179,6 +180,7 @@
|
||||
</plugins>
|
||||
</build>
|
||||
<properties>
|
||||
<!--suppress UnresolvedMavenProperty -->
|
||||
<gpg.passphrase>${env.GPG_PASSPHRASE}</gpg.passphrase>
|
||||
</properties>
|
||||
</profile>
|
||||
|
||||
+10
-4
@@ -1,12 +1,11 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>tjahzi</artifactId>
|
||||
<version>0.7-SNAPSHOT</version>
|
||||
<version>0.8.2-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<name>tjahzi-core</name>
|
||||
@@ -27,7 +26,7 @@
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.netty</groupId>
|
||||
<artifactId>netty-all</artifactId>
|
||||
<artifactId>netty-codec-http</artifactId>
|
||||
<version>4.1.52.Final</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
@@ -113,5 +112,12 @@
|
||||
<version>1.14.3</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.xerial.snappy</groupId>
|
||||
<artifactId>snappy-java</artifactId>
|
||||
<version>1.1.8.4</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -1,8 +1,5 @@
|
||||
package pl.tkowalcz.tjahzi;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufOutputStream;
|
||||
import io.netty.buffer.PooledByteBufAllocator;
|
||||
import logproto.Logproto;
|
||||
import org.agrona.MutableDirectBuffer;
|
||||
import org.agrona.concurrent.Agent;
|
||||
@@ -55,16 +52,14 @@ public class LogBufferAgent implements Agent, MessageHandler {
|
||||
|
||||
long currentTimeMillis = clock.millis();
|
||||
if (exceededBatchSizeThreshold() || exceededWaitTimeThreshold(currentTimeMillis)) {
|
||||
ByteBuf buffer = PooledByteBufAllocator.DEFAULT.buffer();
|
||||
request.build().writeTo(
|
||||
new ByteBufOutputStream(buffer)
|
||||
);
|
||||
try {
|
||||
httpClient.log(request);
|
||||
} finally {
|
||||
request = Logproto.PushRequest.newBuilder();
|
||||
|
||||
httpClient.log(buffer);
|
||||
request = Logproto.PushRequest.newBuilder();
|
||||
|
||||
estimatedBytesPending = 0;
|
||||
timeoutDeadline = currentTimeMillis + batchWaitMillis;
|
||||
estimatedBytesPending = 0;
|
||||
timeoutDeadline = currentTimeMillis + batchWaitMillis;
|
||||
}
|
||||
}
|
||||
|
||||
return workDone;
|
||||
|
||||
@@ -1,12 +1,15 @@
|
||||
package pl.tkowalcz.tjahzi.http;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufOutputStream;
|
||||
import io.netty.buffer.PooledByteBufAllocator;
|
||||
import io.netty.handler.codec.compression.Snappy;
|
||||
import io.netty.handler.codec.http.*;
|
||||
import io.netty.util.ReferenceCountUtil;
|
||||
import logproto.Logproto;
|
||||
import pl.tkowalcz.tjahzi.stats.MonitoringModule;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
|
||||
public class NettyHttpClient implements Closeable {
|
||||
|
||||
@@ -32,21 +35,17 @@ public class NettyHttpClient implements Closeable {
|
||||
}
|
||||
|
||||
public void log(ByteBuf dataBuffer) {
|
||||
ByteBuf output = PooledByteBufAllocator.DEFAULT.buffer();
|
||||
snappy.encode(dataBuffer, output, dataBuffer.readableBytes());
|
||||
dataBuffer.release();
|
||||
|
||||
FullHttpRequest request = new DefaultFullHttpRequest(
|
||||
HttpVersion.HTTP_1_1,
|
||||
HttpMethod.POST,
|
||||
clientConfiguration.getLogEndpoint(),
|
||||
output
|
||||
dataBuffer
|
||||
);
|
||||
|
||||
request.headers()
|
||||
.add(headers)
|
||||
.set(HttpHeaderNames.CONTENT_TYPE, PROTOBUF_MIME_TYPE)
|
||||
.set(HttpHeaderNames.CONTENT_LENGTH, output.readableBytes())
|
||||
.set(HttpHeaderNames.CONTENT_LENGTH, dataBuffer.readableBytes())
|
||||
.set(HttpHeaderNames.HOST, clientConfiguration.getHost());
|
||||
|
||||
lokiConnection.execute(request);
|
||||
@@ -56,4 +55,24 @@ public class NettyHttpClient implements Closeable {
|
||||
public void close() {
|
||||
lokiConnection.close();
|
||||
}
|
||||
|
||||
public void log(Logproto.PushRequest.Builder request) throws IOException {
|
||||
ByteBuf dataBuffer = PooledByteBufAllocator.DEFAULT.buffer();
|
||||
ByteBuf compressedBuffer = PooledByteBufAllocator.DEFAULT.buffer();
|
||||
|
||||
try {
|
||||
request.build().writeTo(
|
||||
new ByteBufOutputStream(dataBuffer)
|
||||
);
|
||||
|
||||
snappy.encode(dataBuffer, compressedBuffer, dataBuffer.readableBytes());
|
||||
log(compressedBuffer);
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
ReferenceCountUtil.safeRelease(dataBuffer);
|
||||
ReferenceCountUtil.safeRelease(compressedBuffer);
|
||||
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
/*
|
||||
* Copyright 2012 The Netty Project
|
||||
*
|
||||
* The Netty Project licenses this file to you 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 pl.tkowalcz.tjahzi.http;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
|
||||
/**
|
||||
* Uncompresses an input {@link ByteBuf} encoded with Snappy compression into an
|
||||
* output {@link ByteBuf}.
|
||||
* <p>
|
||||
* See <a href="https://github.com/google/snappy/blob/master/format_description.txt">snappy format</a>.
|
||||
*/
|
||||
public final class Snappy {
|
||||
|
||||
private static final int MAX_HT_SIZE = 1 << 14;
|
||||
private static final int MIN_COMPRESSIBLE_BYTES = 15;
|
||||
|
||||
// constants for the tag types
|
||||
private static final int LITERAL = 0;
|
||||
private static final int COPY_1_BYTE_OFFSET = 1;
|
||||
private static final int COPY_2_BYTE_OFFSET = 2;
|
||||
|
||||
public void encode(final ByteBuf in, final ByteBuf out, final int length) {
|
||||
// Write the preamble length to the output buffer
|
||||
for (int i = 0; ; i++) {
|
||||
int b = length >>> i * 7;
|
||||
if ((b & 0xFFFFFF80) != 0) {
|
||||
out.writeByte(b & 0x7f | 0x80);
|
||||
} else {
|
||||
out.writeByte(b);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
while (in.readableBytes() > 0) {
|
||||
int min = Math.min(Short.MAX_VALUE, in.readableBytes());
|
||||
|
||||
ByteBuf slice = in.readSlice(min);
|
||||
encode0(slice, out, slice.readableBytes());
|
||||
}
|
||||
}
|
||||
|
||||
private void encode0(final ByteBuf in, final ByteBuf out, final int length) {
|
||||
int inIndex = in.readerIndex();
|
||||
final int baseIndex = inIndex;
|
||||
|
||||
final short[] table = getHashTable(length);
|
||||
final int shift = Integer.numberOfLeadingZeros(table.length) + 1;
|
||||
|
||||
int nextEmit = inIndex;
|
||||
|
||||
if (length - inIndex >= MIN_COMPRESSIBLE_BYTES) {
|
||||
int nextHash = hash(in, ++inIndex, shift);
|
||||
outer:
|
||||
while (true) {
|
||||
int skip = 32;
|
||||
|
||||
int candidate;
|
||||
int nextIndex = inIndex;
|
||||
do {
|
||||
inIndex = nextIndex;
|
||||
int hash = nextHash;
|
||||
int bytesBetweenHashLookups = skip++ >> 5;
|
||||
nextIndex = inIndex + bytesBetweenHashLookups;
|
||||
|
||||
// We need at least 4 remaining bytes to read the hash
|
||||
if (nextIndex > length - 4) {
|
||||
break outer;
|
||||
}
|
||||
|
||||
nextHash = hash(in, nextIndex, shift);
|
||||
|
||||
candidate = baseIndex + table[hash];
|
||||
|
||||
table[hash] = (short) (inIndex - baseIndex);
|
||||
}
|
||||
while (in.getInt(inIndex) != in.getInt(candidate));
|
||||
|
||||
encodeLiteral(in, out, inIndex - nextEmit);
|
||||
|
||||
int insertTail;
|
||||
do {
|
||||
int base = inIndex;
|
||||
int matched = 4 + findMatchingLength(in, candidate + 4, inIndex + 4, length);
|
||||
inIndex += matched;
|
||||
int offset = base - candidate;
|
||||
encodeCopy(out, offset, matched);
|
||||
in.readerIndex(in.readerIndex() + matched);
|
||||
insertTail = inIndex - 1;
|
||||
nextEmit = inIndex;
|
||||
if (inIndex >= length - 4) {
|
||||
break outer;
|
||||
}
|
||||
|
||||
int prevHash = hash(in, insertTail, shift);
|
||||
table[prevHash] = (short) (inIndex - baseIndex - 1);
|
||||
int currentHash = hash(in, insertTail + 1, shift);
|
||||
candidate = baseIndex + table[currentHash];
|
||||
table[currentHash] = (short) (inIndex - baseIndex);
|
||||
}
|
||||
while (in.getInt(insertTail + 1) == in.getInt(candidate));
|
||||
|
||||
nextHash = hash(in, insertTail + 2, shift);
|
||||
++inIndex;
|
||||
}
|
||||
}
|
||||
|
||||
// If there are any remaining characters, write them out as a literal
|
||||
if (nextEmit < length) {
|
||||
encodeLiteral(in, out, length - nextEmit);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Hashes the 4 bytes located at index, shifting the resulting hash into
|
||||
* the appropriate range for our hash table.
|
||||
*
|
||||
* @param in The input buffer to read 4 bytes from
|
||||
* @param index The index to read at
|
||||
* @param shift The shift value, for ensuring that the resulting value is
|
||||
* withing the range of our hash table size
|
||||
* @return A 32-bit hash of 4 bytes located at index
|
||||
*/
|
||||
private static int hash(ByteBuf in, int index, int shift) {
|
||||
return in.getInt(index) * 0x1e35a7bd >>> shift;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an appropriately sized hashtable for the given input size
|
||||
*
|
||||
* @param inputSize The size of our input, ie. the number of bytes we need to encode
|
||||
* @return An appropriately sized empty hashtable
|
||||
*/
|
||||
private static short[] getHashTable(int inputSize) {
|
||||
int htSize = 256;
|
||||
while (htSize < MAX_HT_SIZE && htSize < inputSize) {
|
||||
htSize <<= 1;
|
||||
}
|
||||
return new short[htSize];
|
||||
}
|
||||
|
||||
/**
|
||||
* Iterates over the supplied input buffer between the supplied minIndex and
|
||||
* maxIndex to find how long our matched copy overlaps with an already-written
|
||||
* literal value.
|
||||
*
|
||||
* @param in The input buffer to scan over
|
||||
* @param minIndex The index in the input buffer to start scanning from
|
||||
* @param inIndex The index of the start of our copy
|
||||
* @param maxIndex The length of our input buffer
|
||||
* @return The number of bytes for which our candidate copy is a repeat of
|
||||
*/
|
||||
private static int findMatchingLength(ByteBuf in, int minIndex, int inIndex, int maxIndex) {
|
||||
int matched = 0;
|
||||
|
||||
while (inIndex <= maxIndex - 4 &&
|
||||
in.getInt(inIndex) == in.getInt(minIndex + matched)) {
|
||||
inIndex += 4;
|
||||
matched += 4;
|
||||
}
|
||||
|
||||
while (inIndex < maxIndex && in.getByte(minIndex + matched) == in.getByte(inIndex)) {
|
||||
++inIndex;
|
||||
++matched;
|
||||
}
|
||||
|
||||
return matched;
|
||||
}
|
||||
|
||||
/**
|
||||
* Calculates the minimum number of bits required to encode a value. This can
|
||||
* then in turn be used to calculate the number of septets or octets (as
|
||||
* appropriate) to use to encode a length parameter.
|
||||
*
|
||||
* @param value The value to calculate the minimum number of bits required to encode
|
||||
* @return The minimum number of bits required to encode the supplied value
|
||||
*/
|
||||
private static int bitsToEncode(int value) {
|
||||
int highestOneBit = Integer.highestOneBit(value);
|
||||
int bitLength = 0;
|
||||
while ((highestOneBit >>= 1) != 0) {
|
||||
bitLength++;
|
||||
}
|
||||
|
||||
return bitLength;
|
||||
}
|
||||
|
||||
/**
|
||||
* Writes a literal to the supplied output buffer by directly copying from
|
||||
* the input buffer. The literal is taken from the current readerIndex
|
||||
* up to the supplied length.
|
||||
*
|
||||
* @param in The input buffer to copy from
|
||||
* @param out The output buffer to copy to
|
||||
* @param length The length of the literal to copy
|
||||
*/
|
||||
static void encodeLiteral(ByteBuf in, ByteBuf out, int length) {
|
||||
if (length < 61) {
|
||||
out.writeByte(length - 1 << 2);
|
||||
} else {
|
||||
int bitLength = bitsToEncode(length - 1);
|
||||
int bytesToEncode = 1 + bitLength / 8;
|
||||
out.writeByte(59 + bytesToEncode << 2);
|
||||
for (int i = 0; i < bytesToEncode; i++) {
|
||||
out.writeByte(length - 1 >> i * 8 & 0x0ff);
|
||||
}
|
||||
}
|
||||
|
||||
out.writeBytes(in, length);
|
||||
}
|
||||
|
||||
private static void encodeCopyWithOffset(ByteBuf out, int offset, int length) {
|
||||
if (length < 12 && offset < 2048) {
|
||||
out.writeByte(COPY_1_BYTE_OFFSET | length - 4 << 2 | offset >> 8 << 5);
|
||||
out.writeByte(offset & 0x0ff);
|
||||
} else {
|
||||
out.writeByte(COPY_2_BYTE_OFFSET | length - 1 << 2);
|
||||
out.writeByte(offset & 0x0ff);
|
||||
out.writeByte(offset >> 8 & 0x0ff);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Encodes a series of copies, each at most 64 bytes in length.
|
||||
*
|
||||
* @param out The output buffer to write the copy pointer to
|
||||
* @param offset The offset at which the original instance lies
|
||||
* @param length The length of the original instance
|
||||
*/
|
||||
private static void encodeCopy(ByteBuf out, int offset, int length) {
|
||||
while (length >= 68) {
|
||||
encodeCopyWithOffset(out, offset, 64);
|
||||
length -= 64;
|
||||
}
|
||||
|
||||
if (length > 64) {
|
||||
encodeCopyWithOffset(out, offset, 60);
|
||||
length -= 60;
|
||||
}
|
||||
|
||||
encodeCopyWithOffset(out, offset, length);
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
package pl.tkowalcz.tjahzi;
|
||||
|
||||
import logproto.Logproto;
|
||||
import org.agrona.concurrent.UnsafeBuffer;
|
||||
import org.agrona.concurrent.ringbuffer.ManyToOneRingBuffer;
|
||||
import org.agrona.concurrent.ringbuffer.RingBufferDescriptor;
|
||||
@@ -72,7 +73,7 @@ class LogBufferAgentTest {
|
||||
agent.doWork();
|
||||
|
||||
// Then
|
||||
verify(httpClient).log(any());
|
||||
verify(httpClient).log((Logproto.PushRequest.Builder) any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -179,6 +180,6 @@ class LogBufferAgentTest {
|
||||
agent.doWork();
|
||||
|
||||
// Then
|
||||
verify(httpClient, times(2)).log(any());
|
||||
verify(httpClient, times(2)).log((Logproto.PushRequest.Builder) any());
|
||||
}
|
||||
}
|
||||
@@ -77,8 +77,8 @@ public class ReconnectTest {
|
||||
httpClient,
|
||||
monitoringModule,
|
||||
Map.of(),
|
||||
0,
|
||||
0,
|
||||
10_000,
|
||||
1000,
|
||||
1024 * 1024,
|
||||
false
|
||||
);
|
||||
|
||||
@@ -0,0 +1,44 @@
|
||||
package pl.tkowalcz.tjahzi.http;
|
||||
|
||||
import com.google.common.base.Strings;
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.PooledByteBufAllocator;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.xerial.snappy.Snappy;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
class SnappyTest {
|
||||
|
||||
@Test
|
||||
void shouldCompressAndDecompress() throws IOException {
|
||||
// Given
|
||||
String logLine = "Cupcake ipsum dolor sit amet cake wafer. " +
|
||||
"Souffle jelly beans biscuit topping. " +
|
||||
"Danish bonbon gummies powder caramels. " +
|
||||
"Danish jelly beans sweet roll topping jelly beans oat cake toffee. " +
|
||||
"Chocolate cake sesame snaps brownie biscuit cheesecake. " +
|
||||
"Ice cream dessert sweet donut marshmallow. " +
|
||||
"Muffin bear claw cookie jelly-o sugar plum jelly beans apple pie fruitcake cookie. " +
|
||||
"Tootsie roll carrot cake pastry jujubes jelly beans chupa chups. " +
|
||||
"Souffle cake muffin liquorice tart souffle pie sesame snaps.";
|
||||
|
||||
byte[] expected = Strings.repeat(logLine, 10_000).getBytes();
|
||||
|
||||
ByteBuf input = PooledByteBufAllocator.DEFAULT.heapBuffer();
|
||||
input.writeBytes(expected);
|
||||
|
||||
ByteBuf output = PooledByteBufAllocator.DEFAULT.heapBuffer();
|
||||
|
||||
// When
|
||||
new pl.tkowalcz.tjahzi.http.Snappy().encode(input, output, input.readableBytes());
|
||||
byte[] compressed = new byte[output.readableBytes()];
|
||||
output.readBytes(compressed);
|
||||
|
||||
// Then
|
||||
byte[] actual = Snappy.uncompress(compressed);
|
||||
assertThat(actual).isEqualTo(expected);
|
||||
}
|
||||
}
|
||||
@@ -5,7 +5,7 @@
|
||||
<parent>
|
||||
<groupId>pl.tkowalcz</groupId>
|
||||
<artifactId>tjahzi</artifactId>
|
||||
<version>0.7-SNAPSHOT</version>
|
||||
<version>0.8.2-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<name>tjahzi-loki-protobuf</name>
|
||||
|
||||
Reference in new issue
Block a user