508 lines
14 KiB
Go
508 lines
14 KiB
Go
package otlp
|
|
|
|
import (
|
|
"crypto/subtle"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"git.apinb.com/ops/logs/internal/config"
|
|
"git.apinb.com/ops/logs/internal/impl"
|
|
"git.apinb.com/ops/logs/internal/models"
|
|
"github.com/gin-gonic/gin"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
const (
|
|
maxRecordsPerRequest = 10000
|
|
maxRequestBytes = 32 * 1024 * 1024
|
|
)
|
|
|
|
type anyValue struct {
|
|
StringValue *string `json:"stringValue"`
|
|
BoolValue *bool `json:"boolValue"`
|
|
IntValue json.RawMessage `json:"intValue"`
|
|
DoubleValue *float64 `json:"doubleValue"`
|
|
BytesValue string `json:"bytesValue"`
|
|
ArrayValue *arrayValue `json:"arrayValue"`
|
|
KVListValue *keyValueList `json:"kvlistValue"`
|
|
}
|
|
|
|
type arrayValue struct {
|
|
Values []anyValue `json:"values"`
|
|
}
|
|
|
|
type keyValueList struct {
|
|
Values []keyValue `json:"values"`
|
|
}
|
|
|
|
type keyValue struct {
|
|
Key string `json:"key"`
|
|
Value anyValue `json:"value"`
|
|
}
|
|
|
|
type resource struct {
|
|
Attributes []keyValue `json:"attributes"`
|
|
}
|
|
|
|
type traceRequest struct {
|
|
ResourceSpans []struct {
|
|
Resource resource `json:"resource"`
|
|
ScopeSpans []struct {
|
|
Spans []spanPayload `json:"spans"`
|
|
} `json:"scopeSpans"`
|
|
} `json:"resourceSpans"`
|
|
}
|
|
|
|
type spanPayload struct {
|
|
TraceID string `json:"traceId"`
|
|
SpanID string `json:"spanId"`
|
|
ParentSpanID string `json:"parentSpanId"`
|
|
Name string `json:"name"`
|
|
Kind json.RawMessage `json:"kind"`
|
|
StartTimeUnixNano json.RawMessage `json:"startTimeUnixNano"`
|
|
EndTimeUnixNano json.RawMessage `json:"endTimeUnixNano"`
|
|
Attributes []keyValue `json:"attributes"`
|
|
Status struct {
|
|
Code json.RawMessage `json:"code"`
|
|
Message string `json:"message"`
|
|
} `json:"status"`
|
|
}
|
|
|
|
type logRequest struct {
|
|
ResourceLogs []struct {
|
|
Resource resource `json:"resource"`
|
|
ScopeLogs []struct {
|
|
LogRecords []logPayload `json:"logRecords"`
|
|
} `json:"scopeLogs"`
|
|
} `json:"resourceLogs"`
|
|
}
|
|
|
|
type logPayload struct {
|
|
TimeUnixNano json.RawMessage `json:"timeUnixNano"`
|
|
ObservedTimeUnixNano json.RawMessage `json:"observedTimeUnixNano"`
|
|
SeverityText string `json:"severityText"`
|
|
SeverityNumber json.RawMessage `json:"severityNumber"`
|
|
Body anyValue `json:"body"`
|
|
Attributes []keyValue `json:"attributes"`
|
|
TraceID string `json:"traceId"`
|
|
SpanID string `json:"spanId"`
|
|
}
|
|
|
|
func ReceiveTraces(ctx *gin.Context) {
|
|
if !authorize(ctx) {
|
|
return
|
|
}
|
|
ctx.Request.Body = http.MaxBytesReader(ctx.Writer, ctx.Request.Body, maxRequestBytes)
|
|
var request traceRequest
|
|
if err := ctx.ShouldBindJSON(&request); err != nil {
|
|
ctx.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": "OTLP Trace 请求无效"})
|
|
return
|
|
}
|
|
rows, rejected, message := decodeSpans(request)
|
|
if len(rows) > 0 {
|
|
if err := impl.DBService.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "trace_id"}, {Name: "span_id"}}, DoNothing: true}).
|
|
CreateInBatches(rows, 500).Error; err != nil {
|
|
ctx.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{"error": "Trace 入库失败"})
|
|
return
|
|
}
|
|
}
|
|
ctx.JSON(http.StatusOK, gin.H{"partialSuccess": gin.H{"rejectedSpans": rejected, "errorMessage": message}})
|
|
}
|
|
|
|
func ReceiveLogs(ctx *gin.Context) {
|
|
if !authorize(ctx) {
|
|
return
|
|
}
|
|
ctx.Request.Body = http.MaxBytesReader(ctx.Writer, ctx.Request.Body, maxRequestBytes)
|
|
var request logRequest
|
|
if err := ctx.ShouldBindJSON(&request); err != nil {
|
|
ctx.AbortWithStatusJSON(http.StatusBadRequest, gin.H{"error": "OTLP Log 请求无效"})
|
|
return
|
|
}
|
|
rows, rejected, message := decodeLogs(request)
|
|
if len(rows) > 0 {
|
|
if err := impl.DBService.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "ingest_key"}}, DoNothing: true}).
|
|
CreateInBatches(rows, 500).Error; err != nil {
|
|
ctx.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{"error": "Log 入库失败"})
|
|
return
|
|
}
|
|
}
|
|
ctx.JSON(http.StatusOK, gin.H{"partialSuccess": gin.H{"rejectedLogRecords": rejected, "errorMessage": message}})
|
|
}
|
|
|
|
func authorize(ctx *gin.Context) bool {
|
|
provided := strings.TrimSpace(ctx.GetHeader("X-OTLP-Internal-Key"))
|
|
expected := config.Spec.OTLP.InternalKey
|
|
if provided == "" || len(provided) != len(expected) || subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) != 1 {
|
|
ctx.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "OTLP 接入密钥无效"})
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
func decodeSpans(request traceRequest) ([]models.TraceSpan, int, string) {
|
|
rows := make([]models.TraceSpan, 0)
|
|
rejected := 0
|
|
for _, group := range request.ResourceSpans {
|
|
resourceAttrs := attributesMap(group.Resource.Attributes)
|
|
resourceJSON, _ := json.Marshal(resourceAttrs)
|
|
for _, scope := range group.ScopeSpans {
|
|
for _, input := range scope.Spans {
|
|
if len(rows)+rejected >= maxRecordsPerRequest {
|
|
rejected++
|
|
continue
|
|
}
|
|
row, err := spanModel(input, resourceAttrs, string(resourceJSON))
|
|
if err != nil {
|
|
rejected++
|
|
continue
|
|
}
|
|
rows = append(rows, row)
|
|
}
|
|
}
|
|
}
|
|
message := ""
|
|
if rejected > 0 {
|
|
message = "部分 Span 因标识或时间字段无效被拒绝"
|
|
}
|
|
return rows, rejected, message
|
|
}
|
|
|
|
func spanModel(input spanPayload, resourceAttrs map[string]interface{}, resourceJSON string) (models.TraceSpan, error) {
|
|
traceID, err := normalizeID(input.TraceID, 16)
|
|
if err != nil {
|
|
return models.TraceSpan{}, err
|
|
}
|
|
spanID, err := normalizeID(input.SpanID, 8)
|
|
if err != nil {
|
|
return models.TraceSpan{}, err
|
|
}
|
|
parentID := ""
|
|
if strings.TrimSpace(input.ParentSpanID) != "" {
|
|
parentID, err = normalizeID(input.ParentSpanID, 8)
|
|
if err != nil {
|
|
return models.TraceSpan{}, err
|
|
}
|
|
}
|
|
start, err := nanoTime(input.StartTimeUnixNano)
|
|
if err != nil {
|
|
return models.TraceSpan{}, err
|
|
}
|
|
end, err := nanoTime(input.EndTimeUnixNano)
|
|
if err != nil || end.Before(start) {
|
|
return models.TraceSpan{}, fmt.Errorf("Span 时间无效")
|
|
}
|
|
attrs := attributesMap(input.Attributes)
|
|
attrsJSON, _ := json.Marshal(attrs)
|
|
serviceName := textAttribute(resourceAttrs, "service.name")
|
|
if serviceName == "" {
|
|
serviceName = "unknown_service"
|
|
}
|
|
serviceName = limitCharacters(serviceName, 255)
|
|
operation := strings.TrimSpace(input.Name)
|
|
if operation == "" {
|
|
operation = "unnamed"
|
|
}
|
|
operation = limitCharacters(operation, 512)
|
|
statusCode := statusText(input.Status.Code)
|
|
if httpStatus := intAttribute(attrs, "http.response.status_code", "http.status_code"); statusCode == "unset" && httpStatus >= 500 {
|
|
statusCode = "error"
|
|
}
|
|
now := time.Now().UTC()
|
|
return models.TraceSpan{
|
|
TraceID: traceID, SpanID: spanID, ParentSpanID: parentID, ServiceName: serviceName,
|
|
OperationName: operation, SpanKind: spanKindText(input.Kind), StartTime: start, EndTime: end,
|
|
DurationMs: float64(end.Sub(start).Nanoseconds()) / 1e6, StatusCode: statusCode,
|
|
StatusMessage: strings.TrimSpace(input.Status.Message), BusinessSystemID: uintAttribute(resourceAttrs, attrs, "business.system.id", "business_system_id"),
|
|
ResourceUID: limitCharacters(firstTextAttribute(resourceAttrs, attrs, "resource.uid", "resource_uid"), 255),
|
|
HTTPMethod: limitCharacters(firstTextAttribute(attrs, nil, "http.request.method", "http.method"), 16),
|
|
HTTPRoute: limitCharacters(firstTextAttribute(attrs, nil, "http.route", "url.path", "http.target"), 1024),
|
|
HTTPStatusCode: intAttribute(attrs, "http.response.status_code", "http.status_code"),
|
|
Attributes: string(attrsJSON), ResourceAttrs: resourceJSON, IngestedAt: &now,
|
|
}, nil
|
|
}
|
|
|
|
func decodeLogs(request logRequest) ([]models.LogEvent, int, string) {
|
|
rows := make([]models.LogEvent, 0)
|
|
rejected := 0
|
|
for _, group := range request.ResourceLogs {
|
|
resourceAttrs := attributesMap(group.Resource.Attributes)
|
|
for _, scope := range group.ScopeLogs {
|
|
for _, input := range scope.LogRecords {
|
|
if len(rows)+rejected >= maxRecordsPerRequest {
|
|
rejected++
|
|
continue
|
|
}
|
|
row, err := logModel(input, resourceAttrs)
|
|
if err != nil {
|
|
rejected++
|
|
continue
|
|
}
|
|
rows = append(rows, row)
|
|
}
|
|
}
|
|
}
|
|
message := ""
|
|
if rejected > 0 {
|
|
message = "部分 LogRecord 因标识或内容无效被拒绝"
|
|
}
|
|
return rows, rejected, message
|
|
}
|
|
|
|
func logModel(input logPayload, resourceAttrs map[string]interface{}) (models.LogEvent, error) {
|
|
attrs := attributesMap(input.Attributes)
|
|
detailJSON, _ := json.Marshal(attrs)
|
|
body := scalarText(anyValueToInterface(input.Body))
|
|
if body == "" {
|
|
return models.LogEvent{}, fmt.Errorf("日志正文为空")
|
|
}
|
|
if len(body) > 64*1024 {
|
|
body = limitUTF8Bytes(body, 64*1024)
|
|
}
|
|
traceID := ""
|
|
spanID := ""
|
|
var err error
|
|
if strings.TrimSpace(input.TraceID) != "" {
|
|
traceID, err = normalizeID(input.TraceID, 16)
|
|
if err != nil {
|
|
return models.LogEvent{}, err
|
|
}
|
|
}
|
|
if strings.TrimSpace(input.SpanID) != "" {
|
|
spanID, err = normalizeID(input.SpanID, 8)
|
|
if err != nil {
|
|
return models.LogEvent{}, err
|
|
}
|
|
}
|
|
serviceName := limitCharacters(textAttribute(resourceAttrs, "service.name"), 512)
|
|
resourceUID := limitCharacters(firstTextAttribute(resourceAttrs, attrs, "resource.uid", "resource_uid"), 255)
|
|
var ingestKey *string
|
|
if value := limitCharacters(firstTextAttribute(attrs, nil, "log.record.uid"), 160); value != "" {
|
|
ingestKey = &value
|
|
}
|
|
return models.LogEvent{
|
|
IngestKey: ingestKey, CreatedAt: timeFromLog(input), SourceKind: "otlp", RawPayload: body, NormalizedSummary: body,
|
|
NormalizedDetail: string(detailJSON), DeviceName: serviceName, ResourceType: "service", ResourceUID: resourceUID,
|
|
ResourceID: resourceUID, ResourceName: serviceName, BusinessSystemID: uintAttribute(resourceAttrs, attrs, "business.system.id", "business_system_id"),
|
|
TraceID: traceID, SpanID: spanID, MatchMethod: "otlp", DispatchStatus: "not_applicable",
|
|
SeverityCode: limitCharacters(severityText(input.SeverityText, input.SeverityNumber), 32),
|
|
}, nil
|
|
}
|
|
|
|
func limitCharacters(value string, maximum int) string {
|
|
characters := []rune(value)
|
|
if len(characters) <= maximum {
|
|
return value
|
|
}
|
|
return string(characters[:maximum])
|
|
}
|
|
|
|
func limitUTF8Bytes(value string, maximum int) string {
|
|
if len(value) <= maximum {
|
|
return value
|
|
}
|
|
end := maximum
|
|
for end > 0 && !utf8.ValidString(value[:end]) {
|
|
end--
|
|
}
|
|
return value[:end]
|
|
}
|
|
|
|
func attributesMap(values []keyValue) map[string]interface{} {
|
|
result := make(map[string]interface{}, len(values))
|
|
for _, item := range values {
|
|
if key := strings.TrimSpace(item.Key); key != "" {
|
|
result[key] = anyValueToInterface(item.Value)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func anyValueToInterface(value anyValue) interface{} {
|
|
if value.StringValue != nil {
|
|
return *value.StringValue
|
|
}
|
|
if value.BoolValue != nil {
|
|
return *value.BoolValue
|
|
}
|
|
if value.DoubleValue != nil {
|
|
return *value.DoubleValue
|
|
}
|
|
if len(value.IntValue) > 0 {
|
|
var number json.Number
|
|
if json.Unmarshal(value.IntValue, &number) == nil {
|
|
return number
|
|
}
|
|
var text string
|
|
if json.Unmarshal(value.IntValue, &text) == nil {
|
|
return text
|
|
}
|
|
}
|
|
if value.BytesValue != "" {
|
|
return value.BytesValue
|
|
}
|
|
if value.ArrayValue != nil {
|
|
items := make([]interface{}, 0, len(value.ArrayValue.Values))
|
|
for _, item := range value.ArrayValue.Values {
|
|
items = append(items, anyValueToInterface(item))
|
|
}
|
|
return items
|
|
}
|
|
if value.KVListValue != nil {
|
|
return attributesMap(value.KVListValue.Values)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func normalizeID(value string, size int) (string, error) {
|
|
value = strings.TrimSpace(value)
|
|
if raw, err := hex.DecodeString(value); err == nil && len(raw) == size {
|
|
return strings.ToLower(value), nil
|
|
}
|
|
raw, err := base64.StdEncoding.DecodeString(value)
|
|
if err != nil || len(raw) != size {
|
|
return "", fmt.Errorf("OTLP 标识无效")
|
|
}
|
|
return hex.EncodeToString(raw), nil
|
|
}
|
|
|
|
func nanoTime(raw json.RawMessage) (time.Time, error) {
|
|
value := strings.Trim(strings.TrimSpace(string(raw)), `"`)
|
|
nanos, err := strconv.ParseInt(value, 10, 64)
|
|
if err != nil || nanos <= 0 {
|
|
return time.Time{}, fmt.Errorf("OTLP 时间无效")
|
|
}
|
|
return time.Unix(0, nanos).UTC(), nil
|
|
}
|
|
|
|
func enumValue(raw json.RawMessage) string {
|
|
return strings.Trim(strings.TrimSpace(string(raw)), `"`)
|
|
}
|
|
|
|
func spanKindText(raw json.RawMessage) string {
|
|
switch strings.ToUpper(enumValue(raw)) {
|
|
case "2", "SPAN_KIND_SERVER":
|
|
return "server"
|
|
case "3", "SPAN_KIND_CLIENT":
|
|
return "client"
|
|
case "4", "SPAN_KIND_PRODUCER":
|
|
return "producer"
|
|
case "5", "SPAN_KIND_CONSUMER":
|
|
return "consumer"
|
|
default:
|
|
return "internal"
|
|
}
|
|
}
|
|
|
|
func statusText(raw json.RawMessage) string {
|
|
switch strings.ToUpper(enumValue(raw)) {
|
|
case "1", "STATUS_CODE_OK":
|
|
return "ok"
|
|
case "2", "STATUS_CODE_ERROR":
|
|
return "error"
|
|
default:
|
|
return "unset"
|
|
}
|
|
}
|
|
|
|
func severityText(text string, raw json.RawMessage) string {
|
|
text = strings.ToLower(strings.TrimSpace(text))
|
|
if text != "" {
|
|
if strings.Contains(text, "fatal") {
|
|
return "critical"
|
|
}
|
|
if strings.Contains(text, "error") {
|
|
return "major"
|
|
}
|
|
if strings.Contains(text, "warn") {
|
|
return "warning"
|
|
}
|
|
return "info"
|
|
}
|
|
value, _ := strconv.Atoi(enumValue(raw))
|
|
if value >= 21 {
|
|
return "critical"
|
|
}
|
|
if value >= 17 {
|
|
return "major"
|
|
}
|
|
if value >= 13 {
|
|
return "warning"
|
|
}
|
|
return "info"
|
|
}
|
|
|
|
func timeFromLog(input logPayload) time.Time {
|
|
if value, err := nanoTime(input.TimeUnixNano); err == nil {
|
|
return value
|
|
}
|
|
if value, err := nanoTime(input.ObservedTimeUnixNano); err == nil {
|
|
return value
|
|
}
|
|
return time.Now().UTC()
|
|
}
|
|
|
|
func textAttribute(values map[string]interface{}, key string) string {
|
|
return scalarText(values[key])
|
|
}
|
|
|
|
func firstTextAttribute(primary, secondary map[string]interface{}, keys ...string) string {
|
|
for _, key := range keys {
|
|
if primary != nil {
|
|
if value := scalarText(primary[key]); value != "" {
|
|
return value
|
|
}
|
|
}
|
|
if secondary != nil {
|
|
if value := scalarText(secondary[key]); value != "" {
|
|
return value
|
|
}
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func intAttribute(values map[string]interface{}, keys ...string) int {
|
|
for _, key := range keys {
|
|
value, err := strconv.Atoi(scalarText(values[key]))
|
|
if err == nil && value != 0 {
|
|
return value
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func uintAttribute(primary, secondary map[string]interface{}, keys ...string) *uint {
|
|
value := firstTextAttribute(primary, secondary, keys...)
|
|
parsed, err := strconv.ParseUint(value, 10, 32)
|
|
if err != nil || parsed == 0 {
|
|
return nil
|
|
}
|
|
result := uint(parsed)
|
|
return &result
|
|
}
|
|
|
|
func scalarText(value interface{}) string {
|
|
switch typed := value.(type) {
|
|
case string:
|
|
return typed
|
|
case json.Number:
|
|
return typed.String()
|
|
case float64, bool:
|
|
return fmt.Sprint(typed)
|
|
default:
|
|
if typed == nil {
|
|
return ""
|
|
}
|
|
encoded, _ := json.Marshal(typed)
|
|
return string(encoded)
|
|
}
|
|
}
|