From 72d0a014ff08f42165a7c20eab31d5fd0c586abd Mon Sep 17 00:00:00 2001
From: Alex Van Boxel <alex@vanboxel.be>
Date: Mon, 6 May 2024 07:57:44 +0200
Subject: [PATCH 1/2] [receiver/googlepubsub] fix span and trace id parsing
 crash

---
 .../googlepubsubreceiver-logenhance.yaml      |  29 +
 receiver/googlecloudpubsubreceiver/README.md  |  59 +-
 .../internal/log_entry.go                     | 622 +++++-------------
 .../internal/log_entry_test.go                | 350 +++++++++-
 .../{common_protos.go => log_proto_common.go} |   0
 .../internal/log_proto_payload.go             | 323 +++++++++
 .../internal/log_proto_payload_test.go        |  70 ++
 .../internal/log_translate_options.go         |  40 ++
 .../googlecloudpubsubreceiver/receiver.go     |   2 +-
 9 files changed, 1030 insertions(+), 465 deletions(-)
 create mode 100644 .chloggen/googlepubsubreceiver-logenhance.yaml
 rename receiver/googlecloudpubsubreceiver/internal/{common_protos.go => log_proto_common.go} (100%)
 create mode 100644 receiver/googlecloudpubsubreceiver/internal/log_proto_payload.go
 create mode 100644 receiver/googlecloudpubsubreceiver/internal/log_proto_payload_test.go
 create mode 100644 receiver/googlecloudpubsubreceiver/internal/log_translate_options.go

diff --git a/.chloggen/googlepubsubreceiver-logenhance.yaml b/.chloggen/googlepubsubreceiver-logenhance.yaml
new file mode 100644
index 0000000000000..9354d76b0aeac
--- /dev/null
+++ b/.chloggen/googlepubsubreceiver-logenhance.yaml
@@ -0,0 +1,29 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: googlecloudpubsubreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: fix span and trace id parsing crash when encountering corrupt data in `cloud_logging` encoding
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [ 32007 ]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Fixed span and trace id parsing crash when encountering corrupt data. Enhanced the `cloud_logging` parsing so it's 
+  better testable and include additional tests.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []
diff --git a/receiver/googlecloudpubsubreceiver/README.md b/receiver/googlecloudpubsubreceiver/README.md
index 5e5ee6a2ccc9d..980115ed87f1d 100644
--- a/receiver/googlecloudpubsubreceiver/README.md
+++ b/receiver/googlecloudpubsubreceiver/README.md
@@ -48,10 +48,10 @@ must the `encoding` field in the configuration be set.
 |-----------------------------------|----------------------|-------------------|------------------------------------------------|
 | org.opentelemetry.otlp.traces.v1  | application/protobuf |                   | Decode OTLP trace message                      |
 | org.opentelemetry.otlp.metrics.v1 | application/protobuf |                   | Decode OTLP metric message                     |
-| org.opentelemetry.otlp.logs.v1    | application/json     |                   | Decode OTLP log message                        |
+| org.opentelemetry.otlp.logs.v1    | application/protobuf |                   | Decode OTLP log message                        |
 | -                                 | -                    | otlp_proto_trace  | Decode OTLP trace message                      |
-| -                                 | -                    | otlp_proto_metric | Decode OTLP trace message                      |
-| -                                 | -                    | otlp_proto_log    | Decode OTLP trace message                      |
+| -                                 | -                    | otlp_proto_metric | Decode OTLP metric message                     |
+| -                                 | -                    | otlp_proto_log    | Decode OTLP log message                        |
 | -                                 | -                    | cloud_logging     | Decode [Cloud Logging] [LogEntry] message type |
 | -                                 | -                    | raw_text          | Wrap in an OTLP log message                    |
 
@@ -89,3 +89,56 @@ AND
 attributes.content-type = "application/protobuf"
 ```
 
+# Google Cloud Logging
+
+**Status**: Experimental
+
+Google Cloud logging uses the [LogEntry] to cary log information. In this section the mapping of the fields to 
+OpenTelemetry fields and attributes are documented.
+
+## Semantic Mapping
+
+Some of the attributes can be moved to OpenTelemetry Semantic Conventions. Note however that all the attributes are
+considered experimental and are subject to change.
+
+| Field       | Type      | Description                            | Maps to Unified Model Field    |
+|-------------|-----------|----------------------------------------|--------------------------------|
+| `insert_id` | `boolean` | A unique identifier for the log entry. | `Attributes["log.record.uid"]` |
+
+## Attribute and Field mapping
+
+The rest of the JSON body is either mapped to attributes with an `gcp` prefix or directly mapped to the LogRecord. 
+
+| Field              | Type                     | Description                                                                                                      | Maps to Unified Model Field                   |
+|--------------------|--------------------------|------------------------------------------------------------------------------------------------------------------|-----------------------------------------------|
+| `timestamp`        | `string`                 | The time the event described by the log entry occurred.                                                          | `Timestamp`                                   |
+| `receiveTimestamp` | `string`                 | The time the log entry was received.                                                                             | `ObservedTimestamp`                           |
+| `logName`          | `string`                 | The URL-encoded log ID suffix of the log_name field identifies which log stream this entry belongs to.           | `Attributes["gcp.log_name"]` (string)         |
+| `jsonPayload`      | `google.protobuf.Struct` | The log entry payload, represented as a structure that is expressed as a JSON object.                            | `Body` (KVList)                               |
+| `protoPayload`     | `google.protobuf.Any`    | The log entry payload, represented as a protocol buffer.                                                         | `Body` (KVList, key from JSON representation) |
+| `textPayload`      | `string`                 | The log entry payload, represented as a Unicode string (UTF-8).                                                  | `Body` (string)                               |
+| `trace`            | `string`                 | The trace associated with the log entry, if any.                                                                 | `TraceId`                                     |
+| `spanId`           | `string`                 | The span ID within the trace associated with the log entry.                                                      | `SpanId`                                      |
+| `traceSampled`     | `boolean`                | The sampling decision of the trace associated with the log entry.                                                | `TraceFlags.SAMPLED`                          |
+| `labels`           | `map<string,string>`     | A set of user-defined (key, value) data that provides additional information about the log entry.                | `Attributes["gcp.*"]`                         |
+| `resource`         | `MonitoredResource`      | The monitored resource that produced this log entry.                                                             | `Resource["gcp.*"]`                           |
+| `httpRequest`      | `HttpRequest`            | The HTTP request associated with the log entry, if any.                                                          | `Attributes["gcp.http_request"]` (KVList)     |
+| `operation`        | `LogEntryOperation`      | Information about a operation associated with the log entry.                                                     | `Attributes["gcp.operation"]`  (KVList)       |
+| `sourceLocation`   | `LogEntrySourceLocation` | Source code location information associated with the log entry.                                                  | `Attributes["gcp.source_location"]` (KVList)  |
+| `split`            | `LogSplit`               | Information indicating this LogEntry is part of a sequence of multiple log entries split from a single LogEntry. | `Attributes["gcp.log_split"]`  (KVList)       |
+
+## Severity Mapping
+
+The severity is mapping from [Google Cloud Log Severity](https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity) to the OpenTelemetry Severity Number
+
+| CloudLog         | Severity Number  | CloudLog Description                                                                   |
+|------------------|------------------|----------------------------------------------------------------------------------------|
+| `DEFAULT`(0)     | `UNSPECIFIED`(0) | The log entry has no assigned severity level.                                          |
+| `DEBUG`(100)     | `DEBUG`(5)       | Debug or trace information.                                                            |
+| `INFO`(200)      | `INFO`(9)        | Routine information, such as ongoing status or performance.                            |
+| `NOTICE`(300)    | `INFO2`(10)      | Normal but significant events, such as start up, shut down, or a configuration change. |
+| `WARNING`(400)   | `WARN`(13)       | Warning events might cause problems.                                                   |
+| `ERROR`(500)     | `ERROR`(17)      | Error events are likely to cause problems.                                             |
+| `CRITICAL`(600)  | `FATAL`(21)      | Critical events cause more severe problems or outages.                                 |
+| `ALERT`(700)     | `FATAL2`(22)     | A person must take an action immediately.                                              |
+| `EMERGENCY`(800) | `FATAL4`(24)     | One or more systems are unusable.                                                      |
diff --git a/receiver/googlecloudpubsubreceiver/internal/log_entry.go b/receiver/googlecloudpubsubreceiver/internal/log_entry.go
index e2b86f4ecf1d1..4eb6feca9c69e 100644
--- a/receiver/googlecloudpubsubreceiver/internal/log_entry.go
+++ b/receiver/googlecloudpubsubreceiver/internal/log_entry.go
@@ -4,13 +4,11 @@
 package internal // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudpubsubreceiver/internal"
 
 import (
-	"bytes"
 	"context"
 	"encoding/hex"
 	stdjson "encoding/json"
 	"errors"
 	"fmt"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -24,44 +22,45 @@ import (
 	"google.golang.org/genproto/googleapis/api/monitoredres"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/reflect/protoreflect"
-	"google.golang.org/protobuf/reflect/protoregistry"
-	"google.golang.org/protobuf/types/known/anypb"
 )
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
 
 var invalidTraceID = [16]byte{}
 var invalidSpanID = [8]byte{}
+var errorParsingLogItem = errors.New("error parsing log item")
 
-func cloudLoggingTraceToTraceIDBytes(trace string) [16]byte {
+func cloudLoggingTraceToTraceIDBytes(trace string) ([16]byte, error) {
 	// Format: projects/my-gcp-project/traces/4ebc71f1def9274798cac4e8960d0095
 	lastSlashIdx := strings.LastIndex(trace, "/")
 	if lastSlashIdx == -1 {
-		return invalidTraceID
+		return invalidTraceID, errorParsingLogItem
 	}
 	traceIDStr := trace[lastSlashIdx+1:]
 
-	return traceIDStrTotraceIDBytes(traceIDStr)
+	return traceIDStrToTraceIDBytes(traceIDStr)
 }
 
-func traceIDStrTotraceIDBytes(traceIDStr string) [16]byte {
-	traceIDSlice := [16]byte{}
-	decoded, err := hex.Decode(traceIDSlice[:], []byte(traceIDStr))
-	if err != nil || decoded != 16 {
-		return invalidTraceID
+func traceIDStrToTraceIDBytes(traceIDStr string) ([16]byte, error) {
+	decoded, err := hex.DecodeString(traceIDStr)
+	if err != nil {
+		return invalidTraceID, err
 	}
-
-	return traceIDSlice
+	if len(decoded) != 16 {
+		return invalidTraceID, errorParsingLogItem
+	}
+	return [16]byte(decoded), nil
 }
 
-func spanIDStrToSpanIDBytes(spanIDStr string) [8]byte {
-	spanIDSlice := [8]byte{}
-	decoded, err := hex.Decode(spanIDSlice[:], []byte(spanIDStr))
-	if err != nil || decoded != 8 {
-		return invalidSpanID
+func spanIDStrToSpanIDBytes(spanIDStr string) ([8]byte, error) {
+	decoded, err := hex.DecodeString(spanIDStr)
+	if err != nil {
+		return invalidSpanID, err
 	}
-
-	return spanIDSlice
+	if len(decoded) != 8 {
+		return invalidSpanID, errorParsingLogItem
+	}
+	return [8]byte(decoded), nil
 }
 
 func cloudLoggingSeverityToNumber(severity string) plog.SeverityNumber {
@@ -97,502 +96,231 @@ func getLogEntryDescriptor() protoreflect.MessageDescriptor {
 
 		desc = logEntry.ProtoReflect().Descriptor()
 	})
-
 	return desc
 }
 
-// TranslateLogEntry translates a JSON-encoded LogEntry message into a pair of
-// pcommon.Resource and plog.LogRecord, trying to keep as close as possible to
-// the semantic conventions.
-//
-// For maximum fidelity, the decoding is done according to the protobuf message
-// schema; this ensures that a numeric value in the input is correctly
-// translated to either an integer or a double in the output. It falls back to
-// plain JSON decoding if payload type is not available in the proto registry.
-func TranslateLogEntry(_ context.Context, _ *zap.Logger, data []byte) (pcommon.Resource, plog.LogRecord, error) {
-	lr := plog.NewLogRecord()
-	res := pcommon.NewResource()
-
-	var src map[string]stdjson.RawMessage
-	err := json.Unmarshal(data, &src)
-
-	if err != nil {
-		return res, lr, err
-	}
-
-	resAttrs := res.Attributes()
-	attrs := lr.Attributes()
+type extractFn func(pcommon.Map, plog.LogRecord, pcommon.Map, string, stdjson.RawMessage, Options) error
 
-	for k, v := range src {
-		// Pick out some keys for special handling, and let the rest
-		// pass through to be translated according to the schema.
-		switch k {
-		// Unpack as suggested in the logs data model appendix
-		//   https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/logs/data-model-appendix.md#google-cloud-logging
-		case "insertId":
-			// timestamp -> Attributes[“log.record.uid”]
-			// see: https://github.com/open-telemetry/semantic-conventions/blob/main/model/logs/general.yaml
-			var insertID string
-			err = json.Unmarshal(v, &insertID)
-			if err != nil {
-				return res, lr, err
-			}
-			attrs.PutStr("log.record.uid", insertID)
-			delete(src, k)
-		case "timestamp":
-			// timestamp -> Timestamp
-			var t time.Time
-			err = json.Unmarshal(v, &t)
-			if err != nil {
-				return res, lr, err
-			}
-			lr.SetTimestamp(pcommon.NewTimestampFromTime(t))
-			delete(src, k)
-		case "receiveTimestamp":
-			// timestamp -> Timestamp
-			var t time.Time
-			err = json.Unmarshal(v, &t)
-			if err != nil {
-				return res, lr, err
-			}
-			lr.SetObservedTimestamp(pcommon.NewTimestampFromTime(t))
-			delete(src, k)
-		case "resource":
-			// resource -> Resource
-			// mapping type -> gcp.resource_type
-			// labels -> gcp.<label>
-			var protoRes monitoredres.MonitoredResource
-			err = protojson.Unmarshal(v, &protoRes)
-			if err != nil {
-				return res, lr, err
-			}
-
-			resAttrs.EnsureCapacity(len(protoRes.GetLabels()) + 1)
-			resAttrs.PutStr("gcp.resource_type", protoRes.GetType())
-			for k, v := range protoRes.GetLabels() {
-				resAttrs.PutStr(strcase.ToSnakeWithIgnore(fmt.Sprintf("gcp.%v", k), "."), v)
-			}
-			delete(src, k)
-		case "logName":
-			var logName string
-			err = json.Unmarshal(v, &logName)
-			if err != nil {
-				return res, lr, err
-			}
-			// log_name -> Attributes[“gcp.log_name”]
-			attrs.PutStr("gcp.log_name", logName)
-			delete(src, k)
-		case "jsonPayload", "textPayload":
-			// {json,proto,text}_payload -> Body
-			var payload any
-			err = json.Unmarshal(v, &payload)
-			if err != nil {
-				return res, lr, err
-			}
-			// Note: json.Unmarshal will turn a bare string into a
-			// go string, so this call will correctly set the body
-			// to a string Value.
-			_ = lr.Body().FromRaw(payload)
-			delete(src, k)
-		case "protoPayload":
-			// {json,proto,text}_payload -> Body
-			err = translateInto(lr.Body().SetEmptyMap(), (&anypb.Any{}).ProtoReflect().Descriptor(), v)
-			if err != nil {
-				return res, lr, err
-			}
-			delete(src, k)
-		case "severity":
-			var severity string
-			err = json.Unmarshal(v, &severity)
-			if err != nil {
-				return res, lr, err
+func extractFromFields(_ context.Context, source map[string]stdjson.RawMessage, resourceAttributes pcommon.Map, logRecord plog.LogRecord, attributes pcommon.Map, extractFns map[string]extractFn, options Options) {
+	for key, value := range source {
+		fn := extractFns[key]
+		if fn != nil {
+			err := fn(resourceAttributes, logRecord, attributes, key, value, options)
+			if err == nil {
+				delete(source, key)
 			}
-			// severity -> Severity
-			// According to the spec, this is the original string representation of
-			// the severity as it is known at the source:
-			//   https://opentelemetry.io/docs/reference/specification/logs/data-model/#field-severitytext
-			lr.SetSeverityText(severity)
-			lr.SetSeverityNumber(cloudLoggingSeverityToNumber(severity))
-			delete(src, k)
-		case "trace":
-			var trace string
-			err = json.Unmarshal(v, &trace)
-			if err != nil {
-				return res, lr, err
-			}
-			lr.SetTraceID(cloudLoggingTraceToTraceIDBytes(trace))
-			delete(src, k)
-		case "spanId":
-			var spanID string
-			err = json.Unmarshal(v, &spanID)
-			if err != nil {
-				return res, lr, err
-			}
-			lr.SetSpanID(spanIDStrToSpanIDBytes(spanID))
-			delete(src, k)
-		case "labels":
-			var labels map[string]string
-			err = json.Unmarshal(v, &labels)
-			if err != nil {
-				return res, lr, err
-			}
-			// labels -> Attributes
-			for k, v := range labels {
-				attrs.PutStr(k, v)
-			}
-			delete(src, k)
-		case "httpRequest":
-			httpRequestAttrs := attrs.PutEmptyMap("gcp.http_request")
-			err = translateInto(httpRequestAttrs, getLogEntryDescriptor().Fields().ByJSONName(k).Message(), v, snakeifyKeys)
-			if err != nil {
-				return res, lr, err
-			}
-			delete(src, k)
-		default:
 		}
 	}
-
-	// All other fields -> Attributes["gcp.*"]
-	// At this point we cleared all the fields that have special handling;
-	// translate the rest into the attributes map.
-	_ = translateInto(attrs, getLogEntryDescriptor(), src, preserveDst, prefixKeys("gcp."), snakeifyKeys)
-
-	return res, lr, nil
 }
 
-// should only translate maps?
-
-func snakeify(s string) string {
-	return strcase.ToSnakeWithIgnore(s, ".")
-}
-
-func prefix(p string) func(string) string {
-	return func(s string) string {
-		return p + s
+func handleTimestamp(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	// timestamp -> Timestamp
+	var t time.Time
+	err := json.Unmarshal(value, &t)
+	if err != nil {
+		return err
 	}
+	logRecord.SetTimestamp(pcommon.NewTimestampFromTime(t))
+	return nil
 }
 
-type translateOptions struct {
-	keyMappers  []func(string) string
-	preserveDst bool
-}
-
-type opt func(*translateOptions)
-
-func preserveDst(opts *translateOptions) {
-	opts.preserveDst = true
-}
-
-func snakeifyKeys(opts *translateOptions) {
-	opts.keyMappers = append(opts.keyMappers, snakeify)
+func handleReceiveTimestamp(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	// receiveTimestamp -> Timestamp
+	var t time.Time
+	err := json.Unmarshal(value, &t)
+	if err != nil {
+		return err
+	}
+	logRecord.SetObservedTimestamp(pcommon.NewTimestampFromTime(t))
+	return nil
 }
 
-func prefixKeys(p string) opt {
-	return func(opts *translateOptions) {
-		opts.keyMappers = append(opts.keyMappers, prefix(p))
+// handleInsertID extracts the insertId field from a LogEntry and puts it into the log record attributes.
+// insertId -> Attributes[“log.record.uid”] stability: experimental
+// see https://github.com/open-telemetry/semantic-conventions/blob/main/model/registry/log.yaml
+func handleInsertID(_ pcommon.Map, _ plog.LogRecord, logAttributes pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var insertID string
+	err := json.Unmarshal(value, &insertID)
+	if err != nil {
+		return err
 	}
+	logAttributes.PutStr("log.record.uid", insertID)
+	return nil
 }
 
-func (opts translateOptions) mapKey(s string) string {
-	for _, mapper := range opts.keyMappers {
-		s = mapper(s)
+// handleLogName extracts the logName field from a LogEntry and puts it into the log record attributes.
+// logName -> Attributes[“gcp.log_name”] stability: experimental
+func handleLogName(_ pcommon.Map, _ plog.LogRecord, logAttributes pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var logName string
+	err := json.Unmarshal(value, &logName)
+	if err != nil {
+		return err
 	}
-
-	return s
+	logAttributes.PutStr("gcp.log_name", logName)
+	return nil
 }
 
-func getType(src stdjson.RawMessage) string {
-	dec := stdjson.NewDecoder(bytes.NewReader(src))
-	tok, err := dec.Token()
+func handleResource(resourceAttributes pcommon.Map, _ plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	// resource -> Resource
+	// mapping type -> gcp.resource_type
+	// labels -> gcp.label.<label>
+	var protoRes monitoredres.MonitoredResource
+	err := protojson.Unmarshal(value, &protoRes)
 	if err != nil {
-		return "invalid json"
+		return err
 	}
-	switch t := tok.(type) {
-	case stdjson.Delim:
-		switch t {
-		case '[':
-			return "array"
-		case '{':
-			return "object"
-		default:
-			return "invalid json"
-		}
-	case bool:
-		return "bool"
-	case float64, stdjson.Number:
-		return "number"
-	case string:
-		return "string"
-	case nil:
-		return "null"
-	default:
-		return "unknown"
+	resourceAttributes.PutStr("gcp.resource_type", protoRes.GetType())
+	for k, v := range protoRes.GetLabels() {
+		resourceAttributes.PutStr(strcase.ToSnakeWithIgnore(fmt.Sprintf("gcp.%v", k), "."), v)
 	}
+	return nil
 }
 
-func translateStr(dst pcommon.Value, src stdjson.RawMessage) error {
-	var val string
-	err := json.Unmarshal(src, &val)
+func handleTextPayload(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var textPayload string
+	err := json.Unmarshal(value, &textPayload)
 	if err != nil {
 		return err
 	}
-	dst.SetStr(val)
+	logRecord.Body().SetStr(textPayload)
 	return nil
 }
 
-func translateRaw(dst pcommon.Value, src stdjson.RawMessage) error {
-	var val any
-	err := json.Unmarshal(src, &val)
+func handleJSONPayload(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	// {json,proto,text}_payload -> Body
+	var payload any
+	err := json.Unmarshal(value, &payload)
 	if err != nil {
 		return err
 	}
-	_ = dst.FromRaw(val)
+	// Note: json.Unmarshal will turn a bare string into a
+	// go string, so this call will correctly set the body
+	// to a string Value.
+	_ = logRecord.Body().FromRaw(payload)
 	return nil
 }
 
-func (opts translateOptions) translateValue(dst pcommon.Value, fd protoreflect.FieldDescriptor, src stdjson.RawMessage) error {
-	var err error
-	switch fd.Kind() {
-	case protoreflect.MessageKind:
-		msg := fd.Message()
-		switch fd.Message().FullName() {
-		case "google.protobuf.Duration", "google.protobuf.Timestamp":
-			// protojson represents both of these as strings
-			return translateStr(dst, src)
-		case "google.protobuf.Struct", "google.protobuf.Value":
-			return translateRaw(dst, src)
-		case
-			"google.protobuf.BoolValue",
-			"google.protobuf.BytesValue",
-			"google.protobuf.DoubleValue",
-			"google.protobuf.FloatValue",
-			"google.protobuf.Int32Value",
-			"google.protobuf.Int64Value",
-			"google.protobuf.StringValue",
-			"google.protobuf.UInt32Value",
-			"google.protobuf.UInt64Value":
-			// All the wrapper types have a single field with name
-			// `value` and field number 1, and are represented in
-			// protojson without the wrapping.
-			innerFd := fd.Message().Fields().ByNumber(1)
-			_ = opts.translateValue(dst, innerFd, src)
-		default:
-			var m pcommon.Map
-			switch dst.Type() {
-			case pcommon.ValueTypeMap:
-				m = dst.Map()
-			default:
-				m = dst.SetEmptyMap()
-			}
-			return translateInto(m, msg, src)
-		}
-	case protoreflect.EnumKind:
-		// protojson accepts either string name or enum int value; try both.
-		if translateStr(dst, src) == nil {
-			return nil
-		}
-
-		enum := fd.Enum()
-		var i int32
-		if err = json.Unmarshal(src, &i); err != nil {
-			return fmt.Errorf("wrong type for enum: %v", getType(src))
-		}
-		enumValue := enum.Values().ByNumber(protoreflect.EnumNumber(i))
-		if enumValue == nil {
-			return fmt.Errorf("%v has no enum value for %v", enum.FullName(), i)
-		}
-
-		dst.SetStr(string(enumValue.Name()))
-	case protoreflect.BoolKind:
-		var val bool
-		err = json.Unmarshal(src, &val)
-		if err != nil {
-			return err
-		}
-		dst.SetBool(val)
-	case protoreflect.Int32Kind,
-		protoreflect.Uint32Kind,
-		protoreflect.Sfixed32Kind,
-		protoreflect.Fixed32Kind,
-		protoreflect.Sint32Kind,
-		protoreflect.Int64Kind,
-		protoreflect.Uint64Kind,
-		protoreflect.Sfixed64Kind,
-		protoreflect.Fixed64Kind,
-		protoreflect.Sint64Kind:
-		// The protojson encoding accepts either string or number for
-		// integer types, so try both.
-		var val int64
-		if json.Unmarshal(src, &val) == nil {
-			dst.SetInt(val)
-			return nil
-		}
-
-		var s string
-		if err = json.Unmarshal(src, &s); err != nil {
-			return err
-		}
-		if val, err = strconv.ParseInt(s, 10, 64); err != nil {
-			return err
-		}
-		dst.SetInt(val)
-		return nil
-	case protoreflect.FloatKind, protoreflect.DoubleKind:
-		var val float64
-		err := json.Unmarshal(src, &val)
-		if err != nil {
-			return err
-		}
-		dst.SetDouble(val)
-		return nil
-	case protoreflect.BytesKind:
-		var val []byte
-		err := json.Unmarshal(src, &val)
-		if err != nil {
-			return err
-		}
-		dst.SetEmptyBytes().Append(val...)
-		return nil
-	case protoreflect.StringKind:
-		return translateStr(dst, src)
-	case protoreflect.GroupKind:
-		return errors.New("unexpected group")
-	default:
-		return errors.New("unknown field kind")
+func handleSeverity(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var severity string
+	err := json.Unmarshal(value, &severity)
+	if err != nil {
+		return err
 	}
+	// severity -> Severity
+	// According to the spec, this is the original string representation of
+	// the severity as it is known at the source:
+	//   https://opentelemetry.io/docs/reference/specification/logs/data-model/#field-severitytext
+	logRecord.SetSeverityText(severity)
+	logRecord.SetSeverityNumber(cloudLoggingSeverityToNumber(severity))
 	return nil
 }
 
-func (opts translateOptions) translateList(dst pcommon.Slice, fd protoreflect.FieldDescriptor, src stdjson.RawMessage) error {
-	var msg []stdjson.RawMessage
-	if err := json.Unmarshal(src, &msg); err != nil {
+func handleTrace(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var trace string
+	err := json.Unmarshal(value, &trace)
+	if err != nil {
 		return err
 	}
-
-	for _, v := range msg {
-		err := opts.translateValue(dst.AppendEmpty(), fd, v)
-		if err != nil {
-			return err
-		}
+	traceIDBytes, err := cloudLoggingTraceToTraceIDBytes(trace)
+	if err != nil {
+		return err
 	}
+	logRecord.SetTraceID(traceIDBytes)
 	return nil
 }
 
-func (opts translateOptions) translateMap(dst pcommon.Map, fd protoreflect.FieldDescriptor, src stdjson.RawMessage) error {
-	var msg map[string]stdjson.RawMessage
-	if err := json.Unmarshal(src, &msg); err != nil {
+func handleSpanID(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var spanID string
+	err := json.Unmarshal(value, &spanID)
+	if err != nil {
 		return err
 	}
-	for k, v := range msg {
-		err := opts.translateValue(dst.PutEmpty(k), fd.MapValue(), v)
-		if err != nil {
-			return err
-		}
+	spanIDBytes, err := spanIDStrToSpanIDBytes(spanID)
+	if err != nil {
+		return err
 	}
+	logRecord.SetSpanID(spanIDBytes)
 	return nil
 }
-func translateAny(dst pcommon.Map, src map[string]stdjson.RawMessage) error {
-	// protojson represents Any as the JSON representation of the actual
-	// message, plus a special @type field containing the type URL of the
-	// message.
-	typeURL, ok := src["@type"]
-	if !ok {
-		return errors.New("no @type member in Any message")
-	}
-	var typeURLStr string
-	if err := json.Unmarshal(typeURL, &typeURLStr); err != nil {
-		return err
-	}
-	delete(src, "@type")
-
-	msgType, err := protoregistry.GlobalTypes.FindMessageByURL(typeURLStr)
-	if errors.Is(err, protoregistry.NotFound) {
-		// If we don't have the type, we do a best-effort JSON decode;
-		// some ints might be floats or strings.
-		for k, v := range src {
-			var val any
-			err = json.Unmarshal(v, &val)
-			if err != nil {
-				return nil
-			}
-			_ = dst.PutEmpty(k).FromRaw(val)
-		}
-		return nil
-	}
 
-	err = translateInto(dst, msgType.Descriptor(), src)
+func handleTraceSampled(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var traceSampled bool
+	err := json.Unmarshal(value, &traceSampled)
 	if err != nil {
 		return err
 	}
-
-	dst.PutStr("@type", typeURLStr)
+	var flags plog.LogRecordFlags
+	logRecord.SetFlags(flags.WithIsSampled(traceSampled))
 	return nil
 }
 
-func (opts translateOptions) translateMessage(dst pcommon.Map, desc protoreflect.MessageDescriptor, src map[string]stdjson.RawMessage) error {
-	if !opts.preserveDst {
-		dst.Clear()
+func handleLabels(_ pcommon.Map, _ plog.LogRecord, logAttributes pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	var labels map[string]string
+	err := json.Unmarshal(value, &labels)
+	if err != nil {
+		return err
 	}
 
-	// Handle well-known aggregate types.
-	switch desc.FullName() {
-	case "google.protobuf.Any":
-		return translateAny(dst, src)
-	case "google.protobuf.Struct":
-		for k, v := range src {
-			var val any
-			if err := json.Unmarshal(v, &val); err != nil {
-				return err
-			}
-			_ = dst.PutEmpty(k).FromRaw(val)
-		}
-		return nil
-	case "google.protobuf.Empty":
-		dst.Clear()
-		return nil
-	default:
+	// labels -> Attributes
+	for k, v := range labels {
+		logAttributes.PutStr(strcase.ToSnakeWithIgnore(fmt.Sprintf("gcp.%v", k), "."), v)
 	}
+	return nil
+}
 
-	for k, v := range src {
-		key := opts.mapKey(k)
-		fd := desc.Fields().ByJSONName(k)
-		if fd == nil {
-			return fmt.Errorf("%v has no known field with JSON name %v", desc.FullName(), k)
-		}
-		var err error
-		switch {
-		case fd.IsList():
-			err = opts.translateList(dst.PutEmptySlice(key), fd, v)
-		case fd.IsMap():
-			err = opts.translateMap(dst.PutEmptyMap(key), fd, v)
-		default:
-			err = opts.translateValue(dst.PutEmpty(key), fd, v)
-		}
-		if err != nil {
-			return err
-		}
+func handleHTTPRequest(_ pcommon.Map, _ plog.LogRecord, logAttributes pcommon.Map, key string, value stdjson.RawMessage, _ Options) error {
+	httpRequestAttrs := logAttributes.PutEmptyMap("gcp.http_request")
+	err := translateInto(httpRequestAttrs, getLogEntryDescriptor().Fields().ByJSONName(key).Message(), value, snakeifyKeys)
+	if err != nil {
+		return err
 	}
-
 	return nil
 }
 
-func translateInto(dst pcommon.Map, desc protoreflect.MessageDescriptor, src any, opts ...opt) error {
-	var toTranslate map[string]stdjson.RawMessage
+// TranslateLogEntry translates a JSON-encoded LogEntry message into a pair of
+// pcommon.Resource and plog.LogRecord, trying to keep as close as possible to
+// the semantic conventions.
+//
+// For maximum fidelity, the decoding is done according to the protobuf message
+// schema; this ensures that a numeric value in the input is correctly
+// translated to either an integer or a double in the output. It falls back to
+// plain JSON decoding if payload type is not available in the proto registry.
+func TranslateLogEntry(context context.Context, _ *zap.Logger, data []byte, options Options) (pcommon.Resource, plog.LogRecord, error) {
+	lr := plog.NewLogRecord()
+	res := pcommon.NewResource()
 
-	switch msg := src.(type) {
-	case stdjson.RawMessage:
-		err := json.Unmarshal(msg, &toTranslate)
-		if err != nil {
-			return err
-		}
-	case map[string]stdjson.RawMessage:
-		toTranslate = msg
+	var src map[string]stdjson.RawMessage
+	err := json.Unmarshal(data, &src)
+
+	if err != nil {
+		return res, lr, err
 	}
 
-	options := translateOptions{}
-	for _, opt := range opts {
-		opt(&options)
+	resAttrs := res.Attributes()
+	attrs := lr.Attributes()
+
+	extractFns := map[string]extractFn{
+		"protoPayload":     handleProtoPayload,
+		"receiveTimestamp": handleReceiveTimestamp,
+		"timestamp":        handleTimestamp,
+		"insertId":         handleInsertID,
+		"logName":          handleLogName,
+		"textPayload":      handleTextPayload,
+		"jsonPayload":      handleJSONPayload,
+		"severity":         handleSeverity,
+		"trace":            handleTrace,
+		"spanId":           handleSpanID,
+		"traceSampled":     handleTraceSampled,
+		"labels":           handleLabels,
+		"httpRequest":      handleHTTPRequest,
+		"resource":         handleResource,
 	}
+	extractFromFields(context, src, resAttrs, lr, attrs, extractFns, options)
+
+	// All other fields -> Attributes["gcp.*"]
+	// At this point we cleared all the fields that have special handling;
+	// translate the rest into the attributes map.
+	_ = translateInto(attrs, getLogEntryDescriptor(), src, preserveDst, prefixKeys("gcp."), snakeifyKeys)
 
-	return options.translateMessage(dst, desc, toTranslate)
+	return res, lr, nil
 }
diff --git a/receiver/googlecloudpubsubreceiver/internal/log_entry_test.go b/receiver/googlecloudpubsubreceiver/internal/log_entry_test.go
index cf2df7f95f2cf..0b772b3df75a0 100644
--- a/receiver/googlecloudpubsubreceiver/internal/log_entry_test.go
+++ b/receiver/googlecloudpubsubreceiver/internal/log_entry_test.go
@@ -23,12 +23,16 @@ type Log struct {
 	ObservedTimestamp  string
 	Body               any
 	SeverityText       string
+	SeverityNumber     plog.SeverityNumber
 	Attributes         map[string]any
 	ResourceAttributes map[string]any
 	SpanID             string
 	TraceID            string
 }
 
+type testOptions struct {
+}
+
 func generateLog(t *testing.T, log Log) (pcommon.Resource, plog.LogRecord, error) {
 	res := pcommon.NewResource()
 	assert.NoError(t, res.Attributes().FromRaw(log.ResourceAttributes))
@@ -56,34 +60,310 @@ func generateLog(t *testing.T, log Log) (pcommon.Resource, plog.LogRecord, error
 
 	assert.NoError(t, lr.Body().FromRaw(log.Body))
 	lr.SetSeverityText(log.SeverityText)
+	lr.SetSeverityNumber(log.SeverityNumber)
 
-	lr.SetSpanID(spanIDStrToSpanIDBytes(log.SpanID))
-	lr.SetTraceID(traceIDStrTotraceIDBytes(log.TraceID))
+	spanID, _ := spanIDStrToSpanIDBytes(log.SpanID)
+	lr.SetSpanID(spanID)
+	traceID, _ := traceIDStrToTraceIDBytes(log.TraceID)
+	lr.SetTraceID(traceID)
 
 	return res, lr, nil
 }
 
+func TestCloudLoggingTraceToTraceIDBytes(t *testing.T) {
+	for _, test := range []struct {
+		scenario,
+		in string
+		out [16]byte
+		err string
+	}{
+		// A valid trace
+		{
+			"ValidTrace",
+			"projects/my-gcp-project/traces/1dbe317eb73eb6e3bbb51a2bc3a41e09",
+			[16]uint8{0x1d, 0xbe, 0x31, 0x7e, 0xb7, 0x3e, 0xb6, 0xe3, 0xbb, 0xb5, 0x1a, 0x2b, 0xc3, 0xa4, 0x1e, 0x9},
+			"",
+		},
+		// An invalid traces
+		{
+			"InvalidTrace",
+			"1dbe317eb73eb6e3bbb51a2bc3a41e09",
+			invalidTraceID,
+			errorParsingLogItem.Error(),
+		},
+		{
+			"InvalidHexTrace",
+			"projects/my-gcp-project/traces/xyze317eb73eb6e3bbb51a2bc3a41e09",
+			invalidTraceID,
+			"encoding/hex: invalid byte: U+0078 'x'",
+		},
+		{
+			"InvalidLengthTooLongTrace",
+			"projects/my-gcp-project/traces/1dbe317eb73eb6e3bbb51a2bc3a41e09aa",
+			invalidTraceID,
+			errorParsingLogItem.Error(),
+		},
+		{
+			"InvalidLengthTooShortTrace",
+			"projects/my-gcp-project/traces/1dbe317eb73eb6e3bbb51a2bc3a41e",
+			invalidTraceID,
+			errorParsingLogItem.Error(),
+		},
+	} {
+		t.Run(test.scenario, func(t *testing.T) {
+			out, err := cloudLoggingTraceToTraceIDBytes(test.in)
+			assert.Equal(t, test.out, out)
+			if err != nil {
+				assert.Equal(t, test.err, err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+
+}
+
+func TestSpanIDStrToSpanIDBytes(t *testing.T) {
+	for _, test := range []struct {
+		scenario,
+		in string
+		out [8]byte
+		err string
+	}{
+		// A valid span
+		{
+			"ValidSpan",
+			"3e3a5741b18f0710",
+			[8]uint8{0x3e, 0x3a, 0x57, 0x41, 0xb1, 0x8f, 0x7, 0x10},
+			"",
+		},
+		// Cloud Run sends invalid span id's, make sure we're not crashing,
+		// see https://issuetracker.google.com/issues/338634230?pli=1
+		{
+			"InvalidNumberSpan",
+			"13210305202245662348",
+			invalidSpanID,
+			errorParsingLogItem.Error(),
+		},
+		// A invalid span
+		{
+			"ValidHexTooLongSpan",
+			"3e3a5741b18f0710ab",
+			invalidSpanID,
+			errorParsingLogItem.Error(),
+		},
+		{
+			"ValidHexTooShortSpan",
+			"3e3a5741b18f07",
+			invalidSpanID,
+			errorParsingLogItem.Error(),
+		},
+	} {
+		t.Run(test.scenario, func(t *testing.T) {
+			out, err := spanIDStrToSpanIDBytes(test.in)
+			assert.Equal(t, test.out, out)
+			if err != nil {
+				assert.Equal(t, test.err, err.Error())
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+
+}
+
 func TestTranslateLogEntry(t *testing.T) {
 
 	tests := []struct {
-		input string
-		want  Log
+		scenario string
+		input    string
+		want     Log
 	}{
-		// TODO: Add publicly shareable log test data.
+		{
+			"RealLog",
+			"{\"httpRequest\":{\"latency\":\"0.001901s\",\"protocol\":\"H2C\",\"referer\":\"https://console.cloud.google.com/\",\"remoteIp\":\"1802:2a14:1188:9898:34ab:b5c5:86eb:a142\",\"requestMethod\":\"GET\",\"requestSize\":\"1220\",\"requestUrl\":\"https://example.a.run.app/\",\"responseSize\":\"963\",\"serverIp\":\"2024:6048:3248:42::42\",\"status\":503,\"userAgent\":\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36\"},\"insertId\":\"663760330006df7c63521728\",\"labels\":{\"instanceId\":\"00f46b9285fdad63aa1a6951b9b01126bd635517f264c8f231a40248b473fcdf266eed0fdfebd05fb338203c0e3e5bf6422f0b480445cacc4ec75bd4ebc86a72\"},\"logName\":\"projects/gcp-project/logs/run.googleapis.com%2Frequests\",\"receiveTimestamp\":\"2024-05-05T10:32:19.45570687Z\",\"resource\":{\"labels\":{\"configuration_name\":\"otelcol\",\"location\":\"europe-west1\",\"project_id\":\"gcp-project\",\"revision_name\":\"example-00007-sun\",\"service_name\":\"otelcol\"},\"type\":\"cloud_run_revision\"},\"severity\":\"ERROR\",\"spanId\":\"3e3a5741b18f0710\",\"timestamp\":\"2024-05-05T10:32:19.440152Z\",\"trace\":\"projects/gcp-project/traces/1dbe317eb73eb6e3bbb51a2bc3a41e09\",\"traceSampled\":true}",
+			Log{
+				Timestamp:         "2024-05-05T10:32:19.440152Z",
+				ObservedTimestamp: "",
+				Body:              nil,
+				SeverityText:      "ERROR",
+				SeverityNumber:    plog.SeverityNumberError,
+				Attributes: map[string]any{
+					"gcp.http_request": map[string]any{
+						"latency":        string("0.001901s"),
+						"protocol":       string("H2C"),
+						"referer":        string("https://console.cloud.google.com/"),
+						"remote_ip":      string("1802:2a14:1188:9898:34ab:b5c5:86eb:a142"),
+						"request_method": string("GET"),
+						"request_size":   int64(1220),
+						"request_url":    string("https://example.a.run.app/"),
+						"response_size":  int64(963),
+						"server_ip":      string("2024:6048:3248:42::42"),
+						"status":         int64(503),
+						"user_agent":     string("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"),
+					},
+					"gcp.instance_id": string("00f46b9285fdad63aa1a6951b9b01126bd635517f264c8f231a40248b473fcdf266eed0fdfebd05fb338203c0e3e5bf6422f0b480445cacc4ec75bd4ebc86a72"),
+					"gcp.log_name":    string("projects/gcp-project/logs/run.googleapis.com%2Frequests"),
+					"log.record.uid":  string("663760330006df7c63521728"),
+				},
+				ResourceAttributes: map[string]any{
+					"gcp.configuration_name": string("otelcol"),
+					"gcp.location":           string("europe-west1"),
+					"gcp.project_id":         string("gcp-project"),
+					"gcp.revision_name":      string("example-00007-sun"),
+					"gcp.service_name":       string("otelcol"),
+					"gcp.resource_type":      string("cloud_run_revision"),
+				},
+				SpanID:  "3e3a5741b18f0710",
+				TraceID: "1dbe317eb73eb6e3bbb51a2bc3a41e09",
+			},
+		},
+		// This is the same log as above but one field (span) that causes an error, parsing will continue put added to attribute
+		{
+			"InvalidSpan",
+			"{\"httpRequest\":{\"latency\":\"0.001901s\",\"protocol\":\"H2C\",\"referer\":\"https://console.cloud.google.com/\",\"remoteIp\":\"1802:2a14:1188:9898:34ab:b5c5:86eb:a142\",\"requestMethod\":\"GET\",\"requestSize\":\"1220\",\"requestUrl\":\"https://example.a.run.app/\",\"responseSize\":\"963\",\"serverIp\":\"2024:6048:3248:42::42\",\"status\":503,\"userAgent\":\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36\"},\"insertId\":\"663760330006df7c63521728\",\"labels\":{\"instanceId\":\"00f46b9285fdad63aa1a6951b9b01126bd635517f264c8f231a40248b473fcdf266eed0fdfebd05fb338203c0e3e5bf6422f0b480445cacc4ec75bd4ebc86a72\"},\"logName\":\"projects/gcp-project/logs/run.googleapis.com%2Frequests\",\"receiveTimestamp\":\"2024-05-05T10:32:19.45570687Z\",\"resource\":{\"labels\":{\"configuration_name\":\"otelcol\",\"location\":\"europe-west1\",\"project_id\":\"gcp-project\",\"revision_name\":\"example-00007-sun\",\"service_name\":\"otelcol\"},\"type\":\"cloud_run_revision\"},\"severity\":\"ERROR\",\"spanId\":\"13210305202245662348\",\"timestamp\":\"2024-05-05T10:32:19.440152Z\",\"trace\":\"projects/gcp-project/traces/1dbe317eb73eb6e3bbb51a2bc3a41e09\",\"traceSampled\":true}",
+			Log{
+				Timestamp:         "2024-05-05T10:32:19.440152Z",
+				ObservedTimestamp: "",
+				Body:              nil,
+				SeverityText:      "ERROR",
+				SeverityNumber:    plog.SeverityNumberError,
+				Attributes: map[string]any{
+					"gcp.http_request": map[string]any{
+						"latency":        string("0.001901s"),
+						"protocol":       string("H2C"),
+						"referer":        string("https://console.cloud.google.com/"),
+						"remote_ip":      string("1802:2a14:1188:9898:34ab:b5c5:86eb:a142"),
+						"request_method": string("GET"),
+						"request_size":   int64(1220),
+						"request_url":    string("https://example.a.run.app/"),
+						"response_size":  int64(963),
+						"server_ip":      string("2024:6048:3248:42::42"),
+						"status":         int64(503),
+						"user_agent":     string("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"),
+					},
+					"gcp.instance_id": string("00f46b9285fdad63aa1a6951b9b01126bd635517f264c8f231a40248b473fcdf266eed0fdfebd05fb338203c0e3e5bf6422f0b480445cacc4ec75bd4ebc86a72"),
+					"gcp.log_name":    string("projects/gcp-project/logs/run.googleapis.com%2Frequests"),
+					// this attribute added
+					"gcp.span_id":    string("13210305202245662348"),
+					"log.record.uid": string("663760330006df7c63521728"),
+				},
+				ResourceAttributes: map[string]any{
+					"gcp.configuration_name": string("otelcol"),
+					"gcp.location":           string("europe-west1"),
+					"gcp.project_id":         string("gcp-project"),
+					"gcp.revision_name":      string("example-00007-sun"),
+					"gcp.service_name":       string("otelcol"),
+					"gcp.resource_type":      string("cloud_run_revision"),
+				},
+				SpanID:  "",
+				TraceID: "1dbe317eb73eb6e3bbb51a2bc3a41e09",
+			},
+		},
+		// Payloads
+		{
+			"TextPayload",
+			"{\"textPayload\":\"Hello World\"}",
+			Log{
+				Body: string("Hello World"),
+			},
+		},
+		{
+			"TextPayloadParseError",
+			"{\"textPayload\":42}",
+			Log{
+				Attributes: map[string]any{
+					"gcp.text_payload": nil,
+				},
+			},
+		},
+		{
+			"JsonPayloadParse",
+			"{\"jsonPayload\":{\"foo\":\"bar\"}}",
+			Log{
+				Body: map[string]any{"foo": string("bar")},
+			},
+		},
+		// Severity
+		{
+			"SeverityNumberDebug",
+			"{\"severity\":\"DEBUG\"}",
+			Log{
+				SeverityText:   "DEBUG",
+				SeverityNumber: plog.SeverityNumberDebug,
+			},
+		},
+		{
+			"SeverityNumberInfo",
+			"{\"severity\":\"INFO\"}",
+			Log{
+				SeverityText:   "INFO",
+				SeverityNumber: plog.SeverityNumberInfo,
+			},
+		},
+		{
+			"SeverityNumberInfo2",
+			"{\"severity\":\"NOTICE\"}",
+			Log{
+				SeverityText:   "NOTICE",
+				SeverityNumber: plog.SeverityNumberInfo2,
+			},
+		},
+		{
+			"SeverityNumberWarn",
+			"{\"severity\":\"WARNING\"}",
+			Log{
+				SeverityText:   "WARNING",
+				SeverityNumber: plog.SeverityNumberWarn,
+			},
+		},
+		{
+			"SeverityNumberError",
+			"{\"severity\":\"ERROR\"}",
+			Log{
+				SeverityText:   "ERROR",
+				SeverityNumber: plog.SeverityNumberError,
+			},
+		},
+		{
+			"SeverityNumberFatal",
+			"{\"severity\":\"CRITICAL\"}",
+			Log{
+				SeverityText:   "CRITICAL",
+				SeverityNumber: plog.SeverityNumberFatal,
+			},
+		},
+		{
+			"SeverityNumberFatal2",
+			"{\"severity\":\"ALERT\"}",
+			Log{
+				SeverityText:   "ALERT",
+				SeverityNumber: plog.SeverityNumberFatal2,
+			},
+		},
+		{
+			"SeverityNumberFatal4",
+			"{\"severity\":\"EMERGENCY\"}",
+			Log{
+				SeverityText:   "EMERGENCY",
+				SeverityNumber: plog.SeverityNumberFatal4,
+			},
+		},
 	}
-
 	logger, _ := zap.NewDevelopment()
-
+	var options testOptions
 	for _, tt := range tests {
-		var errs error
-		wantRes, wantLr, err := generateLog(t, tt.want)
-		errs = multierr.Append(errs, err)
+		t.Run(tt.scenario, func(t *testing.T) {
+			var errs error
+			wantRes, wantLr, err := generateLog(t, tt.want)
+			errs = multierr.Append(errs, err)
 
-		gotRes, gotLr, err := TranslateLogEntry(context.TODO(), logger, []byte(tt.input))
-		errs = multierr.Append(errs, err)
-		errs = multierr.Combine(errs, compareResources(wantRes, gotRes), compareLogRecords(wantLr, gotLr))
+			gotRes, gotLr, err := TranslateLogEntry(context.TODO(), logger, []byte(tt.input), options)
+			errs = multierr.Append(errs, err)
+			errs = multierr.Combine(errs, compareResources(wantRes, gotRes), compareLogRecords(wantLr, gotLr))
 
-		require.NoError(t, errs)
+			require.NoError(t, errs)
+		})
 	}
 }
 
@@ -94,8 +374,12 @@ func compareResources(expected, actual pcommon.Resource) error {
 func compareLogRecords(expected, actual plog.LogRecord) error {
 	return multierr.Combine(
 		compare("LogRecord.Timestamp", expected.Timestamp(), actual.Timestamp()),
+		compare("LogRecord.TraceID", expected.TraceID(), actual.TraceID()),
+		compare("LogRecord.SpanID", expected.SpanID(), actual.SpanID()),
 		compare("LogRecord.Attributes", expected.Attributes().AsRaw(), actual.Attributes().AsRaw()),
 		compare("LogRecord.Body", expected.Body().AsRaw(), actual.Body().AsRaw()),
+		compare("LogRecord.SeverityText", expected.SeverityText(), actual.SeverityText()),
+		compare("LogRecord.SeverityNumber", expected.SeverityNumber(), actual.SeverityNumber()),
 	)
 }
 
@@ -105,3 +389,41 @@ func compare(ty string, expected, actual any) error {
 	}
 	return nil
 }
+
+func TestTranslateJsonPayloadLogEntry(t *testing.T) {
+
+	tests := []struct {
+		scenario   string
+		input      string
+		wantAsJSON Log
+	}{
+		{
+			"ProtoPayload",
+			"{\"jsonPayload\": { \"foo\": \"bar\", \"other\": 42} }",
+			Log{
+				Body: map[string]any{
+					"foo":   string("bar"),
+					"other": float64(42),
+				},
+			},
+		},
+	}
+	logger, _ := zap.NewDevelopment()
+	for _, tt := range tests {
+		fn := func(t *testing.T, options testOptions, want Log) {
+			var errs error
+			wantRes, wantLr, err := generateLog(t, want)
+			errs = multierr.Append(errs, err)
+
+			gotRes, gotLr, err := TranslateLogEntry(context.TODO(), logger, []byte(tt.input), options)
+			errs = multierr.Append(errs, err)
+			errs = multierr.Combine(errs, compareResources(wantRes, gotRes), compareLogRecords(wantLr, gotLr))
+
+			require.NoError(t, errs)
+		}
+		var options testOptions
+		t.Run(tt.scenario, func(t *testing.T) {
+			fn(t, options, tt.wantAsJSON)
+		})
+	}
+}
diff --git a/receiver/googlecloudpubsubreceiver/internal/common_protos.go b/receiver/googlecloudpubsubreceiver/internal/log_proto_common.go
similarity index 100%
rename from receiver/googlecloudpubsubreceiver/internal/common_protos.go
rename to receiver/googlecloudpubsubreceiver/internal/log_proto_common.go
diff --git a/receiver/googlecloudpubsubreceiver/internal/log_proto_payload.go b/receiver/googlecloudpubsubreceiver/internal/log_proto_payload.go
new file mode 100644
index 0000000000000..1f24ccc84db31
--- /dev/null
+++ b/receiver/googlecloudpubsubreceiver/internal/log_proto_payload.go
@@ -0,0 +1,323 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudpubsubreceiver/internal"
+
+import (
+	"bytes"
+	stdjson "encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"google.golang.org/protobuf/reflect/protoreflect"
+	"google.golang.org/protobuf/reflect/protoregistry"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+func handleProtoPayloadAsProto(_ pcommon.Map, logRecord plog.LogRecord, _ pcommon.Map, _ string, value stdjson.RawMessage, _ Options) error {
+	err := translateInto(logRecord.Body().SetEmptyMap(), (&anypb.Any{}).ProtoReflect().Descriptor(), value)
+	return err
+}
+
+func handleProtoPayload(resourceAttributes pcommon.Map, logRecord plog.LogRecord, attributes pcommon.Map, key string, value stdjson.RawMessage, options Options) error {
+	return handleProtoPayloadAsProto(resourceAttributes, logRecord, attributes, key, value, options)
+}
+
+func (opts fieldTranslateOptions) translateValue(dst pcommon.Value, fd protoreflect.FieldDescriptor, src stdjson.RawMessage) error {
+	var err error
+	switch fd.Kind() {
+	case protoreflect.MessageKind:
+		msg := fd.Message()
+		switch fd.Message().FullName() {
+		case "google.protobuf.Duration", "google.protobuf.Timestamp":
+			// protojson represents both of these as strings
+			return translateStr(dst, src)
+		case "google.protobuf.Struct", "google.protobuf.Value":
+			return translateRaw(dst, src)
+		case
+			"google.protobuf.BoolValue",
+			"google.protobuf.BytesValue",
+			"google.protobuf.DoubleValue",
+			"google.protobuf.FloatValue",
+			"google.protobuf.Int32Value",
+			"google.protobuf.Int64Value",
+			"google.protobuf.StringValue",
+			"google.protobuf.UInt32Value",
+			"google.protobuf.UInt64Value":
+			// All the wrapper types have a single field with name
+			// `value` and field number 1, and are represented in
+			// protojson without the wrapping.
+			innerFd := fd.Message().Fields().ByNumber(1)
+			_ = opts.translateValue(dst, innerFd, src)
+		default:
+			var m pcommon.Map
+			switch dst.Type() {
+			case pcommon.ValueTypeMap:
+				m = dst.Map()
+			default:
+				m = dst.SetEmptyMap()
+			}
+			return translateInto(m, msg, src)
+		}
+	case protoreflect.EnumKind:
+		// protojson accepts either string name or enum int value; try both.
+		if translateStr(dst, src) == nil {
+			return nil
+		}
+
+		enum := fd.Enum()
+		var i int32
+		if err = json.Unmarshal(src, &i); err != nil {
+			return fmt.Errorf("wrong type for enum: %v", getType(src))
+		}
+		enumValue := enum.Values().ByNumber(protoreflect.EnumNumber(i))
+		if enumValue == nil {
+			return fmt.Errorf("%v has no enum value for %v", enum.FullName(), i)
+		}
+
+		dst.SetStr(string(enumValue.Name()))
+	case protoreflect.BoolKind:
+		var val bool
+		err = json.Unmarshal(src, &val)
+		if err != nil {
+			return err
+		}
+		dst.SetBool(val)
+	case protoreflect.Int32Kind,
+		protoreflect.Uint32Kind,
+		protoreflect.Sfixed32Kind,
+		protoreflect.Fixed32Kind,
+		protoreflect.Sint32Kind,
+		protoreflect.Int64Kind,
+		protoreflect.Uint64Kind,
+		protoreflect.Sfixed64Kind,
+		protoreflect.Fixed64Kind,
+		protoreflect.Sint64Kind:
+		// The protojson encoding accepts either string or number for
+		// integer types, so try both.
+		var val int64
+		if json.Unmarshal(src, &val) == nil {
+			dst.SetInt(val)
+			return nil
+		}
+
+		var s string
+		if err = json.Unmarshal(src, &s); err != nil {
+			return err
+		}
+		if val, err = strconv.ParseInt(s, 10, 64); err != nil {
+			return err
+		}
+		dst.SetInt(val)
+		return nil
+	case protoreflect.FloatKind, protoreflect.DoubleKind:
+		var val float64
+		err := json.Unmarshal(src, &val)
+		if err != nil {
+			return err
+		}
+		dst.SetDouble(val)
+		return nil
+	case protoreflect.BytesKind:
+		var val []byte
+		err := json.Unmarshal(src, &val)
+		if err != nil {
+			return err
+		}
+		dst.SetEmptyBytes().Append(val...)
+		return nil
+	case protoreflect.StringKind:
+		return translateStr(dst, src)
+	case protoreflect.GroupKind:
+		return errors.New("unexpected group")
+	default:
+		return errors.New("unknown field kind")
+	}
+	return nil
+}
+
+func (opts fieldTranslateOptions) translateList(dst pcommon.Slice, fd protoreflect.FieldDescriptor, src stdjson.RawMessage) error {
+	var msg []stdjson.RawMessage
+	if err := json.Unmarshal(src, &msg); err != nil {
+		return err
+	}
+
+	for _, v := range msg {
+		err := opts.translateValue(dst.AppendEmpty(), fd, v)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (opts fieldTranslateOptions) translateMap(dst pcommon.Map, fd protoreflect.FieldDescriptor, src stdjson.RawMessage) error {
+	var msg map[string]stdjson.RawMessage
+	if err := json.Unmarshal(src, &msg); err != nil {
+		return err
+	}
+	for k, v := range msg {
+		err := opts.translateValue(dst.PutEmpty(k), fd.MapValue(), v)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+func translateAny(dst pcommon.Map, src map[string]stdjson.RawMessage) error {
+	// protojson represents Any as the JSON representation of the actual
+	// message, plus a special @type field containing the type URL of the
+	// message.
+	typeURL, ok := src["@type"]
+	if !ok {
+		return errors.New("no @type member in Any message")
+	}
+	var typeURLStr string
+	if err := json.Unmarshal(typeURL, &typeURLStr); err != nil {
+		return err
+	}
+	delete(src, "@type")
+
+	msgType, err := protoregistry.GlobalTypes.FindMessageByURL(typeURLStr)
+	if errors.Is(err, protoregistry.NotFound) {
+		// If we don't have the type, we do a best-effort JSON decode;
+		// some ints might be floats or strings.
+		for k, v := range src {
+			var val any
+			err = json.Unmarshal(v, &val)
+			if err != nil {
+				return nil
+			}
+			_ = dst.PutEmpty(k).FromRaw(val)
+		}
+		return nil
+	}
+
+	err = translateInto(dst, msgType.Descriptor(), src)
+	if err != nil {
+		return err
+	}
+
+	dst.PutStr("@type", typeURLStr)
+	return nil
+}
+
+func translateProtoMessage(dst pcommon.Map, desc protoreflect.MessageDescriptor, src map[string]stdjson.RawMessage, opts fieldTranslateOptions) error {
+	if !opts.preserveDst {
+		dst.Clear()
+	}
+
+	// Handle well-known aggregate types.
+	switch desc.FullName() {
+	case "google.protobuf.Any":
+		return translateAny(dst, src)
+	case "google.protobuf.Struct":
+		for k, v := range src {
+			var val any
+			if err := json.Unmarshal(v, &val); err != nil {
+				return err
+			}
+			_ = dst.PutEmpty(k).FromRaw(val)
+		}
+		return nil
+	case "google.protobuf.Empty":
+		dst.Clear()
+		return nil
+	default:
+	}
+
+	for k, v := range src {
+		key := opts.mapKey(k)
+		fd := desc.Fields().ByJSONName(k)
+		if fd == nil {
+			return fmt.Errorf("%v has no known field with JSON name %v", desc.FullName(), k)
+		}
+		var err error
+		switch {
+		case fd.IsList():
+			err = opts.translateList(dst.PutEmptySlice(key), fd, v)
+		case fd.IsMap():
+			err = opts.translateMap(dst.PutEmptyMap(key), fd, v)
+		default:
+			err = opts.translateValue(dst.PutEmpty(key), fd, v)
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func translateInto(dst pcommon.Map, desc protoreflect.MessageDescriptor, src any, opts ...fieldTranslateFn) error {
+	var toTranslate map[string]stdjson.RawMessage
+
+	switch msg := src.(type) {
+	case stdjson.RawMessage:
+		err := json.Unmarshal(msg, &toTranslate)
+		if err != nil {
+			return err
+		}
+	case map[string]stdjson.RawMessage:
+		toTranslate = msg
+	}
+
+	options := fieldTranslateOptions{}
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	return translateProtoMessage(dst, desc, toTranslate, options)
+}
+
+func translateStr(dst pcommon.Value, src stdjson.RawMessage) error {
+	var val string
+	err := json.Unmarshal(src, &val)
+	if err != nil {
+		return err
+	}
+	dst.SetStr(val)
+	return nil
+}
+
+func translateRaw(dst pcommon.Value, src stdjson.RawMessage) error {
+	var val any
+	err := json.Unmarshal(src, &val)
+	if err != nil {
+		return err
+	}
+	_ = dst.FromRaw(val)
+	return nil
+}
+
+func getType(src stdjson.RawMessage) string {
+	dec := stdjson.NewDecoder(bytes.NewReader(src))
+	tok, err := dec.Token()
+	if err != nil {
+		return "invalid json"
+	}
+	switch t := tok.(type) {
+	case stdjson.Delim:
+		switch t {
+		case '[':
+			return "array"
+		case '{':
+			return "object"
+		default:
+			return "invalid json"
+		}
+	case bool:
+		return "bool"
+	case float64, stdjson.Number:
+		return "number"
+	case string:
+		return "string"
+	case nil:
+		return "null"
+	default:
+		return "unknown"
+	}
+}
diff --git a/receiver/googlecloudpubsubreceiver/internal/log_proto_payload_test.go b/receiver/googlecloudpubsubreceiver/internal/log_proto_payload_test.go
new file mode 100644
index 0000000000000..30b432a4d7508
--- /dev/null
+++ b/receiver/googlecloudpubsubreceiver/internal/log_proto_payload_test.go
@@ -0,0 +1,70 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/multierr"
+	"go.uber.org/zap"
+)
+
+func TestTranslateProtoPayloadLogEntry(t *testing.T) {
+
+	tests := []struct {
+		scenario    string
+		input       string
+		wantAsProto Log
+	}{
+		{
+			"ProtoPayload",
+			"{\"protoPayload\": {  \"@type\": \"type.googleapis.com/google.cloud.audit.AuditLog\",  \"status\": {},  \"authenticationInfo\": {    \"principalEmail\": \"foo@bar.iam.gserviceaccount.com\",    \"serviceAccountDelegationInfo\": [      {        \"firstPartyPrincipal\": {          \"principalEmail\": \"foo@bar.iam.gserviceaccount.com\"        }      }    ],    \"principalSubject\": \"serviceAccount:foo@bar.iam.gserviceaccount.com\"  },  \"requestMetadata\": {    \"callerIp\": \"21.128.18.1\",    \"callerSuppliedUserAgent\": \"grpc-go/1.62.2,gzip(gfe)\",    \"requestAttributes\": {},    \"destinationAttributes\": {}  },  \"serviceName\": \"cloudresourcemanager.googleapis.com\",  \"methodName\": \"SearchProjects\",  \"authorizationInfo\": [    {      \"resourceAttributes\": {}    }  ],  \"resourceName\": \"projects/project-id\",  \"request\": {    \"query\": \"state:active AND NOT projectID:sys-*\",    \"@type\": \"type.googleapis.com/google.cloud.resourcemanager.v3.SearchProjectsRequest\"  }} }",
+			Log{
+				Body: map[string]any{
+					"@type": string("type.googleapis.com/google.cloud.audit.AuditLog"),
+					"authenticationInfo": map[string]any{
+						"principalEmail":               string("foo@bar.iam.gserviceaccount.com"),
+						"principalSubject":             string("serviceAccount:foo@bar.iam.gserviceaccount.com"),
+						"serviceAccountDelegationInfo": []any{map[string]any{"firstPartyPrincipal": map[string]any{"principalEmail": "foo@bar.iam.gserviceaccount.com"}}},
+					},
+					"authorizationInfo": []any{map[string]any{"resourceAttributes": map[string]any{}}},
+					"methodName":        string("SearchProjects"),
+					"request": map[string]any{
+						"@type": string("type.googleapis.com/google.cloud.resourcemanager.v3.SearchProjectsRequest"),
+						"query": string("state:active AND NOT projectID:sys-*"),
+					},
+					"requestMetadata": map[string]any{
+						"callerIp":                string("21.128.18.1"),
+						"callerSuppliedUserAgent": string("grpc-go/1.62.2,gzip(gfe)"),
+						"destinationAttributes":   map[string]any{},
+						"requestAttributes":       map[string]any{},
+					},
+					"resourceName": string("projects/project-id"),
+					"serviceName":  string("cloudresourcemanager.googleapis.com"),
+					"status":       map[string]any{},
+				},
+			},
+		},
+	}
+	logger, _ := zap.NewDevelopment()
+	for _, tt := range tests {
+		fn := func(t *testing.T, options testOptions, want Log) {
+			var errs error
+			wantRes, wantLr, err := generateLog(t, want)
+			errs = multierr.Append(errs, err)
+
+			gotRes, gotLr, err := TranslateLogEntry(context.TODO(), logger, []byte(tt.input), options)
+			errs = multierr.Append(errs, err)
+			errs = multierr.Combine(errs, compareResources(wantRes, gotRes), compareLogRecords(wantLr, gotLr))
+
+			require.NoError(t, errs)
+		}
+		var options testOptions
+		t.Run(tt.scenario, func(t *testing.T) {
+			fn(t, options, tt.wantAsProto)
+		})
+	}
+}
diff --git a/receiver/googlecloudpubsubreceiver/internal/log_translate_options.go b/receiver/googlecloudpubsubreceiver/internal/log_translate_options.go
new file mode 100644
index 0000000000000..945110104f73c
--- /dev/null
+++ b/receiver/googlecloudpubsubreceiver/internal/log_translate_options.go
@@ -0,0 +1,40 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package internal // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/googlecloudpubsubreceiver/internal"
+
+import "github.com/iancoleman/strcase"
+
+type Options any
+
+type fieldTranslateOptions struct {
+	keyMappers  []func(string) string
+	preserveDst bool
+}
+
+type fieldTranslateFn func(*fieldTranslateOptions)
+
+func preserveDst(opts *fieldTranslateOptions) {
+	opts.preserveDst = true
+}
+
+func snakeifyKeys(opts *fieldTranslateOptions) {
+	opts.keyMappers = append(opts.keyMappers, func(s string) string {
+		return strcase.ToSnakeWithIgnore(s, ".")
+	})
+}
+
+func prefixKeys(p string) fieldTranslateFn {
+	return func(opts *fieldTranslateOptions) {
+		opts.keyMappers = append(opts.keyMappers, func(s string) string {
+			return p + s
+		})
+	}
+}
+
+func (opts fieldTranslateOptions) mapKey(s string) string {
+	for _, mapper := range opts.keyMappers {
+		s = mapper(s)
+	}
+	return s
+}
diff --git a/receiver/googlecloudpubsubreceiver/receiver.go b/receiver/googlecloudpubsubreceiver/receiver.go
index 504546201e2e0..e5963128d5bfa 100644
--- a/receiver/googlecloudpubsubreceiver/receiver.go
+++ b/receiver/googlecloudpubsubreceiver/receiver.go
@@ -152,7 +152,7 @@ func (receiver *pubsubReceiver) handleLogStrings(ctx context.Context, message *p
 }
 
 func (receiver *pubsubReceiver) handleCloudLoggingLogEntry(ctx context.Context, message *pubsubpb.ReceivedMessage) error {
-	resource, lr, err := internal.TranslateLogEntry(ctx, receiver.logger, message.Message.Data)
+	resource, lr, err := internal.TranslateLogEntry(ctx, receiver.logger, message.Message.Data, nil)
 
 	lr.SetObservedTimestamp(pcommon.NewTimestampFromTime(time.Now()))
 

From 61c9e9643fa72ab0ab8c7b17169dca1db80167fe Mon Sep 17 00:00:00 2001
From: Alex Van Boxel <alex@vanboxel.be>
Date: Fri, 31 May 2024 10:22:15 +0200
Subject: [PATCH 2/2] Apply suggestions from code review

Co-authored-by: Tiffany Hrabusa <30397949+tiffany76@users.noreply.github.com>
---
 receiver/googlecloudpubsubreceiver/README.md | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/receiver/googlecloudpubsubreceiver/README.md b/receiver/googlecloudpubsubreceiver/README.md
index 980115ed87f1d..27763cbc35696 100644
--- a/receiver/googlecloudpubsubreceiver/README.md
+++ b/receiver/googlecloudpubsubreceiver/README.md
@@ -93,12 +93,12 @@ attributes.content-type = "application/protobuf"
 
 **Status**: Experimental
 
-Google Cloud logging uses the [LogEntry] to cary log information. In this section the mapping of the fields to 
+Google Cloud logging uses the [LogEntry] to carry log information. In this section, the mapping of the fields to 
 OpenTelemetry fields and attributes are documented.
 
 ## Semantic Mapping
 
-Some of the attributes can be moved to OpenTelemetry Semantic Conventions. Note however that all the attributes are
+Some of the attributes can be moved to OpenTelemetry Semantic Conventions. Note, however, that all the attributes are
 considered experimental and are subject to change.
 
 | Field       | Type      | Description                            | Maps to Unified Model Field    |
@@ -107,7 +107,7 @@ considered experimental and are subject to change.
 
 ## Attribute and Field mapping
 
-The rest of the JSON body is either mapped to attributes with an `gcp` prefix or directly mapped to the LogRecord. 
+The rest of the JSON body is either mapped to attributes with a `gcp` prefix or directly mapped to the LogRecord. 
 
 | Field              | Type                     | Description                                                                                                      | Maps to Unified Model Field                   |
 |--------------------|--------------------------|------------------------------------------------------------------------------------------------------------------|-----------------------------------------------|
@@ -129,7 +129,7 @@ The rest of the JSON body is either mapped to attributes with an `gcp` prefix or
 
 ## Severity Mapping
 
-The severity is mapping from [Google Cloud Log Severity](https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity) to the OpenTelemetry Severity Number
+The severity is mapping from [Google Cloud Log Severity](https://cloud.google.com/logging/docs/reference/v2/rest/v2/LogEntry#LogSeverity) to the OpenTelemetry Severity Number.
 
 | CloudLog         | Severity Number  | CloudLog Description                                                                   |
 |------------------|------------------|----------------------------------------------------------------------------------------|