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 2022/06/03 02:54:59 UTC
[skywalking] 01/01: Add on demand logs support to service mesh
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch ondemandlog/als
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit d61fa22c07a3014f43d3d7d2ac2f7f6d59ff6191
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Fri Jun 3 10:54:33 2022 +0800
Add on demand logs support to service mesh
---
apm-protocol/apm-network/src/main/proto | 2 +-
.../oap/server/core/query/type/Logs.java | 3 ++
.../query/graphql/resolver/OndemandLogQuery.java | 37 ++++++++++++++--------
.../oap/query/graphql/type/PodContainers.java | 34 ++++++++++++++++++++
.../src/main/resources/query-protocol | 2 +-
.../envoy/als/LogEntry2MetricsAdapter.java | 21 ++++++++++++
.../receiver/envoy/als/k8s/K8SServiceRegistry.java | 15 ++++++---
.../server/receiver/envoy/als/mx/FieldsHelper.java | 12 +++++++
.../receiver/mesh/TelemetryDataDispatcher.java | 22 +++++++++++++
9 files changed, 128 insertions(+), 20 deletions(-)
diff --git a/apm-protocol/apm-network/src/main/proto b/apm-protocol/apm-network/src/main/proto
index e89af18cc7..8f92ea931a 160000
--- a/apm-protocol/apm-network/src/main/proto
+++ b/apm-protocol/apm-network/src/main/proto
@@ -1 +1 @@
-Subproject commit e89af18cc7be7dc04ec4a7c4ff75f0ae19fa929b
+Subproject commit 8f92ea931a814e0c85e38a741b76c93bf28d296c
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/Logs.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/Logs.java
index 2d5213183e..e9d6759a50 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/Logs.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/Logs.java
@@ -22,11 +22,14 @@ import java.util.ArrayList;
import java.util.List;
import lombok.Getter;
import lombok.Setter;
+import lombok.experimental.Accessors;
@Setter
@Getter
+@Accessors(chain = true)
public class Logs {
private final List<Log> logs;
+ private String errorReason;
public Logs() {
this.logs = new ArrayList<>();
diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/OndemandLogQuery.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/OndemandLogQuery.java
index f453a81355..d149a65628 100644
--- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/OndemandLogQuery.java
+++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/OndemandLogQuery.java
@@ -36,6 +36,7 @@ import org.apache.skywalking.oap.query.graphql.type.InternalLog;
import org.apache.skywalking.oap.query.graphql.type.LogAdapter;
import org.apache.skywalking.oap.query.graphql.type.OndemandContainergQueryCondition;
import org.apache.skywalking.oap.query.graphql.type.OndemandLogQueryCondition;
+import org.apache.skywalking.oap.query.graphql.type.PodContainers;
import org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic.PropertyUtil;
import org.apache.skywalking.oap.server.core.query.input.Duration;
import org.apache.skywalking.oap.server.core.query.type.Attribute;
@@ -65,7 +66,7 @@ public class OndemandLogQuery implements GraphQLQueryResolver {
private final MetadataQueryV2 metadataQuery;
- public List<String> listContainers(final OndemandContainergQueryCondition condition)
+ public PodContainers listContainers(final OndemandContainergQueryCondition condition)
throws IOException {
final ServiceInstance instance =
metadataQuery.getInstance(condition.getServiceInstanceId());
@@ -104,17 +105,18 @@ public class OndemandLogQuery implements GraphQLQueryResolver {
return attributesMap;
}
- public List<String> listContainers(
+ public PodContainers listContainers(
final String namespace,
final String podName) throws IOException {
try {
if (Strings.isNullOrEmpty(namespace) || Strings.isNullOrEmpty(podName)) {
- return Collections.emptyList();
+ return new PodContainers()
+ .setErrorReason("namespace and podName can't be null or empty");
}
final V1Pod pod = kApi().readNamespacedPod(podName, namespace, null);
final V1PodSpec spec = pod.getSpec();
if (isNull(spec)) {
- return Collections.emptyList();
+ return new PodContainers().setErrorReason("No pod spec can be found");
}
final List<String> containers = spec.getContainers().stream()
@@ -127,13 +129,16 @@ public class OndemandLogQuery implements GraphQLQueryResolver {
containers.addAll(init);
}
- return containers;
+ return new PodContainers().setContainers(containers);
} catch (ApiException e) {
log.error("Failed to list containers from Kubernetes, {}", e.getResponseBody(), e);
- Map<String, Object> responseBody = gson.fromJson(e.getResponseBody(), responseType);
- String message = responseBody.getOrDefault("message", e.getCode()).toString();
- throw new RuntimeException(message);
+ if (!Strings.isNullOrEmpty(e.getResponseBody())) {
+ Map<String, Object> responseBody = gson.fromJson(e.getResponseBody(), responseType);
+ String message = responseBody.getOrDefault("message", e.getCode()).toString();
+ return new PodContainers().setErrorReason(message);
+ }
+ return new PodContainers().setErrorReason(e.getMessage());
}
}
@@ -141,15 +146,18 @@ public class OndemandLogQuery implements GraphQLQueryResolver {
final String namespace,
final String podName,
final OndemandLogQueryCondition condition) throws IOException {
+ if (Strings.isNullOrEmpty(namespace) || Strings.isNullOrEmpty(podName)) {
+ return new Logs().setErrorReason("namespace and podName can't be null or empty");
+ }
try {
final V1Pod pod = kApi().readNamespacedPod(podName, namespace, null);
final V1ObjectMeta podMetadata = pod.getMetadata();
if (isNull(podMetadata)) {
- return new Logs();
+ return new Logs().setErrorReason("No pod metadata can be found");
}
final V1PodSpec spec = pod.getSpec();
if (isNull(spec)) {
- return new Logs();
+ return new Logs().setErrorReason("No pod spec can be found");
}
final Duration duration = new Duration();
@@ -188,9 +196,12 @@ public class OndemandLogQuery implements GraphQLQueryResolver {
} catch (ApiException e) {
log.error("Failed to fetch logs from Kubernetes, {}", e.getResponseBody(), e);
- Map<String, Object> responseBody = gson.fromJson(e.getResponseBody(), responseType);
- String message = responseBody.getOrDefault("message", e.getCode()).toString();
- throw new RuntimeException(message);
+ if (!Strings.isNullOrEmpty(e.getResponseBody())) {
+ Map<String, Object> responseBody = gson.fromJson(e.getResponseBody(), responseType);
+ String message = responseBody.getOrDefault("message", e.getCode()).toString();
+ return new Logs().setErrorReason(message);
+ }
+ return new Logs().setErrorReason(e.getMessage());
}
}
diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/PodContainers.java b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/PodContainers.java
new file mode 100644
index 0000000000..de52d7f0e6
--- /dev/null
+++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/type/PodContainers.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.skywalking.oap.query.graphql.type;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.experimental.Accessors;
+
+@Getter
+@Setter
+@Accessors(chain = true)
+public class PodContainers {
+ private String errorReason;
+ private List<String> containers = new ArrayList<>();
+}
diff --git a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol
index faf3a76808..6f23bf5b2b 160000
--- a/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol
+++ b/oap-server/server-query-plugin/query-graphql-plugin/src/main/resources/query-protocol
@@ -1 +1 @@
-Subproject commit faf3a768087825258f64dd782a4dc405dc271cf8
+Subproject commit 6f23bf5b2b510f93320fe1b086075e062b0e8ea2
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java
index dd87332815..29a0ffc46d 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java
@@ -30,6 +30,7 @@ import java.util.List;
import java.util.Optional;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.apm.network.common.v3.DetectPoint;
+import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
@@ -129,6 +130,26 @@ public class LogEntry2MetricsAdapter {
.map(ServiceMetaInfo::getServiceInstanceName)
.ifPresent(builder::setDestServiceInstance);
+ Optional
+ .ofNullable(sourceService)
+ .map(ServiceMetaInfo::getTags)
+ .ifPresent(tags -> {
+ tags.forEach(p -> {
+ builder.addSourceInstanceProperties(
+ KeyStringValuePair.newBuilder().setKey(p.getKey()).setValue(p.getValue()));
+ });
+ });
+
+ Optional
+ .ofNullable(targetService)
+ .map(ServiceMetaInfo::getTags)
+ .ifPresent(tags -> {
+ tags.forEach(p -> {
+ builder.addDestInstanceProperties(
+ KeyStringValuePair.newBuilder().setKey(p.getKey()).setValue(p.getValue()));
+ });
+ });
+
return builder;
}
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java
index 9198397bed..97d98705f1 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java
@@ -34,7 +34,7 @@ import io.kubernetes.client.openapi.models.V1Service;
import io.kubernetes.client.openapi.models.V1ServiceList;
import io.kubernetes.client.util.Config;
import java.io.IOException;
-import java.util.Collections;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -46,6 +46,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
+
import static com.google.common.base.Strings.isNullOrEmpty;
import static java.util.Objects.isNull;
import static java.util.Optional.ofNullable;
@@ -262,14 +263,18 @@ public class K8SServiceRegistry {
));
}
- protected List<ServiceMetaInfo.KeyValue> transformLabelsToTags(final Map<String, String> labels) {
+ protected List<ServiceMetaInfo.KeyValue> transformLabelsToTags(final V1ObjectMeta podMeta) {
+ final Map<String, String> labels = podMeta.getLabels();
+ final List<ServiceMetaInfo.KeyValue> tags = new ArrayList<>();
+ tags.add(new ServiceMetaInfo.KeyValue("pod", podMeta.getName()));
+ tags.add(new ServiceMetaInfo.KeyValue("namespace", podMeta.getNamespace()));
if (isNull(labels)) {
- return Collections.emptyList();
+ return tags;
}
return labels.entrySet()
.stream()
.map(each -> new ServiceMetaInfo.KeyValue(each.getKey(), each.getValue()))
- .collect(Collectors.toList());
+ .collect(Collectors.toCollection(() -> tags));
}
public ServiceMetaInfo findService(final String ip) {
@@ -315,7 +320,7 @@ public class K8SServiceRegistry {
}
serviceMetaInfo.setServiceInstanceName(
String.format("%s.%s", podMetadata.getName(), podMetadata.getNamespace()));
- serviceMetaInfo.setTags(transformLabelsToTags(podMetadata.getLabels()));
+ serviceMetaInfo.setTags(transformLabelsToTags(podMetadata));
return serviceMetaInfo;
});
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java
index e0da74ce12..567870015b 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java
@@ -37,6 +37,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.library.util.ResourceUtils;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
+import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo.KeyValue;
import org.yaml.snakeyaml.Yaml;
import com.google.common.base.Splitter;
@@ -174,6 +175,17 @@ public enum FieldsHelper {
fieldSetterMapping.get(entry.getKey()).accept(serviceMetaInfo, value);
}
}
+ final Map<String, Value> fieldsMap = metadata.getFieldsMap();
+ final List<KeyValue> tags = new ArrayList<>();
+ if (fieldsMap.containsKey("NAME")) {
+ tags.add(new KeyValue("pod", fieldsMap.get("NAME").getStringValue()));
+ }
+ if (fieldsMap.containsKey("NAMESPACE")) {
+ tags.add(new KeyValue("namespace", fieldsMap.get("NAMESPACE").getStringValue()));
+ }
+ if (!tags.isEmpty()) {
+ serviceMetaInfo.setTags(tags);
+ }
}
@RequiredArgsConstructor
diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java
index 3d9cbae287..001b03e3c0 100644
--- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java
+++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java
@@ -23,8 +23,11 @@ import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.analysis.manual.instance.InstanceTraffic;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
import org.apache.skywalking.oap.server.core.config.NamingControl;
import org.apache.skywalking.oap.server.core.source.DetectPoint;
import org.apache.skywalking.oap.server.core.source.Endpoint;
@@ -41,6 +44,7 @@ import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
+import com.google.gson.JsonObject;
/**
* TelemetryDataDispatcher processes the {@link ServiceMeshMetric} format telemetry data, transfers it to source
@@ -110,6 +114,7 @@ public class TelemetryDataDispatcher {
if (org.apache.skywalking.apm.network.common.v3.DetectPoint.server.equals(metrics.getDetectPoint())) {
toService(metrics, minuteTimeBucket);
toServiceInstance(metrics, minuteTimeBucket);
+ toServiceInstanceTraffic(metrics, minuteTimeBucket);
toEndpoint(metrics, minuteTimeBucket);
}
@@ -187,6 +192,23 @@ public class TelemetryDataDispatcher {
SOURCE_RECEIVER.receive(serviceInstance);
}
+ private static void toServiceInstanceTraffic(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ InstanceTraffic instanceTraffic = new InstanceTraffic();
+ instanceTraffic.setTimeBucket(minuteTimeBucket);
+ instanceTraffic.setName(metrics.getDestServiceInstance());
+ instanceTraffic.setServiceId(IDManager.ServiceID.buildId(metrics.getDestServiceName(), true));
+ instanceTraffic.setLastPingTimestamp(TimeBucket.getMinuteTimeBucket(System.currentTimeMillis()));
+ if (metrics.getDestInstancePropertiesList() != null) {
+ final JsonObject properties = new JsonObject();
+ metrics
+ .getDestInstancePropertiesList()
+ .stream()
+ .forEach(it -> properties.addProperty(it.getKey(), it.getValue()));
+ instanceTraffic.setProperties(properties);
+ }
+ MetricsStreamProcessor.getInstance().in(instanceTraffic);
+ }
+
private static void toServiceInstanceRelation(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
ServiceInstanceRelation serviceRelation = new ServiceInstanceRelation();
serviceRelation.setTimeBucket(minuteTimeBucket);