[pulsar][receiver] add apache pulsar receiver

.github/CODEOWNERS

@@ -174,6 +174,7 @@ receiver/postgresqlreceiver/                         @open-telemetry/collector-c
 receiver/prometheusexecreceiver/                     @open-telemetry/collector-contrib-approvers @dmitryax
 receiver/prometheusreceiver/                         @open-telemetry/collector-contrib-approvers @Aneurysm9 @dashpole
 receiver/rabbitmqreceiver/                           @open-telemetry/collector-contrib-approvers @djaglowski @cpheps
+receiver/pulsarreceiver/                             @open-telemetry/collector-contrib-approvers @dmitryax @tjiuming
 receiver/receivercreator/                            @open-telemetry/collector-contrib-approvers @jrcamp
 receiver/redisreceiver/                              @open-telemetry/collector-contrib-approvers @pmcollins @dmitryax
 receiver/riakreceiver/                               @open-telemetry/collector-contrib-approvers @djaglowski @armstrmi

cmd/configschema/go.mod

@@ -425,6 +425,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/postgresqlreceiver v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusexecreceiver v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.57.2 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver v0.0.0-00010101000000-000000000000 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/receivercreator v0.57.2 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/redisreceiver v0.57.2 // indirect
@@ -895,6 +896,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prome
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver => ../../receiver/prometheusreceiver
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver => ../../receiver/pulsarreceiver
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver => ../../receiver/rabbitmqreceiver
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/receivercreator => ../../receiver/receivercreator

go.mod

@@ -126,6 +126,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/postgresqlreceiver v0.57.2
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusexecreceiver v0.57.2
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.57.2
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver v0.57.2
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/receivercreator v0.57.2
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/redisreceiver v0.57.2
@@ -899,6 +900,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prome
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver => ./receiver/prometheusreceiver
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver => ./receiver/pulsarreceiver
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver => ./receiver/rabbitmqreceiver
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/receivercreator => ./receiver/receivercreator

internal/components/components.go

@@ -146,6 +146,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/postgresqlreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusexecreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/rabbitmqreceiver"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/receivercreator"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/redisreceiver"
@@ -246,6 +247,7 @@ func Components() (component.Factories, error) {
 		postgresqlreceiver.NewFactory(),
 		prometheusexecreceiver.NewFactory(),
 		prometheusreceiver.NewFactory(),
+		pulsarreceiver.NewFactory(),
 		rabbitmqreceiver.NewFactory(),
 		receivercreator.NewFactory(),
 		redisreceiver.NewFactory(),

internal/components/receivers_test.go

@@ -246,6 +246,10 @@ func TestDefaultReceivers(t *testing.T) {
 			receiver:     "prometheus_exec",
 			skipLifecyle: true, // Requires running a subproccess that can not be easily set across platforms
 		},
+		{
+			receiver:     "pulsar",
+			skipLifecyle: true, // TODO It requires a running pulsar instance to start successfully.
+		},
 		{
 			receiver: "rabbitmq",
 		},

receiver/pulsarreceiver/Makefile

@@ -0,0 +1,6 @@
+include ../../Makefile.Common
+
+# Remove "-race" from the default set of test arguments.
+# exporter/pulsarexporter tests are failing with the -race check.
+# See https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/12734
+GOTEST_OPT = -v -timeout 300s --tags=$(GO_BUILD_TAGS)

receiver/pulsarreceiver/README.md

@@ -0,0 +1,68 @@
+# Pulsar Receiver
+
+| Status                   |                       |
+| ------------------------ |-----------------------|
+| Stability                | [alpha]               |
+| Supported pipeline types | traces, logs, metrics |
+| Distributions            | [contrib]             |
+
+Pulsar receiver receives logs, metrics, and traces from Pulsar.
+
+Supported pipeline types: logs, metrics, traces
+
+## Getting Started
+
+The following settings can be optionally configured:
+- `endpoint` (default = pulsar://localhost:6650): The url of pulsar cluster.
+- `topic` (default = otlp_spans for traces, otlp_metrics for metrics, otlp_logs for logs): The name of the pulsar topic to consume from.
+- `encoding` (default = otlp_proto): The encoding of the payload sent to pulsar. Available encodings:
+    - `otlp_proto`: the payload is deserialized to `ExportTraceServiceRequest`.
+    - `jaeger_proto`: the payload is deserialized to a single Jaeger proto `Span`.
+    - `jaeger_json`: the payload is deserialized to a single Jaeger JSON Span using `jsonpb`.
+    - `zipkin_proto`: the payload is deserialized into a list of Zipkin proto spans.
+    - `zipkin_json`: the payload is deserialized into a list of Zipkin V2 JSON spans.
+    - `zipkin_thrift`: the payload is deserialized into a list of Zipkin Thrift spans.
+- `consumer_name`: specifies the consumer name.
+- `auth`
+  - `tls`
+    - `cert_file`:
+    - `key_file`:
+  - `token`
+    - `token`
+  - `oauth2`
+    - `issuer_url`:
+    - `client_id`:
+    - `audience`: 
+  - `athenz`
+    - `provider_domain`:
+    - `tenant_domain`:
+    - `tenant_service`:
+    - `private_key`:
+    - `key_id`:
+    - `principal_header`:
+    - `zts_url`:
+- `subscription` (default = otlp_subscription): the subscription name of consumer.
+- `tls_trust_certs_file_path`: path to the CA cert. For a client this verifies the server certificate. Should
+  only be used if `insecure` is set to true.
+- `tls_allow_insecure_connection`: configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)
+
+
+Example configuration:
+```yaml
+receivers:
+  pulsar:
+    endpoint: pulsar://localhost:6650
+    topic: otlp-spans
+    subscription: otlp_spans_sub
+    consumer_name: otlp_spans_sub_1
+    encoding: otlp_proto
+    auth:
+      tls:
+        cert_file: cert.pem
+        key_file: key.pem
+    tls_allow_insecure_connection: false
+    tls_trust_certs_file_path: ca.pem
+```
+
+[alpha]:https://github.com/open-telemetry/opentelemetry-collector#alpha
+[contrib]:https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib

receiver/pulsarreceiver/config.go

@@ -0,0 +1,148 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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.
+
+package pulsarreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver"
+
+import (
+	"errors"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+	"go.opentelemetry.io/collector/config"
+)
+
+type Config struct {
+	config.ReceiverSettings `mapstructure:",squash"`
+	// Configure the service URL for the Pulsar service.
+	Endpoint string `mapstructure:"endpoint"`
+	// The topic of pulsar to consume logs,metrics,traces. (default = "otlp_traces" for traces,
+	// "otlp_metrics" for metrics, "otlp_logs" for logs)
+	Topic string `mapstructure:"topic"`
+	// The Subscription that receiver will be consuming messages from (default "otlp_subscription")
+	Subscription string `mapstructure:"subscription"`
+	// Encoding of the messages (default "otlp_proto")
+	Encoding string `mapstructure:"encoding"`
+	// Name specifies the consumer name.
+	ConsumerName string `mapstructure:"consumer_name"`
+	// Set the path to the trusted TLS certificate file
+	TLSTrustCertsFilePath string `mapstructure:"tls_trust_certs_file_path"`
+	// Configure whether the Pulsar client accept untrusted TLS certificate from broker (default: false)
+	TLSAllowInsecureConnection bool           `mapstructure:"tls_allow_insecure_connection"`
+	Authentication             Authentication `mapstructure:"auth"`
+}
+
+type Authentication struct {
+	TLS    *TLS    `mapstructure:"tls"`
+	Token  *Token  `mapstructure:"Token"`
+	Athenz *Athenz `mapstructure:"athenz"`
+	OAuth2 *OAuth2 `mapstructure:"oauth2"`
+}
+
+type TLS struct {
+	CertFile string `mapstructure:"cert_file"`
+	KeyFile  string `mapstructure:"key_file"`
+}
+
+type Token struct {
+	Token string `mapstructure:"Token"`
+}
+
+type Athenz struct {
+	ProviderDomain  string `mapstructure:"provider_domain"`
+	TenantDomain    string `mapstructure:"tenant_domain"`
+	TenantService   string `mapstructure:"tenant_service"`
+	PrivateKey      string `mapstructure:"private_key"`
+	KeyID           string `mapstructure:"key_id"`
+	PrincipalHeader string `mapstructure:"principal_header"`
+	ZtsURL          string `mapstructure:"zts_url"`
+}
+
+type OAuth2 struct {
+	IssuerURL string `mapstructure:"issuer_url"`
+	ClientID  string `mapstructure:"client_id"`
+	Audience  string `mapstructure:"audience"`
+}
+
+var _ config.Receiver = (*Config)(nil)
+
+// Validate checks the receiver configuration is valid
+func (cfg *Config) Validate() error {
+	return nil
+}
+
+func (cfg *Config) auth() pulsar.Authentication {
+	authentication := cfg.Authentication
+	if authentication.TLS != nil {
+		return pulsar.NewAuthenticationTLS(authentication.TLS.CertFile, authentication.TLS.KeyFile)
+	}
+	if authentication.Token != nil {
+		return pulsar.NewAuthenticationToken(authentication.Token.Token)
+	}
+	if authentication.OAuth2 != nil {
+		return pulsar.NewAuthenticationOAuth2(map[string]string{
+			"issuerUrl": authentication.OAuth2.IssuerURL,
+			"clientId":  authentication.OAuth2.ClientID,
+			"audience":  authentication.OAuth2.Audience,
+		})
+	}
+	if authentication.Athenz != nil {
+		return pulsar.NewAuthenticationAthenz(map[string]string{
+			"providerDomain":  authentication.Athenz.ProviderDomain,
+			"tenantDomain":    authentication.Athenz.TenantDomain,
+			"tenantService":   authentication.Athenz.TenantService,
+			"privateKey":      authentication.Athenz.PrivateKey,
+			"keyId":           authentication.Athenz.KeyID,
+			"principalHeader": authentication.Athenz.PrincipalHeader,
+			"ztsUrl":          authentication.Athenz.ZtsURL,
+		})
+	}
+
+	return nil
+}
+
+func (cfg *Config) clientOptions() pulsar.ClientOptions {
+	url := cfg.Endpoint
+	if len(url) == 0 {
+		url = defaultServiceURL
+	}
+	options := pulsar.ClientOptions{
+		URL: url,
+	}
+
+	options.TLSAllowInsecureConnection = cfg.TLSAllowInsecureConnection
+	if len(cfg.TLSTrustCertsFilePath) > 0 {
+		options.TLSTrustCertsFilePath = cfg.TLSTrustCertsFilePath
+	}
+
+	auth := cfg.auth()
+	options.Authentication = auth
+	return options
+}
+
+func (cfg *Config) consumerOptions() (pulsar.ConsumerOptions, error) {
+	options := pulsar.ConsumerOptions{
+		Type:             pulsar.Failover,
+		Topic:            cfg.Topic,
+		SubscriptionName: cfg.Subscription,
+	}
+
+	if len(cfg.ConsumerName) > 0 {
+		options.Name = cfg.ConsumerName
+	}
+
+	if options.SubscriptionName == "" || options.Topic == "" {
+		return options, errors.New("topic and subscription is required")
+	}
+
+	return options, nil
+}

receiver/pulsarreceiver/config_test.go

@@ -0,0 +1,50 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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.
+
+package pulsarreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/pulsarreceiver"
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/service/servicetest"
+)
+
+func TestLoadConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Receivers[typeStr] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "config.yml"), factories)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(cfg.Receivers))
+
+	r := cfg.Receivers[config.NewComponentID(typeStr)].(*Config)
+
+	assert.Equal(t, &Config{
+		ReceiverSettings:      config.NewReceiverSettings(config.NewComponentID(typeStr)),
+		Topic:                 "otel-pulsar",
+		Endpoint:              "pulsar://localhost:6500",
+		ConsumerName:          "otel-collector",
+		Subscription:          "otel-collector",
+		Encoding:              defaultEncoding,
+		TLSTrustCertsFilePath: "ca.pem",
+		Authentication:        Authentication{TLS: &TLS{CertFile: "cert.pem", KeyFile: "key.pem"}},
+	}, r)
+}