tmp for share

cmd/ingester/app/builder/builder.go

@@ -39,6 +39,8 @@ func CreateConsumer(logger *zap.Logger, metricsFactory metrics.Factory, spanWrit
 		unmarshaller = kafka.NewProtobufUnmarshaller()
 	case kafka.EncodingZipkinThrift:
 		unmarshaller = kafka.NewZipkinThriftUnmarshaller()
+	case kafka.EncodingOtlpProto:
+		unmarshaller = kafka.NewOtlpProtoUnmarshaller()
 	default:
 		return nil, fmt.Errorf(`encoding '%s' not recognised, use one of ("%s")`,
 			options.Encoding, strings.Join(kafka.AllEncodings, "\", \""))

plugin/storage/kafka/options.go

@@ -34,6 +34,7 @@ const (
 	EncodingProto = "protobuf"
 	// EncodingZipkinThrift is used for spans encoded as Zipkin Thrift.
 	EncodingZipkinThrift = "zipkin-thrift"
+	EncodingOtlpJSON     = "otlp-json"
 
 	configPrefix           = "kafka.producer"
 	suffixBrokers          = ".brokers"

plugin/storage/kafka/unmarshaller.go

@@ -20,9 +20,11 @@ import (
 
 	"github.com/gogo/protobuf/jsonpb"
 	"github.com/gogo/protobuf/proto"
+	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
 
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/model/converter/thrift/zipkin"
+	otlp2jaeger "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
 )
 
 // Unmarshaller decodes a byte array to a span
@@ -80,3 +82,29 @@ func (h *ZipkinThriftUnmarshaller) Unmarshal(msg []byte) (*model.Span, error) {
 	}
 	return mSpans[0], err
 }
+
+type OtlpJSONUnmarshaller struct{}
+
+func NewOtlpJSONUnmarshaller() *OtlpJSONUnmarshaller {
+	return &OtlpJSONUnmarshaller{}
+}
+
+func (OtlpJSONUnmarshaller) Unmarshal(buf []byte) (*model.Span, error) {
+	req := ptraceotlp.NewExportRequest()
+	err := req.UnmarshalJSON(buf)
+	if err != nil {
+		return nil, err
+	}
+
+	batch, err := otlp2jaeger.ProtoFromTraces(req.Traces())
+	if err != nil {
+		return nil, err
+	}
+
+	//  check span & raise err
+	// if span {
+
+	// }
+
+	return batch[0].Spans[0], nil
+}