fix(sdk): DSL - Enabled arbitrary ContainerOp names (#4554)

Fixes #4522

sdk/python/kfp/dsl/_container_op.py

@@ -716,11 +716,6 @@ def __init__(self,
                  init_containers: List[UserContainer] = None,
                  sidecars: List[Sidecar] = None,
                  is_exit_handler: bool = False):
-        valid_name_regex = r'^[A-Za-z][A-Za-z0-9\s_-]*$'
-        if not re.match(valid_name_regex, name):
-            raise ValueError(
-                'Only letters, numbers, spaces, "_", and "-"  are allowed in name. Must begin with letter: %s'
-                % (name))
 
         if is_exit_handler:
             warnings.warn('is_exit_handler=True is no longer needed.', DeprecationWarning)

sdk/python/kfp/dsl/_pipeline.py

@@ -16,7 +16,7 @@
 from . import _container_op
 from . import _resource_op
 from . import _ops_group
-from ._component_bridge import _create_container_op_from_component_and_arguments
+from ._component_bridge import _create_container_op_from_component_and_arguments, _sanitize_python_function_name
 from ..components import _components
 from ..components._naming import _make_name_unique_by_adding_index
 import sys
@@ -254,8 +254,13 @@ def add_op(self, op: _container_op.BaseOp, define_only: bool):
     Returns
       op_name: a unique op name.
     """
+    # Sanitizing the op name.
+    # Technically this could be delayed to the compilation stage, but string serialization of PipelineParams make unsanitized names problematic.
+    op_name = _sanitize_python_function_name(op.human_name).replace('_', '-')
     #If there is an existing op with this name then generate a new name.
-    op_name = _make_name_unique_by_adding_index(op.human_name, list(self.ops.keys()), ' ')
+    op_name = _make_name_unique_by_adding_index(op_name, list(self.ops.keys()), ' ')
+    if op_name == '':
+      op_name = _make_name_unique_by_adding_index('task', list(self.ops.keys()), ' ')
 
     self.ops[op_name] = op
     if not define_only:

sdk/python/tests/compiler/compiler_tests.py

@@ -1003,3 +1003,17 @@ def some_pipeline():
         self.assertNotIn(' ', argument['name'], 'The input name "{}" of template "{}" was not sanitized.'.format(argument['name'], template['name']))
       for argument in template['inputs']['artifacts']:
         self.assertNotIn(' ', argument['name'], 'The input name "{}" of template "{}" was not sanitized.'.format(argument['name'], template['name']))
+
+  def test_container_op_with_arbitrary_name(self):
+    def some_pipeline():
+      dsl.ContainerOp(
+        name=r''' !"#$%&'()*+,-./:;<=>?@[\]^_`''',
+        image='alpine:latest',
+      )
+      dsl.ContainerOp(
+        name=r''' !"#$%&'()*+,-./:;<=>?@[\]^_`''',
+        image='alpine:latest',
+      )
+    workflow_dict = compiler.Compiler()._compile(some_pipeline)
+    for template in workflow_dict['spec']['templates']:
+      self.assertNotEqual(template['name'], '')