Cloud Foundry receiver step 2 - implementation

receiver/cloudfoundryreceiver/README.md

@@ -27,28 +27,34 @@ The receiver takes the following configuration options:
 
 | Field | Default | Description |
 | --- | --- | --- |
-| `rlp_gateway_url` | required | URL of the RLP gateway, typically `https://log-stream.<cf-system-domain>` |
-| `rlp_gateway_skip_tls_verify` | `false` | whether to skip TLS verify for the RLP Gateway endpoint |
-| `rlp_gateway_shard_id` | `opentelemetry` | metrics are load balanced among receivers that use the same shard ID, therefore this must only be set if there are multiple receivers which must both receive all the metrics instead of them being balanced between them |
-| `uaa_url` | required | URL of the UAA provider, typically `https://uaa.<cf-system-domain>` |
-| `uaa_skip_tls_verify` | `false` | whether to skip TLS verify for the UAA endpoint |
-| `uaa_username` | required | name of the UAA user (required grant types/authorities described above) |
-| `uaa_password` | required | password of the UAA user |
-| `http_timeout` | `10s` | HTTP socket timeout used for RLP Gateway |
+| `rlp_gateway.endpoint` | required | URL of the RLP gateway, typically `https://log-stream.<cf-system-domain>` |
+| `rlp_gateway.tls.insecure_skip_verify` | `false` | whether to skip TLS verify for the RLP gateway endpoint |
+| `rlp_gateway.shard_id` | `opentelemetry` | metrics are load balanced among receivers that use the same shard ID, therefore this must only be set if there are multiple receivers which must both receive all the metrics instead of them being balanced between them |
+| `uaa.endpoint` | required | URL of the UAA provider, typically `https://uaa.<cf-system-domain>` |
+| `uaa.tls.insecure_skip_verify` | `false` | whether to skip TLS verify for the UAA endpoint |
+| `uaa.username` | required | name of the UAA user (required grant types/authorities described above) |
+| `uaa.password` | required | password of the UAA user |
+
+The `rlp_gateway` configuration section also inherits configuration options from the global from:
+
+- [HTTP Client Configuration](https://github.com/open-telemetry/opentelemetry-collector/tree/main/config/confighttp#client-configuration)
 
 Example:
 
 ```yaml
 receivers:
   cloudfoundry:
-    rlp_gateway_url: "https://log-stream.sys.example.internal"
-    rlp_gateway_skip_tls_verify: false
-    rlp_gateway_shard_id: "opentelemetry"
-    uaa_url: "https://uaa.sys.example.internal"
-    uaa_skip_tls_verify: false
-    uaa_username: "otelclient"
-    uaa_password: "changeit"
-    http_timeout: "20s"
+    rlp_gateway:
+      endpoint: "https://log-stream.sys.example.internal"
+      tls:
+        insecure_skip_verify: false
+      shard_id: "opentelemetry"
+    uaa:
+      endpoint: "https://uaa.sys.example.internal"
+      tls:
+        insecure_skip_verify: false
+      username: "otelclient"
+      password: "changeit"
 ```
 
 The full list of settings exposed for this receiver are documented [here](./config.go)
@@ -63,16 +69,20 @@ origin name is prepended to the metric name with `.` separator. All metrics eith
 ### Attributes
 
 All the metrics have the following attributes:
+
 * `origin` - origin name as documented by Cloud Foundry
 * `source` - for applications, the GUID of the application, otherwise equal to `origin`
 
-For CF/TAS deployed in BOSH, the following attributes are also present, using their canonical BOSH meanings:
+For Cloud Foundry/Tanzu Application Service deployed in BOSH, the following attributes are also present, using their
+canonical BOSH meanings:
+
 * `deployment` - BOSH deployment name
 * `index` - BOSH instance ID (GUID)
 * `ip` - BOSH instance IP
 * `job` - BOSH job name
 
 For metrics originating with `rep` origin name (specific to applications), the following metrics are present:
+
 * `instance_id` - numerical index of the application instance. However, also present for `bbs` origin, where it matches
   the value of `index`
 * `process_id` - process ID (GUID). For a process of type "web" which is the main process of an application, this is

receiver/cloudfoundryreceiver/config.go

@@ -15,40 +15,61 @@
 package cloudfoundryreceiver
 
 import (
+	"errors"
 	"fmt"
 	"net/url"
-	"time"
 
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/confighttp"
 )
 
+type RLPGatewayConfig struct {
+	confighttp.HTTPClientSettings `mapstructure:",squash"`
+	ShardID                       string `mapstructure:"shard_id"`
+}
+
+// LimitedTLSClientSetting is a subset of TLSClientSetting, see LimitedHTTPClientSettings for more details
+type LimitedTLSClientSetting struct {
+	InsecureSkipVerify bool `mapstructure:"insecure_skip_verify"`
+}
+
+// LimitedHTTPClientSettings is a subset of HTTPClientSettings, implemented as a separate type due to the library this
+// configuration is used with not taking a preconfigured http.Client as input, but only taking these specific options
+type LimitedHTTPClientSettings struct {
+	Endpoint   string                  `mapstructure:"endpoint"`
+	TLSSetting LimitedTLSClientSetting `mapstructure:"tls,omitempty"`
+}
+
+type UAAConfig struct {
+	LimitedHTTPClientSettings `mapstructure:",squash"`
+	Username                  string `mapstructure:"username"`
+	Password                  string `mapstructure:"password"`
+}
+
 // Config defines configuration for Collectd receiver.
 type Config struct {
 	config.ReceiverSettings `mapstructure:",squash"`
-
-	RLPGatewayURL           string        `mapstructure:"rlp_gateway_url"`
-	RLPGatewaySkipTLSVerify bool          `mapstructure:"rlp_gateway_skip_tls_verify"`
-	RLPGatewayShardID       string        `mapstructure:"rlp_gateway_shard_id"`
-	UAAUrl                  string        `mapstructure:"uaa_url"`
-	UAASkipTLSVerify        bool          `mapstructure:"uaa_skip_tls_verify"`
-	UAAUsername             string        `mapstructure:"uaa_username"`
-	UAAPassword             string        `mapstructure:"uaa_password"`
-	HTTPTimeout             time.Duration `mapstructure:"http_timeout"`
+	RLPGateway              RLPGatewayConfig `mapstructure:"rlp_gateway"`
+	UAA                     UAAConfig        `mapstructure:"uaa"`
 }
 
 func (c *Config) Validate() error {
-	err := validateURLOption("rlp_gateway_url", c.RLPGatewayURL)
+	err := validateURLOption("rlp_gateway.endpoint", c.RLPGateway.Endpoint)
 	if err != nil {
 		return err
 	}
 
-	err = validateURLOption("uaa_url", c.UAAUrl)
+	err = validateURLOption("uaa.endpoint", c.UAA.Endpoint)
 	if err != nil {
 		return err
 	}
 
-	if c.UAAUsername == "" {
-		return fmt.Errorf("username not specified")
+	if c.UAA.Username == "" {
+		return errors.New("UAA username not specified")
+	}
+
+	if c.UAA.Password == "" {
+		return errors.New("UAA password not specified")
 	}
 
 	return nil

receiver/cloudfoundryreceiver/config_test.go

@@ -16,17 +16,19 @@ package cloudfoundryreceiver
 
 import (
 	"path"
+	"reflect"
 	"testing"
 	"time"
 
 	"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/confighttp"
 	"go.opentelemetry.io/collector/config/configtest"
+	"go.opentelemetry.io/collector/config/configtls"
 )
 
-// todo implement - implement negative tests in stage 2 PR
 func TestLoadConfig(t *testing.T) {
 	factories, err := componenttest.NopFactories()
 	require.NoError(t, err)
@@ -41,19 +43,107 @@ func TestLoadConfig(t *testing.T) {
 	require.Len(t, cfg.Receivers, 2)
 
 	r0 := cfg.Receivers[config.NewComponentID(typeStr)]
-	assert.Equal(t, factory.CreateDefaultConfig(), r0)
+	defaultConfig := factory.CreateDefaultConfig().(*Config)
+	defaultConfig.UAA.Password = "test"
+	assert.Equal(t, defaultConfig, r0)
 
 	r1 := cfg.Receivers[config.NewComponentIDWithName(typeStr, "one")].(*Config)
 	assert.Equal(t,
 		&Config{
-			ReceiverSettings:        config.NewReceiverSettings(config.NewComponentIDWithName(typeStr, "one")),
-			RLPGatewayURL:           "https://log-stream.sys.example.internal",
-			RLPGatewaySkipTLSVerify: true,
-			RLPGatewayShardID:       "otel-test",
-			UAAUrl:                  "https://uaa.sys.example.internal",
-			UAASkipTLSVerify:        true,
-			UAAUsername:             "admin",
-			UAAPassword:             "test",
-			HTTPTimeout:             time.Second * 20,
+			ReceiverSettings: config.NewReceiverSettings(config.NewComponentIDWithName(typeStr, "one")),
+			RLPGateway: RLPGatewayConfig{
+				HTTPClientSettings: confighttp.HTTPClientSettings{
+					Endpoint: "https://log-stream.sys.example.internal",
+					TLSSetting: configtls.TLSClientSetting{
+						InsecureSkipVerify: true,
+					},
+					Timeout: time.Second * 20,
+				},
+				ShardID: "otel-test",
+			},
+			UAA: UAAConfig{
+				LimitedHTTPClientSettings: LimitedHTTPClientSettings{
+					Endpoint: "https://uaa.sys.example.internal",
+					TLSSetting: LimitedTLSClientSetting{
+						InsecureSkipVerify: true,
+					},
+				},
+				Username: "admin",
+				Password: "test",
+			},
 		}, r1)
 }
+
+func TestLoadInvalidConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	require.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Receivers[typeStr] = factory
+	_, err = configtest.LoadConfigAndValidate(path.Join(".", "testdata", "config-invalid.yaml"), factories)
+
+	require.Error(t, err)
+}
+
+func TestInvalidConfigValidation(t *testing.T) {
+	configuration := loadSuccessfulConfig(t)
+	configuration.RLPGateway.Endpoint = "https://[invalid"
+	require.Error(t, configuration.Validate())
+
+	configuration = loadSuccessfulConfig(t)
+	configuration.UAA.Username = ""
+	require.Error(t, configuration.Validate())
+
+	configuration = loadSuccessfulConfig(t)
+	configuration.UAA.Password = ""
+	require.Error(t, configuration.Validate())
+
+	configuration = loadSuccessfulConfig(t)
+	configuration.UAA.Endpoint = "https://[invalid"
+	require.Error(t, configuration.Validate())
+}
+
+func TestHTTPConfigurationStructConsistency(t *testing.T) {
+	// LimitedHTTPClientSettings must have the same structure as HTTPClientSettings, but without the fields that the UAA
+	// library does not support.
+	checkTypeFieldMatch(t, "Endpoint", reflect.TypeOf(LimitedHTTPClientSettings{}), reflect.TypeOf(confighttp.HTTPClientSettings{}))
+	checkTypeFieldMatch(t, "TLSSetting", reflect.TypeOf(LimitedHTTPClientSettings{}), reflect.TypeOf(confighttp.HTTPClientSettings{}))
+	checkTypeFieldMatch(t, "InsecureSkipVerify", reflect.TypeOf(LimitedTLSClientSetting{}), reflect.TypeOf(configtls.TLSClientSetting{}))
+}
+
+func loadSuccessfulConfig(t *testing.T) *Config {
+	configuration := &Config{
+		RLPGateway: RLPGatewayConfig{
+			HTTPClientSettings: confighttp.HTTPClientSettings{
+				Endpoint: "https://log-stream.sys.example.internal",
+				Timeout:  time.Second * 20,
+				TLSSetting: configtls.TLSClientSetting{
+					InsecureSkipVerify: true,
+				},
+			},
+			ShardID: "otel-test",
+		},
+		UAA: UAAConfig{
+			LimitedHTTPClientSettings: LimitedHTTPClientSettings{
+				Endpoint: "https://uaa.sys.example.internal",
+				TLSSetting: LimitedTLSClientSetting{
+					InsecureSkipVerify: true,
+				},
+			},
+			Username: "admin",
+			Password: "test",
+		},
+	}
+
+	require.NoError(t, configuration.Validate())
+	return configuration
+}
+
+func checkTypeFieldMatch(t *testing.T, fieldName string, localType reflect.Type, standardType reflect.Type) {
+	localField, localFieldPresent := localType.FieldByName(fieldName)
+	standardField, standardFieldPresent := standardType.FieldByName(fieldName)
+
+	require.True(t, localFieldPresent, "field %s present in local type", fieldName)
+	require.True(t, standardFieldPresent, "field %s present in standard type", fieldName)
+	require.Equal(t, localField.Tag, standardField.Tag, "field %s tag match", fieldName)
+}

receiver/cloudfoundryreceiver/converter.go

@@ -0,0 +1,68 @@
+// Copyright 2019, 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 cloudfoundryreceiver
+
+import (
+	"time"
+
+	"code.cloudfoundry.org/go-loggregator/rpc/loggregator_v2"
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+const (
+	attributeNamePrefix = "org.cloudfoundry."
+)
+
+func convertEnvelopeToMetrics(envelope *loggregator_v2.Envelope, metricSlice pdata.MetricSlice, startTime time.Time) {
+	namePrefix := envelope.Tags["origin"] + "."
+
+	switch message := envelope.Message.(type) {
+	case *loggregator_v2.Envelope_Log:
+	case *loggregator_v2.Envelope_Counter:
+		metric := metricSlice.AppendEmpty()
+		metric.SetDataType(pdata.MetricDataTypeSum)
+		metric.SetName(namePrefix + message.Counter.GetName())
+		dataPoint := metric.Sum().DataPoints().AppendEmpty()
+		dataPoint.SetDoubleVal(float64(message.Counter.GetTotal()))
+		dataPoint.SetTimestamp(pdata.Timestamp(envelope.GetTimestamp()))
+		dataPoint.SetStartTimestamp(pdata.NewTimestampFromTime(startTime))
+		copyEnvelopeAttributes(dataPoint.Attributes(), envelope)
+	case *loggregator_v2.Envelope_Gauge:
+		for name, value := range message.Gauge.GetMetrics() {
+			metric := metricSlice.AppendEmpty()
+			metric.SetDataType(pdata.MetricDataTypeGauge)
+			metric.SetName(namePrefix + name)
+			dataPoint := metric.Gauge().DataPoints().AppendEmpty()
+			dataPoint.SetDoubleVal(value.Value)
+			dataPoint.SetTimestamp(pdata.Timestamp(envelope.GetTimestamp()))
+			dataPoint.SetStartTimestamp(pdata.NewTimestampFromTime(startTime))
+			copyEnvelopeAttributes(dataPoint.Attributes(), envelope)
+		}
+	}
+}
+
+func copyEnvelopeAttributes(attributes pdata.AttributeMap, envelope *loggregator_v2.Envelope) {
+	for key, value := range envelope.Tags {
+		attributes.InsertString(attributeNamePrefix+key, value)
+	}
+
+	if envelope.SourceId != "" {
+		attributes.InsertString(attributeNamePrefix+"source_id", envelope.SourceId)
+	}
+
+	if envelope.InstanceId != "" {
+		attributes.InsertString(attributeNamePrefix+"instance_id", envelope.InstanceId)
+	}
+}