etl: add inspect APIs as destination-less dry-run

* add Go and Python inspect APIs that runs ETL dry-run without requiring
  callers to provide a destination bucket
* keep inspect on the existing dry-run execution path, with no
  transformed objects written back to storage
* add integration test coverage for validation failures and exact ETL
  error reporting

Signed-off-by: Tony Chen <a122774007@gmail.com>

Author: Nahemah1022

ais/proxy.go

@@ -1644,7 +1644,9 @@ func (p *proxy) _bckpost(w http.ResponseWriter, r *http.Request, msg *apc.ActMsg
 			return
 		}
 		tcbmsg.Prefix = cos.TrimPrefix(tcbmsg.Prefix)
-		if bckFrom.Equal(bckTo, true /*same BID*/, true) {
+		// Same-bucket dry-run is safe for copy and ETL: it exercises the TCB path
+		// without writing objects. ETL inspection uses this request shape.
+		if !tcbmsg.DryRun && bckFrom.Equal(bckTo, true /*same BID*/, true) {
 			if !bckFrom.IsRemote() {
 				p.writeErrf(w, r, "cannot %s bucket %q onto itself", msg.Action, bckFrom.Cname(""))
 				return

ais/test/etl_cp_multiobj_test.go

@@ -65,6 +65,35 @@ func TestETLMultiObj(t *testing.T) {
 	}
 }
 
+func TestETLInspectMultiObj(t *testing.T) {
+	tools.CheckSkip(t, &tools.SkipTestArgs{RequiredDeployment: tools.ClusterTypeK8s})
+
+	var (
+		proxyURL   = tools.RandomProxyURL(t)
+		baseParams = tools.BaseAPIParams(proxyURL)
+		bck        = cmn.Bck{Name: "etl-inspect-objects-" + cos.GenUUID(), Provider: apc.AIS}
+		m          = ioContext{t: t, num: 10, fileSize: 512, fixedSize: true, bck: bck}
+	)
+
+	tools.CreateBucket(t, proxyURL, bck, nil, true /*cleanup*/)
+	m.init(true /*cleanup*/)
+	m.puts()
+
+	initMsg := tetl.InitSpec(t, baseParams, tetl.MD5, etl.Hpush)
+	t.Cleanup(func() { tetl.StopAndDeleteETL(t, baseParams, initMsg.Name()) })
+
+	msg := &cmn.TCOMsg{}
+	msg.Transform.Name = initMsg.Name()
+	msg.ListRange.ObjNames = m.objNames
+	xid, err := api.ETLInspectMultiObj(baseParams, bck, msg)
+	tassert.CheckFatal(t, err)
+	tassert.Fatalf(t, msg.ToBck.IsEmpty() && !msg.DryRun && !msg.ContinueOnError, "inspect must not mutate caller message")
+
+	err = api.WaitForSnapsIdle(baseParams, &xact.ArgsMsg{ID: xid, Kind: apc.ActETLObjects})
+	tassert.CheckFatal(t, err)
+	checkObjectSizes(t, baseParams, bck, m.objNames, int64(m.fileSize))
+}
+
 func testETLMultiObj(t *testing.T, etlName, prefix string, bckFrom cmn.Bck, fileRange, opType string, bcktest testBucketConfig, transform transformFunc) {
 	pt, err := cos.ParseBashTemplate(fileRange)
 	tassert.CheckFatal(t, err)

ais/test/etl_test.go

@@ -355,6 +355,15 @@ func checkETLStats(t *testing.T, xid string, expectedObjCnt int, expectedBytesCn
 		bytes, outBytes, inBytes)
 }
 
+func checkObjectSizes(t *testing.T, bp api.BaseParams, bck cmn.Bck, objNames []string, size int64) {
+	t.Helper()
+	for _, objName := range objNames {
+		oah, err := api.GetObjectWithValidation(bp, bck, objName, nil)
+		tassert.CheckFatal(t, err)
+		tassert.Fatalf(t, oah.Size() == size, "%s: expected size %d, got %d", bck.Cname(objName), size, oah.Size())
+	}
+}
+
 func TestETLObject(t *testing.T) {
 	tools.CheckSkip(t, &tools.SkipTestArgs{RequiredDeployment: tools.ClusterTypeK8s})
 	tetl.CheckNoRunningETLContainers(t, baseParams)
@@ -798,6 +807,34 @@ func TestETLBucketDryRun(t *testing.T) {
 	checkETLStats(t, xid, m.num, uint64(m.num*int(m.fileSize)), false)
 }
 
+func TestETLInspectBucket(t *testing.T) {
+	tools.CheckSkip(t, &tools.SkipTestArgs{RequiredDeployment: tools.ClusterTypeK8s})
+	tetl.CheckNoRunningETLContainers(t, baseParams)
+
+	var (
+		proxyURL   = tools.RandomProxyURL(t)
+		baseParams = tools.BaseAPIParams(proxyURL)
+		bck        = cmn.Bck{Name: "etl-inspect-" + trand.String(5), Provider: apc.AIS}
+		m          = ioContext{t: t, num: 10, fileSize: 512, fixedSize: true, bck: bck}
+	)
+
+	tools.CreateBucket(t, proxyURL, bck, nil, true /*cleanup*/)
+	m.init(true /*cleanup*/)
+	m.puts()
+
+	initMsg := tetl.InitSpec(t, baseParams, tetl.MD5, etl.Hpush)
+	t.Cleanup(func() { tetl.StopAndDeleteETL(t, baseParams, initMsg.Name()) })
+
+	msg := &apc.TCBMsg{Transform: apc.Transform{Name: initMsg.Name()}}
+	xid, err := api.ETLInspectBucket(baseParams, bck, msg)
+	tassert.CheckFatal(t, err)
+	tassert.Fatalf(t, !msg.DryRun && !msg.ContinueOnError, "inspect must not mutate caller message")
+
+	_, err = api.WaitForXactionIC(baseParams, &xact.ArgsMsg{ID: xid, Kind: apc.ActETLBck, Timeout: time.Minute})
+	tassert.CheckFatal(t, err)
+	checkObjectSizes(t, baseParams, bck, m.objNames, int64(m.fileSize))
+}
+
 func TestETLStopAndRestartETL(t *testing.T) {
 	tools.CheckSkip(t, &tools.SkipTestArgs{RequiredDeployment: tools.ClusterTypeK8s})
 	tetl.CheckNoRunningETLContainers(t, baseParams)

api/bucket.go

@@ -5,6 +5,7 @@
 package api
 
 import (
+	"fmt"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -220,6 +221,17 @@ func ETLBucket(bp BaseParams, bckFrom, bckTo cmn.Bck, msg *apc.TCBMsg, fltPresen
 	return tcb(bp, bckFrom, bckTo, jbody, fltPresence...)
 }
 
+// ETLInspectBucket applies ETL to source objects without writing transformed results.
+func ETLInspectBucket(bp BaseParams, bck cmn.Bck, msg *apc.TCBMsg, fltPresence ...int) (string, error) {
+	if msg == nil {
+		return "", fmt.Errorf("%s: nil msg", apc.ActETLBck)
+	}
+	cmsg := *msg
+	cmsg.DryRun = true
+	cmsg.ContinueOnError = true
+	return ETLBucket(bp, bck, bck, &cmsg, fltPresence...)
+}
+
 func tcb(bp BaseParams, bckFrom, bckTo cmn.Bck, jbody []byte, fltPresence ...int) (string, error) {
 	if err := bckTo.Validate(); err != nil {
 		return "", err

api/multiobj.go

@@ -5,6 +5,7 @@
 package api
 
 import (
+	"fmt"
 	"net/http"
 	"strconv"
 
@@ -58,6 +59,18 @@ func ETLMultiObj(bp BaseParams, bckFrom cmn.Bck, msg *cmn.TCOMsg, fltPresence ..
 	return doBckAct(bp, bckFrom, jbody, q)
 }
 
+// ETLInspectMultiObj applies ETL to source objects without writing transformed results.
+func ETLInspectMultiObj(bp BaseParams, bck cmn.Bck, msg *cmn.TCOMsg, fltPresence ...int) (string, error) {
+	if msg == nil {
+		return "", fmt.Errorf("%s: nil msg", apc.ActETLObjects)
+	}
+	cmsg := *msg
+	cmsg.ToBck = bck
+	cmsg.DryRun = true
+	cmsg.ContinueOnError = true
+	return ETLMultiObj(bp, bck, &cmsg, fltPresence...)
+}
+
 func DeleteMultiObj(bp BaseParams, bck cmn.Bck, msg *apc.EvdMsg) (string, error) {
 	bp.Method = http.MethodDelete
 	q := qalloc()

docs/etl.md

@@ -156,6 +156,39 @@ $ ais etl bucket ffmpeg-etl ais://libre-speech ais://libre-speech-transformed --
 $ ais ls ais://libre-speech-transformed | head -5
 ```
 
+### Object Inspection
+
+Object inspection applies an ETL to a bucket or object group without writing
+transformed results. It is useful for validation jobs where the ETL logic either
+accepts the object or raises an error. After the job completes, use ETL error
+reporting to see exactly which objects failed.
+
+Python SDK:
+
+```python
+from aistore.sdk.etl.webserver.fastapi_server import FastAPIServer
+
+etl = client.etl("validation-etl")
+
+@etl.init_class()
+class ValidationETL(FastAPIServer):
+    def transform(self, data: bytes, _path: str, _etl_args: str) -> bytes:
+        if data == b"invalid":
+            raise ValueError("validation failed")
+        return data
+
+job_id = bucket.objects(obj_names=["obj-1", "obj-2"]).inspect(etl_name=etl.name)
+errors = etl.view(job_id=job_id).obj_errors
+print(errors[0]) # obj_name='ais://test-bucket/obj-1' msg='ETL error: {"detail":"Processing error: validation failed"}'
+```
+
+Go API:
+
+```go
+xid, err := api.ETLInspectBucket(bp, bck, msg)
+xid, err := api.ETLInspectMultiObj(bp, bck, msg)
+```
+
 ### Single-Object Transformation
 
 Single-object Transformation allows you to transform one object at a time between any two buckets. It's similar to a regular copy operation, but with an ETL transformation applied in-flight. This is ideal for quick, ad-hoc conversions where creating an entire new bucket isn’t necessary.

python/CHANGELOG.md

@@ -24,6 +24,9 @@ We structure this changelog in accordance with [Keep a Changelog](https://keepac
 - `Job.abort()` mirroring Go's `api.AbortXaction`: stops a job scoped by its
   `id` and/or `kind`. After aborting, `wait()` returns a `WaitResult` with
   `success=False` and the abort error instead of blocking until timeout.
+- ETL inspection APIs: `Bucket.inspect()` and `ObjectGroup.inspect()` run ETL
+  in dry-run mode without writing transformed results, while preserving the
+  existing ETL object-error reporting path.
 
 ### Changed
 

python/aistore/sdk/bucket.py

@@ -951,6 +951,41 @@ def transform(
             HTTP_METHOD_POST, ACT_ETL_BCK, value=value, params=params
         ).text
 
+    def inspect(
+        self,
+        etl_name: str,
+        timeout: str = DEFAULT_ETL_TIMEOUT,
+        prefix_filter: str = "",
+        latest: bool = False,
+        num_workers: Optional[int] = None,
+        etl_pipeline: Optional[List[str]] = None,
+    ) -> str:
+        """
+        Apply ETL to source objects without writing transformed results.
+
+        Args:
+            etl_name (str): Name of the ETL to run.
+            timeout (str, optional): ETL request timeout.
+            prefix_filter (str, optional): Only inspect objects with this prefix.
+            latest (bool, optional): Use the latest remote object version.
+            num_workers (Optional[int], optional): Number of workers per target.
+            etl_pipeline (Optional[List[str]], optional): Ordered ETL pipeline.
+
+        Returns:
+            Job ID (as str) that can be used to check the status of the operation.
+        """
+        return self.transform(
+            etl_name=etl_name,
+            to_bck=self,
+            timeout=timeout,
+            prefix_filter=prefix_filter,
+            latest=latest,
+            dry_run=True,
+            num_workers=num_workers,
+            cont_on_err=True,
+            etl_pipeline=etl_pipeline,
+        )
+
     def put_files(
         self,
         path: Union[str, Path],

python/aistore/sdk/multiobj/object_group.py

@@ -375,6 +375,38 @@ def transform(
             HTTP_METHOD_POST, ACT_TRANSFORM_OBJECTS, value=value
         ).text
 
+    def inspect(
+        self,
+        etl_name: str,
+        timeout: str = DEFAULT_ETL_TIMEOUT,
+        latest: bool = False,
+        num_workers: Optional[int] = None,
+        etl_pipeline: Optional[List[str]] = None,
+    ):
+        """
+        Apply ETL to source objects without writing transformed results.
+
+        Args:
+            etl_name (str): Name of the ETL to run.
+            timeout (str, optional): ETL request timeout.
+            latest (bool, optional): Use the latest remote object version.
+            num_workers (Optional[int], optional): Number of workers per target.
+            etl_pipeline (Optional[List[str]], optional): Ordered ETL pipeline.
+
+        Returns:
+            Job ID (as str) that can be used to check the status of the operation.
+        """
+        return self.transform(
+            to_bck=self.bck,
+            etl_name=etl_name,
+            timeout=timeout,
+            continue_on_error=True,
+            dry_run=True,
+            latest=latest,
+            num_workers=num_workers,
+            etl_pipeline=etl_pipeline,
+        )
+
     # pylint: disable=too-many-arguments,too-many-positional-arguments
     def archive(
         self,

python/tests/integration/sdk/test_etl_ops.py

@@ -205,6 +205,67 @@ def test_init_spec_md5(self):
         # Verify bucket-level transformation and object-level transformation are the same
         self.assertEqual(obj, md5_obj)
 
+    @pytest.mark.etl
+    def test_inspect(self):
+        etl = self.client.etl(self.etl_name)
+
+        @etl.init_class()
+        class MD5Server(FastAPIServer):
+            def transform(self, data: bytes, *_args) -> bytes:
+                return hashlib.md5(data).hexdigest().encode()
+
+        original = self.bucket.object(self.obj_name).get_reader().read_all()
+
+        job_id = self.bucket.inspect(etl_name=etl.name)
+        result = self.client.job(job_id).wait_for_idle(timeout=TEST_TIMEOUT)
+        self.assertTrue(result.success)
+        self.assertEqual(
+            original, self.bucket.object(self.obj_name).get_reader().read_all()
+        )
+
+        job_id = self.bucket.objects(obj_names=[self.obj_name]).inspect(
+            etl_name=etl.name
+        )
+        result = self.client.job(job_id).wait_for_idle(timeout=TEST_TIMEOUT)
+        self.assertTrue(result.success)
+        self.assertEqual(
+            original, self.bucket.object(self.obj_name).get_reader().read_all()
+        )
+
+    @pytest.mark.etl
+    def test_inspect_reports_object_errors(self):
+        obj_names = [f"inspect-{i}" for i in range(10)]
+        for obj_name in obj_names:
+            create_and_put_object(
+                client=self.client,
+                bck=self.bucket.as_model(),
+                obj_name=obj_name,
+                obj_size=self.obj_size,
+            )
+        failed_names = set(obj_names[1::2])
+
+        etl = self.client.etl(self.etl_name)
+
+        @etl.init_class()
+        class ValidationServer(FastAPIServer):
+            def transform(self, data: bytes, path: str, _etl_args: str) -> bytes:
+                if path.rsplit("/", 1)[-1] in failed_names:
+                    raise ValueError("validation failed")
+                return data
+
+        job_id = self.bucket.objects(obj_names=obj_names).inspect(etl_name=etl.name)
+        result = self.client.job(job_id).wait_for_idle(timeout=TEST_TIMEOUT)
+        self.assertTrue(result.success)
+
+        etl_details = etl.view(job_id=job_id)
+        self.assertIsNotNone(etl_details.obj_errors)
+        error_names = sorted(e.obj_name for e in etl_details.obj_errors)
+        expected_errors = sorted(
+            f"{self.bucket.provider.value}://{self.bucket.name}/{obj_name}"
+            for obj_name in failed_names
+        )
+        self.assertEqual(error_names, expected_errors)
+
     @pytest.mark.etl
     def test_init_spec_echo(self):
         # Start ETL with ECHO template