feature: add support for Pulsar

.github/workflows/plugin-tests.yaml

@@ -98,6 +98,7 @@ jobs:
           - echov4
           - rocketmq
           - amqp
+          - pulsar
     steps:
       - uses: actions/checkout@v2
         with:

CHANGES.md

@@ -2,6 +2,11 @@ Changes by Version
 ==================
 Release Notes.
 
+0.5.0
+------------------
+#### Plugins
+* Support [Pulsar](https://github.com/apache/pulsar-client-go) MQ.
+
 0.4.0
 ------------------
 #### Features
@@ -92,7 +97,6 @@ Release Notes.
 
 0.1.0
 ------------------
-
 #### Features
 * Initialize the agent core and user import library.
 * Support gRPC reporter for management, tracing protocols.

docs/en/agent/support-plugins.md

@@ -30,6 +30,7 @@ metrics based on the tracing data.
 * MQ Client
   * `rocketMQ`: [rocketmq-client-go](https://github.com/apache/rocketmq-client-go) tested v2.1.2.
   * `amqp`: [AMQP](https://github.com/rabbitmq/amqp091-go) tested v1.9.0.
+  * `pulsar`: [pulsar-client-go](https://github.com/apache/pulsar-client-go) tested v0.12.0.
 
 # Metrics Plugins
 The meter plugin provides the advanced metrics collections.

go.work

@@ -25,6 +25,7 @@ use (
 	./plugins/echov4
 	./plugins/rocketmq
 	./plugins/amqp
+	./plugins/pulsar
 
 	./test/benchmark-codebase/consumer
 	./test/benchmark-codebase/provider
@@ -56,6 +57,7 @@ use (
 	./test/plugins/scenarios/echov4
 	./test/plugins/scenarios/rocketmq
 	./test/plugins/scenarios/amqp
+	./test/plugins/scenarios/pulsar
 
 	./tools/go-agent
 

plugins/pulsar/go.mod

@@ -0,0 +1,5 @@
+module github.com/apache/skywalking-go/plugins/pulsar
+
+go 1.18
+
+require github.com/apache/pulsar-client-go v0.12.0 // indirect

plugins/pulsar/go.sum

@@ -0,0 +1,2 @@
+github.com/apache/pulsar-client-go v0.12.0 h1:rrMlwpr6IgLRPXLRRh2vSlcw5tGV2PUSjZwmqgh2B2I=
+github.com/apache/pulsar-client-go v0.12.0/go.mod h1:dkutuH4oS2pXiGm+Ti7fQZ4MRjrMPZ8IJeEGAWMeckk=

plugins/pulsar/instrument.go

@@ -0,0 +1,90 @@
+// 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 pulsar
+
+import (
+	"embed"
+
+	"github.com/apache/skywalking-go/plugins/core/instrument"
+)
+
+//go:embed *
+var fs embed.FS
+
+//skywalking:nocopy
+type Instrument struct {
+}
+
+func NewInstrument() *Instrument {
+	return &Instrument{}
+}
+
+func (i *Instrument) Name() string {
+	return "pulsar"
+}
+
+func (i *Instrument) BasePackage() string {
+	return "github.com/apache/pulsar-client-go"
+}
+
+func (i *Instrument) VersionChecker(version string) bool {
+	return true
+}
+
+func (i *Instrument) Points() []*instrument.Point {
+	return []*instrument.Point{
+		{
+			PackagePath: "pulsar",
+			At: instrument.NewMethodEnhance("*partitionProducer", "Send",
+				instrument.WithArgsCount(2),
+				instrument.WithArgType(0, "context.Context"),
+				instrument.WithArgType(1, "*ProducerMessage"),
+				instrument.WithResultCount(2),
+				instrument.WithResultType(0, "MessageID"),
+				instrument.WithResultType(1, "error"),
+			),
+			Interceptor: "SendInterceptor",
+		},
+		{
+			PackagePath: "pulsar",
+			At: instrument.NewMethodEnhance("*partitionProducer", "SendAsync",
+				instrument.WithArgsCount(3),
+				instrument.WithArgType(0, "context.Context"),
+				instrument.WithArgType(1, "*ProducerMessage"),
+				instrument.WithArgType(2, "func(MessageID, *ProducerMessage, error)"),
+				instrument.WithResultCount(0),
+			),
+			Interceptor: "SendAsyncInterceptor",
+		},
+		{
+			PackagePath: "pulsar",
+			At: instrument.NewMethodEnhance("*consumer", "Receive",
+				instrument.WithArgsCount(1),
+				instrument.WithArgType(0, "context.Context"),
+				instrument.WithResultCount(2),
+				instrument.WithResultType(0, "Message"),
+				instrument.WithResultType(1, "error"),
+			),
+			Interceptor: "ReceiveInterceptor",
+		},
+	}
+}
+
+func (i *Instrument) FS() *embed.FS {
+	return &fs
+}

plugins/pulsar/pulsar/receive_consumer.go

@@ -0,0 +1,70 @@
+// 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 pulsar
+
+import (
+	"github.com/apache/pulsar-client-go/pulsar"
+
+	"github.com/apache/skywalking-go/plugins/core/operator"
+	"github.com/apache/skywalking-go/plugins/core/tracing"
+)
+
+const (
+	pulsarReceivePrefix      = "Pulsar/"
+	pulsarReceiveSuffix      = "/Consumer"
+	pulsarReceiveComponentID = 74
+)
+
+type ReceiveInterceptor struct {
+}
+
+func (r *ReceiveInterceptor) BeforeInvoke(invocation operator.Invocation) error {
+	return nil
+}
+
+func (r *ReceiveInterceptor) AfterInvoke(invocation operator.Invocation, result ...interface{}) error {
+	nativeConsumer := invocation.CallerInstance().(*nativeconsumer)
+	topic := nativeConsumer.options.Topic
+	lookup, err := nativeConsumer.client.lookupService.Lookup(topic)
+	if err != nil {
+		return err
+	}
+	message := result[0].(pulsar.Message)
+	peer := lookup.PhysicalAddr.String()
+	operationName := pulsarReceivePrefix + topic + pulsarReceiveSuffix
+
+	span, err := tracing.CreateEntrySpan(operationName, func(headerKey string) (string, error) {
+		return message.Properties()[headerKey], nil
+	},
+		tracing.WithLayer(tracing.SpanLayerMQ),
+		tracing.WithComponent(pulsarReceiveComponentID),
+		tracing.WithTag(tracing.TagMQBroker, lookup.PhysicalAddr.String()),
+		tracing.WithTag(tracing.TagMQTopic, nativeConsumer.topic),
+	)
+	if err != nil {
+		return err
+	}
+
+	if err, ok := result[1].(pulsar.Error); ok {
+		span.Tag(tracing.TagMQStatus, err.Error())
+		span.Error(err.Error())
+	}
+	span.SetPeer(peer)
+	span.End()
+	return nil
+}

plugins/pulsar/pulsar/send_async_producer.go

@@ -0,0 +1,105 @@
+// 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 pulsar
+
+import (
+	"github.com/apache/skywalking-go/plugins/core/operator"
+	"github.com/apache/skywalking-go/plugins/core/tracing"
+)
+
+const (
+	pulsarAsyncPrefix      = "Pulsar/"
+	pulsarAsyncSuffix      = "/AsyncProducer"
+	pulsarCallbackSuffix   = "/Producer/Callback"
+	pulsarAsyncComponentID = 73
+)
+
+type SendAsyncInterceptor struct {
+}
+
+func (s *SendAsyncInterceptor) BeforeInvoke(invocation operator.Invocation) error {
+	nativeProducer := invocation.CallerInstance().(*nativepartitionProducer)
+	topic := nativeProducer.options.Topic
+	msg := invocation.Args()[1].(*ProducerMessage)
+	lookup, err := nativeProducer.client.lookupService.Lookup(topic)
+	if err != nil {
+		return err
+	}
+	peer := lookup.PhysicalAddr.String()
+	operationName := pulsarAsyncPrefix + topic + pulsarAsyncSuffix
+
+	span, err := tracing.CreateExitSpan(operationName, peer, func(headerKey, headerValue string) error {
+		if msg.Properties == nil {
+			msg.Properties = map[string]string{
+				headerKey: headerValue,
+			}
+			return nil
+		}
+		msg.Properties[headerKey] = headerValue
+		return nil
+	},
+		tracing.WithLayer(tracing.SpanLayerMQ),
+		tracing.WithComponent(pulsarAsyncComponentID),
+		tracing.WithTag(tracing.TagMQBroker, lookup.PhysicalAddr.String()),
+		tracing.WithTag(tracing.TagMQTopic, nativeProducer.topic),
+	)
+	if err != nil {
+		return err
+	}
+
+	continueSnapShot := tracing.CaptureContext()
+	zuper := invocation.Args()[2].(func(id MessageID, message *ProducerMessage, err error))
+
+	callbackFunc := func(id MessageID, message *ProducerMessage, err error) {
+		defer tracing.CleanContext()
+		tracing.ContinueContext(continueSnapShot)
+		operationName = pulsarAsyncPrefix + topic + pulsarCallbackSuffix
+
+		localSpan, localErr := tracing.CreateLocalSpan(operationName,
+			tracing.WithComponent(pulsarAsyncComponentID),
+			tracing.WithLayer(tracing.SpanLayerMQ),
+			tracing.WithTag(tracing.TagMQTopic, nativeProducer.topic),
+		)
+		if localErr != nil {
+			zuper(id, message, err)
+			return
+		}
+		if err != nil {
+			span.Error(err.Error())
+		}
+		localSpan.Tag(tracing.TagMQBroker, lookup.PhysicalAddr.String())
+		localSpan.Tag(tracing.TagMQMsgID, id.String())
+
+		zuper(id, message, err)
+		localSpan.SetPeer(peer)
+		localSpan.End()
+	}
+
+	span.SetPeer(peer)
+	invocation.ChangeArg(2, callbackFunc)
+	invocation.SetContext(span)
+	return nil
+}
+
+func (s *SendAsyncInterceptor) AfterInvoke(invocation operator.Invocation, result ...interface{}) error {
+	if invocation.GetContext() == nil {
+		return nil
+	}
+	invocation.GetContext().(tracing.Span).End()
+	return nil
+}