Merge branch 'develop' into 'master'

Release 0.51.0-SNAPSHOT See merge request hercules/hercules!377
vostok · Dec 15, 2021 · d0747df · d0747df
2 parents e6a91a5 + 59451b1
commit d0747df
Show file tree

Hide file tree

Showing 76 changed files with 582 additions and 470 deletions.
diff --git a/Makefile b/Makefile
@@ -1,20 +1,21 @@
 VERSION := $(shell mvn help:evaluate -Dexpression=project.version -q -DforceStdout)
 
-SERVICES := hercules-gate \
-	    hercules-management-api \
-	    hercules-stream-manager \
-	    hercules-stream-api \
-	    hercules-stream-sink \
-	    hercules-timeline-manager \
-	    hercules-timeline-api \
-	    hercules-timeline-sink \
-	    hercules-tracing-api \
-	    hercules-tracing-sink \
-	    hercules-tracing-sink-clickhouse \
-	    hercules-elastic-sink \
-	    hercules-sentry-sink \
-	    hercules-graphite-sink \
-	    hercules-elastic-adapter
+SERVICES := hercules-elastic-adapter \
+        hercules-elastic-sink \
+        hercules-gate \
+        hercules-graphite-adapter \
+        hercules-graphite-sink \
+        hercules-management-api \
+        hercules-sentry-sink \
+        hercules-stream-api \
+        hercules-stream-manager \
+        hercules-stream-sink \
+        hercules-timeline-api \
+        hercules-timeline-manager \
+        hercules-timeline-sink \
+        hercules-tracing-api \
+        hercules-tracing-sink \
+        hercules-tracing-sink-clickhouse
 
 
 .PHONY: all build_images push_images

diff --git a/hercules-application/pom.xml b/hercules-application/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-application/src/main/java/ru/kontur/vostok/hercules/application/Application.java b/hercules-application/src/main/java/ru/kontur/vostok/hercules/application/Application.java
@@ -57,7 +57,6 @@ public static void run(String applicationName, String applicationId, String[] ar
 
         Map<String, String> parameters = ArgsParser.parse(args);
         Properties properties = PropertiesLoader.load(parameters.getOrDefault("application.properties", "file://application.properties"));
-        LOGGER.info(PropertiesUtil.prettyView(properties));
 
         ApplicationConfig applicationConfig = new ApplicationConfig(PropertiesUtil.ofScope(properties, Scopes.APPLICATION));
 

diff --git a/...s-application/src/main/java/ru/kontur/vostok/hercules/application/ApplicationContext.java b/...s-application/src/main/java/ru/kontur/vostok/hercules/application/ApplicationContext.java
@@ -1,5 +1,8 @@
 package ru.kontur.vostok.hercules.application;
 
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
 /**
  * @author Gregory Koshelev
  */
@@ -13,6 +16,8 @@ public class ApplicationContext {
     private final String instanceId;
     private final String hostname;
 
+    private final ConcurrentMap<String, String> additionalContext = new ConcurrentHashMap<>();
+
     public ApplicationContext(
             String applicationName,
             String applicationId,
@@ -63,4 +68,26 @@ public String getInstanceId() {
     public String getHostname() {
         return hostname;
     }
+
+    /**
+     * Returns additional context value by the key.
+     * These values can be exchanged thread-safely between different components.
+     *
+     * @param key the key whose associated value is to be returned
+     * @return the value to which the specified key is mapped, or
+     *         {@code null} if this map contains no mapping for the key
+     */
+    public String get(String key) {
+        return additionalContext.get(key);
+    }
+
+    /**
+     * Associates the specified additional context value with the specified key.
+     *
+     * @param key   key with which the specified value is to be associated
+     * @param value value to be associated with the specified key
+     */
+    public void put(String key, String value) {
+        additionalContext.put(key, value);
+    }
 }
diff --git a/hercules-auth/pom.xml b/hercules-auth/pom.xml
@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-cassandra-sink/pom.xml b/hercules-cassandra-sink/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-cassandra-util/pom.xml b/hercules-cassandra-util/pom.xml
@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-clickhouse-sink/pom.xml b/hercules-clickhouse-sink/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-clickhouse-util/pom.xml b/hercules-clickhouse-util/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-client/pom.xml b/hercules-client/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-configuration/pom.xml b/hercules-configuration/pom.xml
@@ -3,7 +3,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-curator/pom.xml b/hercules-curator/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/hercules-elastic-adapter/pom.xml b/hercules-elastic-adapter/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

diff --git a/...er/src/main/java/ru/kontur/vostok/hercules/elastic/adapter/ElasticAdapterApplication.java b/...er/src/main/java/ru/kontur/vostok/hercules/elastic/adapter/ElasticAdapterApplication.java
@@ -1,11 +1,7 @@
 package ru.kontur.vostok.hercules.elastic.adapter;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import ru.kontur.vostok.hercules.application.Application;
-import ru.kontur.vostok.hercules.configuration.PropertiesLoader;
 import ru.kontur.vostok.hercules.configuration.Scopes;
-import ru.kontur.vostok.hercules.configuration.util.ArgsParser;
 import ru.kontur.vostok.hercules.elastic.adapter.gate.GateSender;
 import ru.kontur.vostok.hercules.elastic.adapter.handler.BulkHandler;
 import ru.kontur.vostok.hercules.elastic.adapter.handler.IndexHandler;
@@ -18,88 +14,32 @@
 import ru.kontur.vostok.hercules.http.handler.InstrumentedRouteHandlerBuilder;
 import ru.kontur.vostok.hercules.util.properties.PropertiesUtil;
 
-import java.util.Map;
 import java.util.Properties;
-import java.util.concurrent.TimeUnit;
 
 /**
  * @author Gregory Koshelev
  */
 public class ElasticAdapterApplication {
-    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticAdapterApplication.class);
-
     private static MetricsCollector metricsCollector;
     private static GateSender gateSender;
     private static IndexManager indexManager;
-    private static HttpServer httpServer;
 
     public static void main(String[] args) {
-        long start = System.currentTimeMillis();
-
-        try {
-            Application.run("Hercules Elastic Adapter", "elastic-adapter", args);
-
-            Map<String, String> parameters = ArgsParser.parse(args);
-
-            Properties properties = PropertiesLoader.load(parameters.getOrDefault("application.properties", "file://application.properties"));
-
+        Application.run("Hercules Elastic Adapter", "elastic-adapter", args, (properties, container) -> {
             Properties metricsProperties = PropertiesUtil.ofScope(properties, Scopes.METRICS);
             Properties gateClientProperties = PropertiesUtil.ofScope(properties, Scopes.GATE_CLIENT);
             Properties indexManagerProperties = PropertiesUtil.ofScope(properties, "index.manager");
             Properties httpServerProperties = PropertiesUtil.ofScope(properties, Scopes.HTTP_SERVER);
 
-            metricsCollector = new MetricsCollector(metricsProperties);
-            metricsCollector.start();
+            metricsCollector = container.register(new MetricsCollector(metricsProperties));
             CommonMetrics.registerCommonMetrics(metricsCollector);
 
-            gateSender = new GateSender(gateClientProperties);
+            gateSender = container.register(new GateSender(gateClientProperties));
 
             indexManager = new IndexManager(indexManagerProperties);
 
-            httpServer = createHttpServer(httpServerProperties);
-            httpServer.start();
-
-        } catch (Throwable t) {
-            LOGGER.error("Cannot start application due to", t);
-            shutdown();
-            return;
-        }
-
-        Runtime.getRuntime().addShutdownHook(new Thread(ElasticAdapterApplication::shutdown));
-
-        LOGGER.info("Elastic Adapter started for {} millis", System.currentTimeMillis() - start);
-    }
-
-    private static void shutdown() {
-        long start = System.currentTimeMillis();
-
-        LOGGER.info("Started Elastic Adapter shutdown");
-
-        try {
-            if (httpServer != null) {
-                httpServer.stop(5_000, TimeUnit.MILLISECONDS);
-            }
-        } catch (Throwable t) {
-            LOGGER.error("Error on http server shutdown", t);
-        }
-
-        try {
-            if (gateSender != null) {
-                gateSender.close();
-            }
-        } catch (Throwable t) {
-            LOGGER.error("Error onn gate client serice shutdown", t);
-        }
-
-        try {
-            if (metricsCollector != null) {
-                metricsCollector.stop();
-            }
-        } catch (Throwable t) {
-            LOGGER.error("Error on metrics collector shutdown", t);
-        }
-
-        LOGGER.info("Finished Elastic Adapter shutdown for {} millis", System.currentTimeMillis() - start);
+            container.register(createHttpServer(httpServerProperties));
+        });
     }
 
     private static HttpServer createHttpServer(Properties httpServerProperties) {

diff --git a/...stic-adapter/src/main/java/ru/kontur/vostok/hercules/elastic/adapter/gate/GateSender.java b/...stic-adapter/src/main/java/ru/kontur/vostok/hercules/elastic/adapter/gate/GateSender.java
@@ -11,14 +11,15 @@
 import ru.kontur.vostok.hercules.util.parameter.Parameter;
 import ru.kontur.vostok.hercules.util.properties.PropertiesUtil;
 
+import java.io.Closeable;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Properties;
 
 /**
  * @author Gregory Koshelev
  */
-public class GateSender {
+public class GateSender implements Closeable {
     private static final Logger LOGGER = LoggerFactory.getLogger(GateSender.class);
 
     private final String apiKey;
@@ -51,6 +52,7 @@ public GateStatus send(List<Event> events, boolean async, String stream) {
         }
     }
 
+    @Override
     public void close() {
         client.close();
     }

diff --git a/hercules-elastic-sink/pom.xml b/hercules-elastic-sink/pom.xml
@@ -5,7 +5,7 @@
     <parent>
         <artifactId>hercules</artifactId>
         <groupId>ru.kontur.vostok.hercules</groupId>
-        <version>0.50.0-SNAPSHOT</version>
+        <version>0.51.0-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -44,6 +44,11 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-inline</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>

diff --git a/...es-elastic-sink/src/main/java/ru/kontur/vostok/hercules/elastic/sink/LeproserySender.java b/...es-elastic-sink/src/main/java/ru/kontur/vostok/hercules/elastic/sink/LeproserySender.java
@@ -2,6 +2,7 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import ru.kontur.vostok.hercules.application.Application;
 import ru.kontur.vostok.hercules.configuration.Scopes;
 import ru.kontur.vostok.hercules.gate.client.GateClient;
 import ru.kontur.vostok.hercules.gate.client.exception.BadRequestException;
@@ -15,12 +16,15 @@
 import ru.kontur.vostok.hercules.protocol.EventBuilder;
 import ru.kontur.vostok.hercules.protocol.TinyString;
 import ru.kontur.vostok.hercules.protocol.Variant;
+import ru.kontur.vostok.hercules.sink.SinkContext;
 import ru.kontur.vostok.hercules.tags.CommonTags;
 import ru.kontur.vostok.hercules.tags.ElasticSearchTags;
 import ru.kontur.vostok.hercules.tags.LogEventTags;
+import ru.kontur.vostok.hercules.util.Lazy;
 import ru.kontur.vostok.hercules.util.concurrent.Topology;
 import ru.kontur.vostok.hercules.util.parameter.Parameter;
 import ru.kontur.vostok.hercules.util.properties.PropertiesUtil;
+import ru.kontur.vostok.hercules.util.text.StringUtil;
 import ru.kontur.vostok.hercules.util.time.TimeUtil;
 import ru.kontur.vostok.hercules.util.validation.ValidationResult;
 import ru.kontur.vostok.hercules.uuid.UuidGenerator;
@@ -48,6 +52,9 @@ class LeproserySender {
     private static final TinyString SERVICE = TinyString.of("service");
     private static final TinyString TEXT = TinyString.of("text");
     private static final TinyString ORIGINAL_INDEX = TinyString.of("original-index");
+    private static final TinyString ELASTIC_SINK_ID = TinyString.of("elastic-sink-id");
+    private static final TinyString ELASTIC_SINK_GROUP_ID = TinyString.of("elastic-sink-groupId");
+    private static final TinyString ELASTIC_SINK_SUBSCRIPTION = TinyString.of("elastic-sink-subscription");
 
     private static final int EMPTY_LEPROSERY_EVENT_SIZE_BYTES = 125;
     private static final int MAX_EVENT_SIZE_BYTES = 500_000;
@@ -57,6 +64,9 @@ class LeproserySender {
     private final String leproseryIndex;
     private final GateClient gateClient;
     private final String leproseryApiKey;
+    private final String sinkId;
+    private final Lazy<String> sinkGroupId;
+    private final Lazy<String> sinkSubscription;
 
     private final Meter sentToLeproseryEventCountMeter;
     private final Meter sentToLeproseryWithErrorsEventCountMeter;
@@ -68,6 +78,9 @@ class LeproserySender {
         this.leproseryStream = PropertiesUtil.get(Props.LEPROSERY_STREAM, properties).get();
         this.leproseryIndex = PropertiesUtil.get(Props.LEPROSERY_INDEX, properties).get();
         this.leproseryApiKey = PropertiesUtil.get(Props.LEPROSERY_API_KEY, properties).get();
+        this.sinkId = StringUtil.getOrDefault(Application.context().getInstanceId(), "null");
+        this.sinkGroupId = new Lazy<>(() -> StringUtil.getOrDefault(Application.context().get(SinkContext.GROUP_ID), "null"));
+        this.sinkSubscription = new Lazy<>(() -> StringUtil.getOrDefault(Application.context().get(SinkContext.SUBSCRIPTION), "null"));
 
         Properties gateProperties = PropertiesUtil.ofScope(properties, Scopes.GATE_CLIENT);
         final String[] urls = PropertiesUtil.get(Props.URLS, gateProperties).get();
@@ -144,6 +157,9 @@ public Optional<Event> toLeproseryEvent(ElasticDocument document, String error)
                         .tag(TEXT, Variant.ofString(textBytes))
                         .tag(ORIGINAL_INDEX, Variant.ofString(document.index()))
                         .tag(ElasticSearchTags.ELK_INDEX_TAG.getName(), Variant.ofString(leproseryIndex))
+                        .tag(ELASTIC_SINK_ID, Variant.ofString(sinkId))
+                        .tag(ELASTIC_SINK_GROUP_ID, Variant.ofString(sinkGroupId.get()))
+                        .tag(ELASTIC_SINK_SUBSCRIPTION, Variant.ofString(sinkSubscription.get()))
                         .build()))
                 .build());
     }