feat: support skywalking integration

deepflowio · Oct 14, 2024 · 01caeb5 · 01caeb5
1 parent 6469152
commit 01caeb5
Show file tree

Hide file tree

Showing 18 changed files with 250 additions and 10 deletions.
diff --git a/agent/Cargo.lock b/agent/Cargo.lock
diff --git a/agent/Cargo.toml b/agent/Cargo.toml
@@ -97,6 +97,7 @@ tonic = "0.8.1"
 wasmtime = "12.0.1"
 wasmtime-wasi = "12.0.1"
 zstd = "0.13.2"
+integration_skywalking = { path = "plugins/integration_skywalking" }
 
 
 [target.'cfg(any(target_os = "linux", target_os = "android"))'.dependencies]

diff --git a/agent/crates/public/src/sender.rs b/agent/crates/public/src/sender.rs
@@ -55,6 +55,7 @@ pub enum SendMessageType {
     // K8sEvent = 16,
     ApplicationLog = 17,
     SyslogDetail = 18,
+    SkyWalking = 19,
 }
 
 impl fmt::Display for SendMessageType {
@@ -78,6 +79,7 @@ impl fmt::Display for SendMessageType {
             Self::AlarmEvent => write!(f, "alarm_event"),
             Self::ApplicationLog => write!(f, "application_log"),
             Self::SyslogDetail => write!(f, "syslog_detail"),
+            Self::SkyWalking => write!(f, "skywalking"),
         }
     }
 }
diff --git a/agent/plugins/integration_skywalking/Cargo.toml b/agent/plugins/integration_skywalking/Cargo.toml
@@ -0,0 +1,9 @@
+[package]
+name = "integration_skywalking"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+hyper = { version = "0.14", features = ["full"] }
+public = { path = "../../crates/public" }
+prost = "0.11"
diff --git a/agent/plugins/integration_skywalking/src/lib.rs b/agent/plugins/integration_skywalking/src/lib.rs
@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2024 Yunshan Networks
+ *
+ * Licensed 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.
+ */
+
+use hyper::{Body, Response, StatusCode};
+use prost::EncodeError;
+use public::{
+    proto::flow_log,
+    queue::DebugSender,
+    sender::{SendMessageType, Sendable},
+};
+use std::net::SocketAddr;
+
+#[derive(Debug, PartialEq)]
+pub struct SkyWalkingExtra(pub flow_log::SkyWalkingExtra);
+
+impl Sendable for SkyWalkingExtra {
+    fn encode(self, _: &mut Vec<u8>) -> Result<usize, EncodeError> {
+        return Ok(0);
+    }
+
+    fn message_type(&self) -> SendMessageType {
+        SendMessageType::SkyWalking
+    }
+}
+
+pub async fn handle_skywalking_request(
+    _: SocketAddr,
+    _: Vec<u8>,
+    _: &str,
+    _: DebugSender<SkyWalkingExtra>,
+) -> Response<Body> {
+    Response::builder()
+        .status(StatusCode::NOT_FOUND)
+        .body(Body::empty())
+        .unwrap()
+}
diff --git a/agent/src/integration_collector.rs b/agent/src/integration_collector.rs
@@ -61,6 +61,7 @@ use crate::{
     policy::PolicyGetter,
 };
 
+use integration_skywalking::{handle_skywalking_request, SkyWalkingExtra};
 use public::{
     counter::{Counter, CounterType, CounterValue, OwnedCountable},
     enums::{CaptureNetworkType, EthernetType, L4Protocol},
@@ -595,6 +596,7 @@ async fn handler(
     telegraf_sender: DebugSender<TelegrafMetric>,
     profile_sender: DebugSender<Profile>,
     application_log_sender: DebugSender<ApplicationLog>,
+    skywalking_sender: DebugSender<SkyWalkingExtra>,
     exception_handler: ExceptionHandler,
     compressed: bool,
     profile_compressed: bool,
@@ -825,6 +827,28 @@ async fn handler(
 
             Ok(Response::builder().body(Body::empty()).unwrap())
         }
+        (
+            &Method::POST,
+            "/v3/segments"
+            | "/skywalking.v3.TraceSegmentReportService/collect"
+            | "/skywalking.v3.TraceSegmentReportService/collectInSync",
+        ) => {
+            if external_trace_integration_disabled {
+                return Ok(Response::builder().body(Body::empty()).unwrap());
+            }
+            let (part, body) = req.into_parts();
+            let whole_body = match aggregate_with_catch_exception(body, &exception_handler).await {
+                Ok(b) => b,
+                Err(e) => {
+                    return Ok(e);
+                }
+            };
+            let data = decode_metric(whole_body, &part.headers)?;
+            Ok(
+                handle_skywalking_request(peer_addr, data, part.uri.path(), skywalking_sender)
+                    .await,
+            )
+        }
         // Return the 404 Not Found for other routes.
         _ => Ok(Response::builder()
             .status(StatusCode::NOT_FOUND)
@@ -925,6 +949,7 @@ pub struct MetricServer {
     telegraf_sender: DebugSender<TelegrafMetric>,
     profile_sender: DebugSender<Profile>,
     application_log_sender: DebugSender<ApplicationLog>,
+    skywalking_sender: DebugSender<SkyWalkingExtra>,
     port: Arc<AtomicU16>,
     exception_handler: ExceptionHandler,
     server_shutdown_tx: Mutex<Option<mpsc::Sender<()>>>,
@@ -952,6 +977,7 @@ impl MetricServer {
         telegraf_sender: DebugSender<TelegrafMetric>,
         profile_sender: DebugSender<Profile>,
         application_log_sender: DebugSender<ApplicationLog>,
+        skywalking_sender: DebugSender<SkyWalkingExtra>,
         port: u16,
         exception_handler: ExceptionHandler,
         compressed: bool,
@@ -980,6 +1006,7 @@ impl MetricServer {
                 telegraf_sender,
                 profile_sender,
                 application_log_sender,
+                skywalking_sender,
                 port: Arc::new(AtomicU16::new(port)),
                 exception_handler,
                 server_shutdown_tx: Default::default(),
@@ -1029,6 +1056,7 @@ impl MetricServer {
         let telegraf_sender = self.telegraf_sender.clone();
         let profile_sender = self.profile_sender.clone();
         let application_log_sender = self.application_log_sender.clone();
+        let skywalking_sender = self.skywalking_sender.clone();
         let port = self.port.clone();
         let monitor_port = Arc::new(AtomicU16::new(port.load(Ordering::Acquire)));
         let (mon_tx, mon_rx) = oneshot::channel();
@@ -1101,6 +1129,7 @@ impl MetricServer {
                     let telegraf_sender = telegraf_sender.clone();
                     let profile_sender = profile_sender.clone();
                     let application_log_sender = application_log_sender.clone();
+                    let skywalking_sender = skywalking_sender.clone();
                     let exception_handler_inner = exception_handler.clone();
                     let counter = counter.clone();
                     let compressed = compressed.clone();
@@ -1118,6 +1147,7 @@ impl MetricServer {
                         let telegraf_sender = telegraf_sender.clone();
                         let profile_sender = profile_sender.clone();
                         let application_log_sender = application_log_sender.clone();
+                        let skywalking_sender = skywalking_sender.clone();
                         let exception_handler = exception_handler_inner.clone();
                         let peer_addr = conn.remote_addr();
                         let counter = counter.clone();
@@ -1141,6 +1171,7 @@ impl MetricServer {
                                     telegraf_sender.clone(),
                                     profile_sender.clone(),
                                     application_log_sender.clone(),
+                                    skywalking_sender.clone(),
                                     exception_handler.clone(),
                                     compressed.load(Ordering::Relaxed),
                                     profile_compressed.load(Ordering::Relaxed),

diff --git a/agent/src/trident.rs b/agent/src/trident.rs
@@ -113,6 +113,7 @@ use crate::{
     utils::environment::{IN_CONTAINER, K8S_WATCH_POLICY},
 };
 
+use integration_skywalking::SkyWalkingExtra;
 use packet_sequence_block::BoxedPacketSequenceBlock;
 use pcap_assembler::{BoxedPcapBatch, PcapAssembler};
 
@@ -1557,6 +1558,7 @@ pub struct AgentComponents {
     pub packet_sequence_uniform_sender: UniformSenderThread<BoxedPacketSequenceBlock>, // Enterprise Edition Feature: packet-sequence
     pub proc_event_uniform_sender: UniformSenderThread<BoxedProcEvents>,
     pub application_log_uniform_sender: UniformSenderThread<ApplicationLog>,
+    pub skywalking_uniform_sender: UniformSenderThread<SkyWalkingExtra>,
     pub exception_handler: ExceptionHandler,
     pub proto_log_sender: DebugSender<BoxAppProtoLogsData>,
     pub pcap_batch_sender: DebugSender<BoxedPcapBatch>,
@@ -2430,6 +2432,32 @@ impl AgentComponents {
             None,
         );
 
+        let skywalking_queue_name = "1-skywalking-to-sender";
+        let (skywalking_sender, skywalking_receiver, counter) = queue::bounded_with_debug(
+            user_config
+                .processors
+                .flow_log
+                .tunning
+                .flow_aggregator_queue_size,
+            skywalking_queue_name,
+            &queue_debugger,
+        );
+        stats_collector.register_countable(
+            &QueueStats {
+                module: skywalking_queue_name,
+                ..Default::default()
+            },
+            Countable::Owned(Box::new(counter)),
+        );
+        let skywalking_uniform_sender = UniformSenderThread::new(
+            skywalking_queue_name,
+            Arc::new(skywalking_receiver),
+            config_handler.sender(),
+            stats_collector.clone(),
+            exception_handler.clone(),
+            None,
+        );
+
         let ebpf_dispatcher_id = dispatcher_components.len();
         #[cfg(any(target_os = "linux", target_os = "android"))]
         let mut ebpf_dispatcher_component = None;
@@ -2672,6 +2700,7 @@ impl AgentComponents {
             telegraf_sender,
             profile_sender,
             application_log_sender,
+            skywalking_sender,
             candidate_config.metric_server.port,
             exception_handler.clone(),
             candidate_config.metric_server.compressed,
@@ -2756,6 +2785,7 @@ impl AgentComponents {
             profile_uniform_sender,
             proc_event_uniform_sender,
             application_log_uniform_sender,
+            skywalking_uniform_sender,
             capture_mode: candidate_config.capture_mode,
             packet_sequence_uniform_output, // Enterprise Edition Feature: packet-sequence
             packet_sequence_uniform_sender, // Enterprise Edition Feature: packet-sequence
@@ -2844,6 +2874,7 @@ impl AgentComponents {
             self.profile_uniform_sender.start();
             self.proc_event_uniform_sender.start();
             self.application_log_uniform_sender.start();
+            self.skywalking_uniform_sender.start();
             if self.config.metric_server.enabled {
                 self.metrics_server_component.start();
             }
@@ -2916,6 +2947,9 @@ impl AgentComponents {
         if let Some(h) = self.application_log_uniform_sender.notify_stop() {
             join_handles.push(h);
         }
+        if let Some(h) = self.skywalking_uniform_sender.notify_stop() {
+            join_handles.push(h);
+        }
         // Enterprise Edition Feature: packet-sequence
         if let Some(h) = self.packet_sequence_uniform_sender.notify_stop() {
             join_handles.push(h);

diff --git a/message/flow_log.proto b/message/flow_log.proto
@@ -296,3 +296,7 @@ message MqttTopic {
     int32 qos = 2; // -1 mean not exist qos
 }
 
+message SkyWalkingExtra {
+    bytes data = 1;
+    bytes peer_ip = 2;
+}
diff --git a/server/go.mod b/server/go.mod
@@ -14,6 +14,7 @@ replace (
 	github.com/deepflowio/deepflow/server/controller/http/service/configuration => ./controller/http/service/configuration
 	github.com/deepflowio/deepflow/server/controller/monitor/license => ./controller/monitor/license
 	github.com/deepflowio/deepflow/server/ingester/config/configdefaults => ./ingester/config/configdefaults
+	github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/sw_import => ./ingester/flow_log/log_data/sw_import
 	github.com/deepflowio/deepflow/server/libs/logger/blocker => ./libs/logger/blocker
 	github.com/deepflowio/deepflow/server/querier/app/distributed_tracing/service/tracemap => ./querier/app/distributed_tracing/service/tracemap
 	github.com/deepflowio/deepflow/server/querier/app/prometheus/router/packet_adapter => ./querier/app/prometheus/router/packet_adapter
@@ -105,6 +106,7 @@ require (
 	github.com/bytedance/sonic v1.11.8
 	github.com/deepflowio/deepflow/server/controller/http/appender v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/controller/http/service/agentlicense v0.0.0-00010101000000-000000000000
+	github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/sw_import v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/libs/logger/blocker v0.0.0-20240822020041-cdaf0f82ce6f
 	github.com/deepflowio/deepflow/server/querier/app/distributed_tracing/service/tracemap v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/querier/app/prometheus/router/packet_adapter v0.0.0-00010101000000-000000000000

diff --git a/server/ingester/flow_log/decoder/decoder.go b/server/ingester/flow_log/decoder/decoder.go
@@ -35,6 +35,7 @@ import (
 	"github.com/deepflowio/deepflow/server/ingester/flow_log/config"
 	"github.com/deepflowio/deepflow/server/ingester/flow_log/dbwriter"
 	"github.com/deepflowio/deepflow/server/ingester/flow_log/log_data"
+	"github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/sw_import"
 	"github.com/deepflowio/deepflow/server/ingester/flow_log/throttler"
 	"github.com/deepflowio/deepflow/server/ingester/flow_tag"
 	"github.com/deepflowio/deepflow/server/libs/codec"
@@ -153,6 +154,7 @@ func (d *Decoder) Run() {
 	decoder := &codec.SimpleDecoder{}
 	pbTaggedFlow := pb.NewTaggedFlow()
 	pbTracesData := &v1.TracesData{}
+	pbSkywalkingData := &pb.SkyWalkingExtra{}
 	for {
 		n := d.inQueue.Gets(buffer)
 		start := time.Now()
@@ -181,6 +183,8 @@ func (d *Decoder) Run() {
 				d.handleOpenTelemetry(decoder, pbTracesData, true)
 			case datatype.MESSAGE_TYPE_PACKETSEQUENCE:
 				d.handleL4Packet(decoder)
+			case datatype.MESSAGE_TYPE_SKYWALKING:
+				d.handleSkyWalking(decoder, pbSkywalkingData, false)
 			default:
 				log.Warningf("unknown msg type: %d", d.msgType)
 
@@ -279,6 +283,51 @@ func (d *Decoder) sendOpenMetetry(tracesData *v1.TracesData) {
 	}
 }
 
+func (d *Decoder) handleSkyWalking(decoder *codec.SimpleDecoder, pbSkyWalkingData *pb.SkyWalkingExtra, compressed bool) {
+	var err error
+	for !decoder.IsEnd() {
+		pbSkyWalkingData.Reset()
+		bytes := decoder.ReadBytes()
+		if len(bytes) > 0 {
+			// universal compression
+			if compressed {
+				bytes, err = decompressOpenTelemetry(bytes)
+			}
+			if err == nil {
+				err = proto.Unmarshal(bytes, pbSkyWalkingData)
+			}
+		}
+		if decoder.Failed() || err != nil {
+			if d.counter.ErrorCount == 0 {
+				log.Errorf("skywalking data decode failed, offset=%d len=%d err: %s", decoder.Offset(), len(decoder.Bytes()), err)
+			}
+			d.counter.ErrorCount++
+			continue
+		}
+		d.sendSkyWalking(pbSkyWalkingData.Data, pbSkyWalkingData.PeerIp)
+	}
+}
+func (d *Decoder) sendSkyWalking(segmentData, peerIP []byte) {
+	if d.debugEnabled {
+		log.Debugf("decoder %d vtap %d recv skywalking data length: %d", d.index, d.agentId, len(segmentData))
+	}
+	d.counter.Count++
+	ls := sw_import.SkyWalkingDataToL7FlowLogs(d.agentId, d.orgId, d.teamId, segmentData, peerIP, d.platformData, d.cfg)
+	for _, l := range ls {
+		l.AddReferenceCount()
+		if !d.throttler.SendWithThrottling(l) {
+			d.counter.DropCount++
+		} else {
+			d.fieldsBuf, d.fieldValuesBuf = d.fieldsBuf[:0], d.fieldValuesBuf[:0]
+			l.GenerateNewFlowTags(d.flowTagWriter.Cache)
+			d.flowTagWriter.WriteFieldsAndFieldValuesInCache()
+			d.appServiceTagWrite(l)
+			d.spanWrite(l)
+		}
+		l.Release()
+	}
+}
+
 func (d *Decoder) handleL4Packet(decoder *codec.SimpleDecoder) {
 	for !decoder.IsEnd() {
 		l4Packet, err := log_data.DecodePacketSequence(d.agentId, d.orgId, d.teamId, decoder)