You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2021/07/13 03:48:59 UTC
[skywalking] branch perf/envoy-log updated (d723fd9 -> 0ebe4c6)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a change to branch perf/envoy-log
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
discard d723fd9 perf: optimize Envoy access log persistence performance
new 0ebe4c6 perf: optimize Envoy access log persistence performance
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (d723fd9)
\
N -- N -- N refs/heads/perf/envoy-log (0ebe4c6)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../skywalking/oap/log/analyzer/dsl/Binding.java | 18 +++++++++---------
1 file changed, 9 insertions(+), 9 deletions(-)
[skywalking] 01/01: perf: optimize Envoy access log persistence
performance
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch perf/envoy-log
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 0ebe4c62d8b14935dcd6deb65ad069ec8fa0c4e1
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Mon Jul 12 21:30:01 2021 +0800
perf: optimize Envoy access log persistence performance
---
.github/workflows/e2e.istio.yaml | 3 ++
.github/workflows/e2e.log.yaml | 3 ++
.../skywalking/oap/log/analyzer/dsl/Binding.java | 47 +++++++++++++++++++---
.../log/analyzer/dsl/spec/filter/FilterSpec.java | 20 ++++++++-
.../analyzer/provider/log/ILogAnalyzerService.java | 7 ++--
.../oap/log/analyzer/provider/log/LogAnalyzer.java | 9 +++--
.../provider/log/LogAnalyzerServiceImpl.java | 5 ++-
.../LogAnalyzerFactory.java} | 18 ++-------
.../provider/log/listener/LogAnalysisListener.java | 3 +-
.../provider/log/listener/LogFilterListener.java | 7 +++-
.../log/listener/RecordAnalysisListener.java | 4 +-
.../log/listener/TrafficAnalysisListener.java | 4 +-
.../src/main/resources/lal/envoy-als.yaml | 12 +++---
.../analysis/manual/log/AbstractLogRecord.java | 1 +
.../agent/kafka/provider/handler/LogHandler.java | 2 +-
.../envoy/persistence/LogsPersistence.java | 22 ++--------
.../envoy/persistence/TCPLogsPersistence.java | 22 ++--------
.../handler/grpc/LogReportServiceHandler.java | 2 +-
.../handler/rest/LogReportServiceRestHandler.java | 2 +-
.../plugin/elasticsearch/query/LogQueryEsDAO.java | 2 +
20 files changed, 114 insertions(+), 81 deletions(-)
diff --git a/.github/workflows/e2e.istio.yaml b/.github/workflows/e2e.istio.yaml
index e635a1d..a8d1622 100644
--- a/.github/workflows/e2e.istio.yaml
+++ b/.github/workflows/e2e.istio.yaml
@@ -23,6 +23,9 @@ on:
- '!**.md'
schedule:
- cron: '0 18 * * *'
+ push:
+ branches:
+ - perf/envoy-log # TODO
env:
SKIP_TEST: true
diff --git a/.github/workflows/e2e.log.yaml b/.github/workflows/e2e.log.yaml
index 116b15e..8027454 100644
--- a/.github/workflows/e2e.log.yaml
+++ b/.github/workflows/e2e.log.yaml
@@ -23,6 +23,9 @@ on:
- '!**.md'
schedule:
- cron: '0 18 * * *'
+ push:
+ branches:
+ - perf/envoy-log # TODO
env:
SW_AGENT_JDK_VERSION: 8
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/Binding.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/Binding.java
index 7502865..55582ac 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/Binding.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/Binding.java
@@ -18,6 +18,10 @@
package org.apache.skywalking.oap.log.analyzer.dsl;
+import com.google.protobuf.Message;
+import groovy.lang.Closure;
+import groovy.lang.GroovyObjectSupport;
+import groovy.lang.MissingPropertyException;
import java.util.Map;
import java.util.regex.Matcher;
import lombok.Getter;
@@ -44,6 +48,7 @@ public class Binding extends groovy.lang.Binding {
setProperty(KEY_LOG, log);
setProperty(KEY_SAVE, true);
setProperty(KEY_ABORT, false);
+ parsed().log = log;
return this;
}
@@ -55,6 +60,15 @@ public class Binding extends groovy.lang.Binding {
return (LogData.Builder) getProperty(KEY_LOG);
}
+ public Binding extraLog(final Message extraLog) {
+ parsed().extraLog = extraLog;
+ return this;
+ }
+
+ public Message extraLog() {
+ return parsed().getExtraLog();
+ }
+
public Binding parsed(final Matcher parsed) {
parsed().matcher = parsed;
return this;
@@ -92,19 +106,32 @@ public class Binding extends groovy.lang.Binding {
return (boolean) getProperty(KEY_ABORT);
}
- public static class Parsed {
+ public static class Parsed extends GroovyObjectSupport {
@Getter
private Matcher matcher;
@Getter
private Map<String, Object> map;
+ @Getter
+ private Message.Builder log;
+
+ @Getter
+ private Message extraLog;
+
public Object getAt(final String key) {
- if (matcher != null) {
- return matcher.group(key);
+ Object result;
+ if (matcher != null && (result = matcher.group(key)) != null) {
+ return result;
+ }
+ if (map != null && (result = map.get(key)) != null) {
+ return result;
}
- if (map != null) {
- return map.get(key);
+ if (extraLog != null && (result = getField(extraLog, key)) != null) {
+ return result;
+ }
+ if (log != null && (result = getField(log, key)) != null) {
+ return result;
}
return null;
}
@@ -113,5 +140,15 @@ public class Binding extends groovy.lang.Binding {
public Object propertyMissing(final String name) {
return getAt(name);
}
+
+ static Object getField(Object obj, String name) {
+ try {
+ Closure<?> c = new Closure<Object>(obj, obj) {
+ };
+ return c.getProperty(name);
+ } catch (MissingPropertyException ignored) {
+ }
+ return null;
+ }
}
}
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/spec/filter/FilterSpec.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/spec/filter/FilterSpec.java
index 43f86a3..0eaceec 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/spec/filter/FilterSpec.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/dsl/spec/filter/FilterSpec.java
@@ -19,13 +19,17 @@
package org.apache.skywalking.oap.log.analyzer.dsl.spec.filter;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.protobuf.Message;
import com.google.protobuf.TextFormat;
import groovy.lang.Closure;
import groovy.lang.DelegatesTo;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import lombok.SneakyThrows;
+import org.apache.skywalking.apm.network.logging.v3.JSONLog;
import org.apache.skywalking.apm.network.logging.v3.LogData;
+import org.apache.skywalking.apm.network.logging.v3.LogDataBody;
import org.apache.skywalking.oap.log.analyzer.dsl.Binding;
import org.apache.skywalking.oap.log.analyzer.dsl.spec.AbstractSpec;
import org.apache.skywalking.oap.log.analyzer.dsl.spec.extractor.ExtractorSpec;
@@ -42,6 +46,8 @@ import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils.toJSON;
+
public class FilterSpec extends AbstractSpec {
private static final Logger LOGGER = LoggerFactory.getLogger(FilterSpec.class);
@@ -143,6 +149,7 @@ public class FilterSpec extends AbstractSpec {
cl.call();
}
+ @SneakyThrows
@SuppressWarnings("unused")
public void sink(@DelegatesTo(SinkSpec.class) final Closure<?> cl) {
if (BINDING.get().shouldAbort()) {
@@ -153,6 +160,7 @@ public class FilterSpec extends AbstractSpec {
final Binding b = BINDING.get();
final LogData.Builder logData = b.log();
+ final Message extraLog = b.extraLog();
if (!b.shouldSave()) {
if (LOGGER.isDebugEnabled()) {
@@ -161,9 +169,19 @@ public class FilterSpec extends AbstractSpec {
return;
}
+ if (extraLog != null) {
+ logData.setBody(
+ LogDataBody.newBuilder()
+ .setJson(
+ JSONLog.newBuilder()
+ .setJson(toJSON(extraLog))
+ .build())
+ .build()
+ );
+ }
factories.stream()
.map(LogAnalysisListenerFactory::create)
- .forEach(it -> it.parse(logData).build());
+ .forEach(it -> it.parse(logData, extraLog).build());
}
@SuppressWarnings("unused")
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java
index 6a8ba52..653cf6d 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java
@@ -17,6 +17,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log;
+import com.google.protobuf.Message;
import org.apache.skywalking.apm.network.logging.v3.LogData;
import org.apache.skywalking.oap.server.library.module.Service;
@@ -25,10 +26,10 @@ import org.apache.skywalking.oap.server.library.module.Service;
*/
public interface ILogAnalyzerService extends Service {
- void doAnalysis(LogData.Builder log);
+ void doAnalysis(LogData.Builder log, Message extraLog);
- default void doAnalysis(LogData logData) {
- doAnalysis(logData.toBuilder());
+ default void doAnalysis(LogData logData, Message extraLog) {
+ doAnalysis(logData.toBuilder(), extraLog);
}
}
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzer.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzer.java
index 172b82f..0b35719 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzer.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzer.java
@@ -17,6 +17,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log;
+import com.google.protobuf.Message;
import java.util.ArrayList;
import java.util.List;
import lombok.RequiredArgsConstructor;
@@ -39,7 +40,7 @@ public class LogAnalyzer {
private final List<LogAnalysisListener> listeners = new ArrayList<>();
- public void doAnalysis(LogData.Builder builder) {
+ public void doAnalysis(LogData.Builder builder, Message extraLog) {
if (StringUtil.isEmpty(builder.getService())) {
// If the service name is empty, the log will be ignored.
log.debug("The log is ignored because the Service name is empty");
@@ -51,12 +52,12 @@ public class LogAnalyzer {
builder.setTimestamp(System.currentTimeMillis());
}
- notifyListener(builder);
+ notifyListener(builder, extraLog);
notifyListenerToBuild();
}
- private void notifyListener(LogData.Builder builder) {
- listeners.forEach(listener -> listener.parse(builder));
+ private void notifyListener(LogData.Builder builder, final Message extraLog) {
+ listeners.forEach(listener -> listener.parse(builder, extraLog));
}
private void notifyListenerToBuild() {
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzerServiceImpl.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzerServiceImpl.java
index ad41a3c..09e1a68 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzerServiceImpl.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/LogAnalyzerServiceImpl.java
@@ -17,6 +17,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log;
+import com.google.protobuf.Message;
import java.util.ArrayList;
import java.util.List;
import lombok.RequiredArgsConstructor;
@@ -32,9 +33,9 @@ public class LogAnalyzerServiceImpl implements ILogAnalyzerService, ILogAnalysis
private final List<LogAnalysisListenerFactory> factories = new ArrayList<>();
@Override
- public void doAnalysis(final LogData.Builder log) {
+ public void doAnalysis(final LogData.Builder log, Message extraLog) {
LogAnalyzer analyzer = new LogAnalyzer(moduleManager, moduleConfig, this);
- analyzer.doAnalysis(log);
+ analyzer.doAnalysis(log, extraLog);
}
@Override
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/analyzer/LogAnalyzerFactory.java
similarity index 61%
copy from oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java
copy to oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/analyzer/LogAnalyzerFactory.java
index 6a8ba52..13b6104 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/ILogAnalyzerService.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/analyzer/LogAnalyzerFactory.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -15,20 +15,8 @@
* limitations under the License.
*/
-package org.apache.skywalking.oap.log.analyzer.provider.log;
+package org.apache.skywalking.oap.log.analyzer.provider.log.analyzer;
-import org.apache.skywalking.apm.network.logging.v3.LogData;
-import org.apache.skywalking.oap.server.library.module.Service;
-
-/**
- * Analyze the collected log data.
- */
-public interface ILogAnalyzerService extends Service {
-
- void doAnalysis(LogData.Builder log);
-
- default void doAnalysis(LogData logData) {
- doAnalysis(logData.toBuilder());
- }
+public class LogAnalyzerFactory {
}
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogAnalysisListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogAnalysisListener.java
index 8e1e96f..e5273e4 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogAnalysisListener.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogAnalysisListener.java
@@ -17,6 +17,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log.listener;
+import com.google.protobuf.Message;
import org.apache.skywalking.apm.network.logging.v3.LogData;
/**
@@ -33,5 +34,5 @@ public interface LogAnalysisListener {
* Parse the raw data from the probe.
* @return {@code this} for chaining.
*/
- LogAnalysisListener parse(LogData.Builder logData);
+ LogAnalysisListener parse(LogData.Builder logData, final Message extraLog);
}
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogFilterListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogFilterListener.java
index e5d715b..27a5ec3 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogFilterListener.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/LogFilterListener.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log.listener;
+import com.google.protobuf.Message;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
@@ -48,8 +49,10 @@ public class LogFilterListener implements LogAnalysisListener {
}
@Override
- public LogAnalysisListener parse(final LogData.Builder logData) {
- dsls.forEach(dsl -> dsl.bind(new Binding().log(logData.build())));
+ public LogAnalysisListener parse(final LogData.Builder logData,
+ final Message extraLog) {
+ dsls.forEach(dsl -> dsl.bind(new Binding().log(logData.build())
+ .extraLog(extraLog)));
return this;
}
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/RecordAnalysisListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/RecordAnalysisListener.java
index cf5b667..f291122 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/RecordAnalysisListener.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/RecordAnalysisListener.java
@@ -17,6 +17,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log.listener;
+import com.google.protobuf.Message;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
@@ -57,7 +58,8 @@ public class RecordAnalysisListener implements LogAnalysisListener {
}
@Override
- public LogAnalysisListener parse(final LogData.Builder logData) {
+ public LogAnalysisListener parse(final LogData.Builder logData,
+ final Message extraLog) {
LogDataBody body = logData.getBody();
log.setUniqueId(UUID.randomUUID().toString().replace("-", ""));
// timestamp
diff --git a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/TrafficAnalysisListener.java b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/TrafficAnalysisListener.java
index 0cf6590..615abd4 100644
--- a/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/TrafficAnalysisListener.java
+++ b/oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/TrafficAnalysisListener.java
@@ -17,6 +17,7 @@
package org.apache.skywalking.oap.log.analyzer.provider.log.listener;
+import com.google.protobuf.Message;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.apm.network.logging.v3.LogData;
import org.apache.skywalking.apm.util.StringUtil;
@@ -61,7 +62,8 @@ public class TrafficAnalysisListener implements LogAnalysisListener {
}
@Override
- public LogAnalysisListener parse(final LogData.Builder logData) {
+ public LogAnalysisListener parse(final LogData.Builder logData,
+ final Message extraLog) {
final long timeBucket = TimeBucket.getTimeBucket(System.currentTimeMillis(), DownSampling.Minute);
// to service traffic
String serviceName = namingControl.formatServiceName(logData.getService());
diff --git a/oap-server/server-bootstrap/src/main/resources/lal/envoy-als.yaml b/oap-server/server-bootstrap/src/main/resources/lal/envoy-als.yaml
index aad0a61..ceb42f4 100644
--- a/oap-server/server-bootstrap/src/main/resources/lal/envoy-als.yaml
+++ b/oap-server/server-bootstrap/src/main/resources/lal/envoy-als.yaml
@@ -17,24 +17,22 @@ rules:
- name: envoy-als
dsl: |
filter {
- json {
- }
// only collect abnormal logs (http status code >= 300, or commonProperties?.responseFlags is not empty)
- if (parsed?.response?.responseCode as Integer < 400 && !parsed?.commonProperties?.responseFlags) {
+ if (parsed?.response?.responseCode?.value as Integer < 400) {
abort {}
}
extractor {
if (parsed?.response?.responseCode) {
- tag 'status.code': parsed?.response?.responseCode as int
+ tag 'status.code': parsed?.response?.responseCode?.value
}
- tag 'response.flag': (parsed?.commonProperties?.responseFlags as Map)?.keySet()
+ tag 'response.flag': parsed?.commonProperties?.responseFlags
}
sink {
sampler {
- if (parsed?.commonProperties?.responseFlags) {
+ if (parsed?.commonProperties?.responseFlags?.toString()) {
// use service:errorCode as sampler id so that each service:errorCode has its own sampler,
// e.g. checkoutservice:[upstreamConnectionFailure], checkoutservice:[upstreamRetryLimitExceeded]
- rateLimit("${log.service}:${(parsed?.commonProperties?.responseFlags as Map)?.keySet()}") {
+ rateLimit("${log.service}:${parsed?.commonProperties?.responseFlags?.toString()}") {
qps 100
}
} else {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java
index 58555c9..8988d00 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/log/AbstractLogRecord.java
@@ -44,6 +44,7 @@ public abstract class AbstractLogRecord extends Record {
public static final String SPAN_ID = "span_id";
public static final String CONTENT_TYPE = "content_type";
public static final String CONTENT = "content";
+ public static final String CONTENT_TYPE_CLASS = "content_type_class";
public static final String TAGS_RAW_DATA = "tags_raw_data";
public static final String TIMESTAMP = "timestamp";
public static final String TAGS = "tags";
diff --git a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java
index 6ec4af5..d7b1760 100644
--- a/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java
+++ b/oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/LogHandler.java
@@ -77,7 +77,7 @@ public class LogHandler implements KafkaHandler {
public void handle(final ConsumerRecord<String, Bytes> record) {
try (HistogramMetrics.Timer ignore = histogram.createTimer()) {
LogData logData = parseConsumerRecord(record);
- logAnalyzerService.doAnalysis(logData);
+ logAnalyzerService.doAnalysis(logData, null);
} catch (Exception e) {
errorCounter.inc();
log.error(e.getMessage(), e);
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java
index 7908ed4..824af8a 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/LogsPersistence.java
@@ -21,9 +21,7 @@ package org.apache.skywalking.oap.server.receiver.envoy.persistence;
import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.logging.v3.JSONLog;
import org.apache.skywalking.apm.network.logging.v3.LogData;
-import org.apache.skywalking.apm.network.logging.v3.LogDataBody;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
import org.apache.skywalking.oap.log.analyzer.module.LogAnalyzerModule;
import org.apache.skywalking.oap.log.analyzer.provider.log.ILogAnalyzerService;
@@ -35,8 +33,6 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapt
import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
-import static org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils.toJSON;
-
/**
* {@code LogsPersistence} analyzes the error logs and persists them to the log system.
*/
@@ -69,7 +65,7 @@ public class LogsPersistence implements ALSHTTPAnalysis {
}
final LogData logData = convertToLogData(entry, result);
- logAnalyzerService.doAnalysis(logData);
+ logAnalyzerService.doAnalysis(logData, entry);
} catch (final Exception e) {
log.error("Failed to persist Envoy access log", e);
}
@@ -81,27 +77,17 @@ public class LogsPersistence implements ALSHTTPAnalysis {
return prev;
}
- public LogData convertToLogData(final HTTPAccessLogEntry logEntry,
- final Result result) throws Exception {
-
+ public LogData convertToLogData(final HTTPAccessLogEntry logEntry, final Result result) {
final ServiceMetaInfo service = result.getService();
- final ServiceMeshMetric.Builder metrics = new LogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart();
+ final ServiceMeshMetric.Builder metrics =
+ new LogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart();
return LogData
.newBuilder()
.setService(service.getServiceName())
.setServiceInstance(service.getServiceInstanceName())
.setTimestamp(metrics.getEndTime())
- .setBody(
- LogDataBody
- .newBuilder()
- .setJson(
- JSONLog
- .newBuilder()
- .setJson(toJSON(logEntry))
- )
- )
.build();
}
}
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java
index 4928639..1d2f67d 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/persistence/TCPLogsPersistence.java
@@ -21,9 +21,7 @@ package org.apache.skywalking.oap.server.receiver.envoy.persistence;
import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.logging.v3.JSONLog;
import org.apache.skywalking.apm.network.logging.v3.LogData;
-import org.apache.skywalking.apm.network.logging.v3.LogDataBody;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
import org.apache.skywalking.oap.log.analyzer.module.LogAnalyzerModule;
import org.apache.skywalking.oap.log.analyzer.provider.log.ILogAnalyzerService;
@@ -35,8 +33,6 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
import org.apache.skywalking.oap.server.receiver.envoy.als.tcp.TCPAccessLogAnalyzer;
import org.apache.skywalking.oap.server.receiver.envoy.als.tcp.TCPLogEntry2MetricsAdapter;
-import static org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils.toJSON;
-
/**
* {@code LogsPersistence} analyzes the error logs and persists them to the log system.
*/
@@ -69,7 +65,7 @@ public class TCPLogsPersistence implements TCPAccessLogAnalyzer {
}
final LogData logData = convertToLogData(entry, result);
- logAnalyzerService.doAnalysis(logData);
+ logAnalyzerService.doAnalysis(logData, entry);
} catch (final Exception e) {
log.error("Failed to persist Envoy access log", e);
}
@@ -81,27 +77,17 @@ public class TCPLogsPersistence implements TCPAccessLogAnalyzer {
return prev;
}
- public LogData convertToLogData(final TCPAccessLogEntry logEntry,
- final Result result) throws Exception {
-
+ public LogData convertToLogData(final TCPAccessLogEntry logEntry, final Result result) {
final ServiceMetaInfo service = result.getService();
- final ServiceMeshMetric.Builder metrics = new TCPLogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart();
+ final ServiceMeshMetric.Builder metrics =
+ new TCPLogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart();
return LogData
.newBuilder()
.setService(service.getServiceName())
.setServiceInstance(service.getServiceInstanceName())
.setTimestamp(metrics.getEndTime())
- .setBody(
- LogDataBody
- .newBuilder()
- .setJson(
- JSONLog
- .newBuilder()
- .setJson(toJSON(logEntry))
- )
- )
.build();
}
}
diff --git a/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/grpc/LogReportServiceHandler.java b/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/grpc/LogReportServiceHandler.java
index e47b363..2760376 100644
--- a/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/grpc/LogReportServiceHandler.java
+++ b/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/grpc/LogReportServiceHandler.java
@@ -89,7 +89,7 @@ public class LogReportServiceHandler extends LogReportServiceGrpc.LogReportServi
try {
LogData.Builder builder = logData.toBuilder();
setServiceName(builder);
- logAnalyzerService.doAnalysis(builder);
+ logAnalyzerService.doAnalysis(builder, null);
} catch (Exception e) {
errorCounter.inc();
log.error(e.getMessage(), e);
diff --git a/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/rest/LogReportServiceRestHandler.java b/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/rest/LogReportServiceRestHandler.java
index 72931ba..6d3da98 100644
--- a/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/rest/LogReportServiceRestHandler.java
+++ b/oap-server/server-receiver-plugin/skywalking-log-recevier-plugin/src/main/java/org/apache/skywalking/oap/server/recevier/log/provider/handler/rest/LogReportServiceRestHandler.java
@@ -81,7 +81,7 @@ public class LogReportServiceRestHandler extends JettyHandler {
ProtoBufJsonUtils.fromJSON(it.toString(), builder);
logs.add(builder);
}
- logs.forEach(logAnalyzerService::doAnalysis);
+ logs.forEach(it -> logAnalyzerService.doAnalysis(it, null));
} catch (final Exception e) {
log.error(e.getMessage(), e);
errorCounter.inc();
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java
index cd0194d..d5f116d 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/LogQueryEsDAO.java
@@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.query;
import java.io.IOException;
import java.util.List;
+import lombok.SneakyThrows;
import org.apache.skywalking.oap.server.core.analysis.manual.log.AbstractLogRecord;
import org.apache.skywalking.oap.server.core.analysis.manual.log.LogRecord;
import org.apache.skywalking.oap.server.core.analysis.manual.searchtag.Tag;
@@ -58,6 +59,7 @@ public class LogQueryEsDAO extends EsDAO implements ILogQueryDAO {
}
@Override
+ @SneakyThrows
public Logs queryLogs(final String serviceId,
final String serviceInstanceId,
final String endpointId,