From 2f2c4464ea1fc322db9d05cc59f36c2d975199f4 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 10 Aug 2023 12:42:12 +0800 Subject: [PATCH 01/16] record time --- main.go | 3 ++- record.go | 14 ++++++++------ 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/main.go b/main.go index eaa8a7a..de76fe3 100644 --- a/main.go +++ b/main.go @@ -97,6 +97,7 @@ func main() { db.Take(&authConfig, "key = ?", "authorization") engine.POST("/v1/*any", func(c *gin.Context) { + begin := time.Now() trackID := uuid.New() // check authorization header if !*noauth { @@ -224,7 +225,7 @@ func main() { if err != nil { log.Println("Failed to read from response tee buffer", err) } - go recordAssistantResponse(contentType, db, trackID, resp) + go recordAssistantResponse(contentType, db, trackID, resp, time.Now().Sub(begin)) }) // --------------------------------- diff --git a/record.go b/record.go index 7cf2ce2..30af0a5 100644 --- a/record.go +++ b/record.go @@ -12,11 +12,12 @@ import ( ) type Record struct { - ID uuid.UUID `gorm:"type:uuid"` - CreatedAt time.Time - IP string - Body string - Response string + ID uuid.UUID `gorm:"type:uuid"` + CreatedAt time.Time + IP string + Body string + Response string + ElapsedTime time.Duration } func recordUserMessage(c *gin.Context, db *gorm.DB, trackID uuid.UUID, body []byte) { @@ -62,7 +63,7 @@ type FetchModeUsage struct { TotalTokens int64 `json:"total_tokens"` } -func recordAssistantResponse(contentType string, db *gorm.DB, trackID uuid.UUID, body []byte) { +func recordAssistantResponse(contentType string, db *gorm.DB, trackID uuid.UUID, body []byte, elapsedTime time.Duration) { result := "" // stream mode if strings.HasPrefix(contentType, "text/event-stream") { @@ -113,6 +114,7 @@ func recordAssistantResponse(contentType string, db *gorm.DB, trackID uuid.UUID, return } record.Response = result + record.ElapsedTime = elapsedTime if db.Save(&record).Error != nil { log.Println("Error to save record:", record) return From f2b144c40950c4963fca16daeaf64929aeb45b68 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 10 Aug 2023 13:32:51 +0800 Subject: [PATCH 02/16] =?UTF-8?q?=E9=83=A8=E7=BD=B2=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 99 ++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 91 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 1be6325..fc511da 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,100 @@ -# API文档 +# API 文档 -本文档提供了使用该负载君恩和能够API的方法和端点的详细说明。 +本文档详细介绍了如何使用负载均衡和能力 API 的方法和端点。 +## 部署方法 + +### 编译 + +以下是编译和运行该负载均衡 API 的步骤： + +1. 首先，确保您已经安装了 golang 和 gcc。 + +2. 克隆本仓库到您的本地机器上。 + +3. 打开终端，并进入到仓库目录中。 + +4. 在终端中执行以下命令来编译代码： + + ``` + make + ``` + + 这将会编译代码并生成可执行文件。 + +5. 编译成功后，您可以直接运行以下命令来启动负载均衡和能力 API： + + ``` + ./openai-api-route + ``` + + 默认情况下，API 将会在本地的 8888 端口进行监听。 + + 如果您希望使用不同的监听地址，可以使用 `-addr` 参数来指定，例如： + + ``` + ./openai-api-route -addr 0.0.0.0:8080 + ``` + + 这将会将监听地址设置为 0.0.0.0:8080。 + +6. 如果数据库不存在，系统会自动创建一个名为 `db.sqlite` 的数据库文件。 + + 如果您希望使用不同的数据库地址，可以使用 `-database` 参数来指定，例如： + + ``` + ./openai-api-route -database /path/to/database.db + ``` + + 这将会将数据库地址设置为 `/path/to/database.db`。 + +7. 现在，您已经成功编译并运行了负载均衡和能力 API。您可以根据需要添加上游、管理上游，并使用 API 进行相关操作。 + +### 运行 + +以下是运行命令的用法： + +``` +Usage of ./openai-api-route: + -add + 添加一个 OpenAI 上游 + -addr string + 监听地址（默认为 ":8888"） + -database string + 数据库地址（默认为 "./db.sqlite"） + -endpoint string + OpenAI API 基地址（默认为 "https://api.openai.com/v1"） + -list + 列出所有上游 + -noauth + 不检查传入的授权头 + -sk string + OpenAI API 密钥（sk-xxxxx） +``` + +您可以直接运行 `./openai-api-route` 命令，如果数据库不存在，系统会自动创建。 + +### 上游管理 + +您可以使用以下命令添加一个上游： + +```bash +./openai-api-route -add -sk sk-xxxxx -endpoint https://api.openai.com/v1 +``` + +您也可以使用 `/admin/upstreams` 的 HTTP 接口进行控制。 + +另外，您还可以直接编辑数据库中的 `openai_upstreams` 表。 ## 身份验证 ### 身份验证中间件流程 1. 从请求头中获取`Authorization`字段的值。 2. 检查`Authorization`字段的值是否以`"Bearer"`开头。 - - 如果不是，则返回错误信息："authorization header should start with 'Bearer'"（HTTP状态码403）。 + - 如果不是，则返回错误信息："authorization header should start with 'Bearer'"（HTTP 状态码 403）。 3. 去除`Authorization`字段值开头的`"Bearer"`和前后的空格。 4. 将剩余的值与预先设置的身份验证配置进行比较。 - - 如果不匹配，则返回错误信息："wrong authorization header"（HTTP状态码403）。 + - 如果不匹配，则返回错误信息："wrong authorization header"（HTTP 状态码 403）。 5. 如果身份验证通过，则返回`nil`。 ## 上游管理 @@ -22,7 +105,7 @@ - 方法: GET - 权限要求: 需要进行身份验证 - 返回数据类型: JSON -- 请求示例: +- 请求示例: ```bash curl -X GET -H "Authorization: Bearer access_token" http://localhost:8080/admin/upstreams ``` @@ -60,7 +143,7 @@ } ``` -### 删除指定ID的上游 +### 删除指定 ID 的上游 - URL: `/admin/upstreams/:id` - 方法: DELETE @@ -77,7 +160,7 @@ } ``` -### 更新指定ID的上游 +### 更新指定 ID 的上游 - URL: `/admin/upstreams/:id` - 方法: PUT @@ -93,4 +176,4 @@ { "message": "success" } - ``` \ No newline at end of file + ``` From 34beab9429528df47e4e7196b8d3a5af3ba506e6 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Fri, 15 Sep 2023 11:57:30 +0800 Subject: [PATCH 03/16] delete api --- main.go | 104 -------------------------------------------------------- 1 file changed, 104 deletions(-) diff --git a/main.go b/main.go index de76fe3..09eacde 100644 --- a/main.go +++ b/main.go @@ -10,7 +10,6 @@ import ( "net/http" "net/http/httputil" "net/url" - "strconv" "strings" "time" @@ -228,108 +227,5 @@ func main() { go recordAssistantResponse(contentType, db, trackID, resp, time.Now().Sub(begin)) }) - // --------------------------------- - // admin APIs - engine.POST("/admin/login", func(c *gin.Context) { - // check authorization headers - if handleAuth(c) != nil { - return - } - c.JSON(200, gin.H{ - "message": "success", - }) - }) - engine.GET("/admin/upstreams", func(c *gin.Context) { - // check authorization headers - if handleAuth(c) != nil { - return - } - upstreams := make([]OPENAI_UPSTREAM, 0) - db.Find(&upstreams) - c.JSON(200, upstreams) - }) - engine.POST("/admin/upstreams", func(c *gin.Context) { - // check authorization headers - if handleAuth(c) != nil { - return - } - newUpstream := OPENAI_UPSTREAM{} - err := c.BindJSON(&newUpstream) - if err != nil { - c.AbortWithError(502, errors.New("can't parse OPENAI_UPSTREAM object")) - return - } - if newUpstream.SK == "" || newUpstream.Endpoint == "" { - c.AbortWithError(403, errors.New("can't create new OPENAI_UPSTREAM with empty sk or endpoint")) - return - } - log.Println("Saveing new OPENAI_UPSTREAM", newUpstream) - err = db.Create(&newUpstream).Error - if err != nil { - c.AbortWithError(403, err) - return - } - }) - engine.DELETE("/admin/upstreams/:id", func(ctx *gin.Context) { - // check authorization headers - if handleAuth(ctx) != nil { - return - } - id, err := strconv.Atoi(ctx.Param("id")) - if err != nil { - ctx.AbortWithError(502, err) - return - } - upstream := OPENAI_UPSTREAM{} - upstream.ID = uint(id) - db.Delete(&upstream) - ctx.JSON(200, gin.H{ - "message": "success", - }) - }) - engine.PUT("/admin/upstreams/:id", func(c *gin.Context) { - // check authorization headers - if handleAuth(c) != nil { - return - } - upstream := OPENAI_UPSTREAM{} - err := c.BindJSON(&upstream) - if err != nil { - c.AbortWithError(502, errors.New("can't parse OPENAI_UPSTREAM object")) - return - } - if upstream.SK == "" || upstream.Endpoint == "" { - c.AbortWithError(403, errors.New("can't create new OPENAI_UPSTREAM with empty sk or endpoint")) - return - } - id, err := strconv.Atoi(c.Param("id")) - if err != nil { - c.AbortWithError(502, err) - return - } - upstream.ID = uint(id) - log.Println("Saveing new OPENAI_UPSTREAM", upstream) - err = db.Create(&upstream).Error - if err != nil { - c.AbortWithError(403, err) - return - } - c.JSON(200, gin.H{ - "message": "success", - }) - }) - engine.GET("/admin/request_records", func(c *gin.Context) { - // check authorization headers - if handleAuth(c) != nil { - return - } - requestRecords := []Record{} - err := db.Order("id desc").Limit(100).Find(&requestRecords).Error - if err != nil { - c.AbortWithError(502, err) - return - } - c.JSON(200, requestRecords) - }) engine.Run(*listenAddr) } From 687fffead3d7cc1d874fde1e12458d27e549a1fd Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Fri, 15 Sep 2023 12:48:18 +0800 Subject: [PATCH 04/16] fix cors --- cors.go | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/cors.go b/cors.go index ec5e2fd..ddae754 100644 --- a/cors.go +++ b/cors.go @@ -6,13 +6,26 @@ import ( // Middleware function to handle CORS requests func handleCORS(c *gin.Context) { - c.Writer.Header().Set("Access-Control-Allow-Origin", "*") - c.Writer.Header().Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") - c.Writer.Header().Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") - if c.Request.Method == "OPTIONS" { + header := c.Writer.Header() + header.Set("Access-Control-Allow-Origin", "*") + header.Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") + header.Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") + c.AbortWithStatus(200) return } + c.Next() + + header := c.Writer.Header() + + header.Del("Access-Control-Allow-Origin") + header.Del("Access-Control-Allow-Methods") + header.Del("Access-Control-Allow-Headers") + + header.Set("Access-Control-Allow-Origin", "*") + header.Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") + header.Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") + } From 7c53581ff27c9bc94ef90826748ec1bb8321edcd Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Fri, 15 Sep 2023 18:00:17 +0800 Subject: [PATCH 05/16] integer id --- main.go | 84 +++++++++++++++++++++++++++++++++++++++++++++---------- record.go | 20 +++---------- 2 files changed, 74 insertions(+), 30 deletions(-) diff --git a/main.go b/main.go index 09eacde..6c1df2b 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "encoding/json" "errors" "flag" "fmt" @@ -14,7 +15,6 @@ import ( "time" "github.com/gin-gonic/gin" - "github.com/google/uuid" "gorm.io/driver/sqlite" "gorm.io/gorm" ) @@ -96,8 +96,11 @@ func main() { db.Take(&authConfig, "key = ?", "authorization") engine.POST("/v1/*any", func(c *gin.Context) { - begin := time.Now() - trackID := uuid.New() + record := Record{ + IP: c.ClientIP(), + CreatedAt: time.Now(), + } + // check authorization header if !*noauth { if handleAuth(c) != nil { @@ -138,6 +141,8 @@ func main() { return } + record.UpstreamID = upstream.ID + // reverse proxy remote, err := url.Parse(upstream.Endpoint) if err != nil { @@ -158,7 +163,7 @@ func main() { } // record chat message from user - go recordUserMessage(c, db, trackID, body) + record.Body = string(body) out.Body = io.NopCloser(bytes.NewReader(body)) @@ -174,18 +179,17 @@ func main() { var buf bytes.Buffer var contentType string proxy.ModifyResponse = func(r *http.Response) error { + record.Status = r.StatusCode if r.StatusCode != 200 { body, err := io.ReadAll(r.Body) if err != nil { - return errors.New("failed to read response from upstream " + err.Error()) + record.Response = "failed to read response from upstream " + err.Error() + return errors.New(record.Response) } - return fmt.Errorf("upstream return '%s' with '%s'", r.Status, string(body)) + record.Response = fmt.Sprintf("upstream return '%s' with '%s'", r.Status, string(body)) + return fmt.Errorf(record.Response) } // count success - go db.Model(&upstream).Updates(map[string]interface{}{ - "success_count": gorm.Expr("success_count + ?", 1), - "last_call_success_time": time.Now(), - }) r.Body = io.NopCloser(io.TeeReader(r.Body, &buf)) contentType = r.Header.Get("content-type") return nil @@ -212,8 +216,6 @@ func main() { ) go sendMatrixMessage(content) if err.Error() != "context canceled" && r.Response.StatusCode != 400 { - // count failed - go db.Model(&upstream).Update("failed_count", gorm.Expr("failed_count + ?", 1)) go sendFeishuMessage(content) } @@ -222,9 +224,63 @@ func main() { proxy.ServeHTTP(c.Writer, c.Request) resp, err := io.ReadAll(io.NopCloser(&buf)) if err != nil { - log.Println("Failed to read from response tee buffer", err) + record.Response = "failed to read response from upstream " + err.Error() + log.Println(record.Response) + } else { + + // record response + // stream mode + if strings.HasPrefix(contentType, "text/event-stream") { + for _, line := range strings.Split(string(resp), "\n") { + chunk := StreamModeChunk{} + line = strings.TrimPrefix(line, "data:") + line = strings.TrimSpace(line) + if line == "" { + continue + } + + err := json.Unmarshal([]byte(line), &chunk) + if err != nil { + log.Println(err) + continue + } + + if len(chunk.Choices) == 0 { + continue + } + record.Response += chunk.Choices[0].Delta.Content + } + } else if strings.HasPrefix(contentType, "application/json") { + var fetchResp FetchModeResponse + err := json.Unmarshal(resp, &fetchResp) + if err != nil { + log.Println("Error parsing fetch response:", err) + return + } + if !strings.HasPrefix(fetchResp.Model, "gpt-") { + log.Println("Not GPT model, skip recording response:", fetchResp.Model) + return + } + if len(fetchResp.Choices) == 0 { + log.Println("Error: fetch response choice length is 0") + return + } + record.Response = fetchResp.Choices[0].Message.Content + } else { + log.Println("Unknown content type", contentType) + return + } + } + + if len(record.Body) > 1024*512 { + record.Body = "" + } + + log.Println("Record result:", record.Response) + record.ElapsedTime = time.Now().Sub(record.CreatedAt) + if db.Create(&record).Error != nil { + log.Println("Error to save record:", record) } - go recordAssistantResponse(contentType, db, trackID, resp, time.Now().Sub(begin)) }) engine.Run(*listenAddr) diff --git a/record.go b/record.go index 30af0a5..5ff89ca 100644 --- a/record.go +++ b/record.go @@ -6,31 +6,19 @@ import ( "strings" "time" - "github.com/gin-gonic/gin" "github.com/google/uuid" "gorm.io/gorm" ) type Record struct { - ID uuid.UUID `gorm:"type:uuid"` + ID int64 `gorm:"primaryKey,autoIncrement"` CreatedAt time.Time IP string - Body string + Body string `gorm:"serializer:json"` Response string ElapsedTime time.Duration -} - -func recordUserMessage(c *gin.Context, db *gorm.DB, trackID uuid.UUID, body []byte) { - bodyStr := string(body) - requestRecord := Record{ - Body: bodyStr, - ID: trackID, - IP: c.ClientIP(), - } - err := db.Create(&requestRecord).Error - if err != nil { - log.Println("Error record request:", err) - } + Status int + UpstreamID uint } type StreamModeChunk struct { From 638feaf381861f2f2b9d0c9484c11e793eb90a04 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Fri, 15 Sep 2023 18:22:21 +0800 Subject: [PATCH 06/16] panic recover --- main.go | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/main.go b/main.go index 6c1df2b..d220661 100644 --- a/main.go +++ b/main.go @@ -100,6 +100,12 @@ func main() { IP: c.ClientIP(), CreatedAt: time.Now(), } + defer func() { + if err := recover(); err != nil { + log.Println("Error:", err) + c.AbortWithError(500, fmt.Errorf("%s", err)) + } + }() // check authorization header if !*noauth { @@ -221,7 +227,16 @@ func main() { log.Println("response is", r.Response) } - proxy.ServeHTTP(c.Writer, c.Request) + + func() { + defer func() { + if err := recover(); err != nil { + log.Println("Panic recover :", err) + } + }() + proxy.ServeHTTP(c.Writer, c.Request) + }() + resp, err := io.ReadAll(io.NopCloser(&buf)) if err != nil { record.Response = "failed to read response from upstream " + err.Error() From 1fcdc59b63753cd834f0f2304a60a68e07056fd7 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Fri, 15 Sep 2023 19:55:05 +0800 Subject: [PATCH 07/16] fix: cors in resp --- cors.go | 31 ------------------------------- main.go | 15 ++++++++++++++- 2 files changed, 14 insertions(+), 32 deletions(-) delete mode 100644 cors.go diff --git a/cors.go b/cors.go deleted file mode 100644 index ddae754..0000000 --- a/cors.go +++ /dev/null @@ -1,31 +0,0 @@ -package main - -import ( - "github.com/gin-gonic/gin" -) - -// Middleware function to handle CORS requests -func handleCORS(c *gin.Context) { - if c.Request.Method == "OPTIONS" { - header := c.Writer.Header() - header.Set("Access-Control-Allow-Origin", "*") - header.Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") - header.Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") - - c.AbortWithStatus(200) - return - } - - c.Next() - - header := c.Writer.Header() - - header.Del("Access-Control-Allow-Origin") - header.Del("Access-Control-Allow-Methods") - header.Del("Access-Control-Allow-Headers") - - header.Set("Access-Control-Allow-Origin", "*") - header.Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") - header.Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") - -} diff --git a/main.go b/main.go index d220661..7719da7 100644 --- a/main.go +++ b/main.go @@ -90,7 +90,13 @@ func main() { }) // CORS handler - engine.Use(handleCORS) + engine.OPTIONS("/v1/*any", func(ctx *gin.Context) { + header := ctx.Writer.Header() + header.Set("Access-Control-Allow-Origin", "*") + header.Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") + header.Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") + ctx.AbortWithStatus(200) + }) // get authorization config from db db.Take(&authConfig, "key = ?", "authorization") @@ -186,6 +192,13 @@ func main() { var contentType string proxy.ModifyResponse = func(r *http.Response) error { record.Status = r.StatusCode + r.Header.Del("Access-Control-Allow-Origin") + r.Header.Del("Access-Control-Allow-Methods") + r.Header.Del("Access-Control-Allow-Headers") + r.Header.Set("Access-Control-Allow-Origin", "*") + r.Header.Set("Access-Control-Allow-Methods", "POST, GET, OPTIONS, PUT, DELETE, PATCH") + r.Header.Set("Access-Control-Allow-Headers", "Origin, Authorization, Content-Type") + if r.StatusCode != 200 { body, err := io.ReadAll(r.Body) if err != nil { From 043fb3db85bb9725d1edafb7336eb1c78fd5ca2e Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Sat, 16 Sep 2023 10:52:33 +0800 Subject: [PATCH 08/16] upstream asis --- auth.go | 3 ++- main.go | 8 ++++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/auth.go b/auth.go index a21c50f..d2d5761 100644 --- a/auth.go +++ b/auth.go @@ -20,7 +20,8 @@ func handleAuth(c *gin.Context) error { authorization = strings.Trim(authorization[len("Bearer"):], " ") log.Println("Received authorization", authorization) - if authorization != authConfig.Value { + + if authConfig.Value != "asis" && authorization != authConfig.Value { err = errors.New("wrong authorization header") c.AbortWithError(403, err) return err diff --git a/main.go b/main.go index 7719da7..531cd86 100644 --- a/main.go +++ b/main.go @@ -185,7 +185,11 @@ func main() { out.URL.Path = in.URL.Path out.Header = http.Header{} out.Header.Set("Host", remote.Host) - out.Header.Set("Authorization", "Bearer "+upstream.SK) + if upstream.SK == "asis" { + out.Header.Set("Authorization", c.Request.Header.Get("Authorization")) + } else { + out.Header.Set("Authorization", "Bearer "+upstream.SK) + } out.Header.Set("Content-Type", c.Request.Header.Get("Content-Type")) } var buf bytes.Buffer @@ -230,7 +234,7 @@ func main() { } content := fmt.Sprintf("[%s] OpenAI 转发出错 ID: %d... 密钥: [%s] 上游: [%s] 错误: %s\n---\n%s", c.ClientIP(), - upstream.ID, upstream.SK[:10], upstream.Endpoint, err.Error(), + upstream.ID, upstream.SK, upstream.Endpoint, err.Error(), strings.Join(upstreamDescriptions, "\n"), ) go sendMatrixMessage(content) From 8342e47e4b26240fd353f9f899ed0ec4ced78fb9 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 28 Sep 2023 15:42:22 +0800 Subject: [PATCH 09/16] asis & readme --- README.md | 82 ++----------------------------------------------------- auth.go | 7 ++++- main.go | 5 ++-- record.go | 17 ++++++------ 4 files changed, 21 insertions(+), 90 deletions(-) diff --git a/README.md b/README.md index fc511da..6cd9fac 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ 这将会编译代码并生成可执行文件。 -5. 编译成功后，您可以直接运行以下命令来启动负载均衡和能力 API： +5. 编译成功后，您可以直接运行以下命令来启动负载均衡 API： ``` ./openai-api-route @@ -85,6 +85,7 @@ Usage of ./openai-api-route: 您也可以使用 `/admin/upstreams` 的 HTTP 接口进行控制。 另外，您还可以直接编辑数据库中的 `openai_upstreams` 表。 + ## 身份验证 ### 身份验证中间件流程 @@ -99,81 +100,4 @@ Usage of ./openai-api-route: ## 上游管理 -### 获取所有上游 - -- URL: `/admin/upstreams` -- 方法: GET -- 权限要求: 需要进行身份验证 -- 返回数据类型: JSON -- 请求示例: - ```bash - curl -X GET -H "Authorization: Bearer access_token" http://localhost:8080/admin/upstreams - ``` -- 返回示例: - ```json - [ - { - "ID": 1, - "SK": "sk_value", - "Endpoint": "endpoint_value" - }, - { - "ID": 2, - "SK": "sk_value", - "Endpoint": "endpoint_value" - } - ] - ``` - -### 创建新的上游 - -- URL: `/admin/upstreams` -- 方法: POST -- 权限要求: 需要进行身份验证 -- 请求数据类型: JSON -- 请求示例: - ```bash - curl -X POST -H "Authorization: Bearer access_token" -H "Content-Type: application/json" -d '{"SK": "sk_value", "Endpoint": "endpoint_value"}' http://localhost:8080/admin/upstreams - ``` -- 返回数据类型: JSON -- 返回示例: - ```json - { - "message": "success" - } - ``` - -### 删除指定 ID 的上游 - -- URL: `/admin/upstreams/:id` -- 方法: DELETE -- 权限要求: 需要进行身份验证 -- 返回数据类型: JSON -- 请求示例: - ```bash - curl -X DELETE -H "Authorization: Bearer access_token" http://localhost:8080/admin/upstreams/1 - ``` -- 返回示例: - ```json - { - "message": "success" - } - ``` - -### 更新指定 ID 的上游 - -- URL: `/admin/upstreams/:id` -- 方法: PUT -- 权限要求: 需要进行身份验证 -- 请求数据类型: JSON -- 请求示例: - ```bash - curl -X PUT -H "Authorization: Bearer access_token" -H "Content-Type: application/json" -d '{"SK": "sk_value", "Endpoint": "endpoint_value"}' http://localhost:8080/admin/upstreams/1 - ``` -- 返回数据类型: JSON -- 返回示例: - ```json - { - "message": "success" - } - ``` +没什么好说的，直接操作数据库 `openai_upstreams` 表，改动立即生效 diff --git a/auth.go b/auth.go index d2d5761..3815e21 100644 --- a/auth.go +++ b/auth.go @@ -21,7 +21,12 @@ func handleAuth(c *gin.Context) error { authorization = strings.Trim(authorization[len("Bearer"):], " ") log.Println("Received authorization", authorization) - if authConfig.Value != "asis" && authorization != authConfig.Value { + if authConfig.Value == "asis" { + log.Println("Authorization is asis, skipping") + return nil + } + + if authorization != authConfig.Value { err = errors.New("wrong authorization header") c.AbortWithError(403, err) return err diff --git a/main.go b/main.go index 531cd86..a76d20f 100644 --- a/main.go +++ b/main.go @@ -103,8 +103,9 @@ func main() { engine.POST("/v1/*any", func(c *gin.Context) { record := Record{ - IP: c.ClientIP(), - CreatedAt: time.Now(), + IP: c.ClientIP(), + CreatedAt: time.Now(), + Authorization: c.Request.Header.Get("Authorization"), } defer func() { if err := recover(); err != nil { diff --git a/record.go b/record.go index 5ff89ca..801c055 100644 --- a/record.go +++ b/record.go @@ -11,14 +11,15 @@ import ( ) type Record struct { - ID int64 `gorm:"primaryKey,autoIncrement"` - CreatedAt time.Time - IP string - Body string `gorm:"serializer:json"` - Response string - ElapsedTime time.Duration - Status int - UpstreamID uint + ID int64 `gorm:"primaryKey,autoIncrement"` + CreatedAt time.Time + IP string + Body string `gorm:"serializer:json"` + Response string + ElapsedTime time.Duration + Status int + UpstreamID uint + Authorization string } type StreamModeChunk struct { From c6cf75d2f6c6b3276380d4337af8926e2cdea467 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 28 Sep 2023 15:45:29 +0800 Subject: [PATCH 10/16] fix: asis --- auth.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/auth.go b/auth.go index 3815e21..a386b3c 100644 --- a/auth.go +++ b/auth.go @@ -21,11 +21,6 @@ func handleAuth(c *gin.Context) error { authorization = strings.Trim(authorization[len("Bearer"):], " ") log.Println("Received authorization", authorization) - if authConfig.Value == "asis" { - log.Println("Authorization is asis, skipping") - return nil - } - if authorization != authConfig.Value { err = errors.New("wrong authorization header") c.AbortWithError(403, err) From f5dc8147e6fa190c48b71cfcfbf2cccac9d75970 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 28 Sep 2023 15:52:49 +0800 Subject: [PATCH 11/16] update: readme introduction --- README.md | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 6cd9fac..4a1d936 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,13 @@ -# API 文档 +# openai-api-route 文档 + +这是一个 OpenAI API 负载均衡的简易工具，使用 golang 原生 reverse proxy 方法转发请求到 OpenAI 上游 + +功能包括： + +- 更改 Authorization 验证头 +- 多种负载均衡策略 +- 记录完整的请求内容、IP 地址、响应时间以及 GPT 回复文本 +- 上游返回错误时发送 飞书 或 Matrix 消息通知 本文档详细介绍了如何使用负载均衡和能力 API 的方法和端点。 From de1f9c1e9419f40a2055a17f00e03a46de0c9408 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 28 Sep 2023 15:57:30 +0800 Subject: [PATCH 12/16] remove upstream count --- main.go | 13 +++---------- structure.go | 9 ++------- 2 files changed, 5 insertions(+), 17 deletions(-) diff --git a/main.go b/main.go index a76d20f..4b64d98 100644 --- a/main.go +++ b/main.go @@ -67,9 +67,9 @@ func main() { if *listMode { result := make([]OPENAI_UPSTREAM, 0) db.Find(&result) - fmt.Println("SK\tEndpoint\tSuccess\tFailed\tLast Success Time") + fmt.Println("SK\tEndpoint") for _, upstream := range result { - fmt.Println(upstream.SK, upstream.Endpoint, upstream.SuccessCount, upstream.FailedCount, upstream.LastCallSuccessTime) + fmt.Println(upstream.SK, upstream.Endpoint) } return } @@ -227,16 +227,9 @@ func main() { // send notification upstreams := []OPENAI_UPSTREAM{} db.Find(&upstreams) - upstreamDescriptions := make([]string, 0) - for _, upstream := range upstreams { - upstreamDescriptions = append(upstreamDescriptions, fmt.Sprintf("ID: %d, %s: %s 成功次数: %d, 失败次数: %d, 最后成功调用: %s", - upstream.ID, upstream.SK, upstream.Endpoint, upstream.SuccessCount, upstream.FailedCount, upstream.LastCallSuccessTime, - )) - } - content := fmt.Sprintf("[%s] OpenAI 转发出错 ID: %d... 密钥: [%s] 上游: [%s] 错误: %s\n---\n%s", + content := fmt.Sprintf("[%s] OpenAI 转发出错 ID: %d... 密钥: [%s] 上游: [%s] 错误: %s", c.ClientIP(), upstream.ID, upstream.SK, upstream.Endpoint, err.Error(), - strings.Join(upstreamDescriptions, "\n"), ) go sendMatrixMessage(content) if err.Error() != "context canceled" && r.Response.StatusCode != 400 { diff --git a/structure.go b/structure.go index ca29e9b..08509ca 100644 --- a/structure.go +++ b/structure.go @@ -1,17 +1,12 @@ package main import ( - "time" - "gorm.io/gorm" ) // one openai upstream contain a pair of key and endpoint type OPENAI_UPSTREAM struct { gorm.Model - SK string `gorm:"index:idx_sk_endpoint,unique"` // key - Endpoint string `gorm:"index:idx_sk_endpoint,unique"` // endpoint - SuccessCount int64 - FailedCount int64 - LastCallSuccessTime time.Time + SK string `gorm:"index:idx_sk_endpoint,unique"` // key + Endpoint string `gorm:"index:idx_sk_endpoint,unique"` // endpoint } From 1e770db740362f67a7742117739d178005a97d61 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Thu, 28 Sep 2023 15:58:11 +0800 Subject: [PATCH 13/16] send notification --- main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main.go b/main.go index 4b64d98..3accd19 100644 --- a/main.go +++ b/main.go @@ -231,8 +231,8 @@ func main() { c.ClientIP(), upstream.ID, upstream.SK, upstream.Endpoint, err.Error(), ) - go sendMatrixMessage(content) if err.Error() != "context canceled" && r.Response.StatusCode != 400 { + go sendMatrixMessage(content) go sendFeishuMessage(content) } From 1a1226568f0f17b95d1b0036cb911fe3da3d1f5f Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Sun, 8 Oct 2023 15:39:12 +0800 Subject: [PATCH 14/16] record error --- main.go | 29 +++++++++++++++-------------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/main.go b/main.go index 3accd19..ec1f721 100644 --- a/main.go +++ b/main.go @@ -210,7 +210,8 @@ func main() { record.Response = "failed to read response from upstream " + err.Error() return errors.New(record.Response) } - record.Response = fmt.Sprintf("upstream return '%s' with '%s'", r.Status, string(body)) + record.Response = fmt.Sprintf("openai-api-route upstream return '%s' with '%s'", r.Status, string(body)) + record.Status = r.StatusCode return fmt.Errorf(record.Response) } // count success @@ -221,22 +222,23 @@ func main() { proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) { log.Println("Error", err, upstream.SK, upstream.Endpoint) + log.Println("debug", r) + // abort to error handle c.AbortWithError(502, err) - // send notification - upstreams := []OPENAI_UPSTREAM{} - db.Find(&upstreams) - content := fmt.Sprintf("[%s] OpenAI 转发出错 ID: %d... 密钥: [%s] 上游: [%s] 错误: %s", - c.ClientIP(), - upstream.ID, upstream.SK, upstream.Endpoint, err.Error(), - ) - if err.Error() != "context canceled" && r.Response.StatusCode != 400 { - go sendMatrixMessage(content) - go sendFeishuMessage(content) + log.Println("response is", r.Response) + + if record.Status == 0 { + record.Status = 502 + } + if record.Response == "" { + record.Response = err.Error() + } + if r.Response != nil { + record.Status = r.Response.StatusCode } - log.Println("response is", r.Response) } func() { @@ -294,7 +296,6 @@ func main() { record.Response = fetchResp.Choices[0].Message.Content } else { log.Println("Unknown content type", contentType) - return } } @@ -302,7 +303,7 @@ func main() { record.Body = "" } - log.Println("Record result:", record.Response) + log.Println("Record result:", record.Status, record.Response) record.ElapsedTime = time.Now().Sub(record.CreatedAt) if db.Create(&record).Error != nil { log.Println("Error to save record:", record) From 34a6bb851584f23a9690b12803b4903ee770f044 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Wed, 18 Oct 2023 14:01:24 +0800 Subject: [PATCH 15/16] add feishu and matrix error report --- main.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/main.go b/main.go index ec1f721..dd17a60 100644 --- a/main.go +++ b/main.go @@ -308,6 +308,11 @@ func main() { if db.Create(&record).Error != nil { log.Println("Error to save record:", record) } + if record.Status != 200 { + errMessage := fmt.Sprintf("IP: %s request %s error %d with %s", record.IP, upstream.Endpoint, record.Status, record.Response) + go sendFeishuMessage(errMessage) + go sendMatrixMessage(errMessage) + } }) engine.Run(*listenAddr) From 449d636749820b8a489d1296f03d78c5305c2543 Mon Sep 17 00:00:00 2001 From: heimoshuiyu Date: Wed, 18 Oct 2023 17:57:46 +0800 Subject: [PATCH 16/16] ignore context cancle --- main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main.go b/main.go index dd17a60..105fe4e 100644 --- a/main.go +++ b/main.go @@ -308,7 +308,7 @@ func main() { if db.Create(&record).Error != nil { log.Println("Error to save record:", record) } - if record.Status != 200 { + if record.Status != 200 && record.Response != "context canceled" { errMessage := fmt.Sprintf("IP: %s request %s error %d with %s", record.IP, upstream.Endpoint, record.Status, record.Response) go sendFeishuMessage(errMessage) go sendMatrixMessage(errMessage)