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/10/28 01:14:55 UTC
[skywalking] branch master updated: Split TCP related service mesh metrics protobuffer definition (#9851)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new fb2ac0e7f9 Split TCP related service mesh metrics protobuffer definition (#9851)
fb2ac0e7f9 is described below
commit fb2ac0e7f95b312e80c0c03eb1e121ef5ab862b8
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Fri Oct 28 09:14:38 2022 +0800
Split TCP related service mesh metrics protobuffer definition (#9851)
---
apm-protocol/apm-network/src/main/proto | 2 +-
docs/en/changes/changes.md | 4 +
docs/en/concepts-and-designs/scope-definitions.md | 141 +++++++++----
.../apache/skywalking/oal/rt/grammar/OALLexer.g4 | 6 +-
.../apache/skywalking/oal/rt/grammar/OALParser.g4 | 6 +-
.../server/core/analysis/DispatcherManager.java | 8 +-
.../instance/TCPInstanceTrafficDispatcher.java} | 29 ++-
.../instance/TCPInstanceUpdateDispatcher.java | 36 ++++
.../TCPServiceInstanceCallRelationDispatcher.java | 62 ++++++
.../service/TCPServiceCallRelationDispatcher.java | 58 ++++++
.../service/TCPServiceTrafficDispatcher.java} | 23 +--
.../oap/server/core/source/DefaultScopeDefine.java | 6 +
.../skywalking/oap/server/core/source/Service.java | 3 -
.../oap/server/core/source/ServiceInstance.java | 4 -
.../core/source/ServiceInstanceRelation.java | 3 -
.../oap/server/core/source/ServiceRelation.java | 3 -
.../core/source/{Service.java => TCPService.java} | 33 +--
...erviceInstance.java => TCPServiceInstance.java} | 32 +--
...lation.java => TCPServiceInstanceRelation.java} | 37 +---
...{Service.java => TCPServiceInstanceUpdate.java} | 62 +-----
...erviceRelation.java => TCPServiceRelation.java} | 33 +--
.../envoy/AccessLogServiceGRPCHandler.java | 56 ++---
.../receiver/envoy/als/AbstractALSAnalyzer.java | 6 +-
.../receiver/envoy/als/AccessLogAnalyzer.java | 15 +-
.../envoy/als/LogEntry2MetricsAdapter.java | 29 +--
.../als/k8s/K8sALSServiceMeshHTTPAnalysis.java | 54 +++--
.../envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java | 38 ++--
.../als/tcp/AbstractTCPAccessLogAnalyzer.java | 6 +-
.../envoy/als/tcp/TCPLogEntry2MetricsAdapter.java | 35 ++--
.../als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java | 54 +++--
.../tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java | 42 ++--
.../envoy/persistence/LogsPersistence.java | 4 +-
.../envoy/persistence/TCPLogsPersistence.java | 10 +-
.../als/k8s/K8SALSServiceMeshHTTPAnalysisTest.java | 20 +-
.../aop/server/receiver/mesh/MeshGRPCHandler.java | 10 +-
.../server/receiver/mesh/MeshReceiverProvider.java | 1 -
.../receiver/mesh/TelemetryDataDispatcher.java | 227 +++++++++++++++++----
.../oap/server/receiver/mesh/MeshDataMock.java | 53 +++--
.../server-starter/src/main/resources/oal/core.oal | 5 +-
.../server-starter/src/main/resources/oal/tcp.oal | 24 +--
.../general/general-instance.json | 1 -
.../mesh/mesh-instance.json | 1 -
.../mesh/mesh-service.json | 10 +-
.../ServiceMeshMetricSenderController.java | 46 +++--
.../java-test-service/e2e-protocol/src/main/proto | 2 +-
45 files changed, 803 insertions(+), 537 deletions(-)
diff --git a/apm-protocol/apm-network/src/main/proto b/apm-protocol/apm-network/src/main/proto
index 906a834dc0..c2c9bbbff4 160000
--- a/apm-protocol/apm-network/src/main/proto
+++ b/apm-protocol/apm-network/src/main/proto
@@ -1 +1 @@
-Subproject commit 906a834dc0c3fb66719b8313f8f9298be3a27349
+Subproject commit c2c9bbbff43f0ad9f917a0a55285538a6b45739e
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index 3200fa1bfb..03066c9ebc 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -71,6 +71,10 @@
* Remove improper self-obs metrics in JvmMetricsHandler(for Kafka channel).
* gRPC stream canceling code is not logged as an error when the client cancels the stream. The client
cancels the stream when the pod is terminated.
+* [**Breaking Change**] Refactor service mesh protobuf definitions and split TCP-related metrics to individual definition.
+* Add `TCP{Service,ServiceInstance,ServiceRelation,ServiceInstanceRelation}` sources and split TCP-related entities out from
+ original `Service,ServiceInstance,ServiceRelation,ServiceInstanceRelation`.
+* [**Breaking Change**] TCP-related source names are changed, fields of TCP-related sources are changed, please refer to the latest `oal/tcp.oal` file.
* Do not log error logs when failed to create ElasticSearch index because the index is created already.
#### UI
diff --git a/docs/en/concepts-and-designs/scope-definitions.md b/docs/en/concepts-and-designs/scope-definitions.md
index b1358e97f2..569cca1bba 100644
--- a/docs/en/concepts-and-designs/scope-definitions.md
+++ b/docs/en/concepts-and-designs/scope-definitions.md
@@ -3,9 +3,9 @@ Using the Aggregation Function, the requests will be grouped by time and **Group
### SCOPE `Service`
-This calculates the metrics data from each request of the service.
+This calculates the metrics data from each request of the service.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service. | | string |
| layer | Layer represents an abstract framework in the computer science, such as operation system(OS_LINUX layer), Kubernetes(k8s layer) | | enum |
@@ -19,14 +19,27 @@ This calculates the metrics data from each request of the service.
| tags | The labels of each request. Each value is made up by `TagKey:TagValue` in the segment. | | `List<String>` |
| tag | The key-value pair of span tags in the segment. | | `Map<String, String>` |
| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string|
-| tcpInfo.receivedBytes | The received bytes of the TCP traffic, if this request is a TCP call. | | long |
-| tcpInfo.sentBytes | The sent bytes of the TCP traffic, if this request is a TCP call. | | long |
+
+### SCOPE `TCPService`
+
+This calculates the metrics data from each request of the TCP service.
+
+| Name | Remarks | Group Key | Type |
+|---|---|---|---|
+| name | The name of the service. | | string |
+| layer | Layer represents an abstract framework in the computer science, such as operation system(OS_LINUX layer), Kubernetes(k8s layer) | | enum |
+| serviceInstanceName | The name of the service instance ID. | | string |
+| tags | The labels of each request. Each value is made up by `TagKey:TagValue` in the segment. | | `List<String>` |
+| tag | The key-value pair of span tags in the segment. | | `Map<String, String>` |
+| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string|
+| receivedBytes | The received bytes of the TCP traffic. | | long |
+| sentBytes | The sent bytes of the TCP traffic. | | long |
### SCOPE `ServiceInstance`
-This calculates the metrics data from each request of the service instance.
+This calculates the metrics data from each request of the service instance.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -39,16 +52,28 @@ This calculates the metrics data from each request of the service instance.
| tags | The labels of each request. Each value is made up by `TagKey:TagValue` in the segment. | | `List<String>` |
| tag | The key-value pair of span tags in the segment. | | `Map<String, String>` |
| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string|
-| tcpInfo.receivedBytes | The received bytes of the TCP traffic, if this request is a TCP call. | | long |
-| tcpInfo.sentBytes | The sent bytes of the TCP traffic, if this request is a TCP call. | | long |
-#### Secondary scopes of `ServiceInstance`
+### SCOPE `TCPServiceInstance`
+
+This calculates the metrics data from each request of the service instance.
+
+| Name | Remarks | Group Key | Type |
+|---|---|---|---|
+| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
+| serviceName | The name of the service. | | string |
+| tags | The labels of each request. Each value is made up by `TagKey:TagValue` in the segment. | | `List<String>` |
+| tag | The key-value pair of span tags in the segment. | | `Map<String, String>` |
+| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string|
+| receivedBytes | The received bytes of the TCP traffic. | | long |
+| sentBytes | The sent bytes of the TCP traffic. | | long |
+
+#### Secondary scopes of `ServiceInstance`
This calculates the metrics data if the service instance is a JVM and collects through javaagent.
1. SCOPE `ServiceInstanceJVMCPU`
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -56,7 +81,7 @@ This calculates the metrics data if the service instance is a JVM and collects t
2. SCOPE `ServiceInstanceJVMMemory`
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -68,7 +93,7 @@ This calculates the metrics data if the service instance is a JVM and collects t
3. SCOPE `ServiceInstanceJVMMemoryPool`
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -80,7 +105,7 @@ This calculates the metrics data if the service instance is a JVM and collects t
4. SCOPE `ServiceInstanceJVMGC`
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -90,7 +115,7 @@ This calculates the metrics data if the service instance is a JVM and collects t
5. SCOPE `ServiceInstanceJVMThread`
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -104,7 +129,7 @@ This calculates the metrics data if the service instance is a JVM and collects t
6. SCOPE `ServiceInstanceJVMClass`
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the service instance, such as `ip:port@Service Name`. **Note**: Currently, the native agent uses `uuid@ipv4` as the instance name, which does not assist in setting up a filter in aggregation. | | string|
| serviceName | The name of the service. | | string |
@@ -114,9 +139,9 @@ This calculates the metrics data if the service instance is a JVM and collects t
### SCOPE `Endpoint`
-This calculates the metrics data from each request of the endpoint in the service.
+This calculates the metrics data from each request of the endpoint in the service.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the endpoint, such as a full path of the HTTP URI. | | string |
| serviceName | The name of the service. | | string |
@@ -130,14 +155,35 @@ This calculates the metrics data from each request of the endpoint in the servic
| tags | The labels of each request. Each value is made up by `TagKey:TagValue` in the segment. | | `List<String>` |
| tag | The key-value pair of span tags in the segment. | | `Map<String, String>` |
| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string|
-| tcpInfo.receivedBytes | The received bytes of the TCP traffic, if this request is a TCP call. | | long |
-| tcpInfo.sentBytes | The sent bytes of the TCP traffic, if this request is a TCP call. | | long |
### SCOPE `ServiceRelation`
This calculates the metrics data from each request between services.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
+|---------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|-----------|------------------------|
+| sourceServiceName | The name of the source service. | | string |
+| sourceServiceInstanceName | The name of the source service instance. | | string |
+| sourceLayer | The layer of the source service. | | enum |
+| destServiceName | The name of the destination service. | | string |
+| destServiceInstanceName | The name of the destination service instance. | | string |
+| destLayer | The layer of the destination service. | | enum |
+| endpoint | The endpoint used in this call. | | string |
+| componentId | The ID of component used in this call. | yes | string |
+| latency | The time taken by each request. | | int |
+| status | Indicates the success or failure of the request. | | bool(true for success) |
+| httpResponseStatusCode | The response code of the HTTP response, and if this request is the HTTP call. E.g. 200, 404, 302 | | int |
+| rpcStatusCode | The string value of the rpc response code. | | string |
+| type | The type of each request, such as Database, HTTP, RPC, or gRPC. | | enum |
+| detectPoint | Where the relation is detected. The value may be client, server, or proxy. | yes | enum |
+| tlsMode | The TLS mode between source and destination services, such as `service_relation_mtls_cpm = from(ServiceRelation.*).filter(tlsMode == "mTLS").cpm()` | | string |
+| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string |
+
+### SCOPE `TCPServiceRelation`
+
+This calculates the metrics data from each request between services.
+
+| Name | Remarks | Group Key | Type |
|---------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------|-----------|------------------------|
| sourceServiceName | The name of the source service. | | string |
| sourceServiceInstanceName | The name of the source service instance. | | string |
@@ -155,14 +201,14 @@ This calculates the metrics data from each request between services.
| detectPoint | Where the relation is detected. The value may be client, server, or proxy. | yes | enum |
| tlsMode | The TLS mode between source and destination services, such as `service_relation_mtls_cpm = from(ServiceRelation.*).filter(tlsMode == "mTLS").cpm()` | | string |
| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string |
-| tcpInfo.receivedBytes | The received bytes of the TCP traffic, if this request is a TCP call. | | long |
-| tcpInfo.sentBytes | The sent bytes of the TCP traffic, if this request is a TCP call. | | long |
+| receivedBytes | The received bytes of the TCP traffic. | | long |
+| sentBytes | The sent bytes of the TCP traffic. | | long |
### SCOPE `ServiceInstanceRelation`
This calculates the metrics data from each request between service instances.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------|------------------------|
| sourceServiceName | The name of the source service. | | string |
| sourceServiceInstanceName | The name of the source service instance. | | string |
@@ -180,17 +226,34 @@ This calculates the metrics data from each request between service instances.
| detectPoint | Where the relation is detected. The value may be client, server, or proxy. | yes | enum |
| tlsMode | The TLS mode between source and destination service instances, such as `service_instance_relation_mtls_cpm = from(ServiceInstanceRelation.*).filter(tlsMode == "mTLS").cpm()` || string | |
| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string |
-| tcpInfo.receivedBytes | The received bytes of the TCP traffic, if this request is a TCP call. | | long |
-| tcpInfo.sentBytes | The sent bytes of the TCP traffic, if this request is a TCP call. | | long |
+
+### SCOPE `TCPServiceInstanceRelation`
+
+This calculates the metrics data from each request between service instances.
+
+| Name | Remarks | Group Key | Type |
+|---------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------|------------------------|
+| sourceServiceName | The name of the source service. | | string |
+| sourceServiceInstanceName | The name of the source service instance. | | string |
+| sourceServiceLayer | The layer of the source service. | | enum |
+| destServiceName | The name of the destination service. | | |
+| destServiceInstanceName | The name of the destination service instance. | | string |
+| destServiceLayer | The layer of the destination service. | | enum |
+| componentId | The ID of the component used in this call. | yes | string |
+| detectPoint | Where the relation is detected. The value may be client, server, or proxy. | yes | enum |
+| tlsMode | The TLS mode between source and destination service instances, such as `service_instance_relation_mtls_cpm = from(ServiceInstanceRelation.*).filter(tlsMode == "mTLS").cpm()` || string | |
+| sideCar.internalErrorCode | The sidecar/gateway proxy internal error code. The value is based on the implementation. | | string |
+| receivedBytes | The received bytes of the TCP traffic. | | long |
+| sentBytes | The sent bytes of the TCP traffic. | | long |
### SCOPE `EndpointRelation`
-This calculates the metrics data of the dependency between endpoints.
-This relation is hard to detect, and it depends on the tracing library to propagate the previous endpoint.
-Therefore, the `EndpointRelation` scope aggregation comes into effect only in services under tracing by SkyWalking native agents,
+This calculates the metrics data of the dependency between endpoints.
+This relation is hard to detect, and it depends on the tracing library to propagate the previous endpoint.
+Therefore, the `EndpointRelation` scope aggregation comes into effect only in services under tracing by SkyWalking native agents,
including auto instrument agents (like Java and .NET), OpenCensus SkyWalking exporter implementation, or other tracing context propagation in SkyWalking specification.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| endpoint | The parent endpoint in the dependency.| | string|
| serviceName | The name of the service. | | string |
@@ -212,7 +275,7 @@ including auto instrument agents (like Java and .NET), OpenCensus SkyWalking exp
This calculates the metrics data from each request of the browser application (browser only).
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The browser application name of each request. | | string |
| count | The number of request, which is fixed at 1. | | int |
@@ -223,7 +286,7 @@ This calculates the metrics data from each request of the browser application (b
This calculates the metrics data from each request of a single version in the browser application (browser only).
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The single version name of each request. | | string |
| serviceName | The name of the browser application. | | string |
@@ -235,7 +298,7 @@ This calculates the metrics data from each request of a single version in the br
This calculates the metrics data from each request of the page in the browser application (browser only).
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The page name of each request. | | string |
| serviceName | The name of the browser application. | | string |
@@ -248,7 +311,7 @@ This calculates the metrics data from each request of the page in the browser ap
This calculates the metrics data from each request of the page in the browser application (browser only).
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The page name of each request. | | string |
| serviceName | The name of the browser application. | | string |
@@ -271,7 +334,7 @@ This calculates the metrics data from each request of the page in the browser ap
This calculates the metrics data from [events](event.md).
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The name of the event. | | string |
| service | The service name to which the event belongs to. | | string |
@@ -286,7 +349,7 @@ This calculates the metrics data from [events](event.md).
This calculates the metrics data from each request of cache system.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The service name of virtual database service. | | string |
| databaseTypeId | The ID of the component used in this call. | | int |
@@ -297,7 +360,7 @@ This calculates the metrics data from each request of cache system.
This calculates the metrics data from slow request of cache system , which is used for `write` or `read` operation.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| databaseServiceId | The service id of virtual cache service. | | string |
| statement | The sql statement . | | string |
@@ -309,7 +372,7 @@ This calculates the metrics data from slow request of cache system , which is us
This calculates the metrics data from each request of cache system.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| name | The service name of virtual cache service. | | string |
| cacheTypeId | The ID of the component used in this call. | | int |
@@ -322,7 +385,7 @@ This calculates the metrics data from each request of cache system.
This calculates the metrics data from slow request of cache system , which is used for `write` or `read` operation.
-| Name | Remarks | Group Key | Type |
+| Name | Remarks | Group Key | Type |
|---|---|---|---|
| cacheServiceId | The service id of virtual cache service. | | string |
| command | The cache command . | | string |
@@ -330,4 +393,4 @@ This calculates the metrics data from slow request of cache system , which is us
| latency | The time taken by each request. | | int(in ms)|
| traceId | The traceId of this slow access| | string|
| status | Indicates the success or failure of the request.| | boolean |
-| operation | Indicates this access is used for `write` or `read` | | string |
\ No newline at end of file
+| operation | Indicates this access is used for `write` or `read` | | string |
diff --git a/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALLexer.g4 b/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALLexer.g4
index 5d39a4b05f..8cd21cbf17 100644
--- a/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALLexer.g4
+++ b/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALLexer.g4
@@ -28,10 +28,14 @@ FROM: 'from';
FILTER: 'filter';
DISABLE: 'disable';
SRC_SERVICE: 'Service';
+SRC_TCP_SERVICE: 'TCPService';
SRC_SERVICE_INSTANCE: 'ServiceInstance';
+SRC_TCP_SERVICE_INSTANCE: 'TCPServiceInstance';
SRC_ENDPOINT: 'Endpoint';
SRC_SERVICE_RELATION: 'ServiceRelation';
+SRC_TCP_SERVICE_RELATION: 'TCPServiceRelation';
SRC_SERVICE_INSTANCE_RELATION: 'ServiceInstanceRelation';
+SRC_TCP_SERVICE_INSTANCE_RELATION: 'TCPServiceInstanceRelation';
SRC_ENDPOINT_RELATION: 'EndpointRelation';
SRC_SERVICE_INSTANCE_JVM_CPU: 'ServiceInstanceJVMCPU';
SRC_SERVICE_INSTANCE_JVM_MEMORY: 'ServiceInstanceJVMMemory';
@@ -142,4 +146,4 @@ fragment Letter
STRING_TO_LONG: '(str->long)';
STRING_TO_LONG_SHORT: '(long)';
STRING_TO_INT: '(str->int)';
-STRING_TO_INT_SHORT: '(int)';
\ No newline at end of file
+STRING_TO_INT_SHORT: '(int)';
diff --git a/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALParser.g4 b/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALParser.g4
index 0c565ef2ad..a0843eade1 100644
--- a/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALParser.g4
+++ b/oap-server/oal-grammar/src/main/antlr4/org/apache/skywalking/oal/rt/grammar/OALParser.g4
@@ -50,8 +50,8 @@ filterExpression
;
source
- : SRC_SERVICE | SRC_DATABASE_ACCESS | SRC_SERVICE_INSTANCE | SRC_ENDPOINT | SRC_CACHE_ACCESS |
- SRC_SERVICE_RELATION | SRC_SERVICE_INSTANCE_RELATION | SRC_ENDPOINT_RELATION |
+ : SRC_SERVICE | SRC_TCP_SERVICE | SRC_DATABASE_ACCESS | SRC_SERVICE_INSTANCE | SRC_TCP_SERVICE_INSTANCE | SRC_ENDPOINT | SRC_CACHE_ACCESS |
+ SRC_SERVICE_RELATION | SRC_TCP_SERVICE_RELATION | SRC_SERVICE_INSTANCE_RELATION | SRC_TCP_SERVICE_INSTANCE_RELATION | SRC_ENDPOINT_RELATION |
SRC_SERVICE_INSTANCE_CLR_CPU | SRC_SERVICE_INSTANCE_CLR_GC | SRC_SERVICE_INSTANCE_CLR_THREAD |
SRC_SERVICE_INSTANCE_JVM_CPU | SRC_SERVICE_INSTANCE_JVM_MEMORY | SRC_SERVICE_INSTANCE_JVM_MEMORY_POOL | SRC_SERVICE_INSTANCE_JVM_GC | SRC_SERVICE_INSTANCE_JVM_THREAD | SRC_SERVICE_INSTANCE_JVM_CLASS |// JVM source of service instance
SRC_ENVOY_INSTANCE_METRIC |
@@ -206,4 +206,4 @@ functionArgCast
castStmt
: STRING_TO_LONG | STRING_TO_LONG_SHORT | STRING_TO_INT | STRING_TO_INT_SHORT
- ;
\ No newline at end of file
+ ;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
index 890d1a4bd6..100f423c64 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
@@ -37,7 +37,7 @@ public class DispatcherManager implements DispatcherDetectorListener {
private static final Logger LOGGER = LoggerFactory.getLogger(DispatcherManager.class);
- private Map<Integer, List<SourceDispatcher>> dispatcherMap;
+ private Map<Integer, List<SourceDispatcher<ISource>>> dispatcherMap;
public DispatcherManager() {
this.dispatcherMap = new HashMap<>();
@@ -48,7 +48,7 @@ public class DispatcherManager implements DispatcherDetectorListener {
return;
}
- List<SourceDispatcher> dispatchers = dispatcherMap.get(source.scope());
+ List<SourceDispatcher<ISource>> dispatchers = dispatcherMap.get(source.scope());
/**
* Dispatcher is only generated by oal script analysis result.
@@ -57,7 +57,7 @@ public class DispatcherManager implements DispatcherDetectorListener {
*/
if (dispatchers != null) {
source.prepare();
- for (SourceDispatcher dispatcher : dispatchers) {
+ for (SourceDispatcher<ISource> dispatcher : dispatchers) {
dispatcher.dispatch(source);
}
}
@@ -106,7 +106,7 @@ public class DispatcherManager implements DispatcherDetectorListener {
int scopeId = dispatcherSource.scope();
- List<SourceDispatcher> dispatchers = this.dispatcherMap.get(scopeId);
+ List<SourceDispatcher<ISource>> dispatchers = this.dispatcherMap.get(scopeId);
if (dispatchers == null) {
dispatchers = new ArrayList<>();
this.dispatcherMap.put(scopeId, dispatchers);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPInfo.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/TCPInstanceTrafficDispatcher.java
similarity index 50%
rename from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPInfo.java
rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/TCPInstanceTrafficDispatcher.java
index 26539e5911..b4e83bafe0 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPInfo.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/TCPInstanceTrafficDispatcher.java
@@ -16,21 +16,20 @@
*
*/
-package org.apache.skywalking.oap.server.core.source;
+package org.apache.skywalking.oap.server.core.analysis.manual.instance;
-import lombok.AllArgsConstructor;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.Setter;
+import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.source.TCPServiceInstance;
-@NoArgsConstructor
-@AllArgsConstructor
-public class TCPInfo {
- @Getter
- @Setter
- private long receivedBytes;
-
- @Getter
- @Setter
- private long sentBytes;
+public class TCPInstanceTrafficDispatcher implements SourceDispatcher<TCPServiceInstance> {
+ @Override
+ public void dispatch(final TCPServiceInstance source) {
+ InstanceTraffic traffic = new InstanceTraffic();
+ traffic.setTimeBucket(source.getTimeBucket());
+ traffic.setName(source.getName());
+ traffic.setServiceId(source.getServiceId());
+ traffic.setLastPingTimestamp(source.getTimeBucket());
+ MetricsStreamProcessor.getInstance().in(traffic);
+ }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/TCPInstanceUpdateDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/TCPInstanceUpdateDispatcher.java
new file mode 100644
index 0000000000..42c99f9011
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/instance/TCPInstanceUpdateDispatcher.java
@@ -0,0 +1,36 @@
+/*
+ * 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.server.core.analysis.manual.instance;
+
+import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.source.TCPServiceInstanceUpdate;
+
+public class TCPInstanceUpdateDispatcher implements SourceDispatcher<TCPServiceInstanceUpdate> {
+ @Override
+ public void dispatch(final TCPServiceInstanceUpdate source) {
+ InstanceTraffic traffic = new InstanceTraffic();
+ traffic.setTimeBucket(source.getTimeBucket());
+ traffic.setName(source.getName());
+ traffic.setServiceId(source.getServiceId());
+ traffic.setLastPingTimestamp(source.getTimeBucket());
+ traffic.setProperties(source.getProperties());
+ MetricsStreamProcessor.getInstance().in(traffic);
+ }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/instance/TCPServiceInstanceCallRelationDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/instance/TCPServiceInstanceCallRelationDispatcher.java
new file mode 100644
index 0000000000..3f06da5494
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/instance/TCPServiceInstanceCallRelationDispatcher.java
@@ -0,0 +1,62 @@
+/*
+ * 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.server.core.analysis.manual.relation.instance;
+
+import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.source.TCPServiceInstanceRelation;
+
+public class TCPServiceInstanceCallRelationDispatcher implements SourceDispatcher<TCPServiceInstanceRelation> {
+
+ @Override
+ public void dispatch(TCPServiceInstanceRelation source) {
+ switch (source.getDetectPoint()) {
+ case SERVER:
+ serverSide(source);
+ break;
+ case CLIENT:
+ clientSide(source);
+ break;
+ }
+ }
+
+ private void serverSide(TCPServiceInstanceRelation source) {
+ ServiceInstanceRelationServerSideMetrics metrics = new ServiceInstanceRelationServerSideMetrics();
+ metrics.setTimeBucket(source.getTimeBucket());
+ metrics.setSourceServiceId(source.getSourceServiceId());
+ metrics.setSourceServiceInstanceId(source.getSourceServiceInstanceId());
+ metrics.setDestServiceId(source.getDestServiceId());
+ metrics.setDestServiceInstanceId(source.getDestServiceInstanceId());
+ metrics.setComponentId(source.getComponentId());
+ metrics.setEntityId(source.getEntityId());
+ MetricsStreamProcessor.getInstance().in(metrics);
+ }
+
+ private void clientSide(TCPServiceInstanceRelation source) {
+ ServiceInstanceRelationClientSideMetrics metrics = new ServiceInstanceRelationClientSideMetrics();
+ metrics.setTimeBucket(source.getTimeBucket());
+ metrics.setSourceServiceId(source.getSourceServiceId());
+ metrics.setSourceServiceInstanceId(source.getSourceServiceInstanceId());
+ metrics.setDestServiceId(source.getDestServiceId());
+ metrics.setDestServiceInstanceId(source.getDestServiceInstanceId());
+ metrics.setComponentId(source.getComponentId());
+ metrics.setEntityId(source.getEntityId());
+ MetricsStreamProcessor.getInstance().in(metrics);
+ }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/service/TCPServiceCallRelationDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/service/TCPServiceCallRelationDispatcher.java
new file mode 100644
index 0000000000..65e9a4534c
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/service/TCPServiceCallRelationDispatcher.java
@@ -0,0 +1,58 @@
+/*
+ * 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.server.core.analysis.manual.relation.service;
+
+import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.source.TCPServiceRelation;
+
+public class TCPServiceCallRelationDispatcher implements SourceDispatcher<TCPServiceRelation> {
+
+ @Override
+ public void dispatch(TCPServiceRelation source) {
+ switch (source.getDetectPoint()) {
+ case SERVER:
+ serverSide(source);
+ break;
+ case CLIENT:
+ clientSide(source);
+ break;
+ }
+ }
+
+ private void serverSide(TCPServiceRelation source) {
+ ServiceRelationServerSideMetrics metrics = new ServiceRelationServerSideMetrics();
+ metrics.setTimeBucket(source.getTimeBucket());
+ metrics.setSourceServiceId(source.getSourceServiceId());
+ metrics.setDestServiceId(source.getDestServiceId());
+ metrics.setComponentId(source.getComponentId());
+ metrics.setEntityId(source.getEntityId());
+ MetricsStreamProcessor.getInstance().in(metrics);
+ }
+
+ private void clientSide(TCPServiceRelation source) {
+ ServiceRelationClientSideMetrics metrics = new ServiceRelationClientSideMetrics();
+ metrics.setTimeBucket(source.getTimeBucket());
+ metrics.setSourceServiceId(source.getSourceServiceId());
+ metrics.setDestServiceId(source.getDestServiceId());
+ metrics.setComponentId(source.getComponentId());
+ metrics.setEntityId(source.getEntityId());
+ MetricsStreamProcessor.getInstance().in(metrics);
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandlerCompat.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/TCPServiceTrafficDispatcher.java
similarity index 53%
rename from oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandlerCompat.java
rename to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/TCPServiceTrafficDispatcher.java
index b6f48b7a2c..7cf009a540 100644
--- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandlerCompat.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/TCPServiceTrafficDispatcher.java
@@ -16,20 +16,19 @@
*
*/
-package org.apache.skywalking.aop.server.receiver.mesh;
+package org.apache.skywalking.oap.server.core.analysis.manual.service;
-import io.grpc.stub.StreamObserver;
-import lombok.RequiredArgsConstructor;
-import org.apache.skywalking.apm.network.servicemesh.v3.MeshProbeDownstream;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
-import org.apache.skywalking.apm.network.servicemesh.v3.compat.ServiceMeshMetricServiceGrpc;
-
-@RequiredArgsConstructor
-public class MeshGRPCHandlerCompat extends ServiceMeshMetricServiceGrpc.ServiceMeshMetricServiceImplBase {
- private final MeshGRPCHandler delegate;
+import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.source.TCPService;
+public class TCPServiceTrafficDispatcher implements SourceDispatcher<TCPService> {
@Override
- public StreamObserver<ServiceMeshMetric> collect(final StreamObserver<MeshProbeDownstream> responseObserver) {
- return delegate.collect(responseObserver);
+ public void dispatch(final TCPService source) {
+ ServiceTraffic traffic = new ServiceTraffic();
+ traffic.setTimeBucket(source.getTimeBucket());
+ traffic.setName(source.getName());
+ traffic.setLayer(source.getLayer());
+ MetricsStreamProcessor.getInstance().in(traffic);
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java
index 05d6e9a302..e8fbf11e1d 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/DefaultScopeDefine.java
@@ -109,6 +109,12 @@ public class DefaultScopeDefine {
public static final int CACHE_ACCESS = 55;
public static final int CACHE_SLOW_ACCESS = 56;
+ public static final int TCP_SERVICE = 57;
+ public static final int TCP_SERVICE_INSTANCE = 58;
+ public static final int TCP_SERVICE_RELATION = 59;
+ public static final int TCP_SERVICE_INSTANCE_RELATION = 60;
+ public static final int TCP_SERVICE_INSTANCE_UPDATE = 61;
+
/**
* Catalog of scope, the metrics processor could use this to group all generated metrics by oal rt.
*/
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
index e3322c13be..1306c140cc 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
@@ -82,9 +82,6 @@ public class Service extends Source {
@Getter
@Setter
private SideCar sideCar = new SideCar();
- @Getter
- @Setter
- private TCPInfo tcpInfo = new TCPInfo();
public String getTag(String key) {
return originalTags.get(key);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java
index 125a45ffd3..6382c80ba5 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java
@@ -87,10 +87,6 @@ public class ServiceInstance extends Source {
@Setter
private SideCar sideCar = new SideCar();
- @Getter
- @Setter
- private TCPInfo tcpInfo = new TCPInfo();
-
@Override
public void prepare() {
serviceId = IDManager.ServiceID.buildId(serviceName, serviceLayer.isNormal());
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
index 5928ad4062..88162b34af 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
@@ -114,9 +114,6 @@ public class ServiceInstanceRelation extends Source {
@Getter
@Setter
private SideCar sideCar = new SideCar();
- @Getter
- @Setter
- private TCPInfo tcpInfo = new TCPInfo();
@Override
public void prepare() {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
index db3d98b001..0b9683d14b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
@@ -104,9 +104,6 @@ public class ServiceRelation extends Source {
@Getter
@Setter
private SideCar sideCar = new SideCar();
- @Getter
- @Setter
- private TCPInfo tcpInfo = new TCPInfo();
@Override
public void prepare() {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPService.java
similarity index 80%
copy from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
copy to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPService.java
index e3322c13be..0a21a91d96 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPService.java
@@ -25,17 +25,17 @@ import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
-import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE;
+import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.TCP_SERVICE;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_CATALOG_NAME;
-@ScopeDeclaration(id = SERVICE, name = "Service", catalog = SERVICE_CATALOG_NAME)
+@ScopeDeclaration(id = TCP_SERVICE, name = "TCPService", catalog = SERVICE_CATALOG_NAME)
@ScopeDefaultColumn.VirtualColumnDefinition(fieldName = "entityId", columnName = "entity_id", isID = true, type = String.class)
-public class Service extends Source {
+public class TCPService extends Source {
private volatile String entityId;
@Override
public int scope() {
- return DefaultScopeDefine.SERVICE;
+ return DefaultScopeDefine.TCP_SERVICE;
}
@Override
@@ -58,33 +58,20 @@ public class Service extends Source {
private String serviceInstanceName;
@Getter
@Setter
- private String endpointName;
- @Getter
- @Setter
- private int latency;
- @Getter
- @Setter
- private boolean status;
- @Getter
- @Setter
- private int httpResponseStatusCode;
- @Getter
- @Setter
- private String rpcStatusCode;
- @Getter
- @Setter
- private RequestType type;
- @Getter
- @Setter
private List<String> tags;
@Setter
private Map<String, String> originalTags;
@Getter
@Setter
private SideCar sideCar = new SideCar();
+
+ @Getter
+ @Setter
+ private long receivedBytes;
+
@Getter
@Setter
- private TCPInfo tcpInfo = new TCPInfo();
+ private long sentBytes;
public String getTag(String key) {
return originalTags.get(key);
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstance.java
similarity index 81%
copy from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java
copy to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstance.java
index 125a45ffd3..294ab32096 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstance.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstance.java
@@ -25,17 +25,17 @@ import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
import org.apache.skywalking.oap.server.core.analysis.Layer;
-import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_INSTANCE;
+import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.TCP_SERVICE_INSTANCE;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_INSTANCE_CATALOG_NAME;
-@ScopeDeclaration(id = SERVICE_INSTANCE, name = "ServiceInstance", catalog = SERVICE_INSTANCE_CATALOG_NAME)
+@ScopeDeclaration(id = TCP_SERVICE_INSTANCE, name = "TCPServiceInstance", catalog = SERVICE_INSTANCE_CATALOG_NAME)
@ScopeDefaultColumn.VirtualColumnDefinition(fieldName = "entityId", columnName = "entity_id", isID = true, type = String.class)
-public class ServiceInstance extends Source {
+public class TCPServiceInstance extends Source {
private volatile String entityId;
@Override
public int scope() {
- return DefaultScopeDefine.SERVICE_INSTANCE;
+ return DefaultScopeDefine.TCP_SERVICE_INSTANCE;
}
@Override
@@ -62,24 +62,6 @@ public class ServiceInstance extends Source {
private Layer serviceLayer;
@Getter
@Setter
- private String endpointName;
- @Getter
- @Setter
- private int latency;
- @Getter
- @Setter
- private boolean status;
- @Getter
- @Setter
- private int httpResponseStatusCode;
- @Getter
- @Setter
- private String rpcStatusCode;
- @Getter
- @Setter
- private RequestType type;
- @Getter
- @Setter
private List<String> tags;
@Setter
private Map<String, String> originalTags;
@@ -89,7 +71,11 @@ public class ServiceInstance extends Source {
@Getter
@Setter
- private TCPInfo tcpInfo = new TCPInfo();
+ private long receivedBytes;
+
+ @Getter
+ @Setter
+ private long sentBytes;
@Override
public void prepare() {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstanceRelation.java
similarity index 83%
copy from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
copy to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstanceRelation.java
index 5928ad4062..1c3bbc691b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstanceRelation.java
@@ -24,17 +24,17 @@ import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
-import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_INSTANCE_RELATION;
+import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.TCP_SERVICE_INSTANCE_RELATION;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_INSTANCE_RELATION_CATALOG_NAME;
-@ScopeDeclaration(id = SERVICE_INSTANCE_RELATION, name = "ServiceInstanceRelation", catalog = SERVICE_INSTANCE_RELATION_CATALOG_NAME)
+@ScopeDeclaration(id = TCP_SERVICE_INSTANCE_RELATION, name = "TCPServiceInstanceRelation", catalog = SERVICE_INSTANCE_RELATION_CATALOG_NAME)
@ScopeDefaultColumn.VirtualColumnDefinition(fieldName = "entityId", columnName = "entity_id", isID = true, type = String.class)
-public class ServiceInstanceRelation extends Source {
+public class TCPServiceInstanceRelation extends Source {
private String entityId;
@Override
public int scope() {
- return DefaultScopeDefine.SERVICE_INSTANCE_RELATION;
+ return DefaultScopeDefine.TCP_SERVICE_INSTANCE_RELATION;
}
@Override
@@ -82,31 +82,9 @@ public class ServiceInstanceRelation extends Source {
private String destServiceInstanceName;
@Getter
@Setter
- private String endpoint;
- @Getter
- @Setter
private int componentId;
@Getter
@Setter
- private int latency;
- @Getter
- @Setter
- private boolean status;
- @Getter
- @Setter
- @Deprecated
- private int responseCode;
- @Getter
- @Setter
- private int httpResponseStatusCode;
- @Getter
- @Setter
- private String rpcStatusCode;
- @Getter
- @Setter
- private RequestType type;
- @Getter
- @Setter
private DetectPoint detectPoint;
@Getter
@Setter
@@ -114,9 +92,14 @@ public class ServiceInstanceRelation extends Source {
@Getter
@Setter
private SideCar sideCar = new SideCar();
+
+ @Getter
+ @Setter
+ private long receivedBytes;
+
@Getter
@Setter
- private TCPInfo tcpInfo = new TCPInfo();
+ private long sentBytes;
@Override
public void prepare() {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstanceUpdate.java
similarity index 52%
copy from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
copy to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstanceUpdate.java
index e3322c13be..a0ae4b7f1c 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Service.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceInstanceUpdate.java
@@ -18,75 +18,35 @@
package org.apache.skywalking.oap.server.core.source;
-import java.util.List;
-import java.util.Map;
+import com.google.gson.JsonObject;
import lombok.Getter;
import lombok.Setter;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
-import org.apache.skywalking.oap.server.core.analysis.Layer;
-import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE;
-import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_CATALOG_NAME;
+import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.TCP_SERVICE_INSTANCE_UPDATE;
-@ScopeDeclaration(id = SERVICE, name = "Service", catalog = SERVICE_CATALOG_NAME)
+@ScopeDeclaration(id = TCP_SERVICE_INSTANCE_UPDATE, name = "TCPServiceInstanceUpdate")
@ScopeDefaultColumn.VirtualColumnDefinition(fieldName = "entityId", columnName = "entity_id", isID = true, type = String.class)
-public class Service extends Source {
- private volatile String entityId;
-
+public class TCPServiceInstanceUpdate extends Source {
@Override
public int scope() {
- return DefaultScopeDefine.SERVICE;
+ return TCP_SERVICE_INSTANCE_UPDATE;
}
@Override
public String getEntityId() {
- if (entityId == null) {
- entityId = IDManager.ServiceID.buildId(name, layer.isNormal());
- }
- return entityId;
+ return IDManager.ServiceInstanceID.buildId(serviceId, name);
}
@Getter
@Setter
- @ScopeDefaultColumn.DefinedByField(columnName = "name", requireDynamicActive = true)
- private String name;
- @Setter
- @Getter
- private Layer layer;
- @Getter
- @Setter
- private String serviceInstanceName;
- @Getter
- @Setter
- private String endpointName;
- @Getter
- @Setter
- private int latency;
- @Getter
- @Setter
- private boolean status;
+ @ScopeDefaultColumn.DefinedByField(columnName = "service_id")
+ private String serviceId;
@Getter
@Setter
- private int httpResponseStatusCode;
- @Getter
- @Setter
- private String rpcStatusCode;
- @Getter
- @Setter
- private RequestType type;
- @Getter
- @Setter
- private List<String> tags;
- @Setter
- private Map<String, String> originalTags;
- @Getter
- @Setter
- private SideCar sideCar = new SideCar();
+ @ScopeDefaultColumn.DefinedByField(columnName = "name", requireDynamicActive = true)
+ private String name;
@Getter
@Setter
- private TCPInfo tcpInfo = new TCPInfo();
-
- public String getTag(String key) {
- return originalTags.get(key);
- }
+ private JsonObject properties;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceRelation.java
similarity index 83%
copy from oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
copy to oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceRelation.java
index db3d98b001..1d42db713b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/TCPServiceRelation.java
@@ -24,17 +24,17 @@ import org.apache.skywalking.oap.server.core.analysis.Layer;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.core.analysis.IDManager;
-import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_RELATION;
+import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.TCP_SERVICE_RELATION;
import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_RELATION_CATALOG_NAME;
-@ScopeDeclaration(id = SERVICE_RELATION, name = "ServiceRelation", catalog = SERVICE_RELATION_CATALOG_NAME)
+@ScopeDeclaration(id = TCP_SERVICE_RELATION, name = "TCPServiceRelation", catalog = SERVICE_RELATION_CATALOG_NAME)
@ScopeDefaultColumn.VirtualColumnDefinition(fieldName = "entityId", columnName = "entity_id", isID = true, type = String.class)
-public class ServiceRelation extends Source {
+public class TCPServiceRelation extends Source {
private String entityId;
@Override
public int scope() {
- return DefaultScopeDefine.SERVICE_RELATION;
+ return DefaultScopeDefine.TCP_SERVICE_RELATION;
}
@Override
@@ -76,27 +76,9 @@ public class ServiceRelation extends Source {
private String destServiceInstanceName;
@Getter
@Setter
- private String endpoint;
- @Getter
- @Setter
private int componentId;
@Getter
@Setter
- private int latency;
- @Getter
- @Setter
- private boolean status;
- @Getter
- @Setter
- private int httpResponseStatusCode;
- @Getter
- @Setter
- private String rpcStatusCode;
- @Getter
- @Setter
- private RequestType type;
- @Getter
- @Setter
private DetectPoint detectPoint;
@Getter
@Setter
@@ -104,9 +86,14 @@ public class ServiceRelation extends Source {
@Getter
@Setter
private SideCar sideCar = new SideCar();
+
+ @Getter
+ @Setter
+ private long receivedBytes;
+
@Getter
@Setter
- private TCPInfo tcpInfo = new TCPInfo();
+ private long sentBytes;
@Override
public void prepare() {
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java
index 4f43a797d2..2406ea1c8b 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/AccessLogServiceGRPCHandler.java
@@ -18,21 +18,15 @@
package org.apache.skywalking.oap.server.receiver.envoy;
-import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
-import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
-import io.envoyproxy.envoy.service.accesslog.v2.AccessLogServiceGrpc;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsResponse;
-import io.grpc.Status;
-import io.grpc.stub.StreamObserver;
import java.util.ArrayList;
import java.util.List;
import java.util.ServiceLoader;
import org.apache.skywalking.aop.server.receiver.mesh.TelemetryDataDispatcher;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
-import org.apache.skywalking.oap.server.library.util.CollectionUtils;
import org.apache.skywalking.oap.server.receiver.envoy.als.ALSHTTPAnalysis;
import org.apache.skywalking.oap.server.receiver.envoy.als.AccessLogAnalyzer;
import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
@@ -44,6 +38,13 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
+import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v2.AccessLogServiceGrpc;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsResponse;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogServiceImplBase {
private static final Logger LOGGER = LoggerFactory.getLogger(AccessLogServiceGRPCHandler.class);
@@ -111,7 +112,7 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
public void onNext(StreamAccessLogsMessage message) {
HistogramMetrics.Timer timer = histogram.createTimer();
try {
- if (isFirst || (alwaysAnalyzeIdentity && message.hasIdentifier())) {
+ if (isFirst || alwaysAnalyzeIdentity && message.hasIdentifier()) {
identifier = message.getIdentifier();
isFirst = false;
role = Role.NONE;
@@ -129,25 +130,29 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
.getId(), role, logCase, message);
}
- List<ServiceMeshMetric.Builder> sourceResult = new ArrayList<>();
+ final ServiceMeshMetrics.Builder sourceResult = ServiceMeshMetrics.newBuilder();
switch (logCase) {
case HTTP_LOGS:
- StreamAccessLogsMessage.HTTPAccessLogEntries logs = message.getHttpLogs();
- counter.inc(logs.getLogEntryCount());
+ final HTTPServiceMeshMetrics.Builder httpMetrics = HTTPServiceMeshMetrics.newBuilder();
+ final StreamAccessLogsMessage.HTTPAccessLogEntries httpLogs = message.getHttpLogs();
- for (final HTTPAccessLogEntry log : logs.getLogEntryList()) {
+ counter.inc(httpLogs.getLogEntryCount());
+
+ for (final HTTPAccessLogEntry httpLog : httpLogs.getLogEntryList()) {
AccessLogAnalyzer.Result result = AccessLogAnalyzer.Result.builder().build();
for (ALSHTTPAnalysis analysis : envoyHTTPAnalysisList) {
- result = analysis.analysis(result, identifier, log, role);
+ result = analysis.analysis(result, identifier, httpLog, role);
}
- if (CollectionUtils.isNotEmpty(result.getMetrics())) {
- sourceResult.addAll(result.getMetrics());
+ if (result.hasResult()) {
+ httpMetrics.addAllMetrics(result.getMetrics().getHttpMetrics().getMetricsList());
}
}
-
+ sourceResult.setHttpMetrics(httpMetrics);
break;
case TCP_LOGS:
- StreamAccessLogsMessage.TCPAccessLogEntries tcpLogs = message.getTcpLogs();
+ final TCPServiceMeshMetrics.Builder tcpMetrics = TCPServiceMeshMetrics.newBuilder();
+ final StreamAccessLogsMessage.TCPAccessLogEntries tcpLogs = message.getTcpLogs();
+
counter.inc(tcpLogs.getLogEntryCount());
for (final TCPAccessLogEntry tcpLog : tcpLogs.getLogEntryList()) {
@@ -155,15 +160,18 @@ public class AccessLogServiceGRPCHandler extends AccessLogServiceGrpc.AccessLogS
for (TCPAccessLogAnalyzer analyzer : envoyTCPAnalysisList) {
result = analyzer.analysis(result, identifier, tcpLog, role);
}
- if (CollectionUtils.isNotEmpty(result.getMetrics())) {
- sourceResult.addAll(result.getMetrics());
+ if (result.hasResult()) {
+ tcpMetrics.addAllMetrics(result.getMetrics().getTcpMetrics().getMetricsList());
}
}
-
+ sourceResult.setTcpMetrics(tcpMetrics);
break;
+ default: // Ignored
}
- sourceDispatcherCounter.inc(sourceResult.size());
- sourceResult.forEach(TelemetryDataDispatcher::process);
+ sourceDispatcherCounter.inc(
+ sourceResult.getHttpMetrics().getMetricsCount() +
+ sourceResult.getTcpMetrics().getMetricsCount());
+ TelemetryDataDispatcher.process(sourceResult.build());
} finally {
timer.finish();
}
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AbstractALSAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AbstractALSAnalyzer.java
index 2bb71afd39..9c468daf5b 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AbstractALSAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AbstractALSAnalyzer.java
@@ -20,18 +20,18 @@ package org.apache.skywalking.oap.server.receiver.envoy.als;
import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
@Slf4j
public abstract class AbstractALSAnalyzer implements ALSHTTPAnalysis {
/**
- * Create an adapter to adapt the {@link HTTPAccessLogEntry log entry} into a {@link ServiceMeshMetric.Builder}.
+ * Create an adapter to adapt the {@link HTTPAccessLogEntry log entry} into a {@link HTTPServiceMeshMetric.Builder}.
*
* @param entry the access log entry that is to be adapted from.
* @param sourceService the source service.
* @param targetService the target/destination service.
- * @return an adapter that adapts {@link HTTPAccessLogEntry log entry} into a {@link ServiceMeshMetric.Builder}.
+ * @return an adapter that adapts {@link HTTPAccessLogEntry log entry} into a {@link HTTPServiceMeshMetric.Builder}.
*/
protected LogEntry2MetricsAdapter newAdapter(
final HTTPAccessLogEntry entry,
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java
index a8aceb90bb..ac9327bdb7 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java
@@ -18,15 +18,14 @@
package org.apache.skywalking.oap.server.receiver.envoy.als;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
+import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import java.util.List;
import lombok.Builder;
import lombok.Data;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
-import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.apache.skywalking.oap.server.library.module.ModuleStartException;
-import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
public interface AccessLogAnalyzer<E> {
String name();
@@ -79,6 +78,10 @@ public interface AccessLogAnalyzer<E> {
/**
* The analyzed metrics result.
*/
- private List<ServiceMeshMetric.Builder> metrics;
+ private ServiceMeshMetrics.Builder metrics;
+
+ public boolean hasResult() {
+ return metrics != null && (metrics.getHttpMetrics().getMetricsCount() > 0 || metrics.getTcpMetrics().getMetricsCount() > 0);
+ }
}
}
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 29a0ffc46d..2cb9cb14d4 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
@@ -31,8 +31,8 @@ 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.HTTPServiceMeshMetric;
import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
import static com.google.common.base.Strings.isNullOrEmpty;
@@ -62,7 +62,7 @@ public class LogEntry2MetricsAdapter {
*
* @return the {@link ServiceMeshMetric.Builder} adapted from the given entry.
*/
- public ServiceMeshMetric.Builder adaptToDownstreamMetrics() {
+ public HTTPServiceMeshMetric.Builder adaptToDownstreamMetrics() {
final AccessLogCommon properties = entry.getCommonProperties();
final long startTime = formatAsLong(properties.getStartTime());
final long duration = formatAsLong(properties.getTimeToLastDownstreamTxByte());
@@ -79,7 +79,7 @@ public class LogEntry2MetricsAdapter {
*
* @return the {@link ServiceMeshMetric.Builder} adapted from the given entry.
*/
- public ServiceMeshMetric.Builder adaptToUpstreamMetrics() {
+ public HTTPServiceMeshMetric.Builder adaptToUpstreamMetrics() {
final AccessLogCommon properties = entry.getCommonProperties();
final long startTime = formatAsLong(properties.getStartTime());
final long outboundStartTime = startTime + formatAsLong(properties.getTimeToFirstUpstreamTxByte());
@@ -92,7 +92,7 @@ public class LogEntry2MetricsAdapter {
.setDetectPoint(DetectPoint.client);
}
- public ServiceMeshMetric.Builder adaptCommonPart() {
+ public HTTPServiceMeshMetric.Builder adaptCommonPart() {
final AccessLogCommon properties = entry.getCommonProperties();
final String endpoint = endpoint();
int responseCode = entry.getResponse().getResponseCode().getValue();
@@ -106,16 +106,17 @@ public class LogEntry2MetricsAdapter {
properties.getTimeToFirstDownstreamTxByte().getNanos()
- properties.getTimeToFirstUpstreamRxByte().getNanos();
- final ServiceMeshMetric.Builder builder =
- ServiceMeshMetric.newBuilder()
- .setEndpoint(endpoint)
- .setResponseCode(Math.toIntExact(responseCode))
- .setStatus(status)
- .setProtocol(protocol)
- .setTlsMode(tlsMode)
- .setInternalErrorCode(internalErrorCode)
- .setInternalRequestLatencyNanos(internalRequestLatencyNanos)
- .setInternalResponseLatencyNanos(internalResponseLatencyNanos);
+ final HTTPServiceMeshMetric.Builder builder =
+ HTTPServiceMeshMetric
+ .newBuilder()
+ .setEndpoint(endpoint)
+ .setResponseCode(Math.toIntExact(responseCode))
+ .setStatus(status)
+ .setProtocol(protocol)
+ .setTlsMode(tlsMode)
+ .setInternalErrorCode(internalErrorCode)
+ .setInternalRequestLatencyNanos(internalRequestLatencyNanos)
+ .setInternalResponseLatencyNanos(internalResponseLatencyNanos);
Optional.ofNullable(sourceService)
.map(ServiceMetaInfo::getServiceName)
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java
index ca460128b8..0552b809e2 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java
@@ -18,27 +18,25 @@
package org.apache.skywalking.oap.server.receiver.envoy.als.k8s;
+import static org.apache.skywalking.oap.server.library.util.StringUtil.isBlank;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.k8s.Addresses.isValid;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
+import org.apache.skywalking.oap.server.receiver.envoy.ServiceMetaInfoFactory;
+import org.apache.skywalking.oap.server.receiver.envoy.als.AbstractALSAnalyzer;
+import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
+import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
import io.envoyproxy.envoy.config.core.v3.Address;
import io.envoyproxy.envoy.config.core.v3.SocketAddress;
import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import java.util.ArrayList;
-import java.util.List;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
-import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
-import org.apache.skywalking.oap.server.receiver.envoy.ServiceMetaInfoFactory;
-import org.apache.skywalking.oap.server.receiver.envoy.als.AbstractALSAnalyzer;
-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.StringUtil.isBlank;
-import static org.apache.skywalking.oap.server.library.util.CollectionUtils.isNotEmpty;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.k8s.Addresses.isValid;
/**
* Analysis log based on ingress and mesh scenarios.
@@ -68,7 +66,7 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
final HTTPAccessLogEntry entry,
final Role role
) {
- if (isNotEmpty(result.getMetrics())) {
+ if (result.hasResult()) {
return result;
}
switch (role) {
@@ -91,7 +89,7 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
return Result.builder().build();
}
- final List<ServiceMeshMetric.Builder> sources = new ArrayList<>();
+ final HTTPServiceMeshMetrics.Builder sources = HTTPServiceMeshMetrics.newBuilder();
final Address downstreamRemoteAddress =
properties.hasDownstreamDirectRemoteAddress()
@@ -106,7 +104,7 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
if (cluster.startsWith("inbound|")) {
// Server side
- final ServiceMeshMetric.Builder metrics;
+ final HTTPServiceMeshMetric.Builder metrics;
if (downstreamService.equals(config.serviceMetaInfoFactory().unknown())) {
// Ingress -> sidecar(server side)
// Mesh telemetry without source, the relation would be generated.
@@ -119,22 +117,22 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
log.debug("Transformed sidecar->sidecar(server side) inbound mesh metrics {}", metrics);
}
- sources.add(metrics);
+ sources.addMetrics(metrics);
} else if (cluster.startsWith("outbound|")) {
// sidecar(client side) -> sidecar
final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress();
if (!isValid(upstreamRemoteAddress)) {
- return Result.builder().metrics(sources).service(localService).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(sources)).service(localService).build();
}
final ServiceMetaInfo destService = find(upstreamRemoteAddress.getSocketAddress().getAddress());
- final ServiceMeshMetric.Builder metric = newAdapter(entry, downstreamService, destService).adaptToUpstreamMetrics();
+ final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, downstreamService, destService).adaptToUpstreamMetrics();
log.debug("Transformed sidecar->sidecar(server side) inbound mesh metric {}", metric);
- sources.add(metric);
+ sources.addMetrics(metric);
}
- return Result.builder().metrics(sources).service(localService).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(sources)).service(localService).build();
}
protected Result analyzeProxy(final HTTPAccessLogEntry entry) {
@@ -150,31 +148,31 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
return Result.builder().build();
}
- final List<ServiceMeshMetric.Builder> result = new ArrayList<>(2);
+ final HTTPServiceMeshMetrics.Builder result = HTTPServiceMeshMetrics.newBuilder();
final SocketAddress downstreamRemoteAddressSocketAddress = downstreamRemoteAddress.getSocketAddress();
final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress());
final SocketAddress downstreamLocalAddressSocketAddress = downstreamLocalAddress.getSocketAddress();
final ServiceMetaInfo ingress = find(downstreamLocalAddressSocketAddress.getAddress());
- final ServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics();
+ final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics();
log.debug("Transformed ingress inbound mesh metric {}", metric);
- result.add(metric);
+ result.addMetrics(metric);
final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress();
final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress());
- final ServiceMeshMetric.Builder outboundMetric =
+ final HTTPServiceMeshMetric.Builder outboundMetric =
newAdapter(entry, ingress, targetService)
.adaptToUpstreamMetrics()
// Can't parse it from tls properties, leave it to Server side.
.setTlsMode(NON_TLS);
log.debug("Transformed ingress outbound mesh metric {}", outboundMetric);
- result.add(outboundMetric);
+ result.addMetrics(outboundMetric);
- return Result.builder().metrics(result).service(ingress).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(result)).service(ingress).build();
}
/**
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
index 047b89ba52..65f543c790 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
@@ -18,27 +18,25 @@
package org.apache.skywalking.oap.server.receiver.envoy.als.mx;
-import com.google.protobuf.Any;
-import com.google.protobuf.TextFormat;
-import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
-import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import java.util.ArrayList;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
import java.util.Base64;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
import org.apache.skywalking.oap.server.receiver.envoy.als.AbstractALSAnalyzer;
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.CollectionUtils.isNotEmpty;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
+import com.google.protobuf.Any;
+import com.google.protobuf.TextFormat;
+import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
+import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import lombok.extern.slf4j.Slf4j;
@Slf4j
public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer {
@@ -73,7 +71,7 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer {
final HTTPAccessLogEntry entry,
final Role role
) {
- if (isNotEmpty(previousResult.getMetrics())) {
+ if (previousResult.hasResult()) {
return previousResult;
}
if (!entry.hasCommonProperties()) {
@@ -92,7 +90,7 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer {
return Result.builder().service(currSvc).build();
}
- final List<ServiceMeshMetric.Builder> result = new ArrayList<>();
+ final HTTPServiceMeshMetrics.Builder httpMetrics = HTTPServiceMeshMetrics.newBuilder();
final AtomicBoolean downstreamExists = new AtomicBoolean();
stateMap.forEach((key, value) -> {
if (!key.equals(UPSTREAM_KEY) && !key.equals(DOWNSTREAM_KEY)) {
@@ -105,33 +103,33 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer {
log.error("Fail to parse metadata {} to FlatNode", Base64.getEncoder().encode(value.toByteArray()));
return;
}
- final ServiceMeshMetric.Builder metrics;
+ final HTTPServiceMeshMetric.Builder metrics;
switch (key) {
case UPSTREAM_KEY:
metrics = newAdapter(entry, currSvc, svc).adaptToUpstreamMetrics().setTlsMode(NON_TLS);
if (log.isDebugEnabled()) {
log.debug("Transformed a {} outbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
}
- result.add(metrics);
+ httpMetrics.addMetrics(metrics);
break;
case DOWNSTREAM_KEY:
metrics = newAdapter(entry, svc, currSvc).adaptToDownstreamMetrics();
if (log.isDebugEnabled()) {
log.debug("Transformed a {} inbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
}
- result.add(metrics);
+ httpMetrics.addMetrics(metrics);
downstreamExists.set(true);
break;
}
});
if (role.equals(Role.PROXY) && !downstreamExists.get()) {
- final ServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
+ final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
if (log.isDebugEnabled()) {
log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric));
}
- result.add(metric);
+ httpMetrics.addMetrics(metric);
}
- return Result.builder().metrics(result).service(currSvc).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(httpMetrics)).service(currSvc).build();
}
protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception {
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/AbstractTCPAccessLogAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/AbstractTCPAccessLogAnalyzer.java
index 05165094ba..a3e719f5cc 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/AbstractTCPAccessLogAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/AbstractTCPAccessLogAnalyzer.java
@@ -21,19 +21,19 @@ package org.apache.skywalking.oap.server.receiver.envoy.als.tcp;
import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
@Slf4j
public abstract class AbstractTCPAccessLogAnalyzer implements TCPAccessLogAnalyzer {
/**
- * Create an adapter to adapt the {@link HTTPAccessLogEntry log entry} into a {@link ServiceMeshMetric.Builder}.
+ * Create an adapter to adapt the {@link HTTPAccessLogEntry log entry} into a {@link TCPServiceMeshMetric.Builder}.
*
* @param entry the access log entry that is to be adapted from.
* @param sourceService the source service.
* @param targetService the target/destination service.
- * @return an adapter that adapts {@link HTTPAccessLogEntry log entry} into a {@link ServiceMeshMetric.Builder}.
+ * @return an adapter that adapts {@link HTTPAccessLogEntry log entry} into a {@link TCPServiceMeshMetric.Builder}.
*/
protected TCPLogEntry2MetricsAdapter newAdapter(
final TCPAccessLogEntry entry,
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/TCPLogEntry2MetricsAdapter.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/TCPLogEntry2MetricsAdapter.java
index 3af653e749..7b3a9ce72f 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/TCPLogEntry2MetricsAdapter.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/TCPLogEntry2MetricsAdapter.java
@@ -25,12 +25,9 @@ import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
import java.util.Optional;
import lombok.RequiredArgsConstructor;
import org.apache.skywalking.apm.network.common.v3.DetectPoint;
-import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
-import org.apache.skywalking.apm.network.servicemesh.v3.TCPInfo;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
-import static org.apache.skywalking.oap.server.library.util.StringUtil.isBlank;
import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.formatAsLong;
import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.parseInternalErrorCode;
import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.parseTLS;
@@ -55,7 +52,7 @@ public class TCPLogEntry2MetricsAdapter {
*
* @return the {@link ServiceMeshMetric.Builder} adapted from the given entry.
*/
- public ServiceMeshMetric.Builder adaptToDownstreamMetrics() {
+ public TCPServiceMeshMetric.Builder adaptToDownstreamMetrics() {
final AccessLogCommon properties = entry.getCommonProperties();
final long startTime = formatAsLong(properties.getStartTime());
final long duration = formatAsLong(properties.getTimeToLastDownstreamTxByte());
@@ -63,7 +60,6 @@ public class TCPLogEntry2MetricsAdapter {
return adaptCommonPart()
.setStartTime(startTime)
.setEndTime(startTime + duration)
- .setLatency((int) Math.max(1L, duration))
.setDetectPoint(DetectPoint.server);
}
@@ -72,7 +68,7 @@ public class TCPLogEntry2MetricsAdapter {
*
* @return the {@link ServiceMeshMetric.Builder} adapted from the given entry.
*/
- public ServiceMeshMetric.Builder adaptToUpstreamMetrics() {
+ public TCPServiceMeshMetric.Builder adaptToUpstreamMetrics() {
final AccessLogCommon properties = entry.getCommonProperties();
final long startTime = formatAsLong(properties.getStartTime());
final long outboundStartTime = startTime + formatAsLong(properties.getTimeToFirstUpstreamTxByte());
@@ -81,11 +77,10 @@ public class TCPLogEntry2MetricsAdapter {
return adaptCommonPart()
.setStartTime(outboundStartTime)
.setEndTime(outboundEndTime)
- .setLatency((int) Math.max(1L, outboundEndTime - outboundStartTime))
.setDetectPoint(DetectPoint.client);
}
- public ServiceMeshMetric.Builder adaptCommonPart() {
+ public TCPServiceMeshMetric.Builder adaptCommonPart() {
final AccessLogCommon properties = entry.getCommonProperties();
final ConnectionProperties connectionProperties = entry.getConnectionProperties();
final String tlsMode = parseTLS(properties.getTlsProperties());
@@ -95,19 +90,15 @@ public class TCPLogEntry2MetricsAdapter {
properties.getTimeToLastDownstreamTxByte().getNanos()
- properties.getTimeToFirstUpstreamRxByte().getNanos();
- final ServiceMeshMetric.Builder builder =
- ServiceMeshMetric.newBuilder()
- .setTlsMode(tlsMode)
- .setProtocol(Protocol.TCP)
- .setStatus(isBlank(internalErrorCode))
- .setTcp(
- TCPInfo.newBuilder()
- .setReceivedBytes(connectionProperties.getReceivedBytes())
- .setSentBytes(connectionProperties.getSentBytes())
- )
- .setInternalErrorCode(internalErrorCode)
- .setInternalRequestLatencyNanos(internalRequestLatencyNanos)
- .setInternalResponseLatencyNanos(internalResponseLatencyNanos);
+ final TCPServiceMeshMetric.Builder builder =
+ TCPServiceMeshMetric
+ .newBuilder()
+ .setTlsMode(tlsMode)
+ .setReceivedBytes(connectionProperties.getReceivedBytes())
+ .setSentBytes(connectionProperties.getSentBytes())
+ .setInternalErrorCode(internalErrorCode)
+ .setInternalRequestLatencyNanos(internalRequestLatencyNanos)
+ .setInternalResponseLatencyNanos(internalResponseLatencyNanos);
Optional.ofNullable(sourceService)
.map(ServiceMetaInfo::getServiceName)
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java
index 39607b78c5..c3ce4ad416 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java
@@ -18,16 +18,10 @@
package org.apache.skywalking.oap.server.receiver.envoy.als.tcp.k8s;
-import io.envoyproxy.envoy.config.core.v3.Address;
-import io.envoyproxy.envoy.config.core.v3.SocketAddress;
-import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
-import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import java.util.ArrayList;
-import java.util.List;
-import lombok.SneakyThrows;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
import org.apache.skywalking.oap.server.receiver.envoy.ServiceMetaInfoFactory;
@@ -35,9 +29,13 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
import org.apache.skywalking.oap.server.receiver.envoy.als.k8s.K8SServiceRegistry;
import org.apache.skywalking.oap.server.receiver.envoy.als.tcp.AbstractTCPAccessLogAnalyzer;
-
-import static org.apache.skywalking.oap.server.library.util.CollectionUtils.isNotEmpty;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
+import io.envoyproxy.envoy.config.core.v3.Address;
+import io.envoyproxy.envoy.config.core.v3.SocketAddress;
+import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
+import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
/**
* Analysis log based on ingress and mesh scenarios.
@@ -67,7 +65,7 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer {
final TCPAccessLogEntry entry,
final Role role
) {
- if (isNotEmpty(previousResult.getMetrics())) {
+ if (previousResult.getMetrics() != null && previousResult.getMetrics().hasTcpMetrics()) {
return previousResult;
}
switch (role) {
@@ -90,7 +88,7 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer {
return previousResult;
}
- final List<ServiceMeshMetric.Builder> sources = new ArrayList<>();
+ final TCPServiceMeshMetrics.Builder sources = TCPServiceMeshMetrics.newBuilder();
final Address downstreamRemoteAddress =
properties.hasDownstreamDirectRemoteAddress()
@@ -102,32 +100,32 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer {
if (cluster.startsWith("inbound|")) {
// Server side
- final ServiceMeshMetric.Builder metrics;
+ final TCPServiceMeshMetric metrics;
if (downstreamService.equals(config.serviceMetaInfoFactory().unknown())) {
// Ingress -> sidecar(server side)
// Mesh telemetry without source, the relation would be generated.
- metrics = newAdapter(entry, null, localService).adaptToDownstreamMetrics();
+ metrics = newAdapter(entry, null, localService).adaptToDownstreamMetrics().build();
log.debug("Transformed ingress->sidecar inbound mesh metrics {}", metrics);
} else {
// sidecar -> sidecar(server side)
- metrics = newAdapter(entry, downstreamService, localService).adaptToDownstreamMetrics();
+ metrics = newAdapter(entry, downstreamService, localService).adaptToDownstreamMetrics().build();
log.debug("Transformed sidecar->sidecar(server side) inbound mesh metrics {}", metrics);
}
- sources.add(metrics);
+ sources.addMetrics(metrics);
} else if (cluster.startsWith("outbound|")) {
// sidecar(client side) -> sidecar
final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress();
final ServiceMetaInfo destService = find(upstreamRemoteAddress.getSocketAddress().getAddress());
- final ServiceMeshMetric.Builder metric = newAdapter(entry, downstreamService, destService).adaptToUpstreamMetrics();
+ final TCPServiceMeshMetric metric = newAdapter(entry, downstreamService, destService).adaptToUpstreamMetrics().build();
log.debug("Transformed sidecar->sidecar(server side) inbound mesh metric {}", metric);
- sources.add(metric);
+ sources.addMetrics(metric);
}
- return Result.builder().metrics(sources).service(localService).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setTcpMetrics(sources)).service(localService).build();
}
protected Result analyzeProxy(final Result previousResult, final TCPAccessLogEntry entry) {
@@ -143,31 +141,31 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer {
return previousResult;
}
- final List<ServiceMeshMetric.Builder> result = new ArrayList<>(2);
+ final TCPServiceMeshMetrics.Builder metrics = TCPServiceMeshMetrics.newBuilder();
final SocketAddress downstreamRemoteAddressSocketAddress = downstreamRemoteAddress.getSocketAddress();
final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress());
final SocketAddress downstreamLocalAddressSocketAddress = downstreamLocalAddress.getSocketAddress();
final ServiceMetaInfo ingress = find(downstreamLocalAddressSocketAddress.getAddress());
- final ServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics();
+ final TCPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics();
log.debug("Transformed ingress inbound mesh metric {}", metric);
- result.add(metric);
+ metrics.addMetrics(metric);
final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress();
final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress());
- final ServiceMeshMetric.Builder outboundMetric =
+ final TCPServiceMeshMetric.Builder outboundMetric =
newAdapter(entry, ingress, targetService)
.adaptToUpstreamMetrics()
// Can't parse it from tls properties, leave it to Server side.
.setTlsMode(NON_TLS);
log.debug("Transformed ingress outbound mesh metric {}", outboundMetric);
- result.add(outboundMetric);
+ metrics.addMetrics(outboundMetric);
- return Result.builder().metrics(result).service(ingress).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setTcpMetrics(metrics)).service(ingress).build();
}
/**
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
index 20772be90b..5aa120a837 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
@@ -18,18 +18,15 @@
package org.apache.skywalking.oap.server.receiver.envoy.als.tcp.mx;
-import com.google.protobuf.Any;
-import com.google.protobuf.TextFormat;
-import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
-import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
-import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
-import java.util.ArrayList;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.DOWNSTREAM_KEY;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.UPSTREAM_KEY;
import java.util.Base64;
-import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
@@ -38,11 +35,12 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo;
import org.apache.skywalking.oap.server.receiver.envoy.als.mx.FieldsHelper;
import org.apache.skywalking.oap.server.receiver.envoy.als.mx.ServiceMetaInfoAdapter;
import org.apache.skywalking.oap.server.receiver.envoy.als.tcp.AbstractTCPAccessLogAnalyzer;
-
-import static org.apache.skywalking.oap.server.library.util.CollectionUtils.isNotEmpty;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.DOWNSTREAM_KEY;
-import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.UPSTREAM_KEY;
+import com.google.protobuf.Any;
+import com.google.protobuf.TextFormat;
+import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
+import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import lombok.extern.slf4j.Slf4j;
@Slf4j
public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyzer {
@@ -72,7 +70,7 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz
final TCPAccessLogEntry entry,
final Role role
) {
- if (isNotEmpty(previousResult.getMetrics())) {
+ if (previousResult.hasResult()) {
return previousResult;
}
if (!entry.hasCommonProperties()) {
@@ -91,7 +89,7 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz
return Result.builder().service(currSvc).build();
}
- final List<ServiceMeshMetric.Builder> result = new ArrayList<>();
+ final TCPServiceMeshMetrics.Builder result = TCPServiceMeshMetrics.newBuilder();
final AtomicBoolean downstreamExists = new AtomicBoolean();
stateMap.forEach((key, value) -> {
if (!key.equals(UPSTREAM_KEY) && !key.equals(DOWNSTREAM_KEY)) {
@@ -104,33 +102,33 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz
log.error("Fail to parse metadata {} to FlatNode", Base64.getEncoder().encode(value.toByteArray()));
return;
}
- final ServiceMeshMetric.Builder metrics;
+ final TCPServiceMeshMetric.Builder metrics;
switch (key) {
case UPSTREAM_KEY:
metrics = newAdapter(entry, currSvc, svc).adaptToUpstreamMetrics().setTlsMode(NON_TLS);
if (log.isDebugEnabled()) {
log.debug("Transformed a {} outbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
}
- result.add(metrics);
+ result.addMetrics(metrics);
break;
case DOWNSTREAM_KEY:
metrics = newAdapter(entry, svc, currSvc).adaptToDownstreamMetrics();
if (log.isDebugEnabled()) {
log.debug("Transformed a {} inbound mesh metrics {}", role, TextFormat.shortDebugString(metrics));
}
- result.add(metrics);
+ result.addMetrics(metrics);
downstreamExists.set(true);
break;
}
});
if (role.equals(Role.PROXY) && !downstreamExists.get()) {
- final ServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
+ final TCPServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics();
if (log.isDebugEnabled()) {
log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric));
}
- result.add(metric);
+ result.addMetrics(metric);
}
- return Result.builder().metrics(result).service(currSvc).build();
+ return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setTcpMetrics(result)).service(currSvc).build();
}
protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception {
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 51c128e36e..00bb3c1cca 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
@@ -22,7 +22,7 @@ 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.LogData;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
import org.apache.skywalking.oap.log.analyzer.module.LogAnalyzerModule;
import org.apache.skywalking.oap.log.analyzer.provider.log.ILogAnalyzerService;
import org.apache.skywalking.oap.server.core.analysis.Layer;
@@ -81,7 +81,7 @@ public class LogsPersistence implements ALSHTTPAnalysis {
public LogData convertToLogData(final HTTPAccessLogEntry logEntry, final Result result) {
final ServiceMetaInfo service = result.getService();
- final ServiceMeshMetric.Builder metrics =
+ final HTTPServiceMeshMetric.Builder metrics =
new LogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart();
return LogData
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 6feadbed3b..ab792eeb0d 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
@@ -18,11 +18,8 @@
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.LogData;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
import org.apache.skywalking.oap.log.analyzer.module.LogAnalyzerModule;
import org.apache.skywalking.oap.log.analyzer.provider.log.ILogAnalyzerService;
import org.apache.skywalking.oap.server.core.analysis.Layer;
@@ -33,6 +30,9 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.Role;
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 io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry;
+import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage;
+import lombok.extern.slf4j.Slf4j;
/**
* {@code LogsPersistence} analyzes the error logs and persists them to the log system.
@@ -81,7 +81,7 @@ public class TCPLogsPersistence implements TCPAccessLogAnalyzer {
public LogData convertToLogData(final TCPAccessLogEntry logEntry, final Result result) {
final ServiceMetaInfo service = result.getService();
- final ServiceMeshMetric.Builder metrics =
+ final TCPServiceMeshMetric.Builder metrics =
new TCPLogEntry2MetricsAdapter(logEntry, null, null).adaptCommonPart();
return LogData
diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SALSServiceMeshHTTPAnalysisTest.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SALSServiceMeshHTTPAnalysisTest.java
index 84ffff66f4..a70fa735ab 100644
--- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SALSServiceMeshHTTPAnalysisTest.java
+++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SALSServiceMeshHTTPAnalysisTest.java
@@ -24,7 +24,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import org.apache.skywalking.apm.network.common.v3.DetectPoint;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig;
import org.apache.skywalking.oap.server.receiver.envoy.MetricServiceGRPCHandlerTestMain;
@@ -86,14 +86,14 @@ public class K8SALSServiceMeshHTTPAnalysisTest {
AccessLogAnalyzer.Result result = this.analysis.analysis(AccessLogAnalyzer.Result.builder().build(), requestBuilder.getIdentifier(), requestBuilder.getHttpLogs().getLogEntry(0), Role.PROXY);
- Assert.assertEquals(2, result.getMetrics().size());
+ Assert.assertEquals(2, result.getMetrics().getHttpMetrics().getMetricsCount());
- ServiceMeshMetric.Builder incoming = result.getMetrics().get(0);
+ HTTPServiceMeshMetric incoming = result.getMetrics().getHttpMetrics().getMetrics(0);
Assert.assertEquals("UNKNOWN", incoming.getSourceServiceName());
Assert.assertEquals("ingress", incoming.getDestServiceName());
Assert.assertEquals(DetectPoint.server, incoming.getDetectPoint());
- ServiceMeshMetric.Builder outgoing = result.getMetrics().get(1);
+ HTTPServiceMeshMetric outgoing = result.getMetrics().getHttpMetrics().getMetrics(1);
Assert.assertEquals("ingress", outgoing.getSourceServiceName());
Assert.assertEquals("productpage", outgoing.getDestServiceName());
Assert.assertEquals(DetectPoint.client, outgoing.getDetectPoint());
@@ -108,9 +108,9 @@ public class K8SALSServiceMeshHTTPAnalysisTest {
AccessLogAnalyzer.Result result = this.analysis.analysis(AccessLogAnalyzer.Result.builder().build(), requestBuilder.getIdentifier(), requestBuilder.getHttpLogs().getLogEntry(0), Role.SIDECAR);
- Assert.assertEquals(1, result.getMetrics().size());
+ Assert.assertEquals(1, result.getMetrics().getHttpMetrics().getMetricsCount());
- ServiceMeshMetric.Builder incoming = result.getMetrics().get(0);
+ HTTPServiceMeshMetric incoming = result.getMetrics().getHttpMetrics().getMetrics(0);
Assert.assertEquals("", incoming.getSourceServiceName());
Assert.assertEquals("productpage", incoming.getDestServiceName());
Assert.assertEquals(DetectPoint.server, incoming.getDetectPoint());
@@ -125,9 +125,9 @@ public class K8SALSServiceMeshHTTPAnalysisTest {
AccessLogAnalyzer.Result result = this.analysis.analysis(AccessLogAnalyzer.Result.builder().build(), requestBuilder.getIdentifier(), requestBuilder.getHttpLogs().getLogEntry(0), Role.SIDECAR);
- Assert.assertEquals(1, result.getMetrics().size());
+ Assert.assertEquals(1, result.getMetrics().getHttpMetrics().getMetricsCount());
- ServiceMeshMetric.Builder incoming = result.getMetrics().get(0);
+ HTTPServiceMeshMetric incoming = result.getMetrics().getHttpMetrics().getMetrics(0);
Assert.assertEquals("productpage", incoming.getSourceServiceName());
Assert.assertEquals("review", incoming.getDestServiceName());
Assert.assertEquals(DetectPoint.server, incoming.getDetectPoint());
@@ -142,9 +142,9 @@ public class K8SALSServiceMeshHTTPAnalysisTest {
AccessLogAnalyzer.Result result = this.analysis.analysis(AccessLogAnalyzer.Result.builder().build(), requestBuilder.getIdentifier(), requestBuilder.getHttpLogs().getLogEntry(0), Role.SIDECAR);
- Assert.assertEquals(1, result.getMetrics().size());
+ Assert.assertEquals(1, result.getMetrics().getHttpMetrics().getMetricsCount());
- ServiceMeshMetric.Builder incoming = result.getMetrics().get(0);
+ HTTPServiceMeshMetric incoming = result.getMetrics().getHttpMetrics().getMetrics(0);
Assert.assertEquals("productpage", incoming.getSourceServiceName());
Assert.assertEquals("detail", incoming.getDestServiceName());
Assert.assertEquals(DetectPoint.client, incoming.getDetectPoint());
diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java
index cb568dd0d6..42e2fb421e 100644
--- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java
+++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java
@@ -21,7 +21,7 @@ package org.apache.skywalking.aop.server.receiver.mesh;
import io.grpc.Status;
import io.grpc.stub.StreamObserver;
import org.apache.skywalking.apm.network.servicemesh.v3.MeshProbeDownstream;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetricServiceGrpc;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.Logger;
@@ -35,15 +35,15 @@ public class MeshGRPCHandler extends ServiceMeshMetricServiceGrpc.ServiceMeshMet
}
@Override
- public StreamObserver<ServiceMeshMetric> collect(StreamObserver<MeshProbeDownstream> responseObserver) {
- return new StreamObserver<ServiceMeshMetric>() {
+ public StreamObserver<ServiceMeshMetrics> collect(StreamObserver<MeshProbeDownstream> responseObserver) {
+ return new StreamObserver<ServiceMeshMetrics>() {
@Override
- public void onNext(ServiceMeshMetric metrics) {
+ public void onNext(ServiceMeshMetrics metrics) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Received mesh metrics: {}", metrics);
}
- TelemetryDataDispatcher.process(metrics.toBuilder());
+ TelemetryDataDispatcher.process(metrics);
}
@Override
diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java
index 7c04d2e8a1..97c0a34cf6 100644
--- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java
@@ -68,7 +68,6 @@ public class MeshReceiverProvider extends ModuleProvider {
.getService(GRPCHandlerRegister.class);
MeshGRPCHandler meshGRPCHandler = new MeshGRPCHandler(getManager());
service.addHandler(meshGRPCHandler);
- service.addHandler(new MeshGRPCHandlerCompat(meshGRPCHandler));
}
@Override
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 c037764523..93a8cece9a 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
@@ -19,8 +19,10 @@
package org.apache.skywalking.aop.server.receiver.mesh;
import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
+import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric;
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;
@@ -36,6 +38,11 @@ import org.apache.skywalking.oap.server.core.source.ServiceInstanceRelation;
import org.apache.skywalking.oap.server.core.source.ServiceInstanceUpdate;
import org.apache.skywalking.oap.server.core.source.ServiceRelation;
import org.apache.skywalking.oap.server.core.source.SourceReceiver;
+import org.apache.skywalking.oap.server.core.source.TCPService;
+import org.apache.skywalking.oap.server.core.source.TCPServiceInstance;
+import org.apache.skywalking.oap.server.core.source.TCPServiceInstanceRelation;
+import org.apache.skywalking.oap.server.core.source.TCPServiceInstanceUpdate;
+import org.apache.skywalking.oap.server.core.source.TCPServiceRelation;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.util.StringUtil;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
@@ -51,6 +58,8 @@ import com.google.gson.JsonObject;
*/
@Slf4j
public class TelemetryDataDispatcher {
+ private static final int TCP_COMPONENT = 110; // Defined in component-libraries.yml
+
private static SourceReceiver SOURCE_RECEIVER;
private static NamingControl NAME_LENGTH_CONTROL;
private static HistogramMetrics MESH_ANALYSIS_METRICS;
@@ -78,36 +87,99 @@ public class TelemetryDataDispatcher {
);
}
- public static void process(ServiceMeshMetric.Builder data) {
- try (HistogramMetrics.Timer ignored = MESH_ANALYSIS_METRICS.createTimer()) {
- if (data.getSourceServiceName() != null) {
- data.setSourceServiceName(NAME_LENGTH_CONTROL.formatServiceName(data.getSourceServiceName()));
- }
- if (data.getSourceServiceInstance() != null) {
- data.setSourceServiceInstance(NAME_LENGTH_CONTROL.formatInstanceName(data.getSourceServiceInstance()));
- }
- if (data.getDestServiceName() != null) {
- data.setDestServiceName(NAME_LENGTH_CONTROL.formatServiceName(data.getDestServiceName()));
- }
- if (data.getDestServiceInstance() != null) {
- data.setDestServiceInstance(NAME_LENGTH_CONTROL.formatInstanceName(data.getDestServiceInstance()));
- }
- if (data.getEndpoint() != null) {
- data.setEndpoint(NAME_LENGTH_CONTROL.formatEndpointName(data.getDestServiceName(), data.getEndpoint()));
+ public static void process(ServiceMeshMetrics metrics) {
+ dispatchHTTPMetrics(metrics);
+ dispatchTCPMetrics(metrics);
+ }
+
+ private static void dispatchTCPMetrics(ServiceMeshMetrics metrics) {
+ metrics.getTcpMetrics().getMetricsList().forEach(m -> {
+ final TCPServiceMeshMetric.Builder data = m.toBuilder();
+ try (HistogramMetrics.Timer ignored = MESH_ANALYSIS_METRICS.createTimer()) {
+ if (data.getSourceServiceName() != null) {
+ data.setSourceServiceName(
+ NAME_LENGTH_CONTROL.formatServiceName(data.getSourceServiceName()));
+ }
+ if (data.getSourceServiceInstance() != null) {
+ data.setSourceServiceInstance(
+ NAME_LENGTH_CONTROL.formatInstanceName(data.getSourceServiceInstance()));
+ }
+ if (data.getDestServiceName() != null) {
+ data.setDestServiceName(
+ NAME_LENGTH_CONTROL.formatServiceName(data.getDestServiceName()));
+ }
+ if (data.getDestServiceInstance() != null) {
+ data.setDestServiceInstance(
+ NAME_LENGTH_CONTROL.formatInstanceName(data.getDestServiceInstance()));
+ }
+ if (data.getInternalErrorCode() == null) {
+ // Add this since 8.2.0, set the default value.
+ data.setInternalErrorCode(Const.EMPTY_STRING);
+ }
+
+ dispatchTCPMetrics(data);
+ } catch (Exception e) {
+ MESH_ERROR_METRICS.inc();
+ log.error(e.getMessage(), e);
}
- if (data.getInternalErrorCode() == null) {
- // Add this since 8.2.0, set the default value.
- data.setInternalErrorCode(Const.EMPTY_STRING);
+ });
+ }
+
+ private static void dispatchHTTPMetrics(ServiceMeshMetrics metrics) {
+ metrics.getHttpMetrics().getMetricsList().forEach(m -> {
+ final HTTPServiceMeshMetric.Builder data = m.toBuilder();
+ try (HistogramMetrics.Timer ignored = MESH_ANALYSIS_METRICS.createTimer()) {
+ if (data.getSourceServiceName() != null) {
+ data.setSourceServiceName(
+ NAME_LENGTH_CONTROL.formatServiceName(data.getSourceServiceName()));
+ }
+ if (data.getSourceServiceInstance() != null) {
+ data.setSourceServiceInstance(
+ NAME_LENGTH_CONTROL.formatInstanceName(data.getSourceServiceInstance()));
+ }
+ if (data.getDestServiceName() != null) {
+ data.setDestServiceName(
+ NAME_LENGTH_CONTROL.formatServiceName(data.getDestServiceName()));
+ }
+ if (data.getDestServiceInstance() != null) {
+ data.setDestServiceInstance(
+ NAME_LENGTH_CONTROL.formatInstanceName(data.getDestServiceInstance()));
+ }
+ if (data.getEndpoint() != null) {
+ data.setEndpoint(NAME_LENGTH_CONTROL
+ .formatEndpointName(data.getDestServiceName(), data.getEndpoint()));
+ }
+ if (data.getInternalErrorCode() == null) {
+ // Add this since 8.2.0, set the default value.
+ data.setInternalErrorCode(Const.EMPTY_STRING);
+ }
+
+ dispatchHTTPMetrics(data);
+ } catch (Exception e) {
+ MESH_ERROR_METRICS.inc();
+ log.error(e.getMessage(), e);
}
+ });
+ }
+
+ private static void dispatchTCPMetrics(TCPServiceMeshMetric.Builder metrics) {
+ long minuteTimeBucket = TimeBucket.getMinuteTimeBucket(metrics.getStartTime());
- doDispatch(data);
- } catch (Exception e) {
- MESH_ERROR_METRICS.inc();
- log.error(e.getMessage(), e);
+ if (org.apache.skywalking.apm.network.common.v3.DetectPoint.server.equals(metrics.getDetectPoint())) {
+ toTCPService(metrics, minuteTimeBucket);
+ toTCPServiceInstance(metrics, minuteTimeBucket);
+ toTCPServiceInstanceTraffic(metrics, minuteTimeBucket);
+ }
+
+ String sourceService = metrics.getSourceServiceName();
+ // Don't generate relation, if no source.
+ if (StringUtil.isNotEmpty(sourceService)) {
+ toTCPServiceRelation(metrics, minuteTimeBucket);
+ toTCPServiceInstanceRelation(metrics, minuteTimeBucket);
}
}
- static void doDispatch(ServiceMeshMetric.Builder metrics) {
+ static void dispatchHTTPMetrics(HTTPServiceMeshMetric.Builder metrics) {
long minuteTimeBucket = TimeBucket.getMinuteTimeBucket(metrics.getStartTime());
if (org.apache.skywalking.apm.network.common.v3.DetectPoint.server.equals(metrics.getDetectPoint())) {
@@ -125,7 +197,7 @@ public class TelemetryDataDispatcher {
}
}
- private static void toService(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ private static void toService(HTTPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
Service service = new Service();
service.setTimeBucket(minuteTimeBucket);
service.setName(metrics.getDestServiceName());
@@ -139,13 +211,24 @@ public class TelemetryDataDispatcher {
service.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
service.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
service.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
- service.getTcpInfo().setReceivedBytes(metrics.getTcp().getReceivedBytes());
- service.getTcpInfo().setSentBytes(metrics.getTcp().getSentBytes());
SOURCE_RECEIVER.receive(service);
}
- private static void toServiceRelation(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ private static void toTCPService(TCPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ TCPService service = new TCPService();
+ service.setTimeBucket(minuteTimeBucket);
+ service.setName(metrics.getDestServiceName());
+ service.setLayer(Layer.MESH);
+ service.setServiceInstanceName(metrics.getDestServiceInstance());
+ service.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
+ service.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
+ service.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
+
+ SOURCE_RECEIVER.receive(service);
+ }
+
+ private static void toServiceRelation(HTTPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
ServiceRelation serviceRelation = new ServiceRelation();
serviceRelation.setTimeBucket(minuteTimeBucket);
serviceRelation.setSourceServiceName(metrics.getSourceServiceName());
@@ -165,13 +248,30 @@ public class TelemetryDataDispatcher {
serviceRelation.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
serviceRelation.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
serviceRelation.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
- serviceRelation.getTcpInfo().setReceivedBytes(metrics.getTcp().getReceivedBytes());
- serviceRelation.getTcpInfo().setSentBytes(metrics.getTcp().getSentBytes());
SOURCE_RECEIVER.receive(serviceRelation);
}
- private static void toServiceInstance(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ private static void toTCPServiceRelation(TCPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ TCPServiceRelation serviceRelation = new TCPServiceRelation();
+ serviceRelation.setTimeBucket(minuteTimeBucket);
+ serviceRelation.setSourceServiceName(metrics.getSourceServiceName());
+ serviceRelation.setSourceLayer(Layer.MESH);
+ serviceRelation.setSourceServiceInstanceName(metrics.getSourceServiceInstance());
+ serviceRelation.setDestServiceName(metrics.getDestServiceName());
+ serviceRelation.setDestLayer(Layer.MESH);
+ serviceRelation.setDestServiceInstanceName(metrics.getDestServiceInstance());
+ serviceRelation.setDetectPoint(detectPointMapping(metrics.getDetectPoint()));
+ serviceRelation.setComponentId(TCP_COMPONENT);
+ serviceRelation.setTlsMode(metrics.getTlsMode());
+ serviceRelation.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
+ serviceRelation.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
+ serviceRelation.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
+
+ SOURCE_RECEIVER.receive(serviceRelation);
+ }
+
+ private static void toServiceInstance(HTTPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
ServiceInstance serviceInstance = new ServiceInstance();
serviceInstance.setTimeBucket(minuteTimeBucket);
serviceInstance.setName(metrics.getDestServiceInstance());
@@ -185,13 +285,24 @@ public class TelemetryDataDispatcher {
serviceInstance.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
serviceInstance.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
serviceInstance.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
- serviceInstance.getTcpInfo().setReceivedBytes(metrics.getTcp().getReceivedBytes());
- serviceInstance.getTcpInfo().setSentBytes(metrics.getTcp().getSentBytes());
SOURCE_RECEIVER.receive(serviceInstance);
}
- private static void toServiceInstanceTraffic(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ private static void toTCPServiceInstance(TCPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ TCPServiceInstance serviceInstance = new TCPServiceInstance();
+ serviceInstance.setTimeBucket(minuteTimeBucket);
+ serviceInstance.setName(metrics.getDestServiceInstance());
+ serviceInstance.setServiceName(metrics.getDestServiceName());
+ serviceInstance.setServiceLayer(Layer.MESH);
+ serviceInstance.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
+ serviceInstance.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
+ serviceInstance.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
+
+ SOURCE_RECEIVER.receive(serviceInstance);
+ }
+
+ private static void toServiceInstanceTraffic(HTTPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
ServiceInstanceUpdate instanceTraffic = new ServiceInstanceUpdate();
instanceTraffic.setTimeBucket(minuteTimeBucket);
instanceTraffic.setName(metrics.getDestServiceInstance());
@@ -207,7 +318,23 @@ public class TelemetryDataDispatcher {
SOURCE_RECEIVER.receive(instanceTraffic);
}
- private static void toServiceInstanceRelation(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ private static void toTCPServiceInstanceTraffic(TCPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ TCPServiceInstanceUpdate instanceTraffic = new TCPServiceInstanceUpdate();
+ instanceTraffic.setTimeBucket(minuteTimeBucket);
+ instanceTraffic.setName(metrics.getDestServiceInstance());
+ instanceTraffic.setServiceId(IDManager.ServiceID.buildId(metrics.getDestServiceName(), true));
+ if (metrics.getDestInstancePropertiesList() != null) {
+ final JsonObject properties = new JsonObject();
+ metrics
+ .getDestInstancePropertiesList()
+ .stream()
+ .forEach(it -> properties.addProperty(it.getKey(), it.getValue()));
+ instanceTraffic.setProperties(properties);
+ }
+ SOURCE_RECEIVER.receive(instanceTraffic);
+ }
+
+ private static void toServiceInstanceRelation(HTTPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
ServiceInstanceRelation serviceRelation = new ServiceInstanceRelation();
serviceRelation.setTimeBucket(minuteTimeBucket);
serviceRelation.setSourceServiceInstanceName(metrics.getSourceServiceInstance());
@@ -228,13 +355,30 @@ public class TelemetryDataDispatcher {
serviceRelation.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
serviceRelation.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
serviceRelation.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
- serviceRelation.getTcpInfo().setReceivedBytes(metrics.getTcp().getReceivedBytes());
- serviceRelation.getTcpInfo().setSentBytes(metrics.getTcp().getSentBytes());
SOURCE_RECEIVER.receive(serviceRelation);
}
- private static void toEndpoint(ServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ private static void toTCPServiceInstanceRelation(TCPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
+ TCPServiceInstanceRelation serviceRelation = new TCPServiceInstanceRelation();
+ serviceRelation.setTimeBucket(minuteTimeBucket);
+ serviceRelation.setSourceServiceInstanceName(metrics.getSourceServiceInstance());
+ serviceRelation.setSourceServiceName(metrics.getSourceServiceName());
+ serviceRelation.setSourceServiceLayer(Layer.MESH);
+ serviceRelation.setDestServiceInstanceName(metrics.getDestServiceInstance());
+ serviceRelation.setDestServiceLayer(Layer.MESH);
+ serviceRelation.setDestServiceName(metrics.getDestServiceName());
+ serviceRelation.setDetectPoint(detectPointMapping(metrics.getDetectPoint()));
+ serviceRelation.setComponentId(TCP_COMPONENT);
+ serviceRelation.setTlsMode(metrics.getTlsMode());
+ serviceRelation.getSideCar().setInternalErrorCode(metrics.getInternalErrorCode());
+ serviceRelation.getSideCar().setInternalRequestLatencyNanos(metrics.getInternalRequestLatencyNanos());
+ serviceRelation.getSideCar().setInternalResponseLatencyNanos(metrics.getInternalResponseLatencyNanos());
+
+ SOURCE_RECEIVER.receive(serviceRelation);
+ }
+
+ private static void toEndpoint(HTTPServiceMeshMetric.Builder metrics, long minuteTimeBucket) {
if (StringUtil.isEmpty(metrics.getEndpoint())) {
return;
}
@@ -261,8 +405,6 @@ public class TelemetryDataDispatcher {
return RequestType.gRPC;
case HTTP:
return RequestType.HTTP;
- case TCP:
- return RequestType.TCP;
case UNRECOGNIZED:
default:
return RequestType.RPC;
@@ -277,9 +419,6 @@ public class TelemetryDataDispatcher {
case HTTP:
// HTTP in component-libraries.yml
return 49;
- case TCP:
- // TCP in component-libraries.yml
- return 110;
case UNRECOGNIZED:
default:
// RPC in component-libraries.yml
diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/mesh/MeshDataMock.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/mesh/MeshDataMock.java
index 69e6d1db87..ad18591928 100644
--- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/mesh/MeshDataMock.java
+++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/mesh/MeshDataMock.java
@@ -23,9 +23,11 @@ import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import java.util.concurrent.TimeUnit;
import org.apache.skywalking.apm.network.common.v3.DetectPoint;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.MeshProbeDownstream;
import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetricServiceGrpc;
public class MeshDataMock {
@@ -34,27 +36,36 @@ public class MeshDataMock {
public static void main(String[] args) throws InterruptedException {
ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).usePlaintext().build();
- long startTimestamp = System.currentTimeMillis();
- //long startTimestamp = new DateTime().minusDays(2).getMillis();
-
- final StreamObserver<ServiceMeshMetric> meshObserver = createMeshObserver(channel);
+ final StreamObserver<ServiceMeshMetrics> meshObserver = createMeshObserver(channel);
for (int i = 0; i < 50; i++) {
- meshObserver.onNext(ServiceMeshMetric.newBuilder()
- .setSourceServiceName("e2e-test-source-service")
- .setSourceServiceInstance("e2e-test-source-service-instance")
- .setDestServiceName("Extra model column are the column defined by in the codes, These columns of model are not required logically in aggregation or further query,")
- .setDestServiceInstance("Extra model column are the column defined by in the codes, These columns of model are not required logically in aggregation or further query,")
- .setEndpoint("Extra model column are the column defined by in the codes, These columns of model are not required logically in aggregation or further query,")
- .setStartTime(System.currentTimeMillis() - 1000L)
- .setEndTime(System.currentTimeMillis() - 500L + i)
- .setLatency(2000)
- .setResponseCode(200)
- .setStatus(true)
- .setProtocol(Protocol.HTTP)
- .setDetectPoint(DetectPoint.server)
- .setInternalErrorCode("rate_limited")
- .build());
+ meshObserver.onNext(
+ ServiceMeshMetrics
+ .newBuilder()
+ .setHttpMetrics(
+ HTTPServiceMeshMetrics
+ .newBuilder()
+ .addMetrics(
+ HTTPServiceMeshMetric
+ .newBuilder()
+ .setSourceServiceName("e2e-test-source-service")
+ .setSourceServiceInstance("e2e-test-source-service-instance")
+ .setDestServiceName(
+ "Extra model column are the column defined by in the codes, These columns of model are not required logically in aggregation or further query,")
+ .setDestServiceInstance(
+ "Extra model column are the column defined by in the codes, These columns of model are not required logically in aggregation or further query,")
+ .setEndpoint(
+ "Extra model column are the column defined by in the codes, These columns of model are not required logically in aggregation or further query,")
+ .setStartTime(System.currentTimeMillis() - 1000L)
+ .setEndTime(System.currentTimeMillis() - 500L + i)
+ .setLatency(2000)
+ .setResponseCode(200)
+ .setStatus(true)
+ .setProtocol(Protocol.HTTP)
+ .setDetectPoint(DetectPoint.server)
+ .setInternalErrorCode("rate_limited")
+ .build()))
+ .build());
}
meshObserver.onCompleted();
@@ -63,7 +74,7 @@ public class MeshDataMock {
}
}
- private static StreamObserver<ServiceMeshMetric> createMeshObserver(ManagedChannel channel) {
+ private static StreamObserver<ServiceMeshMetrics> createMeshObserver(ManagedChannel channel) {
ServiceMeshMetricServiceGrpc.ServiceMeshMetricServiceStub stub = ServiceMeshMetricServiceGrpc.newStub(
channel);
return stub.collect(new StreamObserver<MeshProbeDownstream>() {
diff --git a/oap-server/server-starter/src/main/resources/oal/core.oal b/oap-server/server-starter/src/main/resources/oal/core.oal
index 10ed0a9bf5..6bbd2ccc91 100755
--- a/oap-server/server-starter/src/main/resources/oal/core.oal
+++ b/oap-server/server-starter/src/main/resources/oal/core.oal
@@ -16,9 +16,6 @@
*
*/
-// For services using protocols HTTP 1/2, gRPC, RPC, etc., the cpm metrics means "calls per minute",
-// for services that are built on top of TCP, the cpm means "packages per minute".
-
// Service scope metrics
service_resp_time = from(Service.latency).longAvg();
service_sla = from(Service.*).percent(status == true);
@@ -85,4 +82,4 @@ cache_write_percentile = from(CacheAccess.latency).filter(operation == VirtualCa
cache_access_resp_time = from(CacheAccess.latency).longAvg();
cache_access_sla = from(CacheAccess.*).percent(status == true);
cache_access_cpm = from(CacheAccess.*).cpm();
-cache_access_percentile = from(CacheAccess.latency).percentile(10);
\ No newline at end of file
+cache_access_percentile = from(CacheAccess.latency).percentile(10);
diff --git a/oap-server/server-starter/src/main/resources/oal/tcp.oal b/oap-server/server-starter/src/main/resources/oal/tcp.oal
index 0ce8b6ca18..7934c56c3d 100644
--- a/oap-server/server-starter/src/main/resources/oal/tcp.oal
+++ b/oap-server/server-starter/src/main/resources/oal/tcp.oal
@@ -17,16 +17,16 @@
*/
// TCP services' metrics
-service_throughput_received = from(Service.tcpInfo.receivedBytes).filter(type == RequestType.TCP).longAvg();
-service_throughput_sent = from(Service.tcpInfo.sentBytes).filter(type == RequestType.TCP).longAvg();
-service_relation_client_received = from(ServiceRelation.tcpInfo.receivedBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.CLIENT).longAvg();
-service_relation_client_sent = from(ServiceRelation.tcpInfo.sentBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.CLIENT).longAvg();
-service_relation_server_received = from(ServiceRelation.tcpInfo.receivedBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.SERVER).longAvg();
-service_relation_server_sent = from(ServiceRelation.tcpInfo.sentBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.SERVER).longAvg();
+service_throughput_received = from(TCPService.receivedBytes).longAvg();
+service_throughput_sent = from(TCPService.sentBytes).longAvg();
+service_relation_client_received = from(TCPServiceRelation.receivedBytes).filter(detectPoint == DetectPoint.CLIENT).longAvg();
+service_relation_client_sent = from(TCPServiceRelation.sentBytes).filter(detectPoint == DetectPoint.CLIENT).longAvg();
+service_relation_server_received = from(TCPServiceRelation.receivedBytes).filter(detectPoint == DetectPoint.SERVER).longAvg();
+service_relation_server_sent = from(TCPServiceRelation.sentBytes).filter(detectPoint == DetectPoint.SERVER).longAvg();
-service_instance_throughput_received = from(ServiceInstance.tcpInfo.receivedBytes).filter(type == RequestType.TCP).longAvg();
-service_instance_throughput_sent = from(ServiceInstance.tcpInfo.sentBytes).filter(type == RequestType.TCP).longAvg();
-service_instance_relation_client_received = from(ServiceInstanceRelation.tcpInfo.receivedBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.CLIENT).longAvg();
-service_instance_relation_client_sent = from(ServiceInstanceRelation.tcpInfo.sentBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.CLIENT).longAvg();
-service_instance_relation_server_received = from(ServiceInstanceRelation.tcpInfo.receivedBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.SERVER).longAvg();
-service_instance_relation_server_sent = from(ServiceInstanceRelation.tcpInfo.sentBytes).filter(type == RequestType.TCP).filter(detectPoint == DetectPoint.SERVER).longAvg();
+service_instance_throughput_received = from(TCPServiceInstance.receivedBytes).longAvg();
+service_instance_throughput_sent = from(TCPServiceInstance.sentBytes).longAvg();
+service_instance_relation_client_received = from(TCPServiceInstanceRelation.receivedBytes).filter(detectPoint == DetectPoint.CLIENT).longAvg();
+service_instance_relation_client_sent = from(TCPServiceInstanceRelation.sentBytes).filter(detectPoint == DetectPoint.CLIENT).longAvg();
+service_instance_relation_server_received = from(TCPServiceInstanceRelation.receivedBytes).filter(detectPoint == DetectPoint.SERVER).longAvg();
+service_instance_relation_server_sent = from(TCPServiceInstanceRelation.sentBytes).filter(detectPoint == DetectPoint.SERVER).longAvg();
diff --git a/oap-server/server-starter/src/main/resources/ui-initialized-templates/general/general-instance.json b/oap-server/server-starter/src/main/resources/ui-initialized-templates/general/general-instance.json
index 71f11f6612..87ba2d627a 100644
--- a/oap-server/server-starter/src/main/resources/ui-initialized-templates/general/general-instance.json
+++ b/oap-server/server-starter/src/main/resources/ui-initialized-templates/general/general-instance.json
@@ -137,7 +137,6 @@
"type": "Widget",
"widget": {
"title": "Service Instance Load (calls / min)",
- "tips": "For HTTP 1/2, gRPC, RPC services, this means Calls Per Minute (calls / min), for TCP services, this means Packets Per Minute (PPM)",
"name": "Instance_Load"
},
"graph": {
diff --git a/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-instance.json b/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-instance.json
index 1fd4baf1ef..e1b59522da 100644
--- a/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-instance.json
+++ b/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-instance.json
@@ -172,7 +172,6 @@
"type": "Widget",
"widget": {
"title": "Service Instance Load (calls or packets / min)",
- "tips": "For HTTP 1/2, gRPC, RPC services, this means Calls Per Minute (calls / min), for TCP services, this means Packets Per Minute (packets / min)",
"name": "Instance_Load"
},
"graph": {
diff --git a/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-service.json b/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-service.json
index 066a23da66..37c3b7066c 100644
--- a/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-service.json
+++ b/oap-server/server-starter/src/main/resources/ui-initialized-templates/mesh/mesh-service.json
@@ -99,8 +99,7 @@
"i": "2",
"type": "Widget",
"widget": {
- "title": "Service Load (calls or packets / min)",
- "tips": "For HTTP 1/2, gRPC, RPC services, this means Calls Per Minute (calls / min), for TCP services, this means Packets Per Minute (packets / min)"
+ "title": "Service Load (calls or packets / min)"
},
"graph": {
"type": "Card",
@@ -343,7 +342,6 @@
"type": "Widget",
"widget": {
"title": "Service Load (calls or packets / min)",
- "tips": "For HTTP 1/2, gRPC, RPC services, this means Calls Per Minute (calls / min), for TCP services, this means Packets Per Minute (packets / min)",
"name": "Service_Load"
},
"graph": {
@@ -440,8 +438,7 @@
"i": "11",
"type": "Widget",
"widget": {
- "title": "Service Instances Load (calls or packets / min)",
- "tips": "For HTTP 1/2, gRPC, RPC services, this means Calls Per Minute (calls / min), for TCP services, this means Packets Per Minute (packets / min)"
+ "title": "Service Instances Load (calls or packets / min)"
},
"graph": {
"type": "TopList"
@@ -567,8 +564,7 @@
"i": "22",
"type": "Widget",
"widget": {
- "title": "Endpoint Load in Current Service (calls or packets / min)",
- "tips": "For HTTP 1/2, gRPC, RPC services, this means Calls Per Minute (calls / min), for TCP services, this means Packets Per Minute (packets / min)"
+ "title": "Endpoint Load in Current Service (calls or packets / min)"
},
"graph": {
"type": "TopList",
diff --git a/test/e2e-v2/java-test-service/e2e-mock-sender/src/main/java/org/apache/skywalking/e2e/controller/ServiceMeshMetricSenderController.java b/test/e2e-v2/java-test-service/e2e-mock-sender/src/main/java/org/apache/skywalking/e2e/controller/ServiceMeshMetricSenderController.java
index 5477a71d4b..1b094c40c8 100644
--- a/test/e2e-v2/java-test-service/e2e-mock-sender/src/main/java/org/apache/skywalking/e2e/controller/ServiceMeshMetricSenderController.java
+++ b/test/e2e-v2/java-test-service/e2e-mock-sender/src/main/java/org/apache/skywalking/e2e/controller/ServiceMeshMetricSenderController.java
@@ -26,9 +26,11 @@ import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.concurrent.CountDownLatch;
import org.apache.skywalking.apm.network.common.v3.DetectPoint;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.MeshProbeDownstream;
import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
-import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
+import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics;
import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetricServiceGrpc;
import org.apache.skywalking.e2e.E2EConfiguration;
import org.springframework.web.bind.annotation.PathVariable;
@@ -54,34 +56,42 @@ public class ServiceMeshMetricSenderController {
@PostMapping("/sendMetrics4TTL/{metricsTTL}")
public String sendMetrics4TTL(@PathVariable("metricsTTL") int metricsTTL) throws Exception {
- final ServiceMeshMetric.Builder builder =
- ServiceMeshMetric.newBuilder()
- .setSourceServiceName("e2e-test-source-service")
- .setSourceServiceInstance("e2e-test-source-service-instance")
- .setDestServiceName("e2e-test-dest-service")
- .setDestServiceInstance("e2e-test-dest-service-instance")
- .setEndpoint("e2e/test")
- .setLatency(2000)
- .setResponseCode(200)
- .setStatus(SUCCESS)
- .setProtocol(Protocol.HTTP)
- .setDetectPoint(DetectPoint.server);
+ final HTTPServiceMeshMetric.Builder builder =
+ HTTPServiceMeshMetric
+ .newBuilder()
+ .setSourceServiceName("e2e-test-source-service")
+ .setSourceServiceInstance("e2e-test-source-service-instance")
+ .setDestServiceName("e2e-test-dest-service")
+ .setDestServiceInstance("e2e-test-dest-service-instance")
+ .setEndpoint("e2e/test")
+ .setLatency(2000)
+ .setResponseCode(200)
+ .setStatus(SUCCESS)
+ .setProtocol(Protocol.HTTP)
+ .setDetectPoint(DetectPoint.server);
final LocalDateTime now = LocalDateTime.now(ZoneOffset.UTC);
final LocalDateTime startTime = now.minusDays(metricsTTL + 1);
final LocalDateTime endTime = startTime.plusMinutes(1);
- sendMetrics(builder
- .setStartTime(startTime.toEpochSecond(ZoneOffset.UTC) * 1000)
- .setEndTime(endTime.toEpochSecond(ZoneOffset.UTC) * 1000).build());
+ sendMetrics(ServiceMeshMetrics
+ .newBuilder()
+ .setHttpMetrics(
+ HTTPServiceMeshMetrics
+ .newBuilder()
+ .addMetrics(
+ builder
+ .setStartTime(startTime.toEpochSecond(ZoneOffset.UTC) * 1000)
+ .setEndTime(endTime.toEpochSecond(ZoneOffset.UTC) * 1000)))
+ .build());
return "Metrics send success!";
}
- void sendMetrics(final ServiceMeshMetric metrics) throws InterruptedException {
+ void sendMetrics(final ServiceMeshMetrics metrics) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
- StreamObserver<ServiceMeshMetric> collect = grpcStub.collect(new StreamObserver<MeshProbeDownstream>() {
+ StreamObserver<ServiceMeshMetrics> collect = grpcStub.collect(new StreamObserver<MeshProbeDownstream>() {
@Override
public void onNext(final MeshProbeDownstream meshProbeDownstream) {
diff --git a/test/e2e-v2/java-test-service/e2e-protocol/src/main/proto b/test/e2e-v2/java-test-service/e2e-protocol/src/main/proto
index fbbe955545..c2c9bbbff4 160000
--- a/test/e2e-v2/java-test-service/e2e-protocol/src/main/proto
+++ b/test/e2e-v2/java-test-service/e2e-protocol/src/main/proto
@@ -1 +1 @@
-Subproject commit fbbe955545fd2c942ca59cd05720a084d010bb8a
+Subproject commit c2c9bbbff43f0ad9f917a0a55285538a6b45739e