Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

LC-416 added json logging for elk and setting mdc values #100

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
+74 −2
Draft

LC-416 added json logging for elk and setting mdc values #100

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
aec5963
Add json logging for elk including setting mdc values for run id, doc…
kiratraynor Mar 5, 2024
d6647ea
Ensures log level is debug for additional json logs.
kiratraynor Mar 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion lucille-core/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,12 @@
</exclusions>
</dependency>


<!-- dependency to allow json logging for log4j -->
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-layout-template-json</artifactId>
<version>2.22.1</version>
</dependency>

<dependency>
<groupId>org.apache.kafka</groupId>
Expand Down
2 changes: 1 addition & 1 deletion lucille-core/src/main/java/com/kmwllc/lucille/core/Indexer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ private void checkForDoc() {
sendToIndexWithAccounting(batch.flushIfExpired());
return;
}

log.debug("Indexing Doc {}", doc.getId());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the call to sendToIndexWithAccounting is not necessarily the time when the doc is sent to the search engine
batch.add(doc) will typically add the doc to a batch being built up, and return an empty list so the sendToIndexWithAccounting call is effectively a no-op
batch.add(doc) returns a non-empty list only in certain situations, like when the batch is full and ready to be flushed

sendToIndexWithAccounting(batch.add(doc));
}

Expand Down
4 changes: 4 additions & 0 deletions lucille-core/src/main/java/com/kmwllc/lucille/core/PublisherImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.slf4j.MDC;

/**
* Publisher implementation that maintains an in-memory list of pending documents.
Expand Down Expand Up @@ -93,6 +94,7 @@ public PublisherImpl(Config config, PublisherMessenger messenger, String runId,

@Override
public void publish(Document document) throws Exception {
MDC.put("docId", document.getId());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

in local mode, the publisher will be running in one thread, some number of workers will be running in their own threads, and an indexer will be running in yet another thread. I don't think setting the docId in the MDC here will propagate the change to other threads where docId is needed, but even if it did, the propogated docId would be "right" for those other threads because everything is happening concurrently -- one doc3 might be published at the same time worker 1 is processing doc2 and the indexer is indexing doc1...

if (firstDocStopWatch.isStarted()) {
firstDocStopWatch.stop();
log.info("First doc published after " + firstDocStopWatch.getTime(TimeUnit.MILLISECONDS) + " ms");
Expand All @@ -111,6 +113,7 @@ public void publish(Document document) throws Exception {
}

private void publishInternal(Document document) throws Exception {
log.debug("Publishing document with id {}", document.getId());
if (!isCollapsing) {
sendForProcessing(document);
return;
Expand All @@ -122,6 +125,7 @@ private void publishInternal(Document document) throws Exception {
}

if (previousDoc.getId().equals(document.getId())) {
log.debug("Sending document.");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this clause isn't when we're sending the document
this clause is when we're "collapsing" the current document into the previous one because they had the same IDs

previousDoc.setOrAddAll(document);
} else {
sendForProcessing(previousDoc);
Expand Down
2 changes: 2 additions & 0 deletions lucille-core/src/main/java/com/kmwllc/lucille/core/Runner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
import java.io.StringWriter;
import java.util.*;
import java.util.concurrent.TimeUnit;
import org.slf4j.MDC;

/**
* Executes a Lucille run. A run is a sequential execution of one or more Connectors.
Expand Down Expand Up @@ -219,6 +220,7 @@ public static Map<String, List<Exception>> validatePipelines(Config config) thro
*/
public static RunResult run(Config config, RunType type) throws Exception {
String runId = UUID.randomUUID().toString();
MDC.put("runId", runId);
log.info("Starting run with id " + runId);

List<Connector> connectors = Connector.fromConfig(config);
Expand Down
2 changes: 2 additions & 0 deletions lucille-core/src/main/java/com/kmwllc/lucille/core/Stage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.slf4j.MDC;

/**
* An operation that can be performed on a Document.<br>
Expand Down Expand Up @@ -287,6 +288,7 @@ public void initialize(int position, String metricsPrefix) throws StageException
if (name == null) {
this.name = "stage_" + position;
}
MDC.put("stage", this.name);

MetricRegistry metrics = SharedMetricRegistries.getOrCreate(LogUtils.METRICS_REG);
this.timer = metrics.timer(metricsPrefix + ".stage." + name + ".processDocumentTime");
Expand Down
10 changes: 10 additions & 0 deletions lucille-core/src/main/resources/log4j2.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,22 @@
</Policies>
<DefaultRolloverStrategy max="4" />
</RollingFile>
<RollingFile name="elk" fileName="./log/com.kmwllc.lucille-json.log" filePattern="./log/com.kmwllc.lucille-json-%i.log">
<JsonTemplateLayout eventTemplateUri="classpath:EcsLayout.json"/>
<Policies>
<SizeBasedTriggeringPolicy size="10 KB" />
</Policies>
<DefaultRolloverStrategy max="20" />
</RollingFile>
</Appenders>
<Loggers>
<Root level="INFO">
<AppenderRef ref="console" />
<AppenderRef ref="file" />
</Root>
<!-- <Root level="DEBUG">-->
<!-- <AppenderRef ref="elk" />-->
<!-- </Root>-->
<Logger name="com.kmwllc.com.kmwllc.lucille.core.Heartbeat" level="INFO" additivity="false">
<AppenderRef ref="heartbeat" />
</Logger>
Expand Down
49 changes: 49 additions & 0 deletions lucille-core/src/test/resources/EcsLayout.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
{
"@timestamp": {
"$resolver": "timestamp",
"pattern": {
"format": "yy/MM/dd'T'HH:mm:ss'Z'",
"timeZone": "UTC"
}
},
"ecs.version": "1.2.0",
"log.level": {
"$resolver": "level",
"field": "name"
},
"message": {
"$resolver": "message",
"stringified": true
},
"process.thread.name": {
"$resolver": "thread",
"field": "name"
},
"log.logger": {
"$resolver": "logger",
"field": "name"
},
"labels": {
"$resolver": "mdc",
"flatten": true,
"stringified": true
},
"tags": {
"$resolver": "ndc"
},
"error.type": {
"$resolver": "exception",
"field": "className"
},
"error.message": {
"$resolver": "exception",
"field": "message"
},
"error.stack_trace": {
"$resolver": "exception",
"field": "stackTrace",
"stackTrace": {
"stringified": true
}
}
}
Loading