etl: include object path from control message in ws:// handler args

Nahemah1022 · Nahemah1022 · commit 82fcb58b08f3 · 2025-05-14T09:46:59.000-07:00
* replace logging and error response with `nlog` and `cmn.WriteErr`

Signed-off-by: Tony Chen &lt;a122774007@gmail.com&gt;
diff --git a/ext/etl/webserver/cmn.go b/ext/etl/webserver/cmn.go
@@ -8,23 +8,14 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"log"
 	"net/http"
-	"runtime/debug"
 	"strconv"
 
 	"github.com/NVIDIA/aistore/cmn/cos"
 )
 
 const GetContentType = "binary/octet-stream"
 
-func invalidMsgHandler(w http.ResponseWriter, format string, a ...any) {
-	logErrorf(format, a...)
-	w.Header().Set(cos.HdrContentType, "text/plain")
-	w.WriteHeader(http.StatusBadRequest)
-	fmt.Fprintf(w, format, a...)
-}
-
 func setResponseHeaders(header http.Header, size int64) {
 	header.Set(cos.HdrContentLength, strconv.FormatInt(size, 10))
 	header.Set(cos.HdrContentType, GetContentType)
@@ -49,7 +40,3 @@ func wrapHTTPError(resp *http.Response, err error) (*http.Response, error) {
 
 	return resp, nil
 }
-
-func logErrorf(format string, a ...any) {
-	log.Printf(string(debug.Stack())+" : "+format, a...)
-}
diff --git a/ext/etl/webserver/webserver.go b/ext/etl/webserver/webserver.go
@@ -12,13 +12,14 @@ import (
 	"net/url"
 	"os"
 	"path"
-	"path/filepath"
 	"strings"
 
 	"github.com/gorilla/websocket"
 
 	"github.com/NVIDIA/aistore/api/apc"
+	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
+	"github.com/NVIDIA/aistore/cmn/nlog"
 	"github.com/NVIDIA/aistore/ext/etl"
 )
 
@@ -72,7 +73,7 @@ func (*etlServerBase) healthHandler(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusOK)
 		w.Write([]byte("Running"))
 	default:
-		invalidMsgHandler(w, "invalid http method %s", r.Method)
+		cmn.WriteErr405(w, r, r.Method)
 	}
 }
 
@@ -83,7 +84,7 @@ func (base *etlServerBase) handler(w http.ResponseWriter, r *http.Request) {
 	case http.MethodGet:
 		base.getHandler(w, r)
 	default:
-		invalidMsgHandler(w, "invalid http method %s", r.Method)
+		cmn.WriteErr405(w, r, r.Method)
 	}
 }
 
@@ -100,16 +101,16 @@ func (base *etlServerBase) putHandler(w http.ResponseWriter, r *http.Request) {
 	case etl.ArgTypeFQN:
 		objReader, err = base.getFQNReader(r.URL.Path)
 		if err != nil {
-			invalidMsgHandler(w, "GET from FQN failed; err %v", err)
+			cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "GET from FQN failed", err))
 			return
 		}
 	default:
-		invalidMsgHandler(w, "invalid arg_type: %s", base.argType)
+		cmn.WriteErrMsg(w, r, "invalid arg_type: "+base.argType)
 	}
 
 	transformedReader, err := base.Transform(objReader, r.URL.Path, r.URL.Query().Get(apc.QparamETLTransformArgs))
 	if err != nil {
-		invalidMsgHandler(w, "failed to transform: %v", err)
+		cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "failed to transform", err))
 		return
 	}
 	objReader.Close() // objReader is supposed to be fully consumed by the custom transform function - safe to close at this point
@@ -129,14 +130,14 @@ func (base *etlServerBase) putHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if _, err := io.Copy(w, transformedReader); err != nil {
-		invalidMsgHandler(w, "%v", err)
+		cmn.WriteErr(w, r, err)
 	}
 }
 
 // GET /
 func (base *etlServerBase) getHandler(w http.ResponseWriter, r *http.Request) {
 	if base.aisTargetURL == "" {
-		invalidMsgHandler(w, "missing env variable AIS_TARGET_URL")
+		cmn.WriteErrMsg(w, r, "missing env variable AIS_TARGET_URL")
 		return
 	}
 
@@ -153,30 +154,30 @@ func (base *etlServerBase) getHandler(w http.ResponseWriter, r *http.Request) {
 	case etl.ArgTypeDefault, etl.ArgTypeURL:
 		resp, err := wrapHTTPError(base.client.Get(fmt.Sprintf("%s/%s", base.aisTargetURL, p))) //nolint:bodyclose // is closed from objReader
 		if err != nil {
-			invalidMsgHandler(w, "GET from AIStore failed; err %v", err)
+			cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "GET from AIStore failed", err))
 			return
 		}
 		objReader = resp.Body
 	case etl.ArgTypeFQN:
 		objReader, err = base.getFQNReader(r.URL.Path)
 		if err != nil {
-			invalidMsgHandler(w, "GET from FQN failed; err %v", err)
+			cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "GET from FQN failed", err))
 			return
 		}
 	default:
-		invalidMsgHandler(w, "invalid arg_type: %s", base.argType)
+		cmn.WriteErrMsg(w, r, "invalid arg_type: "+base.argType)
 	}
 
 	transformedReader, err := base.Transform(objReader, r.URL.Path, r.URL.Query().Get(apc.QparamETLTransformArgs))
 	if err != nil {
-		invalidMsgHandler(w, "failed to transform: %v", err)
+		cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "failed to transform", err))
 		return
 	}
 	objReader.Close() // objReader is supposed to be fully consumed by the custom transform function - safe to close at this point
 
 	if directPutURL := r.Header.Get(apc.HdrNodeURL); directPutURL != "" {
 		if err := base.handleDirectPut(directPutURL, transformedReader); err != nil {
-			invalidMsgHandler(w, "%v", err)
+			w.WriteHeader(http.StatusBadRequest)
 		} else {
 			setResponseHeaders(w.Header(), 0)
 			w.WriteHeader(http.StatusNoContent)
@@ -185,7 +186,7 @@ func (base *etlServerBase) getHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if _, err := io.Copy(w, transformedReader); err != nil {
-		logErrorf("%v", err)
+		cmn.WriteErr(w, r, err)
 	}
 }
 
@@ -195,7 +196,7 @@ func (base *etlServerBase) websocketHandler(w http.ResponseWriter, r *http.Reque
 	}
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
-		logErrorf("failed to upgrade to websocket: %v", err)
+		nlog.Errorln("failed to upgrade to websocket", err)
 		return
 	}
 	defer conn.Close()
@@ -209,32 +210,32 @@ func (base *etlServerBase) websocketHandler(w http.ResponseWriter, r *http.Reque
 			if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
 				break
 			}
-			log.Printf("error reading control message: %v", err)
+			nlog.Errorln("error reading control message", err)
 			break
 		}
 
 		if ctrl.FQN != "" {
 			fqn, err := url.PathUnescape(ctrl.FQN)
 			if err != nil {
-				logErrorf("failed to unescape: %v", err)
+				nlog.Errorln("failed to unescape", err)
 				break
 			}
 			reader, err = base.getFQNReader(fqn)
 			if err != nil {
-				logErrorf("failed to read FQN: %v", err)
+				nlog.Errorln("failed to read FQN", err)
 				break
 			}
 		} else {
 			_, reader, err = conn.NextReader()
 			if err != nil {
-				logErrorf("failed to read binary: %v", err)
+				nlog.Errorln("failed to read binary", err)
 				break
 			}
 		}
 
-		transformed, err := base.Transform(io.NopCloser(reader), "", ctrl.Targs)
+		transformed, err := base.Transform(io.NopCloser(reader), ctrl.Path, ctrl.Targs)
 		if err != nil {
-			logErrorf("transform error: %v", err)
+			nlog.Errorln("transform error", err)
 			break
 		}
 
@@ -244,23 +245,23 @@ func (base *etlServerBase) websocketHandler(w http.ResponseWriter, r *http.Reque
 				conn.WriteMessage(websocket.TextMessage, cos.UnsafeB("direct put success"))
 				continue
 			}
-			logErrorf("direct put failed: %v", err)
+			nlog.Errorln("direct put failed", err)
 		}
 
 		writer, err := conn.NextWriter(websocket.BinaryMessage)
 		if err != nil {
-			logErrorf("write error: %v", err)
+			nlog.Errorln("write error", err)
 			break
 		}
 		if _, err := io.Copy(writer, transformed); err != nil {
-			logErrorf("copy error: %v", err)
+			nlog.Errorln("copy error", err)
 		}
 		writer.Close()
 	}
 }
 
 func (*etlServerBase) getFQNReader(urlPath string) (io.ReadCloser, error) {
-	fqn := filepath.Join("/", strings.TrimLeft(urlPath, "/"))
+	fqn := cos.JoinWords(urlPath)
 	return os.Open(fqn)
 }
 

Original file line number	Diff line number	Diff line change
`@@ -12,13 +12,14 @@ import (`
`12`	`12`	`"net/url"`
`13`	`13`	`"os"`
`14`	`14`	`"path"`
`15`		`- "path/filepath"`
`16`	`15`	`"strings"`
`17`	`16`
`18`	`17`	`"github.com/gorilla/websocket"`
`19`	`18`
`20`	`19`	`"github.com/NVIDIA/aistore/api/apc"`
	`20`	`+ "github.com/NVIDIA/aistore/cmn"`
`21`	`21`	`"github.com/NVIDIA/aistore/cmn/cos"`
	`22`	`+ "github.com/NVIDIA/aistore/cmn/nlog"`
`22`	`23`	`"github.com/NVIDIA/aistore/ext/etl"`
`23`	`24`	`)`
`24`	`25`
`@@ -72,7 +73,7 @@ func (etlServerBase) healthHandler(w http.ResponseWriter, r http.Request) {`
`72`	`73`	`w.WriteHeader(http.StatusOK)`
`73`	`74`	`w.Write([]byte("Running"))`
`74`	`75`	`default:`
`75`		`- invalidMsgHandler(w, "invalid http method %s", r.Method)`
	`76`	`+ cmn.WriteErr405(w, r, r.Method)`
`76`	`77`	`}`
`77`	`78`	`}`
`78`	`79`
`@@ -83,7 +84,7 @@ func (base etlServerBase) handler(w http.ResponseWriter, r http.Request) {`
`83`	`84`	`case http.MethodGet:`
`84`	`85`	`base.getHandler(w, r)`
`85`	`86`	`default:`
`86`		`- invalidMsgHandler(w, "invalid http method %s", r.Method)`
	`87`	`+ cmn.WriteErr405(w, r, r.Method)`
`87`	`88`	`}`
`88`	`89`	`}`
`89`	`90`
`@@ -100,16 +101,16 @@ func (base etlServerBase) putHandler(w http.ResponseWriter, r http.Request) {`
`100`	`101`	`case etl.ArgTypeFQN:`
`101`	`102`	`objReader, err = base.getFQNReader(r.URL.Path)`
`102`	`103`	`if err != nil {`
`103`		`- invalidMsgHandler(w, "GET from FQN failed; err %v", err)`
	`104`	`+ cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "GET from FQN failed", err))`
`104`	`105`	`return`
`105`	`106`	`}`
`106`	`107`	`default:`
`107`		`- invalidMsgHandler(w, "invalid arg_type: %s", base.argType)`
	`108`	`+ cmn.WriteErrMsg(w, r, "invalid arg_type: "+base.argType)`
`108`	`109`	`}`
`109`	`110`
`110`	`111`	`transformedReader, err := base.Transform(objReader, r.URL.Path, r.URL.Query().Get(apc.QparamETLTransformArgs))`
`111`	`112`	`if err != nil {`
`112`		`- invalidMsgHandler(w, "failed to transform: %v", err)`
	`113`	`+ cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "failed to transform", err))`
`113`	`114`	`return`
`114`	`115`	`}`
`115`	`116`	`objReader.Close() // objReader is supposed to be fully consumed by the custom transform function - safe to close at this point`
`@@ -129,14 +130,14 @@ func (base etlServerBase) putHandler(w http.ResponseWriter, r http.Request) {`
`129`	`130`	`}`
`130`	`131`
`131`	`132`	`if _, err := io.Copy(w, transformedReader); err != nil {`
`132`		`- invalidMsgHandler(w, "%v", err)`
	`133`	`+ cmn.WriteErr(w, r, err)`
`133`	`134`	`}`
`134`	`135`	`}`
`135`	`136`
`136`	`137`	`// GET /`
`137`	`138`	`func (base etlServerBase) getHandler(w http.ResponseWriter, r http.Request) {`
`138`	`139`	`if base.aisTargetURL == "" {`
`139`		`- invalidMsgHandler(w, "missing env variable AIS_TARGET_URL")`
	`140`	`+ cmn.WriteErrMsg(w, r, "missing env variable AIS_TARGET_URL")`
`140`	`141`	`return`
`141`	`142`	`}`
`142`	`143`
`@@ -153,30 +154,30 @@ func (base etlServerBase) getHandler(w http.ResponseWriter, r http.Request) {`
`153`	`154`	`case etl.ArgTypeDefault, etl.ArgTypeURL:`
`154`	`155`	`resp, err := wrapHTTPError(base.client.Get(fmt.Sprintf("%s/%s", base.aisTargetURL, p))) //nolint:bodyclose // is closed from objReader`
`155`	`156`	`if err != nil {`
`156`		`- invalidMsgHandler(w, "GET from AIStore failed; err %v", err)`
	`157`	`+ cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "GET from AIStore failed", err))`
`157`	`158`	`return`
`158`	`159`	`}`
`159`	`160`	`objReader = resp.Body`
`160`	`161`	`case etl.ArgTypeFQN:`
`161`	`162`	`objReader, err = base.getFQNReader(r.URL.Path)`
`162`	`163`	`if err != nil {`
`163`		`- invalidMsgHandler(w, "GET from FQN failed; err %v", err)`
	`164`	`+ cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "GET from FQN failed", err))`
`164`	`165`	`return`
`165`	`166`	`}`
`166`	`167`	`default:`
`167`		`- invalidMsgHandler(w, "invalid arg_type: %s", base.argType)`
	`168`	`+ cmn.WriteErrMsg(w, r, "invalid arg_type: "+base.argType)`
`168`	`169`	`}`
`169`	`170`
`170`	`171`	`transformedReader, err := base.Transform(objReader, r.URL.Path, r.URL.Query().Get(apc.QparamETLTransformArgs))`
`171`	`172`	`if err != nil {`
`172`		`- invalidMsgHandler(w, "failed to transform: %v", err)`
	`173`	`+ cmn.WriteErr(w, r, fmt.Errorf("[%s] %w", "failed to transform", err))`
`173`	`174`	`return`
`174`	`175`	`}`
`175`	`176`	`objReader.Close() // objReader is supposed to be fully consumed by the custom transform function - safe to close at this point`
`176`	`177`
`177`	`178`	`if directPutURL := r.Header.Get(apc.HdrNodeURL); directPutURL != "" {`
`178`	`179`	`if err := base.handleDirectPut(directPutURL, transformedReader); err != nil {`
`179`		`- invalidMsgHandler(w, "%v", err)`
	`180`	`+ w.WriteHeader(http.StatusBadRequest)`
`180`	`181`	`} else {`
`181`	`182`	`setResponseHeaders(w.Header(), 0)`
`182`	`183`	`w.WriteHeader(http.StatusNoContent)`
`@@ -185,7 +186,7 @@ func (base etlServerBase) getHandler(w http.ResponseWriter, r http.Request) {`
`185`	`186`	`}`
`186`	`187`
`187`	`188`	`if _, err := io.Copy(w, transformedReader); err != nil {`
`188`		`- logErrorf("%v", err)`
	`189`	`+ cmn.WriteErr(w, r, err)`
`189`	`190`	`}`
`190`	`191`	`}`
`191`	`192`
`@@ -195,7 +196,7 @@ func (base etlServerBase) websocketHandler(w http.ResponseWriter, r http.Reque`
`195`	`196`	`}`
`196`	`197`	`conn, err := upgrader.Upgrade(w, r, nil)`
`197`	`198`	`if err != nil {`
`198`		`- logErrorf("failed to upgrade to websocket: %v", err)`
	`199`	`+ nlog.Errorln("failed to upgrade to websocket", err)`
`199`	`200`	`return`
`200`	`201`	`}`
`201`	`202`	`defer conn.Close()`
`@@ -209,32 +210,32 @@ func (base etlServerBase) websocketHandler(w http.ResponseWriter, r http.Reque`
`209`	`210`	`if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {`
`210`	`211`	`break`
`211`	`212`	`}`
`212`		`- log.Printf("error reading control message: %v", err)`
	`213`	`+ nlog.Errorln("error reading control message", err)`
`213`	`214`	`break`
`214`	`215`	`}`
`215`	`216`
`216`	`217`	`if ctrl.FQN != "" {`
`217`	`218`	`fqn, err := url.PathUnescape(ctrl.FQN)`
`218`	`219`	`if err != nil {`
`219`		`- logErrorf("failed to unescape: %v", err)`
	`220`	`+ nlog.Errorln("failed to unescape", err)`
`220`	`221`	`break`
`221`	`222`	`}`
`222`	`223`	`reader, err = base.getFQNReader(fqn)`
`223`	`224`	`if err != nil {`
`224`		`- logErrorf("failed to read FQN: %v", err)`
	`225`	`+ nlog.Errorln("failed to read FQN", err)`
`225`	`226`	`break`
`226`	`227`	`}`
`227`	`228`	`} else {`
`228`	`229`	`_, reader, err = conn.NextReader()`
`229`	`230`	`if err != nil {`
`230`		`- logErrorf("failed to read binary: %v", err)`
	`231`	`+ nlog.Errorln("failed to read binary", err)`
`231`	`232`	`break`
`232`	`233`	`}`
`233`	`234`	`}`
`234`	`235`
`235`		`- transformed, err := base.Transform(io.NopCloser(reader), "", ctrl.Targs)`
	`236`	`+ transformed, err := base.Transform(io.NopCloser(reader), ctrl.Path, ctrl.Targs)`
`236`	`237`	`if err != nil {`
`237`		`- logErrorf("transform error: %v", err)`
	`238`	`+ nlog.Errorln("transform error", err)`
`238`	`239`	`break`
`239`	`240`	`}`
`240`	`241`
`@@ -244,23 +245,23 @@ func (base etlServerBase) websocketHandler(w http.ResponseWriter, r http.Reque`
`244`	`245`	`conn.WriteMessage(websocket.TextMessage, cos.UnsafeB("direct put success"))`
`245`	`246`	`continue`
`246`	`247`	`}`
`247`		`- logErrorf("direct put failed: %v", err)`
	`248`	`+ nlog.Errorln("direct put failed", err)`
`248`	`249`	`}`
`249`	`250`
`250`	`251`	`writer, err := conn.NextWriter(websocket.BinaryMessage)`
`251`	`252`	`if err != nil {`
`252`		`- logErrorf("write error: %v", err)`
	`253`	`+ nlog.Errorln("write error", err)`
`253`	`254`	`break`
`254`	`255`	`}`
`255`	`256`	`if _, err := io.Copy(writer, transformed); err != nil {`
`256`		`- logErrorf("copy error: %v", err)`
	`257`	`+ nlog.Errorln("copy error", err)`
`257`	`258`	`}`
`258`	`259`	`writer.Close()`
`259`	`260`	`}`
`260`	`261`	`}`
`261`	`262`
`262`	`263`	`func (*etlServerBase) getFQNReader(urlPath string) (io.ReadCloser, error) {`
`263`		`- fqn := filepath.Join("/", strings.TrimLeft(urlPath, "/"))`
	`264`	`+ fqn := cos.JoinWords(urlPath)`
`264`	`265`	`return os.Open(fqn)`
`265`	`266`	`}`
`266`	`267`