Merge branch 'develop' into 'master'

Release 0.36.0-SNAPSHOT

See merge request hercules/hercules!232

hercules-application/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-auth/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-cassandra-sink/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-cassandra-sink/src/main/java/ru/kontur/vostok/hercules/cassandra/sink/CassandraSender.java

@@ -75,8 +75,6 @@ public CassandraSender(Properties properties, MetricsCollector metricsCollector)
 
     @Override
     public void start() {
-        cassandraConnector.connect();
-
         CqlSession session = cassandraConnector.session();
         preparedStatement = session.prepare(query());
 

hercules-cassandra-util/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-cassandra-util/src/main/java/ru/kontur/vostok/hercules/cassandra/util/CassandraConnector.java

@@ -67,9 +67,11 @@ public CassandraConnector(Properties properties) {
         this.consistencyLevel = PropertiesUtil.get(Props.CONSISTENCY_LEVEL, properties).get();
 
         this.batchSizeBytesLimit = PropertiesUtil.get(Props.BATCH_SIZE_BYTES_LIMIT, properties).get();
+
+        init();
     }
 
-    public void connect() {
+    private void init() {
         DriverConfigLoader configLoader = DriverConfigLoader.programmaticBuilder().
                 withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofMillis(requestTimeoutMs)).
                 withString(DefaultDriverOption.REQUEST_CONSISTENCY, consistencyLevel).

hercules-clickhouse-sink/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-clickhouse-sink/src/main/java/ru/kontur/vostok/hercules/clickhouse/sink/ClickHouseSender.java

@@ -3,6 +3,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import ru.kontur.vostok.hercules.clickhouse.util.ClickHouseConnector;
+import ru.kontur.vostok.hercules.configuration.Scopes;
 import ru.kontur.vostok.hercules.health.AutoMetricStopwatch;
 import ru.kontur.vostok.hercules.health.MetricsCollector;
 import ru.kontur.vostok.hercules.health.Timer;
@@ -42,7 +43,7 @@ public abstract class ClickHouseSender extends Sender {
     public ClickHouseSender(Properties properties, MetricsCollector metricsCollector) {
         super(properties, metricsCollector);
 
-        Properties clickhouseProperties = PropertiesUtil.ofScope(properties, "clickhouse");
+        Properties clickhouseProperties = PropertiesUtil.ofScope(properties, Scopes.CLICKHOUSE);
         this.connector = new ClickHouseConnector(clickhouseProperties);
 
         this.processingTimeMsTimer = metricsCollector.timer("processingTimeMs");

hercules-clickhouse-util/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-configuration/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-configuration/src/main/java/ru/kontur/vostok/hercules/configuration/Scopes.java

@@ -7,6 +7,7 @@ public final class Scopes {
     public static final String ZOOKEEPER = "zk";
     public static final String KAFKA = "kafka";
     public static final String CASSANDRA = "cassandra";
+    public static final String CLICKHOUSE = "clickhouse";
 
     public static final String CURATOR = "curator";
 

hercules-curator/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-elastic-sink/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-gate-client/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-gate/README.md

@@ -141,6 +141,10 @@ HTTP Server binds on host:port are defined in Main Application settings.
 
 `http.server.throttling.requestTimeout` - timeout for request, which capacity throttling more then permissible, default value: `5000`
 
+### Validation settings
+
+`validation.max.event.size` - max size of Hercules event, value must be consistent with broker setting `max.message.bytes`, default value: `500000`
+
 ### Kafka Producer settings
 See Producer's Config from Apache Kafka documentation. Main settings are presented below.
 
@@ -218,6 +222,8 @@ http.server.connection.threshold=100000
 http.server.throttling.capacity=1073741824
 http.server.throttling.requestTimeout=10000
 
+validation.max.event.size=500000
+
 producer.bootstrap.servers=localhost:9092
 producer.acks=all
 producer.batch.size=65536

hercules-gate/application.properties

@@ -6,6 +6,8 @@ http.server.connection.threshold=100000
 http.server.throttling.capacity=1073741824
 http.server.throttling.requestTimeout=10000
 
+validation.max.event.size=500000
+
 producer.bootstrap.servers=localhost:9092
 producer.acks=all
 producer.batch.size=65536

hercules-gate/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-gate/src/main/java/ru/kontur/vostok/hercules/gate/GateApplication.java

@@ -11,6 +11,7 @@
 import ru.kontur.vostok.hercules.configuration.Scopes;
 import ru.kontur.vostok.hercules.configuration.util.ArgsParser;
 import ru.kontur.vostok.hercules.curator.CuratorClient;
+import ru.kontur.vostok.hercules.gate.validation.EventValidator;
 import ru.kontur.vostok.hercules.health.CommonMetrics;
 import ru.kontur.vostok.hercules.health.MetricsCollector;
 import ru.kontur.vostok.hercules.http.HttpServer;
@@ -45,6 +46,7 @@ public class GateApplication {
     private static MetricsCollector metricsCollector;
     private static HttpServer server;
     private static EventSender eventSender;
+    private static EventValidator eventValidator;
     private static CuratorClient curatorClient;
     private static AuthManager authManager;
     private static AuthValidationManager authValidationManager;
@@ -65,12 +67,14 @@ public static void main(String[] args) {
             Properties curatorProperties = PropertiesUtil.ofScope(properties, Scopes.CURATOR);
             Properties metricsProperties = PropertiesUtil.ofScope(properties, Scopes.METRICS);
             Properties sdProperties = PropertiesUtil.ofScope(properties, Scopes.SERVICE_DISCOVERY);
+            Properties validationProperties = PropertiesUtil.ofScope(properties, "validation");
 
             metricsCollector = new MetricsCollector(metricsProperties);
             metricsCollector.start();
             CommonMetrics.registerCommonMetrics(metricsCollector);
 
             eventSender = new EventSender(producerProperties, new HashPartitioner(new NaiveHasher()), metricsCollector);
+            eventValidator = new EventValidator(validationProperties);
 
             curatorClient = new CuratorClient(curatorProperties);
             curatorClient.start();
@@ -164,13 +168,13 @@ private static void shutdown() {
         LOGGER.info("Finished Gateway shutdown for {}  millis", System.currentTimeMillis() - start);
     }
 
-    private static HttpServer createHttpServer(Properties httpServerProperies) {
+    private static HttpServer createHttpServer(Properties httpServerProperties) {
         StreamRepository streamRepository = new StreamRepository(curatorClient);
         StreamStorage streamStorage = new StreamStorage(streamRepository, 30_000L /* TODO: for test usages; It should be moved to configuration */);
 
-        Properties throttlingProperties = PropertiesUtil.ofScope(httpServerProperies, Scopes.THROTTLING);
+        Properties throttlingProperties = PropertiesUtil.ofScope(httpServerProperties, Scopes.THROTTLING);
 
-        SendRequestProcessor sendRequestProcessor = new SendRequestProcessor(eventSender, metricsCollector);
+        SendRequestProcessor sendRequestProcessor = new SendRequestProcessor(eventSender, eventValidator, metricsCollector);
         CapacityThrottle<HttpServerRequest, SendContext> throttle = new CapacityThrottle<>(
                 throttlingProperties,
                 new DefaultHttpServerRequestWeigher(),
@@ -180,7 +184,7 @@ private static HttpServer createHttpServer(Properties httpServerProperies) {
         metricsCollector.gauge("throttling.totalCapacity", throttle::totalCapacity);
         metricsCollector.gauge("throttling.availableCapacity", throttle::availableCapacity);
 
-        long maxContentLength = PropertiesUtil.get(HttpServer.Props.MAX_CONTENT_LENGTH, httpServerProperies).get();
+        long maxContentLength = PropertiesUtil.get(HttpServer.Props.MAX_CONTENT_LENGTH, httpServerProperties).get();
 
         AuthProvider authProvider = new AuthProvider(new AdminAuthManager(Collections.emptySet()), authManager);
         HandlerWrapper authHandlerWrapper = new OrdinaryAuthHandlerWrapper(authProvider);
@@ -190,15 +194,15 @@ private static HttpServer createHttpServer(Properties httpServerProperies) {
         HttpHandler sendHandler = authHandlerWrapper.wrap(
                 new GateHandler(authProvider, throttle, authValidationManager, streamStorage, false, maxContentLength, metricsCollector));
 
-        RouteHandler handler = new InstrumentedRouteHandlerBuilder(httpServerProperies, metricsCollector).
+        RouteHandler handler = new InstrumentedRouteHandlerBuilder(httpServerProperties, metricsCollector).
                 post("/stream/sendAsync", sendAsyncHandler).
                 post("/stream/send", sendHandler).
                 build();
 
         return new UndertowHttpServer(
                 Application.application().getConfig().getHost(),
                 Application.application().getConfig().getPort(),
-                httpServerProperies,
+                httpServerProperties,
                 handler);
     }
 }

hercules-gate/src/main/java/ru/kontur/vostok/hercules/gate/SendRequestProcessor.java

@@ -48,19 +48,20 @@ public class SendRequestProcessor implements RequestProcessor<HttpServerRequest,
     private final EventSender eventSender;
     private final TimeSource time;
 
-    private final EventValidator eventValidator = new EventValidator();
+    private final EventValidator eventValidator;
     private final Lz4Decompressor lz4Decompressor = new Lz4Decompressor();
 
     private final Timer readEventsDurationMsTimer;
     private final Meter sentEventsMeter;
     private final Timer decompressionTimeMsTimer;
 
-    public SendRequestProcessor(EventSender eventSender, MetricsCollector metricsCollector) {
-        this(eventSender, metricsCollector, TimeSource.SYSTEM);
+    public SendRequestProcessor(EventSender eventSender, EventValidator eventValidator, MetricsCollector metricsCollector) {
+        this(eventSender, eventValidator, metricsCollector, TimeSource.SYSTEM);
     }
 
-    SendRequestProcessor(EventSender eventSender, MetricsCollector metricsCollector, TimeSource time) {
+    SendRequestProcessor(EventSender eventSender, EventValidator eventValidator, MetricsCollector metricsCollector, TimeSource time) {
         this.eventSender = eventSender;
+        this.eventValidator = eventValidator;
         this.time = time;
 
         this.readEventsDurationMsTimer = metricsCollector.timer(this.getClass().getSimpleName() + ".readEventsDurationMs");

hercules-gate/src/main/java/ru/kontur/vostok/hercules/gate/validation/EventValidator.java

@@ -3,13 +3,24 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import ru.kontur.vostok.hercules.protocol.Event;
+import ru.kontur.vostok.hercules.util.parameter.Parameter;
+import ru.kontur.vostok.hercules.util.properties.PropertiesUtil;
+import ru.kontur.vostok.hercules.util.validation.IntegerValidators;
+
+import java.util.Properties;
 
 /**
  * @author Gregory Koshelev
  */
 public class EventValidator {
     private static final Logger LOGGER = LoggerFactory.getLogger(EventValidator.class);
 
+    private final int maxEventSize;
+
+    public EventValidator(Properties properties) {
+        this.maxEventSize = PropertiesUtil.get(Props.MAX_EVENT_SIZE, properties).get();
+    }
+
     public boolean validate(Event event) {
         if (event.getVersion() != 1) { // Gate supports Event of version 1 only
             LOGGER.warn("Event version != 1");
@@ -21,6 +32,20 @@ public boolean validate(Event event) {
             return false;
         }
 
+        int eventSize = event.getBytes().length;
+        if (eventSize > maxEventSize) {
+            LOGGER.warn("Event size = {} bytes, more then limit = {}", eventSize, maxEventSize);
+            return false;
+        }
+
         return true;
     }
+
+    private static class Props {
+        static final Parameter<Integer> MAX_EVENT_SIZE =
+                Parameter.integerParameter("max.event.size").
+                        withValidator(IntegerValidators.positive()).
+                        withDefault(500_000).
+                        build();
+    }
 }

hercules-graphite-sink/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-health/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-http/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-init/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-json/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-kafka-util/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

hercules-management-api/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.35.1-SNAPSHOT</version>
+        <version>0.36.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>