Add support for logs in cloudfoundryreceiver

receiver/cloudfoundryreceiver/README.md

@@ -3,12 +3,14 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [beta]: metrics   |
+| Stability     | [development]: logs   |
+|               | [beta]: metrics   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fcloudfoundry%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fcloudfoundry) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fcloudfoundry%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fcloudfoundry) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@pellared](https://www.github.com/pellared), [@crobert-1](https://www.github.com/crobert-1) |
 | Emeritus      | [@agoallikmaa](https://www.github.com/agoallikmaa) |
 
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
 [beta]: https://github.com/open-telemetry/opentelemetry-collector#beta
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 <!-- end autogenerated section -->

receiver/cloudfoundryreceiver/config.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"strings"
 
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configopaque"
@@ -47,6 +48,10 @@ func (c *Config) Validate() error {
 		return err
 	}
 
+	if strings.TrimSpace(c.RLPGateway.ShardID) == "" {
+		return errors.New("shardID cannot be empty")
+	}
+
 	err = validateURLOption("uaa.endpoint", c.UAA.Endpoint)
 	if err != nil {
 		return err

receiver/cloudfoundryreceiver/config_test.go

@@ -63,6 +63,31 @@ func TestLoadConfig(t *testing.T) {
 			id:           component.NewIDWithName(metadata.Type, "invalid"),
 			errorMessage: "failed to parse rlp_gateway.endpoint as url: parse \"https://[invalid\": missing ']' in host",
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "shardidnotdefined"),
+			expected: &Config{
+				RLPGateway: RLPGatewayConfig{
+					ClientConfig: confighttp.ClientConfig{
+						Endpoint: "https://log-stream.sys.example.internal",
+						TLSSetting: configtls.ClientConfig{
+							InsecureSkipVerify: true,
+						},
+						Timeout: time.Second * 20,
+					},
+					ShardID: "opentelemetry",
+				},
+				UAA: UAAConfig{
+					LimitedClientConfig: LimitedClientConfig{
+						Endpoint: "https://uaa.sys.example.internal",
+						TLSSetting: LimitedTLSClientSetting{
+							InsecureSkipVerify: true,
+						},
+					},
+					Username: "admin",
+					Password: "test",
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.id.String(), func(t *testing.T) {
@@ -96,6 +121,14 @@ func TestInvalidConfigValidation(t *testing.T) {
 	configuration.UAA.Password = ""
 	require.Error(t, configuration.Validate())
 
+	configuration = loadSuccessfulConfig(t)
+	configuration.RLPGateway.ShardID = ""
+	require.Error(t, configuration.Validate())
+
+	configuration = loadSuccessfulConfig(t)
+	configuration.RLPGateway.ShardID = ""
+	require.Error(t, configuration.Validate())
+
 	configuration = loadSuccessfulConfig(t)
 	configuration.UAA.Endpoint = "https://[invalid"
 	require.Error(t, configuration.Validate())

receiver/cloudfoundryreceiver/converter.go

@@ -4,11 +4,14 @@
 package cloudfoundryreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/cloudfoundryreceiver"
 
 import (
+	"strings"
 	"time"
 
 	"code.cloudfoundry.org/go-loggregator/rpc/loggregator_v2"
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/otel/trace"
 )
 
 const (
@@ -19,7 +22,6 @@ func convertEnvelopeToMetrics(envelope *loggregator_v2.Envelope, metricSlice pme
 	namePrefix := envelope.Tags["origin"] + "."
 
 	switch message := envelope.Message.(type) {
-	case *loggregator_v2.Envelope_Log:
 	case *loggregator_v2.Envelope_Counter:
 		metric := metricSlice.AppendEmpty()
 		metric.SetName(namePrefix + message.Counter.GetName())
@@ -41,6 +43,27 @@ func convertEnvelopeToMetrics(envelope *loggregator_v2.Envelope, metricSlice pme
 	}
 }
 
+func convertEnvelopeToLogs(envelope *loggregator_v2.Envelope, logSlice plog.LogRecordSlice, startTime time.Time) {
+	switch envelope.Message.(type) {
+	case *loggregator_v2.Envelope_Log:
+		log := logSlice.AppendEmpty()
+		log.SetTimestamp(pcommon.Timestamp(envelope.GetTimestamp()))
+		log.SetObservedTimestamp(pcommon.NewTimestampFromTime(startTime))
+		log.Body().SetStr(string(envelope.GetLog().GetPayload()))
+		switch envelope.GetLog().GetType() {
+		case loggregator_v2.Log_OUT:
+			log.SetSeverityText(plog.SeverityNumberInfo.String())
+			log.SetSeverityNumber(plog.SeverityNumberInfo)
+		case loggregator_v2.Log_ERR:
+			log.SetSeverityText(plog.SeverityNumberError.String())
+			log.SetSeverityNumber(plog.SeverityNumberError)
+		}
+		copyEnvelopeAttributes(log.Attributes(), envelope)
+		_ = parseLogTracingFields(log)
+	default:
+	}
+}
+
 func copyEnvelopeAttributes(attributes pcommon.Map, envelope *loggregator_v2.Envelope) {
 	for key, value := range envelope.Tags {
 		attributes.PutStr(attributeNamePrefix+key, value)
@@ -54,3 +77,38 @@ func copyEnvelopeAttributes(attributes pcommon.Map, envelope *loggregator_v2.Env
 		attributes.PutStr(attributeNamePrefix+"instance_id", envelope.InstanceId)
 	}
 }
+
+func parseLogTracingFields(log plog.LogRecord) error {
+	if value, found := log.Attributes().Get("org.cloudfoundry.source_type"); !found || value.AsString() != "RTR" {
+		return nil
+	}
+	s := log.Body().AsString()
+	quoted := false
+	a := strings.FieldsFunc(s, func(r rune) bool {
+		if r == '"' {
+			quoted = !quoted
+		}
+		return !quoted && r == ' '
+	})
+
+	traceIDStr := strings.Split(a[21], ":")[1]
+	traceIDStr = strings.Trim(traceIDStr, "\"")
+
+	spanIDStr := strings.Split(a[22], ":")[1]
+	spanIDStr = strings.Trim(spanIDStr, "\"")
+
+	traceID, err := trace.TraceIDFromHex(traceIDStr)
+	if err != nil {
+		return err
+	}
+
+	spanID, err := trace.SpanIDFromHex(spanIDStr)
+	if err != nil {
+		return err
+	}
+
+	log.SetTraceID([16]byte(traceID))
+	log.SetSpanID([8]byte(spanID))
+
+	return nil
+}

receiver/cloudfoundryreceiver/converter_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 )
 
@@ -142,6 +143,50 @@ func TestConvertGaugeEnvelope(t *testing.T) {
 	assertAttributes(t, dataPoint.Attributes(), expectedAttributes)
 }
 
+func TestConvertLogsEnvelope(t *testing.T) {
+	now := time.Now()
+	before := time.Now().Add(-time.Second)
+
+	envelope := loggregator_v2.Envelope{
+		Timestamp: before.UnixNano(),
+		SourceId:  "uaa",
+		Tags: map[string]string{
+			"origin":     "gorouter",
+			"deployment": "cf",
+			"job":        "router",
+			"index":      "bc276108-8282-48a5-bae7-c009c4392246",
+			"ip":         "10.244.0.34",
+		},
+		Message: &loggregator_v2.Envelope_Log{
+			Log: &loggregator_v2.Log{
+				Payload: []byte("log message payload"),
+				Type:    loggregator_v2.Log_OUT,
+			},
+		},
+	}
+
+	logSlice := plog.NewLogRecordSlice()
+
+	convertEnvelopeToLogs(&envelope, logSlice, now)
+
+	require.Equal(t, 1, logSlice.Len())
+
+	log := logSlice.At(0)
+	assert.Equal(t, "log message payload", log.Body().AsString())
+	assert.Equal(t, plog.SeverityNumberInfo.String(), log.SeverityText())
+	assert.Equal(t, pcommon.NewTimestampFromTime(before), log.Timestamp())
+	assert.Equal(t, pcommon.NewTimestampFromTime(now), log.ObservedTimestamp())
+
+	assertAttributes(t, log.Attributes(), map[string]string{
+		"org.cloudfoundry.source_id":  "uaa",
+		"org.cloudfoundry.origin":     "gorouter",
+		"org.cloudfoundry.deployment": "cf",
+		"org.cloudfoundry.job":        "router",
+		"org.cloudfoundry.index":      "bc276108-8282-48a5-bae7-c009c4392246",
+		"org.cloudfoundry.ip":         "10.244.0.34",
+	})
+}
+
 func assertAttributes(t *testing.T, attributes pcommon.Map, expected map[string]string) {
 	assert.Equal(t, len(expected), attributes.Len())
 

receiver/cloudfoundryreceiver/doc.go

@@ -4,7 +4,7 @@
 //go:generate mdatagen metadata.yaml
 
 // Package cloudfoundryreceiver implements a receiver that can be used by the
-// Opentelemetry collector to receive Cloud Foundry metrics via its Reverse
+// Opentelemetry collector to receive Cloud Foundry metrics and logs via its Reverse
 // Log Proxy (RLP) Gateway component. The protocol is handled by the
 // go-loggregator library, which uses HTTP to connect to the gateway and receive
 // JSON-protobuf encoded v2 Envelope messages as documented by loggregator-api.

receiver/cloudfoundryreceiver/factory.go

@@ -28,7 +28,8 @@ func NewFactory() receiver.Factory {
 	return receiver.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability))
+		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
+		receiver.WithLogs(createLogsReceiver, metadata.LogsStability))
 }
 
 func createDefaultConfig() component.Config {
@@ -61,5 +62,15 @@ func createMetricsReceiver(
 	nextConsumer consumer.Metrics,
 ) (receiver.Metrics, error) {
 	c := cfg.(*Config)
-	return newCloudFoundryReceiver(params, *c, nextConsumer)
+	return newCloudFoundryMetricsReceiver(params, *c, nextConsumer)
+}
+
+func createLogsReceiver(
+	_ context.Context,
+	params receiver.CreateSettings,
+	cfg component.Config,
+	nextConsumer consumer.Logs,
+) (receiver.Logs, error) {
+	c := cfg.(*Config)
+	return newCloudFoundryLogsReceiver(params, *c, nextConsumer)
 }

receiver/cloudfoundryreceiver/factory_test.go

@@ -21,12 +21,22 @@ func TestCreateDefaultConfig(t *testing.T) {
 	assert.NoError(t, componenttest.CheckConfigStruct(cfg))
 }
 
-func TestCreateReceiver(t *testing.T) {
+func TestCreateMetricsReceiver(t *testing.T) {
 	factory := NewFactory()
 	cfg := factory.CreateDefaultConfig()
 
 	params := receivertest.NewNopCreateSettings()
 	tReceiver, err := factory.CreateMetricsReceiver(context.Background(), params, cfg, consumertest.NewNop())
 	assert.NoError(t, err)
-	assert.NotNil(t, tReceiver, "receiver creation failed")
+	assert.NotNil(t, tReceiver, "metrics receiver creation failed")
+}
+
+func TestCreateLogsReceiver(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+
+	params := receivertest.NewNopCreateSettings()
+	tReceiver, err := factory.CreateLogsReceiver(context.Background(), params, cfg, consumertest.NewNop())
+	assert.NoError(t, err)
+	assert.NotNil(t, tReceiver, "logs receiver creation failed")
 }

receiver/cloudfoundryreceiver/metadata.yaml

@@ -5,6 +5,7 @@ status:
   class: receiver
   stability:
     beta: [metrics]
+    development: [logs]
   distributions: [contrib]
   codeowners:
     active: [pellared, crobert-1]