etl: introduce new etl init spec YAML format

Nahemah1022 · Nahemah1022 · commit 312af7b9b3cd · 2025-05-20T11:27:16.000-07:00
- command is always optional - enforce all images to have a default
  command
- environment variables are specific to the chosen transformer, and
  should be validated by the transformer itself during initialization.

Echo Transformer:
```
name: echo
runtime:
  image: aistorage/transformer_echo:latest
  # command is optional, default to `uvicorn` for FastAPI

```
Transformer that requires environment vars:
```
name: hash-with-args
runtime:
  image: aistorage/transformer_hash_with_args:latest
  # command is also optional
  env:
    - name: seed
      value: "1234"
```
Python Runtime:
```
name: user-defined-transform
runtime:
  image: aistorage/runtime_python:3.13
  env:
    - name: SERVER
      value: PICKLED_ETL_SERVER_INSTANCE
    - name: SERVER_TYPE
      value: FastAPI
    - name: DEPENDENCIES
      value: PARSED_REQUIREMENTS.TXT
  # default command: ./bootstrap.sh
```
```
echo ${SERVER} &gt; /server.py
echo ${DEPENDENCIES} &gt; /requirements.txt
pip install --target='/runtime' -r /requirements.txt
export PYTHONPATH=/runtime

case $SERVER_TYPE in
  FastAPIServer)
    exec uvicorn fastapi_server:app --host 0.0.0.0 --port 8000 --workers 4 --no-access-log
    ;;
  FlaskServer)
    exec gunicorn flask_server:app --bind 0.0.0.0:8000 --workers 4
    ;;
  *)
    exec python code.py
    ;;
esac
```

TODOs:

* replace `InitCode` with a special case of this new format.
* unit/integration tests against this new `ETLSpec` format.

Signed-off-by: Tony Chen &lt;a122774007@gmail.com&gt;
diff --git a/ais/test/etl_test.go b/ais/test/etl_test.go
@@ -340,6 +340,7 @@ func TestETLInline(t *testing.T) {
 
 		tests = []testObjConfig{
 			{transformer: tetl.MD5, comm: etl.Hpush},
+			{transformer: tetl.MD5ETLSpec, comm: etl.Hpush},
 		}
 	)
 
@@ -431,12 +432,15 @@ func TestETLInlineObjWithArgs(t *testing.T) {
 		transformer = tetl.HashWithArgs
 
 		tests = []struct {
-			name     string
-			commType string
-			onlyLong bool
+			name        string
+			commType    string
+			transformer string
+			onlyLong    bool
 		}{
-			{name: "etl-args-hpush", commType: etl.Hpush},
-			{name: "etl-args-hpull", commType: etl.Hpull},
+			{name: "etl-args-hpush", commType: etl.Hpush, transformer: tetl.HashWithArgs},
+			{name: "etl-args-hpush", commType: etl.Hpush, transformer: tetl.HashWithArgsETLSpec},
+			{name: "etl-args-hpull", commType: etl.Hpull, transformer: tetl.HashWithArgs},
+			{name: "etl-args-hpull", commType: etl.Hpull, transformer: tetl.HashWithArgsETLSpec},
 		}
 	)
 
@@ -538,9 +542,10 @@ func TestETLAnyToAnyBucket(t *testing.T) {
 		bcktests = []testBucketConfig{{false, false, false}}
 
 		tests = []testObjConfig{
-			{transformer: tetl.Echo, comm: etl.WebSocket, onlyLong: true}, // TODO -- FIXME: re-enable this test after echo transformer updated
+			{transformer: tetl.Echo, comm: etl.WebSocket, onlyLong: true},
 			{transformer: tetl.Echo, comm: etl.Hpull},
 			{transformer: tetl.Echo, comm: etl.Hpush},
+			{transformer: tetl.EchoETLSpec, comm: etl.Hpush},
 		}
 	)
 
diff --git a/ais/tgtetl.go b/ais/tgtetl.go
@@ -188,20 +188,13 @@ func (t *target) inlineETL(w http.ResponseWriter, r *http.Request, dpq *dpq, lom
 	// NOTE:
 	// - poll for a while here for a possible abort error (e.g., pod runtime error)
 	// - and notice hardcoded timeout
-	if abortErr := xetl.AbortedAfter(etl.DefaultObjTimeout); abortErr != nil {
+	if abortErr := xetl.AbortedAfter(etl.DefaultAbortTimeout); abortErr != nil {
 		t.writeErr(w, r, abortErr, ecode)
 		return
 	}
 
-	ectx := &cmn.ETLErrCtx{
-		ETLName:          dpq.etl.name,
-		ETLTransformArgs: dpq.etl.targs,
-		PodName:          comm.PodName(),
-		SvcName:          comm.SvcName(),
-	}
-	errV := cmn.NewErrETL(ectx, err.Error(), ecode)
-	xetl.AddErr(errV)
-	t.writeErr(w, r, errV, ecode)
+	xetl.AddErr(err)
+	t.writeErr(w, r, err, ecode)
 }
 
 func (t *target) logsETL(w http.ResponseWriter, r *http.Request, etlName string) {
diff --git a/cmd/cli/cli/const.go b/cmd/cli/cli/const.go
@@ -1102,9 +1102,8 @@ var (
 	// ETL
 	etlExtFlag  = cli.StringFlag{Name: "ext", Usage: "Mapping from old to new extensions of transformed objects' names"}
 	etlNameFlag = cli.StringFlag{
-		Name:     "name",
-		Usage:    "unique ETL name (leaving this field empty will have unique ID auto-generated)",
-		Required: true,
+		Name:  "name",
+		Usage: "unique ETL name (leaving this field empty will have unique ID auto-generated)",
 	}
 	etlObjectRequestTimeout = DurationFlag{
 		Name: "object-timeout",
diff --git a/cmn/k8s/util.go b/cmn/k8s/util.go
@@ -20,6 +20,9 @@ const (
 func ValidateEtlName(name string) error {
 	const prefix = "ETL name %q "
 	l := len(name)
+	if l == 0 {
+		return fmt.Errorf(prefix+"is empty", name)
+	}
 	if l < shortNameETL {
 		return fmt.Errorf(prefix+"is too short", name)
 	}
diff --git a/ext/etl/api.go b/ext/etl/api.go
@@ -12,6 +12,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/feat"
@@ -20,6 +21,8 @@ import (
 
 	jsoniter "github.com/json-iterator/go"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/yaml"
 	"k8s.io/client-go/kubernetes/scheme"
 )
 
@@ -44,8 +47,10 @@ const (
 )
 
 const (
-	DefaultInitTimeout = 45 * time.Second
-	DefaultObjTimeout  = 10 * time.Second
+	DefaultInitTimeout   = 45 * time.Second
+	DefaultObjTimeout    = 10 * time.Second
+	DefaultAbortTimeout  = 2 * time.Second
+	DefaultContainerPort = 8000
 )
 
 // enum ETL lifecycle status (see docs/etl.md#etl-pod-lifecycle for details)
@@ -94,18 +99,30 @@ type (
 
 	// and implementations
 	InitMsgBase struct {
-		EtlName          string       `json:"id"`
-		CommTypeX        string       `json:"communication"` // enum commTypes
-		ArgTypeX         string       `json:"argument"`      // enum argTypes
-		InitTimeout      cos.Duration `json:"init_timeout,omitempty"`
-		ObjTimeout       cos.Duration `json:"obj_timeout,omitempty"`
-		SupportDirectPut bool         `json:"support_direct_put,omitempty"`
+		EtlName          string       `json:"name" yaml:"name"`
+		CommTypeX        string       `json:"communication" yaml:"communication"`
+		ArgTypeX         string       `json:"argument" yaml:"argument"`
+		InitTimeout      cos.Duration `json:"init_timeout,omitempty" yaml:"init_timeout,omitempty"`
+		ObjTimeout       cos.Duration `json:"obj_timeout,omitempty" yaml:"obj_timeout,omitempty"`
+		SupportDirectPut bool         `json:"support_direct_put,omitempty" yaml:"support_direct_put,omitempty"`
 	}
 	InitSpecMsg struct {
 		Spec []byte `json:"spec"`
 		InitMsgBase
 	}
 
+	// ETLSpec is a YAML representation of the ETL pod spec.
+	ETLSpec struct {
+		InitMsgBase             // included all optional fields from InitMsgBase
+		Runtime     RuntimeSpec `json:"runtime" yaml:"runtime"`
+	}
+
+	RuntimeSpec struct {
+		Image   string          `json:"image" yaml:"image"`
+		Command []string        `json:"command,omitempty" yaml:"command,omitempty"`
+		Env     []corev1.EnvVar `json:"env,omitempty" yaml:"env,omitempty"`
+	}
+
 	// ========================================================================================
 	// InitCodeMsg carries the name of the transforming function;
 	// the `Transform` function is mandatory and cannot be "" (empty) - it _will_ be called
@@ -280,6 +297,11 @@ func (m *InitMsgBase) validate(detail string) error {
 			"and that your ETL server properly implements the direct put mechanism")
 		return cmn.NewErrUnsuppErr(err)
 	}
+
+	if !strings.HasSuffix(m.CommTypeX, CommTypeSeparator) {
+		m.CommTypeX += CommTypeSeparator
+	}
+
 	// NOTE: default timeout
 	if m.InitTimeout == 0 {
 		m.InitTimeout = cos.Duration(DefaultInitTimeout)
@@ -319,9 +341,9 @@ func (m *InitSpecMsg) Validate() error {
 	errCtx := &cmn.ETLErrCtx{ETLName: m.Name()}
 
 	// Check pod specification constraints.
-	pod, err := ParsePodSpec(errCtx, m.Spec)
+	pod, err := m.ParsePodSpec()
 	if err != nil {
-		return err
+		return cmn.NewErrETLf(errCtx, "failed to parse pod spec: %v\n%q", err, string(m.Spec))
 	}
 	if len(pod.Spec.Containers) != 1 {
 		return cmn.NewErrETLf(errCtx, "unsupported number of containers (%d), expected: 1", len(pod.Spec.Containers))
@@ -353,37 +375,93 @@ func (m *InitSpecMsg) Validate() error {
 		return cmn.NewErrETLf(errCtx, "readinessProbe port must be the %q port", k8s.Default)
 	}
 
-	if m.CommTypeX == "" {
-		comm, found := pod.ObjectMeta.Annotations[CommTypeAnnotation]
-		if !found {
-			return cmn.NewErrETLf(errCtx, "annotations.communication_type must be provided, or specified in the init message")
-		}
-		m.CommTypeX = comm
-	}
-
 	if dp, found := pod.ObjectMeta.Annotations[SupportDirectPutAnnotation]; found {
 		m.SupportDirectPut, err = cos.ParseBool(dp)
 		if err != nil {
 			return err
 		}
 	}
 
-	if !strings.HasSuffix(m.CommTypeX, CommTypeSeparator) {
-		m.CommTypeX += CommTypeSeparator
+	return m.InitMsgBase.validate(m.String())
+}
+
+func (s *ETLSpec) Validate() error {
+	errCtx := &cmn.ETLErrCtx{ETLName: s.Name()}
+	if s.Runtime.Image == "" {
+		return cmn.NewErrETLf(errCtx, "runtime.image must be specified")
 	}
+	return nil
+}
 
-	return m.InitMsgBase.validate(m.String())
+// ParsePodSpec parses `m.Spec` into a Kubernetes Pod object.
+// First, it tries to parse as an ETLSpec. If that fails, it falls back to a full Pod spec.
+func (m *InitSpecMsg) ParsePodSpec() (*corev1.Pod, error) {
+	if pod, err := m.parseAsETLSpec(); err == nil {
+		return pod, nil
+	}
+	return m.parseAsFullPodSpec()
+}
+
+func (m *InitSpecMsg) parseAsETLSpec() (*corev1.Pod, error) {
+	var etlSpec ETLSpec
+	if err := yaml.Unmarshal(m.Spec, &etlSpec); err != nil {
+		return nil, err
+	}
+	if err := etlSpec.Validate(); err != nil {
+		return nil, err
+	}
+	pod := &corev1.Pod{
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            etlSpec.EtlName,
+				Image:           etlSpec.Runtime.Image,
+				ImagePullPolicy: corev1.PullAlways,
+				Ports:           []corev1.ContainerPort{{Name: k8s.Default, ContainerPort: DefaultContainerPort}},
+				ReadinessProbe: &corev1.Probe{
+					ProbeHandler: corev1.ProbeHandler{
+						HTTPGet: &corev1.HTTPGetAction{
+							Path: "/" + apc.ETLHealth,
+							Port: intstr.FromString(k8s.Default),
+						},
+					},
+				},
+				Command: etlSpec.Runtime.Command,
+				Env:     etlSpec.Runtime.Env,
+			}},
+		},
+	}
+
+	// Override optional InitMsg fields if present in etlSpec
+	if etlSpec.EtlName != "" {
+		m.EtlName = etlSpec.EtlName
+	}
+	if commType := etlSpec.CommType(); commType != "" {
+		m.CommTypeX = commType
+	}
+	if argType := etlSpec.ArgType(); argType != "" {
+		m.ArgTypeX = argType
+	}
+	if etlSpec.IsDirectPut() {
+		m.SupportDirectPut = true
+	}
+	if etlSpec.InitTimeout != 0 {
+		m.InitTimeout = etlSpec.InitTimeout
+	}
+	if etlSpec.ObjTimeout != 0 {
+		m.ObjTimeout = etlSpec.ObjTimeout
+	}
+	return pod, nil
 }
 
-func ParsePodSpec(errCtx *cmn.ETLErrCtx, spec []byte) (*corev1.Pod, error) {
-	obj, _, err := scheme.Codecs.UniversalDeserializer().Decode(spec, nil, nil)
+func (m *InitSpecMsg) parseAsFullPodSpec() (*corev1.Pod, error) {
+	obj, _, err := scheme.Codecs.UniversalDeserializer().Decode(m.Spec, nil, nil)
 	if err != nil {
-		return nil, cmn.NewErrETLf(errCtx, "failed to parse pod spec: %v\n%q", err, string(spec))
+		return nil, err
 	}
 	pod, ok := obj.(*corev1.Pod)
 	if !ok {
 		kind := obj.GetObjectKind().GroupVersionKind().Kind
-		return nil, cmn.NewErrETL(errCtx, "expected pod spec, got: "+kind)
+		return nil, errors.New("expected pod spec, got: " + kind)
 	}
 	return pod, nil
 }
diff --git a/ext/etl/boot.go b/ext/etl/boot.go
@@ -46,8 +46,8 @@ type etlBootstrapper struct {
 }
 
 func (b *etlBootstrapper) createPodSpec() (err error) {
-	if b.pod, err = ParsePodSpec(b.errCtx, b.msg.Spec); err != nil {
-		return
+	if b.pod, err = b.msg.ParsePodSpec(); err != nil {
+		return cmn.NewErrETLf(b.errCtx, "failed to parse pod spec: %v\n%q", err, string(b.msg.Spec))
 	}
 	b.originalPodName = b.pod.GetName()
 	b.errCtx.ETLName = b.originalPodName
@@ -60,6 +60,7 @@ func (b *etlBootstrapper) _prepSpec() (err error) {
 	b.pod.SetName(k8s.CleanName(b.msg.Name() + "-" + core.T.SID()))
 	b.errCtx.PodName = b.pod.GetName()
 	b.pod.APIVersion = "v1"
+	b.pod.Kind = "Pod"
 
 	// The following combination of Affinity and Anti-Affinity provides for:
 	// 1. The ETL container is always scheduled on the target invoking it.
diff --git a/ext/etl/communicator.go b/ext/etl/communicator.go
@@ -200,7 +200,6 @@ func doWithTimeout(reqArgs *cmn.HreqArgs, getBody getBodyFunc, timeout time.Dura
 	ecode, err = cmn.NetworkCallWithRetry(&cmn.RetryArgs{
 		Call:      rtyr.call,
 		SoftErr:   10,
-		HardErr:   2,
 		Verbosity: cmn.RetryLogVerbose,
 		Sleep:     max(cmn.Rom.MaxKeepalive(), time.Second*5),
 	})
diff --git a/ext/etl/transform.go b/ext/etl/transform.go
@@ -277,6 +277,9 @@ func start(msg *InitSpecMsg, xid, secret string, opts StartOpts, config *cmn.Con
 		pw = newPodWatcher(podName, boot)
 		comm = newCommunicator(newAborter(msg.Name()), boot, pw)
 
+		if comm == nil {
+			return podName, svcName, nil, err
+		}
 		if err := mgr.add(msg.Name(), comm); err != nil {
 			return podName, svcName, nil, err
 		}
diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md
@@ -13,6 +13,7 @@ We structure this changelog in accordance with [Keep a Changelog](https://keepac
 - Improve `ObjectFileReader` logging to include the full exception details and traceback when retrying and resuming.
 - Update `ObjectFileReader` resume logic to accommodate for limitation w/ the `Streaming-Cold-GET` and read range as to not cause a timeout. 
   - If the remote object is not cached, resuming via stream is not possible and reading must restart from the beginning.
+- Replace `id` fields with `name` in `InitSpecETLArgs` and `InitCodeETLArgs` types to align with updated ETL init API spec.
 
 ## [1.13.8] - 2025-05-15
 
diff --git a/python/aistore/sdk/types.py b/python/aistore/sdk/types.py
@@ -316,7 +316,7 @@ class ETLDetails(BaseModel):
     Represents the API response of queries on single ETL details
     """
 
-    id: str
+    name: str
     communication: str
     init_timeout: Optional[str]
     obj_timeout: Optional[str]
@@ -361,7 +361,7 @@ class InitSpecETLArgs(InitETLArgs):
 
     def as_dict(self):
         return {
-            "id": self.etl_name,
+            "name": self.etl_name,
             "init_timeout": self.init_timeout,
             "obj_timeout": self.obj_timeout,
             "communication": f"{self.communication_type}://",
@@ -383,7 +383,7 @@ class InitCodeETLArgs(InitETLArgs):
 
     def as_dict(self):
         dict_rep = {
-            "id": self.etl_name,
+            "name": self.etl_name,
             "runtime": self.runtime,
             "communication": f"{self.communication_type}://",
             "init_timeout": self.init_timeout,
diff --git a/python/tests/unit/sdk/test_etl.py b/python/tests/unit/sdk/test_etl.py
@@ -67,7 +67,7 @@ def init_spec_exec_assert(self, expected_action, **kwargs):
         expected_action["spec"] = base64.b64encode(
             template.encode(UTF_ENCODING)
         ).decode(UTF_ENCODING)
-        expected_action["id"] = self.etl_name
+        expected_action["name"] = self.etl_name
 
         expected_response_text = self.etl_name
         mock_response = Mock()
@@ -181,7 +181,7 @@ def encode_fn(preimported_modules, func, comm_type):
         return base64.b64encode(template).decode(UTF_ENCODING)
 
     def init_code_exec_assert(self, expected_action, **kwargs):
-        expected_action["id"] = self.etl_name
+        expected_action["name"] = self.etl_name
 
         expected_response_text = "response text"
         mock_response = Mock()
diff --git a/tools/tetl/etl.go b/tools/tetl/etl.go

Original file line number	Diff line number	Diff line change
`@@ -340,6 +340,7 @@ func TestETLInline(t *testing.T) {`
`340`	`340`
`341`	`341`	`tests = []testObjConfig{`
`342`	`342`	`{transformer: tetl.MD5, comm: etl.Hpush},`
	`343`	`+ {transformer: tetl.MD5ETLSpec, comm: etl.Hpush},`
`343`	`344`	`}`
`344`	`345`	`)`
`345`	`346`
`@@ -431,12 +432,15 @@ func TestETLInlineObjWithArgs(t *testing.T) {`
`431`	`432`	`transformer = tetl.HashWithArgs`
`432`	`433`
`433`	`434`	`tests = []struct {`
`434`		`- name string`
`435`		`- commType string`
`436`		`- onlyLong bool`
	`435`	`+ name string`
	`436`	`+ commType string`
	`437`	`+ transformer string`
	`438`	`+ onlyLong bool`
`437`	`439`	`}{`
`438`		`- {name: "etl-args-hpush", commType: etl.Hpush},`
`439`		`- {name: "etl-args-hpull", commType: etl.Hpull},`
	`440`	`+ {name: "etl-args-hpush", commType: etl.Hpush, transformer: tetl.HashWithArgs},`
	`441`	`+ {name: "etl-args-hpush", commType: etl.Hpush, transformer: tetl.HashWithArgsETLSpec},`
	`442`	`+ {name: "etl-args-hpull", commType: etl.Hpull, transformer: tetl.HashWithArgs},`
	`443`	`+ {name: "etl-args-hpull", commType: etl.Hpull, transformer: tetl.HashWithArgsETLSpec},`
`440`	`444`	`}`
`441`	`445`	`)`
`442`	`446`
`@@ -538,9 +542,10 @@ func TestETLAnyToAnyBucket(t *testing.T) {`
`538`	`542`	`bcktests = []testBucketConfig{{false, false, false}}`
`539`	`543`
`540`	`544`	`tests = []testObjConfig{`
`541`		`- {transformer: tetl.Echo, comm: etl.WebSocket, onlyLong: true}, // TODO -- FIXME: re-enable this test after echo transformer updated`
	`545`	`+ {transformer: tetl.Echo, comm: etl.WebSocket, onlyLong: true},`
`542`	`546`	`{transformer: tetl.Echo, comm: etl.Hpull},`
`543`	`547`	`{transformer: tetl.Echo, comm: etl.Hpush},`
	`548`	`+ {transformer: tetl.EchoETLSpec, comm: etl.Hpush},`
`544`	`549`	`}`
`545`	`550`	`)`
`546`	`551`
Original file line number	Diff line number	Diff line change
`@@ -277,6 +277,9 @@ func start(msg InitSpecMsg, xid, secret string, opts StartOpts, config cmn.Con`
`277`	`277`	`pw = newPodWatcher(podName, boot)`
`278`	`278`	`comm = newCommunicator(newAborter(msg.Name()), boot, pw)`
`279`	`279`
	`280`	`+ if comm == nil {`
	`281`	`+ return podName, svcName, nil, err`
	`282`	`+ }`
`280`	`283`	`if err := mgr.add(msg.Name(), comm); err != nil {`
`281`	`284`	`return podName, svcName, nil, err`
`282`	`285`	`}`