add opentelemetry output

codec/oltp_encoder.go

@@ -0,0 +1,113 @@
+package codec
+
+import (
+	"encoding/hex"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/plog/plogotlp"
+	"strings"
+	"time"
+)
+
+const (
+	resourcePrefix       = "oltp_resource_"
+	scopePrefix          = "oltp_scope_"
+	scopeName            = "oltp_scope_name"
+	scopeVersion         = "oltp_scope_version"
+	scopeSchemaUrl       = "oltp_scope_schemaurl"
+	recordPrefix         = "oltp_logrecords_"
+	recordTime           = "oltp_logrecords_time"
+	recordSeverity       = "oltp_logrecords_severity"
+	recordSeverityNumber = "oltp_logrecords_severitynumber"
+	recordFlags          = "oltp_logrecords_flags"
+	recordTraceId        = "oltp_logrecords_traceid"
+	recordSpanId         = "oltp_logrecords_spanid"
+	recordBody           = "oltp_logrecords_body"
+)
+
+type OltpEncoder struct{}
+
+func (o *OltpEncoder) Encode(event map[string]interface{}) (plogotlp.ExportRequest, error) {
+	logs := plog.NewLogs()
+	rsLogs := logs.ResourceLogs().AppendEmpty()
+	scopeLog := rsLogs.ScopeLogs().AppendEmpty()
+	logRecord := scopeLog.LogRecords().AppendEmpty()
+	logRecord.SetObservedTimestamp(pcommon.Timestamp(time.Now().UnixNano()))
+
+	for k, v := range event {
+		// TODO: just support string, do no not use reflect
+		switch k {
+		case scopeName:
+			if _, ok := v.(string); ok {
+				scopeLog.Scope().SetName(v.(string))
+			}
+		case scopeVersion:
+			if _, ok := v.(string); ok {
+				scopeLog.Scope().SetVersion(v.(string))
+			}
+		case scopeSchemaUrl:
+			if _, ok := v.(string); ok {
+				scopeLog.SetSchemaUrl(v.(string))
+			}
+		case recordTime:
+			if _, ok := v.(uint64); ok {
+				logRecord.SetTimestamp(pcommon.Timestamp(v.(uint64)))
+			}
+		case recordSeverity:
+			if _, ok := v.(string); ok {
+				logRecord.SetSeverityText(v.(string))
+			}
+		case recordSeverityNumber:
+			if _, ok := v.(int32); ok {
+				logRecord.SetSeverityNumber(plog.SeverityNumber(v.(int32)))
+			}
+		case recordFlags:
+			if _, ok := v.(uint32); ok {
+				logRecord.SetFlags(plog.LogRecordFlags(v.(uint32)))
+			}
+		case recordTraceId:
+			if _, ok := v.(string); ok {
+				bytes, err := hex.DecodeString(v.(string))
+				if err != nil {
+					continue
+				}
+				if len(bytes) == 16 {
+					var traceId [16]byte
+					copy(traceId[:], bytes)
+					logRecord.SetTraceID(traceId)
+				}
+			}
+		case recordSpanId:
+			if _, ok := v.(string); ok {
+				bytes, err := hex.DecodeString(v.(string))
+				if err != nil {
+					continue
+				}
+				if len(bytes) == 8 {
+					var spanId [8]byte
+					copy(spanId[:], bytes)
+					logRecord.SetSpanID(spanId)
+				}
+			}
+		case recordBody:
+			if _, ok := v.(string); ok {
+				logRecord.Body().SetStr(v.(string))
+			}
+		default:
+			if strings.HasPrefix(k, resourcePrefix) {
+				if _, ok := v.(string); ok {
+					rsLogs.Resource().Attributes().PutStr(k[len(resourcePrefix):], v.(string))
+				}
+			} else if strings.HasPrefix(k, scopePrefix) {
+				if _, ok := v.(string); ok {
+					scopeLog.Scope().Attributes().PutStr(k[len(scopePrefix):], v.(string))
+				}
+			} else if strings.HasPrefix(k, recordPrefix) {
+				if _, ok := v.(string); ok {
+					logRecord.Attributes().PutStr(k[len(recordPrefix):], v.(string))
+				}
+			}
+		}
+	}
+	return plogotlp.NewExportRequestFromLogs(logs), nil
+}

go.mod

@@ -1,30 +1,63 @@
 module github.com/childe/gohangout
 
-go 1.13
+go 1.18
 
 require (
 	github.com/ClickHouse/clickhouse-go v1.5.4
 	github.com/Masterminds/sprig/v3 v3.2.2
-	github.com/bkaradzic/go-lz4 v1.0.1-0.20160924222819-7224d8d8f27e // indirect
 	github.com/childe/healer v0.5.5
 	github.com/fsnotify/fsnotify v1.5.1
-	github.com/golang/snappy v0.0.4 // indirect
-	github.com/google/uuid v1.3.0 // indirect
-	github.com/huandu/xstrings v1.3.2 // indirect
-	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/ipipdotnet/datx-go v0.0.0-20181123035258-af996d4701a0
 	github.com/ipipdotnet/ipdb-go v1.3.1
 	github.com/json-iterator/go v1.1.12
 	github.com/magiconair/properties v1.8.6
-	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852
 	github.com/prometheus/client_golang v1.12.1
-	github.com/shopspring/decimal v1.3.1 // indirect
 	github.com/spf13/cast v1.4.1
 	github.com/yalp/jsonpath v0.0.0-20180802001716-5cc68e5049a0
+	github.com/youmark/pkcs8 v0.0.0-20201027041543-1326539a0a0a
+	go.opentelemetry.io/collector/pdata v0.66.0
+	google.golang.org/grpc v1.57.0
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/klog/v2 v2.100.1
 )
 
+require (
+	github.com/Masterminds/goutils v1.1.1 // indirect
+	github.com/Masterminds/semver/v3 v3.1.1 // indirect
+	github.com/aviddiviner/go-murmur v0.0.0-20150519214947-b9740d71e571 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/bkaradzic/go-lz4 v1.0.1-0.20160924222819-7224d8d8f27e // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect
+	github.com/go-logr/logr v1.2.0 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/glog v1.1.0 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/golang/snappy v0.0.2 // indirect
+	github.com/google/uuid v1.3.1 // indirect
+	github.com/huandu/xstrings v1.3.2 // indirect
+	github.com/imdario/mergo v0.3.12 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.32.1 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/shopspring/decimal v1.3.1 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.8.0 // indirect
+	golang.org/x/crypto v0.12.0 // indirect
+	golang.org/x/net v0.14.0 // indirect
+	golang.org/x/sys v0.11.0 // indirect
+	golang.org/x/text v0.12.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20230822172742-b8732ec3820d // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
+)
+
 replace github.com/spf13/cast v1.4.1 => github.com/oasisprotocol/cast v0.0.0-20220606122631-eba453e69641