From 38a86a5408c30b805bce2b5c79b7e8000e917a68 Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Wed, 14 Jun 2023 04:03:22 +0000 Subject: [PATCH] Support transmit the Continuous Profiling protocol (#137) --- CHANGES.md | 1 + go.mod | 4 +- go.sum | 20 ++++-- .../grpc/nativeebpfprofiling/forwarder.go | 32 +++++++--- .../continuous_profiling_service.go | 61 +++++++++++++++++++ .../grpc/nativeebpfprofiling/receiver.go | 13 ++-- 6 files changed, 110 insertions(+), 21 deletions(-) create mode 100644 plugins/receiver/grpc/nativeebpfprofiling/continuous_profiling_service.go diff --git a/CHANGES.md b/CHANGES.md index f2e2bea8..427bcb26 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -10,6 +10,7 @@ Release Notes. * Update the base docker image. * Add timeout configuration for gRPC client. * Reduce log print when the enqueue data to the pipeline error. +* Support transmit the Continuous Profiling protocol. #### Bug Fixes * Fix [CVE-2022-41721](https://avd.aquasec.com/nvd/cve-2022-41721). diff --git a/go.mod b/go.mod index ee4b5291..dddee91b 100644 --- a/go.mod +++ b/go.mod @@ -19,10 +19,10 @@ require ( golang.org/x/mod v0.9.0 golang.org/x/text v0.8.0 google.golang.org/grpc v1.53.0 - google.golang.org/protobuf v1.29.1 + google.golang.org/protobuf v1.30.0 gopkg.in/yaml.v3 v3.0.1 k8s.io/apimachinery v0.26.2 - skywalking.apache.org/repo/goapi v0.0.0-20221115073815-1d2a7c96c0b1 + skywalking.apache.org/repo/goapi v0.0.0-20230531132709-826aefddf3cb ) require ( diff --git a/go.sum b/go.sum index b805f2ac..2452a59a 100644 --- a/go.sum +++ b/go.sum @@ -423,6 +423,7 @@ github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9de github.com/yuin/goldmark v1.1.32/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/yusufpapurcu/wmi v1.2.2 h1:KBNDSne4vP5mbSWnJbO+51IMOXJB67QiYCSBrubbPRg= github.com/yusufpapurcu/wmi v1.2.2/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU= @@ -448,6 +449,7 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.7.0 h1:AvwMYaRytfdeVt3u6mLaxYtErKYjxA2OXjJ1HHq6t3A= golang.org/x/crypto v0.7.0/go.mod h1:pYwdfH91IfpZVANVyUOhSIPZaFoJGxTFbZhFTx+dXZU= @@ -486,6 +488,7 @@ golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.9.0 h1:KENHtAZL2y3NLMYZeHY9DW8HW8V+kQyJsY/V9JlKvCs= golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -527,6 +530,7 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ= golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= @@ -551,6 +555,7 @@ golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -597,12 +602,14 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= golang.org/x/term v0.6.0 h1:clScbb1cHjoCkyRbWwBEUZ5H/tIFu5TAXIqaZD0Gcjw= golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U= golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= @@ -614,6 +621,7 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68= golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= @@ -674,6 +682,7 @@ golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4f golang.org/x/tools v0.0.0-20210108195828-e2f9c7f1fc8e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.0/go.mod h1:xkSsbof2nBLbhDlRMhhhyNLN/zl3eTqcnHD5viDpcZ0= golang.org/x/tools v0.1.3/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= @@ -781,9 +790,8 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= -google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= -google.golang.org/protobuf v1.29.1 h1:7QBf+IK2gx70Ap/hDsOmam3GE0v9HicjfEdAxE62UoM= -google.golang.org/protobuf v1.29.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng= +google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= @@ -849,5 +857,5 @@ sigs.k8s.io/structured-merge-diff/v4 v4.2.3 h1:PRbqxJClWWYMNV1dhaG4NsibJbArud9kF sigs.k8s.io/structured-merge-diff/v4 v4.2.3/go.mod h1:qjx8mGObPmV2aSZepjQjbmb2ihdVs8cGKBraizNC69E= sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo= sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8= -skywalking.apache.org/repo/goapi v0.0.0-20221115073815-1d2a7c96c0b1 h1:AeRofErSIajLM+TNbMi0XXV9DZ4WAz5I5lM5gNTDB74= -skywalking.apache.org/repo/goapi v0.0.0-20221115073815-1d2a7c96c0b1/go.mod h1:lxmYWY1uAP5SLVKNymAyDzn7KG6dhPWN+pYHmyt+0vo= +skywalking.apache.org/repo/goapi v0.0.0-20230531132709-826aefddf3cb h1:rsExxPGSCqiTScUfph4R3uNfQbVvaqMXYz84Hx3W6NI= +skywalking.apache.org/repo/goapi v0.0.0-20230531132709-826aefddf3cb/go.mod h1:bW4dg0GUN4rMCMS8DLlaY3ZiKUAJ1fQYKoZ91Bl0kTk= diff --git a/plugins/forwarder/grpc/nativeebpfprofiling/forwarder.go b/plugins/forwarder/grpc/nativeebpfprofiling/forwarder.go index b9a9852c..c458e93f 100644 --- a/plugins/forwarder/grpc/nativeebpfprofiling/forwarder.go +++ b/plugins/forwarder/grpc/nativeebpfprofiling/forwarder.go @@ -41,7 +41,8 @@ const ( type Forwarder struct { config.CommonFields - profilingClient profiling.EBPFProfilingServiceClient + profilingClient profiling.EBPFProfilingServiceClient + continuousClient profiling.ContinuousProfilingServiceClient } func (f *Forwarder) Name() string { @@ -67,6 +68,7 @@ func (f *Forwarder) Prepare(connection interface{}) error { f.Name(), reflect.TypeOf(connection).String()) } f.profilingClient = profiling.NewEBPFProfilingServiceClient(client) + f.continuousClient = profiling.NewContinuousProfilingServiceClient(client) return nil } @@ -108,15 +110,27 @@ func (f *Forwarder) ForwardType() v1.SniffType { } func (f *Forwarder) SyncForward(e *v1.SniffData) (*v1.SniffData, error) { - query := e.GetEBPFProfilingTaskQuery() - if query == nil { - return nil, fmt.Errorf("unsupport data") - } - commands, err := f.profilingClient.QueryTasks(context.Background(), query) - if err != nil { - return nil, err + switch data := e.GetData().(type) { + case *v1.SniffData_EBPFProfilingTaskQuery: + commands, err := f.profilingClient.QueryTasks(context.Background(), data.EBPFProfilingTaskQuery) + if err != nil { + return nil, err + } + return &v1.SniffData{Data: &v1.SniffData_Commands{Commands: commands}}, nil + case *v1.SniffData_ContinuousProfilingPolicyQuery: + commands, err := f.continuousClient.QueryPolicies(context.Background(), data.ContinuousProfilingPolicyQuery) + if err != nil { + return nil, err + } + return &v1.SniffData{Data: &v1.SniffData_Commands{Commands: commands}}, nil + case *v1.SniffData_ContinuousProfilingReport: + commands, err := f.continuousClient.ReportProfilingTask(context.Background(), data.ContinuousProfilingReport) + if err != nil { + return nil, err + } + return &v1.SniffData{Data: &v1.SniffData_Commands{Commands: commands}}, nil } - return &v1.SniffData{Data: &v1.SniffData_Commands{Commands: commands}}, nil + return nil, fmt.Errorf("unsupported data type %T", e.GetData()) } func (f *Forwarder) SupportedSyncInvoke() bool { diff --git a/plugins/receiver/grpc/nativeebpfprofiling/continuous_profiling_service.go b/plugins/receiver/grpc/nativeebpfprofiling/continuous_profiling_service.go new file mode 100644 index 00000000..f7674ca0 --- /dev/null +++ b/plugins/receiver/grpc/nativeebpfprofiling/continuous_profiling_service.go @@ -0,0 +1,61 @@ +// Licensed to 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. Apache Software Foundation (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 nativeebpfprofiling + +import ( + "context" + + module "github.com/apache/skywalking-satellite/internal/satellite/module/api" + + common_v3 "skywalking.apache.org/repo/goapi/collect/common/v3" + v3 "skywalking.apache.org/repo/goapi/collect/ebpf/profiling/v3" + sniffer "skywalking.apache.org/repo/goapi/satellite/data/v1" +) + +type ContinuousProfilingReportService struct { + module.SyncInvoker + + v3.UnimplementedContinuousProfilingServiceServer +} + +func (c *ContinuousProfilingReportService) QueryPolicies(ctx context.Context, query *v3.ContinuousProfilingPolicyQuery) (*common_v3.Commands, error) { + event := &sniffer.SniffData{ + Data: &sniffer.SniffData_ContinuousProfilingPolicyQuery{ + ContinuousProfilingPolicyQuery: query, + }, + } + data, err := c.SyncInvoker.SyncInvoke(event) + if err != nil { + return nil, err + } + return data.GetCommands(), nil +} + +func (c *ContinuousProfilingReportService) ReportProfilingTask(ctx context.Context, + report *v3.ContinuousProfilingReport) (*common_v3.Commands, error) { + event := &sniffer.SniffData{ + Data: &sniffer.SniffData_ContinuousProfilingReport{ + ContinuousProfilingReport: report, + }, + } + data, err := c.SyncInvoker.SyncInvoke(event) + if err != nil { + return nil, err + } + return data.GetCommands(), nil +} diff --git a/plugins/receiver/grpc/nativeebpfprofiling/receiver.go b/plugins/receiver/grpc/nativeebpfprofiling/receiver.go index 9018dd4e..6486b35f 100644 --- a/plugins/receiver/grpc/nativeebpfprofiling/receiver.go +++ b/plugins/receiver/grpc/nativeebpfprofiling/receiver.go @@ -36,7 +36,9 @@ const ( type Receiver struct { config.CommonFields grpcreceiver.CommonGRPCReceiverFields - service *ProfilingReportService // The gRPC request handler for profiling data. + + profilingService *ProfilingReportService + continuousService *ContinuousProfilingReportService } func (r *Receiver) Name() string { @@ -58,12 +60,15 @@ func (r *Receiver) DefaultConfig() string { func (r *Receiver) RegisterHandler(server interface{}) { r.CommonGRPCReceiverFields = *grpcreceiver.InitCommonGRPCReceiverFields(server) - r.service = &ProfilingReportService{receiveChannel: r.OutputChannel} - v3.RegisterEBPFProfilingServiceServer(r.Server, r.service) + r.profilingService = &ProfilingReportService{receiveChannel: r.OutputChannel} + r.continuousService = &ContinuousProfilingReportService{} + v3.RegisterEBPFProfilingServiceServer(r.Server, r.profilingService) + v3.RegisterContinuousProfilingServiceServer(r.Server, r.continuousService) } func (r *Receiver) RegisterSyncInvoker(invoker module.SyncInvoker) { - r.service.SyncInvoker = invoker + r.profilingService.SyncInvoker = invoker + r.continuousService.SyncInvoker = invoker } func (r *Receiver) Channel() <-chan *v1.SniffData {