mirror of
https://github.com/hanzoai/otel-collector.git
synced 2026-11-03 02:20:04 +00:00
zeekayandHanzo Dev
c16ed9fe26
Modules declare go 1.27.1 and the build images follow. go fix applied the modernizers; omitzero is not applied, since it changes what the JSON says. Co-authored-by: Hanzo Dev <[email protected]>
86 lines
2.1 KiB
Go
86 lines
2.1 KiB
Go
// Copyright The OpenTelemetry Authors
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
package o11ykafkaexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/kafkaexporter"
|
|
|
|
import (
|
|
"errors"
|
|
|
|
"github.com/goccy/go-json"
|
|
|
|
"github.com/Shopify/sarama"
|
|
"go.opentelemetry.io/collector/pdata/pcommon"
|
|
"go.opentelemetry.io/collector/pdata/plog"
|
|
)
|
|
|
|
var errUnsupported = errors.New("unsupported serialization")
|
|
|
|
type rawMarshaler struct {
|
|
}
|
|
|
|
func newRawMarshaler() rawMarshaler {
|
|
return rawMarshaler{}
|
|
}
|
|
|
|
func (r rawMarshaler) Marshal(logs plog.Logs, topic string) ([]*sarama.ProducerMessage, error) {
|
|
var messages []*sarama.ProducerMessage
|
|
for i := 0; i < logs.ResourceLogs().Len(); i++ {
|
|
rl := logs.ResourceLogs().At(i)
|
|
for j := 0; j < rl.ScopeLogs().Len(); j++ {
|
|
sl := rl.ScopeLogs().At(j)
|
|
for k := 0; k < sl.LogRecords().Len(); k++ {
|
|
lr := sl.LogRecords().At(k)
|
|
b, err := r.logBodyAsBytes(lr.Body())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(b) == 0 {
|
|
continue
|
|
}
|
|
|
|
messages = append(messages, &sarama.ProducerMessage{
|
|
Topic: topic,
|
|
Value: sarama.ByteEncoder(b),
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
return messages, nil
|
|
}
|
|
|
|
func (r rawMarshaler) logBodyAsBytes(value pcommon.Value) ([]byte, error) {
|
|
switch value.Type() {
|
|
case pcommon.ValueTypeStr:
|
|
return r.interfaceAsBytes(value.Str())
|
|
case pcommon.ValueTypeBytes:
|
|
return value.Bytes().AsRaw(), nil
|
|
case pcommon.ValueTypeBool:
|
|
return r.interfaceAsBytes(value.Bool())
|
|
case pcommon.ValueTypeDouble:
|
|
return r.interfaceAsBytes(value.Double())
|
|
case pcommon.ValueTypeInt:
|
|
return r.interfaceAsBytes(value.Int())
|
|
case pcommon.ValueTypeEmpty:
|
|
return []byte{}, nil
|
|
case pcommon.ValueTypeSlice:
|
|
return r.interfaceAsBytes(value.Slice().AsRaw())
|
|
case pcommon.ValueTypeMap:
|
|
return r.interfaceAsBytes(value.Map().AsRaw())
|
|
default:
|
|
return nil, errUnsupported
|
|
}
|
|
}
|
|
|
|
func (r rawMarshaler) interfaceAsBytes(value any) ([]byte, error) {
|
|
if value == nil {
|
|
return []byte{}, nil
|
|
}
|
|
res, err := json.Marshal(value)
|
|
return res, err
|
|
}
|
|
|
|
func (r rawMarshaler) Encoding() string {
|
|
return "raw"
|
|
}
|