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:58 UTC
[skywalking] branch ondemandlog/als created (now d61fa22c07)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a change to branch ondemandlog/als
in repository https://gitbox.apache.org/repos/asf/skywalking.git
at d61fa22c07 Add on demand logs support to service mesh
This branch includes the following new commits:
new d61fa22c07 Add on demand logs support to service mesh
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.
[skywalking] 01/01: Add on demand logs support to service mesh
Posted by ke...@apache.org.
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);