update logs

api.go

@@ -24,6 +24,7 @@ import (
 	"crypto/subtle"
 	"encoding/json"
 	"fmt"
+	"mime"
 	"net/http"
 	"strconv"
 	"strings"
@@ -33,6 +34,7 @@ import (
 	"runtime/pprof"
 
 	"github.com/golang-jwt/jwt/v5"
+	"github.com/gorilla/websocket"
 )
 
 const (
@@ -190,6 +192,10 @@ func (cr *Cluster) apiAuthHandleFunc(next http.HandlerFunc) http.Handler {
 }
 
 func (cr *Cluster) initAPIv0() http.Handler {
+	wsUpgrader := websocket.Upgrader{
+		HandshakeTimeout: time.Minute,
+	}
+
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", func(rw http.ResponseWriter, req *http.Request) {
 		writeJson(rw, http.StatusNotFound, Map{
@@ -236,17 +242,54 @@ func (cr *Cluster) initAPIv0() http.Handler {
 			})
 			return
 		}
-		username, password := config.Dashboard.Username, config.Dashboard.Password
-		if username == "" || password == "" {
+
+		var (
+			authUser, authPass string
+		)
+		ct, _, err := mime.ParseMediaType(req.Header.Get("Content-Type"))
+		if err != nil {
+			writeJson(rw, http.StatusBadRequest, Map{
+				"error":        "Unexpected Content-Type",
+				"content-type": req.Header.Get("Content-Type"),
+				"message":      err.Error(),
+			})
+			return
+		}
+		switch ct {
+		case "application/x-www-form-urlencoded":
+			authUser = req.PostFormValue("username")
+			authPass = req.PostFormValue("password")
+		case "application/json":
+			var data struct {
+				User string `json:"username"`
+				Pass string `json:"password"`
+			}
+			if err := json.NewDecoder(req.Body).Decode(&data); err != nil {
+				writeJson(rw, http.StatusBadRequest, Map{
+					"error":   "Cannot decode json body",
+					"message": err.Error(),
+				})
+				return
+			}
+			authUser, authPass = data.User, data.Pass
+		default:
+			writeJson(rw, http.StatusBadRequest, Map{
+				"error":        "Unexpected Content-Type",
+				"content-type": ct,
+			})
+			return
+		}
+
+		expectUsername, expectPassword := config.Dashboard.Username, config.Dashboard.Password
+		if expectUsername == "" || expectPassword == "" {
 			writeJson(rw, http.StatusUnauthorized, Map{
 				"error": "The username or password was not set on the server",
 			})
 			return
 		}
-		user := req.Header.Get("X-Username")
-		pass := req.Header.Get("X-Password")
-		if subtle.ConstantTimeCompare(([]byte)(username), ([]byte)(user)) == 0 ||
-			subtle.ConstantTimeCompare(([]byte)(password), ([]byte)(pass)) == 0 {
+		expectPassword = asSha256Hex(expectPassword)
+		if subtle.ConstantTimeCompare(([]byte)(expectUsername), ([]byte)(authUser)) == 0 ||
+			subtle.ConstantTimeCompare(([]byte)(expectPassword), ([]byte)(authPass)) == 0 {
 			writeJson(rw, http.StatusUnauthorized, Map{
 				"error": "The username or password is incorrect",
 			})
@@ -255,7 +298,8 @@ func (cr *Cluster) initAPIv0() http.Handler {
 		token, err := cr.generateToken(cli)
 		if err != nil {
 			writeJson(rw, http.StatusInternalServerError, Map{
-				"error": err.Error(),
+				"error":   "Cannot generate token",
+				"message": err.Error(),
 			})
 			return
 		}
@@ -264,8 +308,13 @@ func (cr *Cluster) initAPIv0() http.Handler {
 		})
 	})
 	mux.Handle("/log", cr.apiAuthHandleFunc(func(rw http.ResponseWriter, req *http.Request) {
-		rw.WriteHeader(http.StatusOK)
-		e := json.NewEncoder(rw)
+		conn, err := wsUpgrader.Upgrade(rw, req, nil)
+		if err != nil {
+			http.Error(rw, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		defer conn.Close()
+
 		ctx, cancel := context.WithCancel(req.Context())
 		defer cancel()
 
@@ -274,27 +323,38 @@ func (cr *Cluster) initAPIv0() http.Handler {
 			level = LogLevelDebug
 		}
 
+		type logObj struct {
+			Time  int64  `json:"time"`
+			Level string `json:"lvl"`
+			Log   string `json:"log"`
+		}
+		c := make(chan *logObj, 64)
 		unregister := RegisterLogMonitor(level, func(ts int64, level LogLevel, log string) {
-			type logObj struct {
-				Time  int64  `json:"time"`
-				Level string `json:"lvl"`
-				Log   string `json:"log"`
-			}
-			var v = logObj{
+			select {
+			case c <- &logObj{
 				Time:  ts,
 				Level: level.String(),
 				Log:   log,
-			}
-			if err := e.Encode(v); err != nil {
-				e.Encode(err.Error())
-				cancel()
-				return
+			}:
+			default:
 			}
 		})
 		defer unregister()
 
-		select {
-		case <-ctx.Done():
+		for {
+			select {
+			case v := <-c:
+				if err := conn.WriteJSON(v); err != nil {
+					return
+				}
+			case <-time.After(time.Minute):
+				if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(time.Second*15)); err != nil {
+					logErrorf("Error when sending ping message on log socket: %v", err)
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
 		}
 	}))
 	mux.Handle("/pprof", cr.apiAuthHandleFunc(func(rw http.ResponseWriter, req *http.Request) {

go.mod

@@ -4,11 +4,14 @@ go 1.21.6
 
 require (
 	github.com/LiterMC/socket.io v0.1.7
+	github.com/golang-jwt/jwt/v5 v5.2.0
+	github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79
 	github.com/hamba/avro/v2 v2.18.0
 	github.com/klauspost/compress v1.17.4
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/redis/go-redis/v9 v9.4.0
 	github.com/studio-b12/gowebdav v0.9.0
+	github.com/vbauerster/mpb/v8 v8.7.2
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -17,16 +20,13 @@ require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
-	github.com/golang-jwt/jwt/v5 v5.2.0 // indirect
 	github.com/gorilla/websocket v1.5.1 // indirect
-	github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/mattn/go-runewidth v0.0.15 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/rivo/uniseg v0.4.4 // indirect
-	github.com/vbauerster/mpb/v8 v8.7.2 // indirect
 	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
 )

go.sum

@@ -1,17 +1,3 @@
-github.com/LiterMC/socket.io v0.1.0 h1:p3SGNJRKaTldk5Weye1EvKG92l02fLyRgRDmkcLzC7U=
-github.com/LiterMC/socket.io v0.1.0/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
-github.com/LiterMC/socket.io v0.1.1 h1:3DDMHFIG73HlUfjrH8bm1WPGHR+bEWbxQEofUr8pQeg=
-github.com/LiterMC/socket.io v0.1.1/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
-github.com/LiterMC/socket.io v0.1.2 h1:iCWxwjqEiGXDm8v8b3hoXO1SDnfpqFEBjAVOaNQE7KY=
-github.com/LiterMC/socket.io v0.1.2/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
-github.com/LiterMC/socket.io v0.1.3 h1:bqpPBwgocbLgxYVlilUwcNluQXJqKqWGcitovb/SCxc=
-github.com/LiterMC/socket.io v0.1.3/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
-github.com/LiterMC/socket.io v0.1.4 h1:2FyJsbLkRYwqOZOtHEPGmdnIRSbtwI02QnYeYh7oACY=
-github.com/LiterMC/socket.io v0.1.4/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
-github.com/LiterMC/socket.io v0.1.5 h1:EaCqdCqQuG+Jms+q1Rq9Hivfz0sYTE94bD7Ml9RwKVI=
-github.com/LiterMC/socket.io v0.1.5/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
-github.com/LiterMC/socket.io v0.1.6 h1:SbgYRcbQUVtsQr4e7uvr2vtN5Ybxztjvlv7ASH4S320=
-github.com/LiterMC/socket.io v0.1.6/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
 github.com/LiterMC/socket.io v0.1.7 h1:HP4ZJQ1bLOjH2NxalgqSTnU/ukpivYoFjIZLy4zsoqA=
 github.com/LiterMC/socket.io v0.1.7/go.mod h1:60lM7+qdBnP64Fk2fB6WmAS6HxI6WCdhlcvaSnutx50=
 github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1ow=

handler.go

@@ -20,6 +20,7 @@
 package main
 
 import (
+	"context"
 	"crypto"
 	"encoding/hex"
 	"encoding/json"
@@ -49,7 +50,6 @@ type statusResponseWriter struct {
 	http.ResponseWriter
 	status int
 	wrote  int64
-	info   map[string]any
 }
 
 func (w *statusResponseWriter) WriteHeader(status int) {
@@ -63,16 +63,13 @@ func (w *statusResponseWriter) Write(buf []byte) (n int, err error) {
 	return
 }
 
-func (w *statusResponseWriter) SetInfo(key string, value any) {
-	if w.info == nil {
-		w.info = make(map[string]any)
-	}
-	w.info[key] = value
-}
+const (
+	AccessLogExtraCtxKey = "handle.access.extra"
+)
 
-func SetAccessInfo(rw http.ResponseWriter, key string, value any) {
-	if srw, ok := rw.(*statusResponseWriter); ok {
-		srw.SetInfo(key, value)
+func SetAccessInfo(req *http.Request, key string, value any) {
+	if info, ok := req.Context().Value(AccessLogExtraCtxKey).(map[string]any); ok {
+		info[key] = value
 	}
 }
 
@@ -157,11 +154,13 @@ func (cr *Cluster) GetHandler() (handler http.Handler) {
 				UA:     ua,
 			})
 
+			extraInfoMap := make(map[string]any)
+			req = req.WithContext(context.WithValue(req.Context(), AccessLogExtraCtxKey, extraInfoMap))
 			next.ServeHTTP(srw, req)
 
 			used := time.Since(start)
-			LogAccess(LogLevelInfo, &accessRecord{
-				Type:    "served",
+			accRec := &accessRecord{
+				Type:    "access",
 				Status:  srw.status,
 				Used:    used,
 				Content: srw.wrote,
@@ -170,8 +169,13 @@ func (cr *Cluster) GetHandler() (handler http.Handler) {
 				Method:  req.Method,
 				URI:     req.RequestURI,
 				UA:      ua,
-				Extra:   srw.info,
-			})
+				Extra:   extraInfoMap,
+			}
+			if len(extraInfoMap) > 0 {
+				accRec.Extra = extraInfoMap
+			}
+			LogAccess(LogLevelInfo, accRec)
+
 			if srw.status < 200 && 400 <= srw.status {
 				return
 			}
@@ -206,7 +210,8 @@ func (cr *Cluster) GetHandler() (handler http.Handler) {
 				case <-updateTicker.C:
 					cr.stats.mux.Lock()
 
-					logInfof("Served %d requests, %s, used %.2fs, %s/s", total, bytesToUnit(totalBytes), totalUsed, bytesToUnit(totalBytes/60))
+					logInfof("Served %d requests, total responsed body = %s, total used CPU time = %.2fs",
+						total, bytesToUnit(totalBytes), totalUsed)
 					for ua, v := range uas {
 						if ua == "" {
 							ua = "[Unknown]"
@@ -234,7 +239,7 @@ func (cr *Cluster) GetHandler() (handler http.Handler) {
 					select {
 					case <-cr.WaitForEnable():
 						disabled = cr.Disabled()
-					case <-time.After(time.Minute * 10):
+					case <-time.After(time.Hour):
 						return
 					}
 				}
@@ -382,7 +387,7 @@ func (cr *Cluster) handleDownload(rw http.ResponseWriter, req *http.Request, has
 		return true
 	})
 	if storage != nil {
-		SetAccessInfo(rw, "storage", storage.String())
+		SetAccessInfo(req, "storage", storage.String())
 	}
 	if err != nil {
 		logDebugf("[handler]: failed to serve download: %v", err)

logger.go

@@ -362,7 +362,9 @@ func LogAccess(level LogLevel, data any) {
 
 	var buf [512]byte
 	bts := bytes.NewBuffer(buf[:0])
-	json.NewEncoder(bts).Encode(data)
+	e := json.NewEncoder(bts)
+	e.SetEscapeHTML(false)
+	e.Encode(data)
 
 	var s string
 	if fs, ok := data.(fmt.Stringer); ok {

util.go

@@ -23,8 +23,10 @@ import (
 	"context"
 	"crypto"
 	crand "crypto/rand"
+	"crypto/sha256"
 	"crypto/x509"
 	"encoding/base64"
+	"encoding/hex"
 	"encoding/pem"
 	"errors"
 	"fmt"
@@ -611,6 +613,11 @@ func HexTo256(s string) (n int) {
 	return hexToNumMap[s[0]]*0x10 + hexToNumMap[s[1]]
 }
 
+func asSha256Hex(s string) string {
+	buf := sha256.Sum256(([]byte)(s))
+	return hex.EncodeToString(buf[:])
+}
+
 func genRandB64(n int) (s string, err error) {
 	buf := make([]byte, n)
 	if _, err = crand.Read(buf); err != nil {