[receiver/skywalking] Create skywalking component folder/structure

CHANGELOG.md

@@ -12,6 +12,7 @@
 - `dynatraceexporter`: Validate QueueSettings and perform config validation in Validate() instead (#8020)
 - `sapmexporter`: Add validation for `sending_queue` setting (#8023)
 - `signalfxexporter`: Add validation for `sending_queue` setting (#8026)
+- `skywalkingreceiver`: Add new skywalking receiver component folder and structure (#8107)
 
 ### 🛑 Breaking changes 🛑
 

cmd/configschema/go.mod

@@ -437,7 +437,7 @@ require (
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
-	golang.org/x/sys v0.0.0-20220207234003-57398862261d // indirect
+	golang.org/x/sys v0.0.0-20220209214540-3681064d5158 // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
@@ -446,7 +446,7 @@ require (
 	gonum.org/v1/gonum v0.9.3 // indirect
 	google.golang.org/api v0.68.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220207185906-7721543eae58 // indirect
+	google.golang.org/genproto v0.0.0-20220218161850-94dd64e39d7c // indirect
 	google.golang.org/grpc v1.44.0 // indirect
 	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/DataDog/dd-trace-go.v1 v1.36.0 // indirect

go.mod

@@ -451,7 +451,7 @@ require (
 	gonum.org/v1/gonum v0.9.3 // indirect
 	google.golang.org/api v0.68.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220207185906-7721543eae58 // indirect
+	google.golang.org/genproto v0.0.0-20220218161850-94dd64e39d7c // indirect
 	google.golang.org/grpc v1.44.0 // indirect
 	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/DataDog/dd-trace-go.v1 v1.36.0 // indirect
@@ -474,7 +474,7 @@ require (
 	sigs.k8s.io/json v0.0.0-20211020170558-c049b76a60c6 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
 	sigs.k8s.io/yaml v1.2.0 // indirect
-	skywalking.apache.org/repo/goapi v0.0.0-20211122071111-ffc517fbfe21 // indirect
+	skywalking.apache.org/repo/goapi v0.0.0-20220121092418-9c455d0dda3f // indirect
 )
 
 // Replace references to modules that are in this repository with their relateive paths
@@ -752,6 +752,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/signa
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/simpleprometheusreceiver => ./receiver/simpleprometheusreceiver
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver => ./receiver/skywalkingreceiver
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/splunkhecreceiver => ./receiver/splunkhecreceiver
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver => ./receiver/statsdreceiver

receiver/skywalkingreceiver/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

receiver/skywalkingreceiver/README.md

@@ -0,0 +1,24 @@
+# Skywalking Receiver
+
+Receives trace data in [Skywalking](https://skywalking.apache.org/) format.
+
+Supported pipeline types: traces
+
+## Getting Started
+
+Examples:
+
+```yaml
+receivers:
+  skywalking:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:11800
+      http:
+        endpoint: 0.0.0.0:12800
+
+service:
+  pipelines:
+    traces:
+      receivers: [skywalking]
+```

receiver/skywalkingreceiver/config.go

@@ -0,0 +1,77 @@
+// Copyright  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 skywalkingreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver"
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/confighttp"
+)
+
+const (
+	// The config field id to load the protocol map from
+	protocolsFieldName = "protocols"
+)
+
+// Protocols is the configuration for the supported protocols.
+type Protocols struct {
+	GRPC *configgrpc.GRPCServerSettings `mapstructure:"grpc"`
+	HTTP *confighttp.HTTPServerSettings `mapstructure:"http"`
+}
+
+// Config defines configuration for skywalking receiver.
+type Config struct {
+	config.ReceiverSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
+	Protocols               `mapstructure:"protocols"`
+}
+
+var _ config.Receiver = (*Config)(nil)
+var _ config.Unmarshallable = (*Config)(nil)
+
+// Validate checks the receiver configuration is valid
+func (cfg *Config) Validate() error {
+	if cfg.GRPC == nil && cfg.HTTP == nil {
+		return fmt.Errorf("must specify at least one protocol when using the Skywalking receiver")
+	}
+
+	return nil
+}
+
+// Unmarshal a config.Parser into the config struct.
+func (cfg *Config) Unmarshal(componentParser *config.Map) error {
+	if componentParser == nil || len(componentParser.AllKeys()) == 0 {
+		return fmt.Errorf("empty config for Jaeger receiver")
+	}
+
+	// UnmarshalExact will not set struct properties to nil even if no key is provided,
+	// so set the protocol structs to nil where the keys were omitted.
+	err := componentParser.UnmarshalExact(cfg)
+	if err != nil {
+		return err
+	}
+
+	protocols, err := componentParser.Sub(protocolsFieldName)
+	if err != nil {
+		return err
+	}
+
+	if !protocols.IsSet(protoGRPC) {
+		cfg.GRPC = nil
+	}
+
+	return nil
+}

receiver/skywalkingreceiver/config_test.go

@@ -0,0 +1,71 @@
+// Copyright  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 skywalkingreceiver
+
+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/config/configgrpc"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/config/confignet"
+	"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.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	assert.Equal(t, len(cfg.Receivers), 3)
+
+	r1 := cfg.Receivers[config.NewComponentIDWithName(typeStr, "customname")].(*Config)
+	assert.Equal(t, r1,
+		&Config{
+			ReceiverSettings: config.NewReceiverSettings(config.NewComponentIDWithName(typeStr, "customname")),
+			Protocols: Protocols{
+				HTTP: &confighttp.HTTPServerSettings{
+					Endpoint: "0.0.0.0:12801",
+				},
+			},
+		})
+
+	rDefaults := cfg.Receivers[config.NewComponentIDWithName(typeStr, "defaults")].(*Config)
+	assert.Equal(t, rDefaults,
+		&Config{
+			ReceiverSettings: config.NewReceiverSettings(config.NewComponentIDWithName(typeStr, "defaults")),
+			Protocols: Protocols{
+				GRPC: &configgrpc.GRPCServerSettings{
+					NetAddr: confignet.NetAddr{
+						Endpoint:  defaultGRPCBindEndpoint,
+						Transport: "tcp",
+					},
+				},
+				HTTP: &confighttp.HTTPServerSettings{
+					Endpoint: defaultHTTPBindEndpoint,
+				},
+			},
+		})
+}

receiver/skywalkingreceiver/factory.go

@@ -0,0 +1,114 @@
+// Copyright  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 skywalkingreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/skywalkingreceiver"
+
+// This file implements factory for skywalking receiver.
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strconv"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/config/confignet"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/receiver/receiverhelper"
+)
+
+const (
+	typeStr = "skywalking"
+
+	// Protocol values.
+	protoGRPC = "grpc"
+
+	// Default endpoints to bind to.
+	defaultGRPCBindEndpoint = "0.0.0.0:11800"
+	defaultHTTPBindEndpoint = "0.0.0.0:12800"
+)
+
+// NewFactory creates a new Jaeger receiver factory.
+func NewFactory() component.ReceiverFactory {
+	return receiverhelper.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		receiverhelper.WithTraces(createTracesReceiver))
+}
+
+// CreateDefaultConfig creates the default configuration for Skywalking receiver.
+func createDefaultConfig() config.Receiver {
+	return &Config{
+		ReceiverSettings: config.NewReceiverSettings(config.NewComponentID(typeStr)),
+		Protocols: Protocols{
+			GRPC: &configgrpc.GRPCServerSettings{
+				NetAddr: confignet.NetAddr{
+					Endpoint:  defaultGRPCBindEndpoint,
+					Transport: "tcp",
+				},
+			},
+			HTTP: &confighttp.HTTPServerSettings{
+				Endpoint: defaultHTTPBindEndpoint,
+			},
+		},
+	}
+}
+
+// createTracesReceiver creates a trace receiver based on provided config.
+func createTracesReceiver(
+	_ context.Context,
+	set component.ReceiverCreateSettings,
+	cfg config.Receiver,
+	nextConsumer consumer.Traces,
+) (component.TracesReceiver, error) {
+
+	// Convert settings in the source config to configuration struct
+	// that Skywalking receiver understands.
+	rCfg := cfg.(*Config)
+
+	var config configuration
+	// Set ports
+	if rCfg.Protocols.GRPC != nil {
+		config.CollectorGRPCServerSettings = *rCfg.Protocols.GRPC
+		config.CollectorGRPCPort, _ = extractPortFromEndpoint(rCfg.Protocols.GRPC.NetAddr.Endpoint)
+	}
+
+	if rCfg.Protocols.HTTP != nil {
+		config.CollectorHTTPSettings = *rCfg.Protocols.HTTP
+		config.CollectorHTTPPort, _ = extractPortFromEndpoint(rCfg.Protocols.HTTP.Endpoint)
+	}
+
+	// Create the receiver.
+	return newSkywalkingReceiver(rCfg.ID(), &config, nextConsumer, set), nil
+}
+
+// extract the port number from string in "address:port" format. If the
+// port number cannot be extracted returns an error.
+func extractPortFromEndpoint(endpoint string) (int, error) {
+	_, portStr, err := net.SplitHostPort(endpoint)
+	if err != nil {
+		return 0, fmt.Errorf("endpoint is not formatted correctly: %s", err.Error())
+	}
+	port, err := strconv.ParseInt(portStr, 10, 0)
+	if err != nil {
+		return 0, fmt.Errorf("endpoint port is not a number: %s", err.Error())
+	}
+	if port < 1 || port > 65535 {
+		return 0, fmt.Errorf("port number must be between 1 and 65535")
+	}
+	return int(port), nil
+}