finish service log module

2026-06-04 10:13:53 +08:00 · 2025-04-29 19:08:02 +08:00
parent a3bebde83c
commit 604a8312ef
15 changed files with 346 additions and 38 deletions
@@ -62,6 +62,7 @@ var logFormatter = map[string]interface{}{
 		"$proxy_host",
 		"$proxy_header",
 		"$proxy_addr",
+		"$response_header",
 		"$response_headers",
 		"$status",
 		"$content_type",
@@ -259,3 +259,42 @@ type RestLogItem struct {
 	ResponseTime string         `json:"response_time"`
 	Traffic      string         `json:"traffic"`
 }
+
+type RestLogInfo struct {
+	Id               string         `json:"id"`
+	API              auto.Label     `json:"api" aolabel:"api"`
+	Consumer         auto.Label     `json:"consumer" aolabel:"service"`
+	IsSystemConsumer bool           `json:"is_system_consumer"`
+	Status           int64          `json:"status"`
+	Ip               string         `json:"ip"`
+	ResponseTime     string         `json:"response_time"`
+	Traffic          string         `json:"traffic"`
+	LogTime          auto.TimeLabel `json:"log_time"`
+	Request          OriginRequest  `json:"request"`
+	Response         OriginRequest  `json:"response"`
+}
+
+type AILogInfo struct {
+	Id               string          `json:"id"`
+	API              auto.Label      `json:"api" aolabel:"api"`
+	Consumer         auto.Label      `json:"consumer" aolabel:"service"`
+	IsSystemConsumer bool            `json:"is_system_consumer"`
+	Status           int64           `json:"status"`
+	Ip               string          `json:"ip"`
+	Provider         auto.Label      `json:"provider" aolabel:"ai_provider"`
+	Model            string          `json:"model"`
+	LogTime          auto.TimeLabel  `json:"log_time"`
+	Request          OriginAIRequest `json:"request"`
+	Response         OriginAIRequest `json:"response"`
+}
+
+type OriginRequest struct {
+	Header string `json:"header"`
+	Origin string `json:"origin"`
+}
+
+type OriginAIRequest struct {
+	OriginRequest
+	Body  string `json:"body"`
+	Token int64  `json:"token"`
+}
@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/url"
 	"sort"
 	"strings"
 	"time"
@@ -95,6 +96,111 @@ type imlServiceModule struct {
 	transaction store.ITransaction `autowired:""`
 }

+func formatHeader(header string) string {
+	result, err := url.QueryUnescape(header)
+	if err != nil {
+		return header
+	}
+	result = strings.ReplaceAll(result, "&", "\n")
+	result = strings.ReplaceAll(result, "=", ": ")
+	return result
+}
+
+func (i *imlServiceModule) RestLogInfo(ctx context.Context, serviceId string, logId string) (*service_dto.RestLogInfo, error) {
+	c, err := i.clusterService.Get(ctx, cluster.DefaultClusterID)
+	if err != nil {
+		return nil, fmt.Errorf("cluster %s not found", cluster.DefaultClusterID)
+	}
+	info, err := i.logService.LogInfo(ctx, "loki", c.Cluster, logId)
+	if err != nil {
+		return nil, err
+	}
+	if info.Service != serviceId {
+		return nil, errors.New("service not match")
+	}
+
+	logInfo := &service_dto.RestLogInfo{
+		Id:           info.ID,
+		API:          auto.UUID(info.API),
+		Consumer:     auto.UUID(info.Consumer),
+		Status:       info.StatusCode,
+		Ip:           info.RemoteIP,
+		ResponseTime: common.FormatTime(info.ResponseTime),
+		Traffic:      common.FormatByte(info.Traffic),
+		LogTime:      auto.TimeLabel(info.RecordTime),
+		Request: service_dto.OriginRequest{
+			Header: formatHeader(info.RequestHeader),
+			Origin: info.RequestBody,
+		},
+		Response: service_dto.OriginRequest{
+			Header: formatHeader(info.ResponseHeader),
+			Origin: info.ResponseBody,
+		},
+	}
+
+	if info.Consumer == "apipark-global" {
+		logInfo.IsSystemConsumer = true
+		logInfo.Consumer = auto.Label{
+			Id:   info.Consumer,
+			Name: "System Consumer",
+		}
+	}
+	return logInfo, nil
+}
+
+func (i *imlServiceModule) AILogInfo(ctx context.Context, serviceId string, logId string) (*service_dto.AILogInfo, error) {
+	c, err := i.clusterService.Get(ctx, cluster.DefaultClusterID)
+	if err != nil {
+		return nil, fmt.Errorf("cluster %s not found", cluster.DefaultClusterID)
+	}
+	info, err := i.logService.LogInfo(ctx, "loki", c.Cluster, logId)
+	if err != nil {
+		return nil, err
+	}
+	if info.Service != serviceId {
+		return nil, errors.New("service not match")
+	}
+	response, err := parseAIResponse(info.ResponseBody)
+	if err != nil {
+		response = info.ResponseBody
+	}
+
+	logInfo := &service_dto.AILogInfo{
+		Id:       info.ID,
+		API:      auto.UUID(info.API),
+		Consumer: auto.UUID(info.Consumer),
+		Status:   info.StatusCode,
+		Ip:       info.RemoteIP,
+		Provider: auto.UUID(info.AIProvider),
+		Model:    info.AIModel,
+		LogTime:  auto.TimeLabel(info.RecordTime),
+		Request: service_dto.OriginAIRequest{
+			OriginRequest: service_dto.OriginRequest{
+				Header: formatHeader(info.RequestHeader),
+				Origin: info.RequestBody,
+			},
+			Body:  parseAIRequest(info.RequestBody),
+			Token: info.TotalToken,
+		},
+		Response: service_dto.OriginAIRequest{
+			OriginRequest: service_dto.OriginRequest{
+				Header: formatHeader(info.ResponseHeader),
+				Origin: info.ResponseBody,
+			},
+			Body:  response,
+			Token: info.TotalToken,
+		},
+	}
+	if info.Consumer == "apipark-global" {
+		logInfo.IsSystemConsumer = true
+		logInfo.Consumer = auto.Label{
+			Id:   info.Consumer,
+			Name: "System Consumer",
+		}
+	}
+	return logInfo, nil
+}
+
 func (i *imlServiceModule) RestLogs(ctx context.Context, serviceId string, start int64, end int64, page int, size int) ([]*service_dto.RestLogItem, int64, error) {
 	list, total, err := i.logService.LogRecordsByService(ctx, serviceId, time.Unix(start, 0), time.Unix(end, 0), page, size)
 	if err != nil {
@@ -42,6 +42,8 @@ type IServiceModule interface {
 type ILogModule interface {
 	AILogs(ctx context.Context, serviceId string, start int64, end int64, page int, size int) ([]*service_dto.AILogItem, int64, error)
 	RestLogs(ctx context.Context, serviceId string, start int64, end int64, page int, size int) ([]*service_dto.RestLogItem, int64, error)
+	RestLogInfo(ctx context.Context, serviceId string, logId string) (*service_dto.RestLogInfo, error)
+	AILogInfo(ctx context.Context, serviceId string, logId string) (*service_dto.AILogInfo, error)
 }

 type IServiceDocModule interface {
@@ -0,0 +1,115 @@
+package service
+
+import (
+	"bufio"
+	"encoding/json"
+	"strings"
+)
+
+// ChatCompletionChunk represents the structure of a single chunk in the streaming response
+type ChatCompletionChunk struct {
+	Object  string   `json:"object"`
+	Choices []Choice `json:"choices"`
+}
+
+// ChatCompletion represents the structure of a non-streaming response
+type ChatCompletion struct {
+	Object  string       `json:"object"`
+	Choices []FullChoice `json:"choices"`
+}
+
+// Choice represents a choice in the streaming chunk
+type Choice struct {
+	Delta        Delta   `json:"delta"`
+	FinishReason *string `json:"finish_reason"`
+}
+
+// FullChoice represents a choice in the non-streaming response
+type FullChoice struct {
+	Message Message `json:"message"`
+}
+
+// Delta represents the delta content in a streaming choice
+type Delta struct {
+	Content string `json:"content"`
+	Role    string `json:"role,omitempty"`
+}
+
+// Message represents the message content in a non-streaming choice
+type Message struct {
+	Content string `json:"content"`
+	Role    string `json:"role"`
+}
+
+// ParseAIResponse parses both streaming and non-streaming AI responses and returns the concatenated content
+func parseAIResponse(input string) (string, error) {
+	// First, try to parse as a non-streaming response
+	var nonStreaming ChatCompletion
+	if err := json.Unmarshal([]byte(input), &nonStreaming); err == nil && nonStreaming.Object == "chat.completion" {
+		var result strings.Builder
+		for _, choice := range nonStreaming.Choices {
+			result.WriteString(choice.Message.Content)
+		}
+		return result.String(), nil
+	}
+
+	// If not non-streaming, parse as streaming response
+	var result strings.Builder
+	scanner := bufio.NewScanner(strings.NewReader(input))
+
+	for scanner.Scan() {
+		line := scanner.Text()
+		// Skip empty lines or [DONE]
+		if line == "" || line == "data: [DONE]" {
+			continue
+		}
+
+		// Check if line starts with "data: "
+		if !strings.HasPrefix(line, "data: ") {
+			continue
+		}
+
+		// Extract JSON data
+		jsonData := strings.TrimPrefix(line, "data: ")
+		var chunk ChatCompletionChunk
+		if err := json.Unmarshal([]byte(jsonData), &chunk); err != nil {
+			return "", err
+		}
+
+		// Process each choice
+		for _, choice := range chunk.Choices {
+			// Append content from delta
+			result.WriteString(choice.Delta.Content)
+			// Check if this is the final chunk
+			if choice.FinishReason != nil && *choice.FinishReason == "stop" {
+				return result.String(), nil
+			}
+		}
+	}
+
+	if err := scanner.Err(); err != nil {
+		return "", err
+	}
+
+	return result.String(), nil
+}
+
+func parseAIRequest(ori string) string {
+	type aiRequest struct {
+		Messages []struct {
+			Role    string `json:"role"`
+			Content string `json:"content"`
+		} `json:"messages"`
+	}
+	var req aiRequest
+	err := json.Unmarshal([]byte(ori), &req)
+	if err != nil {
+		return ori
+	}
+	size := len(req.Messages)
+	if size == 0 {
+		return ""
+	}
+
+	return req.Messages[size-1].Content
+}