From f200c02920a3058ba3fc790848f740d880a6532b Mon Sep 17 00:00:00 2001 From: Chanu Ollala Date: Thu, 23 Jul 2026 17:57:25 -0700 Subject: [PATCH] etl: acknowledge completed direct puts * target returns 204 with Ais-Direct-Put-Complete and stored length * Go ETL webserver propagates the marker and length, including zero * preserve legacy response handling and add completion coverage Signed-off-by: Chanu Ollala --- ais/tgtetl.go | 13 ++- api/apc/headers.go | 4 + ext/etl/webserver/webserver.go | 32 ++++- ext/etl/webserver/webserver_internal_test.go | 116 +++++++++++++++++++ 4 files changed, 158 insertions(+), 7 deletions(-) diff --git a/ais/tgtetl.go b/ais/tgtetl.go index 6779f387c2..96846885ea 100644 --- a/ais/tgtetl.go +++ b/ais/tgtetl.go @@ -9,6 +9,7 @@ import ( "fmt" "net/http" "net/url" + "strconv" "strings" "github.com/NVIDIA/aistore/api/apc" @@ -329,6 +330,9 @@ func (t *target) getObjectETL(w http.ResponseWriter, r *http.Request, dpq *dpq, // PUT /v1/etl/_object///?uuid= // Handles PUT requests from ETL containers (K8s Pods). +// Acknowledges a successful direct put with 204 (No Content) and the +// (apc.HdrDirectPutComplete, apc.HdrDirectPutLength) response headers; +// ETL webservers propagate this ack back through the pipeline. func (t *target) putObjectETL(w http.ResponseWriter, r *http.Request, dpq *dpq, bck *meta.Bck, objName string) { config := cmn.GCO.Get() lom := core.AllocLOM(objName) @@ -344,10 +348,17 @@ func (t *target) putObjectETL(w http.ResponseWriter, r *http.Request, dpq *dpq, } } ecode, err := t.putObject(w, r, dpq, lom, config) - core.FreeLOM(lom) if err != nil { + core.FreeLOM(lom) t.writeErr(w, r, err, ecode) + return } + size := lom.Lsize() + core.FreeLOM(lom) + + w.Header().Set(apc.HdrDirectPutComplete, "true") + w.Header().Set(apc.HdrDirectPutLength, strconv.FormatInt(size, 10)) + w.WriteHeader(http.StatusNoContent) } // HEAD /v1/etl/_object/// diff --git a/api/apc/headers.go b/api/apc/headers.go index 02fbd94ebf..e06a5a45bc 100644 --- a/api/apc/headers.go +++ b/api/apc/headers.go @@ -131,6 +131,10 @@ const ( // ETL HdrETLPodInfo = aisPrefix + "ETL-Pod-Info" // serialized etl.Info HdrDirectPutLength = aisPrefix + "Direct-Put-Length" + // Set by the target (alongside 204 and HdrDirectPutLength) after storing an + // object via the direct-put endpoint. Presence-based - the value is ignored; + // intermediate ETL webservers propagate it back through the pipeline. + HdrDirectPutComplete = aisPrefix + "Direct-Put-Complete" // ETL → AIS retry contract: emitted by the ETL webserver alongside HTTP 503 // to signal that the ETL bailed on a transient direct-put failure without // trying locally (one-shot body case). AIS retries the whole PUT against diff --git a/ext/etl/webserver/webserver.go b/ext/etl/webserver/webserver.go index fb9c79d9e3..0fcb133a3a 100644 --- a/ext/etl/webserver/webserver.go +++ b/ext/etl/webserver/webserver.go @@ -79,6 +79,7 @@ type ( StatusCode int Size int64 DirectPutLength int64 + Complete bool // ack carried apc.HdrDirectPutComplete; propagate it Body io.ReadCloser } ) @@ -310,7 +311,12 @@ func (base *etlServerBase) handleDirectPut(w http.ResponseWriter, transformedRea if dresp.Body != nil { setResponseHeaders(w.Header(), dresp.Size) } - if dresp.DirectPutLength > 0 { + if dresp.Complete { + // delivered ack: propagate the marker and the length verbatim, + // including a length of 0 (empty object stored) + w.Header().Set(apc.HdrDirectPutComplete, "true") + w.Header().Set(apc.HdrDirectPutLength, strconv.FormatInt(dresp.DirectPutLength, 10)) + } else if dresp.DirectPutLength > 0 { w.Header().Set(apc.HdrDirectPutLength, strconv.FormatInt(dresp.DirectPutLength, 10)) } w.WriteHeader(dresp.StatusCode) @@ -447,6 +453,22 @@ func (base *etlServerBase) directPut(directPutURL string, r io.ReadCloser, size length = directPutLength } + // delivered ack from the target (directly or propagated by a downstream + // stage). Presence-based: the value is ignored, and the marker decides + // regardless of status (the target pairs it with 204). + if _, ok := resp.Header[apc.HdrDirectPutComplete]; ok { + resp.Body.Close() + return &directPutResponse{ + StatusCode: http.StatusNoContent, + DirectPutLength: int64(length), + Complete: true, + Body: nil, + }, nil + } + + // Legacy handling below, unchanged: kept for targets that predate + // apc.HdrDirectPutComplete; to be phased out with them. + // delivered to target, no content if resp.StatusCode == http.StatusNoContent { resp.Body.Close() @@ -458,11 +480,9 @@ func (base *etlServerBase) directPut(directPutURL string, r io.ReadCloser, size } if resp.StatusCode == http.StatusOK { - // NOTE: keyed on Content-Length, not the read body (contrast Python's - // handle_direct_put_response): a chunked 200 (ContentLength == -1) is - // always forwarded as content - an empty transform result is a valid - // object. Python currently maps any empty 200 body to 204/"delivered"; - // to be aligned with this behavior in a separate PR. + // NOTE: keyed on Content-Length, not the read body: a chunked 200 + // (ContentLength == -1) is always forwarded as content because an + // empty transform result is a valid object. // from target, no content if resp.ContentLength == 0 { diff --git a/ext/etl/webserver/webserver_internal_test.go b/ext/etl/webserver/webserver_internal_test.go index c852dc419d..1a3da5e8ec 100644 --- a/ext/etl/webserver/webserver_internal_test.go +++ b/ext/etl/webserver/webserver_internal_test.go @@ -282,6 +282,117 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) { directPutPath = "ais@#test/obj" ) + t.Run("204 with Direct-Put-Complete propagates marker and length", func(t *testing.T) { + const directPutLength = 1234 + directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method) + w.Header().Set(apc.HdrDirectPutComplete, "true") + w.Header().Set(apc.HdrDirectPutLength, strconv.Itoa(directPutLength)) + w.WriteHeader(http.StatusNoContent) + })) + defer directPutTargetServer.Close() + + var ( + content = []byte("test bytes") + req = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content)) + w = httptest.NewRecorder() + ) + req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}} + + svr.putHandler(w, req) + + resp := w.Result() + defer resp.Body.Close() + + tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode) + _, marked := resp.Header[apc.HdrDirectPutComplete] + tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated") + got := resp.Header.Get(apc.HdrDirectPutLength) + tassert.Fatalf(t, got == strconv.Itoa(directPutLength), "expected Direct-Put-Length %d, got %q", directPutLength, got) + }) + + t.Run("Direct-Put-Complete with length 0 propagates 0", func(t *testing.T) { + directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method) + w.Header().Set(apc.HdrDirectPutComplete, "true") + w.Header().Set(apc.HdrDirectPutLength, "0") // empty object stored + w.WriteHeader(http.StatusNoContent) + })) + defer directPutTargetServer.Close() + + var ( + content = []byte("test bytes") + req = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content)) + w = httptest.NewRecorder() + ) + req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}} + + svr.putHandler(w, req) + + resp := w.Result() + defer resp.Body.Close() + + tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode) + _, marked := resp.Header[apc.HdrDirectPutComplete] + tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated") + got := resp.Header.Get(apc.HdrDirectPutLength) + tassert.Fatalf(t, got == "0", "expected Direct-Put-Length 0, got %q", got) + }) + + t.Run("Direct-Put-Complete with empty value still delivered", func(t *testing.T) { + directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method) + w.Header().Set(apc.HdrDirectPutComplete, "") // presence-based: value is ignored + w.WriteHeader(http.StatusNoContent) + })) + defer directPutTargetServer.Close() + + var ( + content = []byte("test bytes") + req = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content)) + w = httptest.NewRecorder() + ) + req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}} + + svr.putHandler(w, req) + + resp := w.Result() + defer resp.Body.Close() + + tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode) + _, marked := resp.Header[apc.HdrDirectPutComplete] + tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated") + }) + + t.Run("marked 200 normalized to 204", func(t *testing.T) { + directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method) + w.Header().Set(apc.HdrDirectPutComplete, "true") + w.Header().Set(apc.HdrDirectPutLength, "5") + w.WriteHeader(http.StatusOK) // the marker decides regardless of status + })) + defer directPutTargetServer.Close() + + var ( + content = []byte("test bytes") + req = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content)) + w = httptest.NewRecorder() + ) + req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}} + + svr.putHandler(w, req) + + resp := w.Result() + defer resp.Body.Close() + + tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode) + _, marked := resp.Header[apc.HdrDirectPutComplete] + tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated") + got := resp.Header.Get(apc.HdrDirectPutLength) + tassert.Fatalf(t, got == "5", "expected Direct-Put-Length 5, got %q", got) + }) + + // legacy path: bare 204 ack from a target that predates Direct-Put-Complete t.Run("204 forwards Direct-Put-Length", func(t *testing.T) { const directPutLength = 1234 directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -304,6 +415,8 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) { defer resp.Body.Close() tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode) + _, marked := resp.Header[apc.HdrDirectPutComplete] + tassert.Fatalf(t, !marked, "legacy delivery must not gain Direct-Put-Complete") got := resp.Header.Get(apc.HdrDirectPutLength) tassert.Fatalf(t, got == strconv.Itoa(directPutLength), "expected Direct-Put-Length %d, got %q", directPutLength, got) }) @@ -368,6 +481,7 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) { tassert.Fatalf(t, bytes.Equal(result, expected), "expected body %s, got %s", expected, result) }) + // legacy path: bare 200 ack from a target that predates Direct-Put-Complete t.Run("200 empty body maps to 204", func(t *testing.T) { nextStageServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method) @@ -388,6 +502,8 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) { defer resp.Body.Close() tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode) + _, marked := resp.Header[apc.HdrDirectPutComplete] + tassert.Fatalf(t, !marked, "legacy delivery must not gain Direct-Put-Complete") got := resp.Header.Get(apc.HdrDirectPutLength) tassert.Fatalf(t, got == strconv.Itoa(len(content)), "expected Direct-Put-Length %d, got %q", len(content), got) })