We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
AI Platform Pusher fails on 2nd run
I am imitating `_ai_platform_serving_args` of [taxi_pipeline_kubeflow.py](https://github.com/tensorflow/tfx/blob/46bb4f975c36ea1defde4b3c33553e088b3dc5b8/tfx/examples/chicago_taxi_pipeline/taxi_pipeline_kubeflow.py#L85)

When I execute the Pusher for 2nd time, it fails on the following error, as opposed to creating a new version under the same model name.

```
Traceback (most recent call last):
  File "/opt/venv/lib/python3.6/site-packages/tfx/extensions/google_cloud_ai_platform/runner.py", line 176, in deploy_model_for_cmle_serving
    api.projects().models().create(body=body, parent=parent).execute()
  File "/opt/venv/lib/python3.6/site-packages/googleapiclient/_helpers.py", line 130, in positional_wrapper
    return wrapped(*args, **kwargs)
  File "/opt/venv/lib/python3.6/site-packages/googleapiclient/http.py", line 856, in execute
    raise HttpError(resp, content, uri=self.uri)
googleapiclient.errors.HttpError: <HttpError 409 when requesting 
https://ml.googleapis.com/v1/projects/xxxxxxxx/models?alt=json
 returned "Field: model.name Error: A model with the same name already exists.". Details: "[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'model.name', 'description': 'A model with the same name already exists.'}]}]">
```

 I would expect that a new version with `v{{unix timestamp}}` to be created under the same model name.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `tfx/extensions/google_cloud_ai_platform/runner.py`
Content:
```
1 # Copyright 2019 Google LLC. All Rights Reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Helper class to start TFX training jobs on CMLE."""
15 from __future__ import absolute_import
16 from __future__ import division
17 from __future__ import print_function
18 
19 import datetime
20 import json
21 import sys
22 import time
23 from googleapiclient import discovery
24 from googleapiclient import errors
25 import tensorflow as tf
26 from typing import Any, Dict, List, Text
27 
28 from tfx import types
29 from tfx import version
30 from tfx.types import artifact_utils
31 
32 _POLLING_INTERVAL_IN_SECONDS = 30
33 
34 # TODO(b/139934802) Ensure mirroring of released TFX containers in Docker Hub
35 # and gcr.io/tfx-oss-public/ registries.
36 _TFX_IMAGE = 'gcr.io/tfx-oss-public/tfx:%s' % (version.__version__)
37 
38 
39 def _get_tf_runtime_version() -> Text:
40   # runtimeVersion should be same as <major>.<minor> of currently
41   # installed tensorflow version.
42   return '.'.join(tf.__version__.split('.')[0:2])
43 
44 
45 def _get_caip_python_version() -> Text:
46   """Returns supported python version on Cloud AI Platform.
47 
48   See
49   https://cloud.google.com/ml-engine/docs/tensorflow/versioning#set-python-version-training
50 
51   Returns:
52     '2.7' for PY2 or '3.5' for PY3.
53   """
54   return {2: '2.7', 3: '3.5'}[sys.version_info.major]
55 
56 
57 def start_cmle_training(input_dict: Dict[Text, List[types.Artifact]],
58                         output_dict: Dict[Text, List[types.Artifact]],
59                         exec_properties: Dict[Text, Any],
60                         executor_class_path: Text, training_inputs: Dict[Text,
61                                                                          Any]):
62   """Start a trainer job on CMLE.
63 
64   This is done by forwarding the inputs/outputs/exec_properties to the
65   tfx.scripts.run_executor module on a CMLE training job interpreter.
66 
67   Args:
68     input_dict: Passthrough input dict for tfx.components.Trainer.executor.
69     output_dict: Passthrough input dict for tfx.components.Trainer.executor.
70     exec_properties: Passthrough input dict for tfx.components.Trainer.executor.
71     executor_class_path: class path for TFX core default trainer.
72     training_inputs: Training input for CMLE training job. 'pythonModule',
73       'pythonVersion' and 'runtimeVersion' will be inferred by the runner. For
74       the full set of parameters supported, refer to
75         https://cloud.google.com/ml-engine/docs/tensorflow/deploying-models#creating_a_model_version.
76 
77   Returns:
78     None
79   Raises:
80     RuntimeError: if the Google Cloud AI Platform training job failed.
81   """
82   training_inputs = training_inputs.copy()
83   # Remove cmle_args from exec_properties so CMLE trainer doesn't call itself
84   for gaip_training_key in ['cmle_training_args', 'gaip_training_args']:
85     if gaip_training_key in exec_properties.get('custom_config'):
86       exec_properties['custom_config'].pop(gaip_training_key)
87 
88   json_inputs = artifact_utils.jsonify_artifact_dict(input_dict)
89   tf.logging.info('json_inputs=\'%s\'.', json_inputs)
90   json_outputs = artifact_utils.jsonify_artifact_dict(output_dict)
91   tf.logging.info('json_outputs=\'%s\'.', json_outputs)
92   json_exec_properties = json.dumps(exec_properties)
93   tf.logging.info('json_exec_properties=\'%s\'.', json_exec_properties)
94 
95   # Configure CMLE job
96   api_client = discovery.build('ml', 'v1')
97 
98   # We use custom containers to launch training on AI Platform, which invokes
99   # the specified image using the container's entrypoint. The default
100   # entrypoint for TFX containers is to call scripts/run_executor.py. The
101   # arguments below are passed to this run_executor entry to run the executor
102   # specified in `executor_class_path`.
103   job_args = [
104       '--executor_class_path', executor_class_path, '--inputs', json_inputs,
105       '--outputs', json_outputs, '--exec-properties', json_exec_properties
106   ]
107 
108   if not training_inputs.get('masterConfig'):
109     training_inputs['masterConfig'] = {
110         'imageUri': _TFX_IMAGE,
111     }
112 
113   training_inputs['args'] = job_args
114 
115   # Pop project_id so CMLE doesn't complain about an unexpected parameter.
116   # It's been a stowaway in cmle_args and has finally reached its destination.
117   project = training_inputs.pop('project')
118   project_id = 'projects/{}'.format(project)
119 
120   job_name = 'tfx_' + datetime.datetime.now().strftime('%Y%m%d%H%M%S')
121   job_spec = {'jobId': job_name, 'trainingInput': training_inputs}
122 
123   # Submit job to CMLE
124   tf.logging.info('Submitting job=\'{}\', project=\'{}\' to CMLE.'.format(
125       job_name, project))
126   request = api_client.projects().jobs().create(
127       body=job_spec, parent=project_id)
128   request.execute()
129 
130   # Wait for CMLE job to finish
131   job_id = '{}/jobs/{}'.format(project_id, job_name)
132   request = api_client.projects().jobs().get(name=job_id)
133   response = request.execute()
134   while response['state'] not in ('SUCCEEDED', 'FAILED'):
135     time.sleep(_POLLING_INTERVAL_IN_SECONDS)
136     response = request.execute()
137 
138   if response['state'] == 'FAILED':
139     err_msg = 'Job \'{}\' did not succeed.  Detailed response {}.'.format(
140         job_name, response)
141     tf.logging.error(err_msg)
142     raise RuntimeError(err_msg)
143 
144   # CMLE training complete
145   tf.logging.info('Job \'{}\' successful.'.format(job_name))
146 
147 
148 def deploy_model_for_cmle_serving(serving_path: Text, model_version: Text,
149                                   cmle_serving_args: Dict[Text, Any]):
150   """Deploys a model for serving with CMLE.
151 
152   Args:
153     serving_path: The path to the model. Must be a GCS URI.
154     model_version: Version of the model being deployed. Must be different from
155       what is currently being served.
156     cmle_serving_args: Dictionary containing arguments for pushing to CMLE. For
157       the full set of parameters supported, refer to
158       https://cloud.google.com/ml-engine/reference/rest/v1/projects.models.versions#Version
159 
160   Raises:
161     RuntimeError: if an error is encountered when trying to push.
162   """
163   tf.logging.info(
164       'Deploying to model with version {} to CMLE for serving: {}'.format(
165           model_version, cmle_serving_args))
166 
167   model_name = cmle_serving_args['model_name']
168   project_id = cmle_serving_args['project_id']
169   runtime_version = _get_tf_runtime_version()
170   python_version = _get_caip_python_version()
171 
172   api = discovery.build('ml', 'v1')
173   body = {'name': model_name}
174   parent = 'projects/{}'.format(project_id)
175   try:
176     api.projects().models().create(body=body, parent=parent).execute()
177   except errors.HttpError as e:
178     # If the error is to create an already existing model, it's ok to ignore.
179     # TODO(b/135211463): Remove the disable once the pytype bug is fixed.
180     if e.resp.status == '409':  # pytype: disable=attribute-error
181       tf.logging.warn('Model {} already exists'.format(model_name))
182     else:
183       raise RuntimeError('CMLE Push failed: {}'.format(e))
184 
185   body = {
186       'name': 'v{}'.format(model_version),
187       'deployment_uri': serving_path,
188       'runtime_version': runtime_version,
189       'python_version': python_version,
190   }
191 
192   # Push to CMLE, and record the operation name so we can poll for its state.
193   model_name = 'projects/{}/models/{}'.format(project_id, model_name)
194   response = api.projects().models().versions().create(
195       body=body, parent=model_name).execute()
196   op_name = response['name']
197 
198   while True:
199     deploy_status = api.projects().operations().get(name=op_name).execute()
200     if deploy_status.get('done'):
201       break
202     if 'error' in deploy_status:
203       # The operation completed with an error.
204       tf.logging.error(deploy_status['error'])
205       raise RuntimeError(
206           'Failed to deploy model to CMLE for serving: {}'.format(
207               deploy_status['error']))
208 
209     time.sleep(_POLLING_INTERVAL_IN_SECONDS)
210     tf.logging.info('Model still being deployed...')
211 
212   tf.logging.info(
213       'Successfully deployed model {} with version {}, serving from {}'.format(
214           model_name, model_version, serving_path))
215
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/tfx/extensions/google_cloud_ai_platform/runner.py b/tfx/extensions/google_cloud_ai_platform/runner.py
--- a/tfx/extensions/google_cloud_ai_platform/runner.py
+++ b/tfx/extensions/google_cloud_ai_platform/runner.py
@@ -177,7 +177,7 @@
   except errors.HttpError as e:
     # If the error is to create an already existing model, it's ok to ignore.
     # TODO(b/135211463): Remove the disable once the pytype bug is fixed.
-    if e.resp.status == '409':  # pytype: disable=attribute-error
+    if e.resp.status == 409:  # pytype: disable=attribute-error
       tf.logging.warn('Model {} already exists'.format(model_name))
     else:
       raise RuntimeError('CMLE Push failed: {}'.format(e))

{"golden_diff": "diff --git a/tfx/extensions/google_cloud_ai_platform/runner.py b/tfx/extensions/google_cloud_ai_platform/runner.py\n--- a/tfx/extensions/google_cloud_ai_platform/runner.py\n+++ b/tfx/extensions/google_cloud_ai_platform/runner.py\n@@ -177,7 +177,7 @@\n   except errors.HttpError as e:\n     # If the error is to create an already existing model, it's ok to ignore.\n     # TODO(b/135211463): Remove the disable once the pytype bug is fixed.\n-    if e.resp.status == '409':  # pytype: disable=attribute-error\n+    if e.resp.status == 409:  # pytype: disable=attribute-error\n       tf.logging.warn('Model {} already exists'.format(model_name))\n     else:\n       raise RuntimeError('CMLE Push failed: {}'.format(e))\n", "issue": "AI Platform Pusher fails on 2nd run\nI am imitating `_ai_platform_serving_args` of [taxi_pipeline_kubeflow.py](https://github.com/tensorflow/tfx/blob/46bb4f975c36ea1defde4b3c33553e088b3dc5b8/tfx/examples/chicago_taxi_pipeline/taxi_pipeline_kubeflow.py#L85)\r\n\r\nWhen I execute the Pusher for 2nd time, it fails on the following error, as opposed to creating a new version under the same model name.\r\n\r\n```\r\nTraceback (most recent call last):\r\n  File \"/opt/venv/lib/python3.6/site-packages/tfx/extensions/google_cloud_ai_platform/runner.py\", line 176, in deploy_model_for_cmle_serving\r\n    api.projects().models().create(body=body, parent=parent).execute()\r\n  File \"/opt/venv/lib/python3.6/site-packages/googleapiclient/_helpers.py\", line 130, in positional_wrapper\r\n    return wrapped(*args, **kwargs)\r\n  File \"/opt/venv/lib/python3.6/site-packages/googleapiclient/http.py\", line 856, in execute\r\n    raise HttpError(resp, content, uri=self.uri)\r\ngoogleapiclient.errors.HttpError: <HttpError 409 when requesting \r\nhttps://ml.googleapis.com/v1/projects/xxxxxxxx/models?alt=json\r\n returned \"Field: model.name Error: A model with the same name already exists.\". Details: \"[{'@type': 'type.googleapis.com/google.rpc.BadRequest', 'fieldViolations': [{'field': 'model.name', 'description': 'A model with the same name already exists.'}]}]\">\r\n```\r\n\r\n I would expect that a new version with `v{{unix timestamp}}` to be created under the same model name.\n", "before_files": [{"content": "# Copyright 2019 Google LLC. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Helper class to start TFX training jobs on CMLE.\"\"\"\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport datetime\nimport json\nimport sys\nimport time\nfrom googleapiclient import discovery\nfrom googleapiclient import errors\nimport tensorflow as tf\nfrom typing import Any, Dict, List, Text\n\nfrom tfx import types\nfrom tfx import version\nfrom tfx.types import artifact_utils\n\n_POLLING_INTERVAL_IN_SECONDS = 30\n\n# TODO(b/139934802) Ensure mirroring of released TFX containers in Docker Hub\n# and gcr.io/tfx-oss-public/ registries.\n_TFX_IMAGE = 'gcr.io/tfx-oss-public/tfx:%s' % (version.__version__)\n\n\ndef _get_tf_runtime_version() -> Text:\n  # runtimeVersion should be same as <major>.<minor> of currently\n  # installed tensorflow version.\n  return '.'.join(tf.__version__.split('.')[0:2])\n\n\ndef _get_caip_python_version() -> Text:\n  \"\"\"Returns supported python version on Cloud AI Platform.\n\n  See\n  https://cloud.google.com/ml-engine/docs/tensorflow/versioning#set-python-version-training\n\n  Returns:\n    '2.7' for PY2 or '3.5' for PY3.\n  \"\"\"\n  return {2: '2.7', 3: '3.5'}[sys.version_info.major]\n\n\ndef start_cmle_training(input_dict: Dict[Text, List[types.Artifact]],\n                        output_dict: Dict[Text, List[types.Artifact]],\n                        exec_properties: Dict[Text, Any],\n                        executor_class_path: Text, training_inputs: Dict[Text,\n                                                                         Any]):\n  \"\"\"Start a trainer job on CMLE.\n\n  This is done by forwarding the inputs/outputs/exec_properties to the\n  tfx.scripts.run_executor module on a CMLE training job interpreter.\n\n  Args:\n    input_dict: Passthrough input dict for tfx.components.Trainer.executor.\n    output_dict: Passthrough input dict for tfx.components.Trainer.executor.\n    exec_properties: Passthrough input dict for tfx.components.Trainer.executor.\n    executor_class_path: class path for TFX core default trainer.\n    training_inputs: Training input for CMLE training job. 'pythonModule',\n      'pythonVersion' and 'runtimeVersion' will be inferred by the runner. For\n      the full set of parameters supported, refer to\n        https://cloud.google.com/ml-engine/docs/tensorflow/deploying-models#creating_a_model_version.\n\n  Returns:\n    None\n  Raises:\n    RuntimeError: if the Google Cloud AI Platform training job failed.\n  \"\"\"\n  training_inputs = training_inputs.copy()\n  # Remove cmle_args from exec_properties so CMLE trainer doesn't call itself\n  for gaip_training_key in ['cmle_training_args', 'gaip_training_args']:\n    if gaip_training_key in exec_properties.get('custom_config'):\n      exec_properties['custom_config'].pop(gaip_training_key)\n\n  json_inputs = artifact_utils.jsonify_artifact_dict(input_dict)\n  tf.logging.info('json_inputs=\\'%s\\'.', json_inputs)\n  json_outputs = artifact_utils.jsonify_artifact_dict(output_dict)\n  tf.logging.info('json_outputs=\\'%s\\'.', json_outputs)\n  json_exec_properties = json.dumps(exec_properties)\n  tf.logging.info('json_exec_properties=\\'%s\\'.', json_exec_properties)\n\n  # Configure CMLE job\n  api_client = discovery.build('ml', 'v1')\n\n  # We use custom containers to launch training on AI Platform, which invokes\n  # the specified image using the container's entrypoint. The default\n  # entrypoint for TFX containers is to call scripts/run_executor.py. The\n  # arguments below are passed to this run_executor entry to run the executor\n  # specified in `executor_class_path`.\n  job_args = [\n      '--executor_class_path', executor_class_path, '--inputs', json_inputs,\n      '--outputs', json_outputs, '--exec-properties', json_exec_properties\n  ]\n\n  if not training_inputs.get('masterConfig'):\n    training_inputs['masterConfig'] = {\n        'imageUri': _TFX_IMAGE,\n    }\n\n  training_inputs['args'] = job_args\n\n  # Pop project_id so CMLE doesn't complain about an unexpected parameter.\n  # It's been a stowaway in cmle_args and has finally reached its destination.\n  project = training_inputs.pop('project')\n  project_id = 'projects/{}'.format(project)\n\n  job_name = 'tfx_' + datetime.datetime.now().strftime('%Y%m%d%H%M%S')\n  job_spec = {'jobId': job_name, 'trainingInput': training_inputs}\n\n  # Submit job to CMLE\n  tf.logging.info('Submitting job=\\'{}\\', project=\\'{}\\' to CMLE.'.format(\n      job_name, project))\n  request = api_client.projects().jobs().create(\n      body=job_spec, parent=project_id)\n  request.execute()\n\n  # Wait for CMLE job to finish\n  job_id = '{}/jobs/{}'.format(project_id, job_name)\n  request = api_client.projects().jobs().get(name=job_id)\n  response = request.execute()\n  while response['state'] not in ('SUCCEEDED', 'FAILED'):\n    time.sleep(_POLLING_INTERVAL_IN_SECONDS)\n    response = request.execute()\n\n  if response['state'] == 'FAILED':\n    err_msg = 'Job \\'{}\\' did not succeed.  Detailed response {}.'.format(\n        job_name, response)\n    tf.logging.error(err_msg)\n    raise RuntimeError(err_msg)\n\n  # CMLE training complete\n  tf.logging.info('Job \\'{}\\' successful.'.format(job_name))\n\n\ndef deploy_model_for_cmle_serving(serving_path: Text, model_version: Text,\n                                  cmle_serving_args: Dict[Text, Any]):\n  \"\"\"Deploys a model for serving with CMLE.\n\n  Args:\n    serving_path: The path to the model. Must be a GCS URI.\n    model_version: Version of the model being deployed. Must be different from\n      what is currently being served.\n    cmle_serving_args: Dictionary containing arguments for pushing to CMLE. For\n      the full set of parameters supported, refer to\n      https://cloud.google.com/ml-engine/reference/rest/v1/projects.models.versions#Version\n\n  Raises:\n    RuntimeError: if an error is encountered when trying to push.\n  \"\"\"\n  tf.logging.info(\n      'Deploying to model with version {} to CMLE for serving: {}'.format(\n          model_version, cmle_serving_args))\n\n  model_name = cmle_serving_args['model_name']\n  project_id = cmle_serving_args['project_id']\n  runtime_version = _get_tf_runtime_version()\n  python_version = _get_caip_python_version()\n\n  api = discovery.build('ml', 'v1')\n  body = {'name': model_name}\n  parent = 'projects/{}'.format(project_id)\n  try:\n    api.projects().models().create(body=body, parent=parent).execute()\n  except errors.HttpError as e:\n    # If the error is to create an already existing model, it's ok to ignore.\n    # TODO(b/135211463): Remove the disable once the pytype bug is fixed.\n    if e.resp.status == '409':  # pytype: disable=attribute-error\n      tf.logging.warn('Model {} already exists'.format(model_name))\n    else:\n      raise RuntimeError('CMLE Push failed: {}'.format(e))\n\n  body = {\n      'name': 'v{}'.format(model_version),\n      'deployment_uri': serving_path,\n      'runtime_version': runtime_version,\n      'python_version': python_version,\n  }\n\n  # Push to CMLE, and record the operation name so we can poll for its state.\n  model_name = 'projects/{}/models/{}'.format(project_id, model_name)\n  response = api.projects().models().versions().create(\n      body=body, parent=model_name).execute()\n  op_name = response['name']\n\n  while True:\n    deploy_status = api.projects().operations().get(name=op_name).execute()\n    if deploy_status.get('done'):\n      break\n    if 'error' in deploy_status:\n      # The operation completed with an error.\n      tf.logging.error(deploy_status['error'])\n      raise RuntimeError(\n          'Failed to deploy model to CMLE for serving: {}'.format(\n              deploy_status['error']))\n\n    time.sleep(_POLLING_INTERVAL_IN_SECONDS)\n    tf.logging.info('Model still being deployed...')\n\n  tf.logging.info(\n      'Successfully deployed model {} with version {}, serving from {}'.format(\n          model_name, model_version, serving_path))\n", "path": "tfx/extensions/google_cloud_ai_platform/runner.py"}], "after_files": [{"content": "# Copyright 2019 Google LLC. All Rights Reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Helper class to start TFX training jobs on CMLE.\"\"\"\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport datetime\nimport json\nimport sys\nimport time\nfrom googleapiclient import discovery\nfrom googleapiclient import errors\nimport tensorflow as tf\nfrom typing import Any, Dict, List, Text\n\nfrom tfx import types\nfrom tfx import version\nfrom tfx.types import artifact_utils\n\n_POLLING_INTERVAL_IN_SECONDS = 30\n\n# TODO(b/139934802) Ensure mirroring of released TFX containers in Docker Hub\n# and gcr.io/tfx-oss-public/ registries.\n_TFX_IMAGE = 'gcr.io/tfx-oss-public/tfx:%s' % (version.__version__)\n\n\ndef _get_tf_runtime_version() -> Text:\n  # runtimeVersion should be same as <major>.<minor> of currently\n  # installed tensorflow version.\n  return '.'.join(tf.__version__.split('.')[0:2])\n\n\ndef _get_caip_python_version() -> Text:\n  \"\"\"Returns supported python version on Cloud AI Platform.\n\n  See\n  https://cloud.google.com/ml-engine/docs/tensorflow/versioning#set-python-version-training\n\n  Returns:\n    '2.7' for PY2 or '3.5' for PY3.\n  \"\"\"\n  return {2: '2.7', 3: '3.5'}[sys.version_info.major]\n\n\ndef start_cmle_training(input_dict: Dict[Text, List[types.Artifact]],\n                        output_dict: Dict[Text, List[types.Artifact]],\n                        exec_properties: Dict[Text, Any],\n                        executor_class_path: Text, training_inputs: Dict[Text,\n                                                                         Any]):\n  \"\"\"Start a trainer job on CMLE.\n\n  This is done by forwarding the inputs/outputs/exec_properties to the\n  tfx.scripts.run_executor module on a CMLE training job interpreter.\n\n  Args:\n    input_dict: Passthrough input dict for tfx.components.Trainer.executor.\n    output_dict: Passthrough input dict for tfx.components.Trainer.executor.\n    exec_properties: Passthrough input dict for tfx.components.Trainer.executor.\n    executor_class_path: class path for TFX core default trainer.\n    training_inputs: Training input for CMLE training job. 'pythonModule',\n      'pythonVersion' and 'runtimeVersion' will be inferred by the runner. For\n      the full set of parameters supported, refer to\n        https://cloud.google.com/ml-engine/docs/tensorflow/deploying-models#creating_a_model_version.\n\n  Returns:\n    None\n  Raises:\n    RuntimeError: if the Google Cloud AI Platform training job failed.\n  \"\"\"\n  training_inputs = training_inputs.copy()\n  # Remove cmle_args from exec_properties so CMLE trainer doesn't call itself\n  for gaip_training_key in ['cmle_training_args', 'gaip_training_args']:\n    if gaip_training_key in exec_properties.get('custom_config'):\n      exec_properties['custom_config'].pop(gaip_training_key)\n\n  json_inputs = artifact_utils.jsonify_artifact_dict(input_dict)\n  tf.logging.info('json_inputs=\\'%s\\'.', json_inputs)\n  json_outputs = artifact_utils.jsonify_artifact_dict(output_dict)\n  tf.logging.info('json_outputs=\\'%s\\'.', json_outputs)\n  json_exec_properties = json.dumps(exec_properties)\n  tf.logging.info('json_exec_properties=\\'%s\\'.', json_exec_properties)\n\n  # Configure CMLE job\n  api_client = discovery.build('ml', 'v1')\n\n  # We use custom containers to launch training on AI Platform, which invokes\n  # the specified image using the container's entrypoint. The default\n  # entrypoint for TFX containers is to call scripts/run_executor.py. The\n  # arguments below are passed to this run_executor entry to run the executor\n  # specified in `executor_class_path`.\n  job_args = [\n      '--executor_class_path', executor_class_path, '--inputs', json_inputs,\n      '--outputs', json_outputs, '--exec-properties', json_exec_properties\n  ]\n\n  if not training_inputs.get('masterConfig'):\n    training_inputs['masterConfig'] = {\n        'imageUri': _TFX_IMAGE,\n    }\n\n  training_inputs['args'] = job_args\n\n  # Pop project_id so CMLE doesn't complain about an unexpected parameter.\n  # It's been a stowaway in cmle_args and has finally reached its destination.\n  project = training_inputs.pop('project')\n  project_id = 'projects/{}'.format(project)\n\n  job_name = 'tfx_' + datetime.datetime.now().strftime('%Y%m%d%H%M%S')\n  job_spec = {'jobId': job_name, 'trainingInput': training_inputs}\n\n  # Submit job to CMLE\n  tf.logging.info('Submitting job=\\'{}\\', project=\\'{}\\' to CMLE.'.format(\n      job_name, project))\n  request = api_client.projects().jobs().create(\n      body=job_spec, parent=project_id)\n  request.execute()\n\n  # Wait for CMLE job to finish\n  job_id = '{}/jobs/{}'.format(project_id, job_name)\n  request = api_client.projects().jobs().get(name=job_id)\n  response = request.execute()\n  while response['state'] not in ('SUCCEEDED', 'FAILED'):\n    time.sleep(_POLLING_INTERVAL_IN_SECONDS)\n    response = request.execute()\n\n  if response['state'] == 'FAILED':\n    err_msg = 'Job \\'{}\\' did not succeed.  Detailed response {}.'.format(\n        job_name, response)\n    tf.logging.error(err_msg)\n    raise RuntimeError(err_msg)\n\n  # CMLE training complete\n  tf.logging.info('Job \\'{}\\' successful.'.format(job_name))\n\n\ndef deploy_model_for_cmle_serving(serving_path: Text, model_version: Text,\n                                  cmle_serving_args: Dict[Text, Any]):\n  \"\"\"Deploys a model for serving with CMLE.\n\n  Args:\n    serving_path: The path to the model. Must be a GCS URI.\n    model_version: Version of the model being deployed. Must be different from\n      what is currently being served.\n    cmle_serving_args: Dictionary containing arguments for pushing to CMLE. For\n      the full set of parameters supported, refer to\n      https://cloud.google.com/ml-engine/reference/rest/v1/projects.models.versions#Version\n\n  Raises:\n    RuntimeError: if an error is encountered when trying to push.\n  \"\"\"\n  tf.logging.info(\n      'Deploying to model with version {} to CMLE for serving: {}'.format(\n          model_version, cmle_serving_args))\n\n  model_name = cmle_serving_args['model_name']\n  project_id = cmle_serving_args['project_id']\n  runtime_version = _get_tf_runtime_version()\n  python_version = _get_caip_python_version()\n\n  api = discovery.build('ml', 'v1')\n  body = {'name': model_name}\n  parent = 'projects/{}'.format(project_id)\n  try:\n    api.projects().models().create(body=body, parent=parent).execute()\n  except errors.HttpError as e:\n    # If the error is to create an already existing model, it's ok to ignore.\n    # TODO(b/135211463): Remove the disable once the pytype bug is fixed.\n    if e.resp.status == 409:  # pytype: disable=attribute-error\n      tf.logging.warn('Model {} already exists'.format(model_name))\n    else:\n      raise RuntimeError('CMLE Push failed: {}'.format(e))\n\n  body = {\n      'name': 'v{}'.format(model_version),\n      'deployment_uri': serving_path,\n      'runtime_version': runtime_version,\n      'python_version': python_version,\n  }\n\n  # Push to CMLE, and record the operation name so we can poll for its state.\n  model_name = 'projects/{}/models/{}'.format(project_id, model_name)\n  response = api.projects().models().versions().create(\n      body=body, parent=model_name).execute()\n  op_name = response['name']\n\n  while True:\n    deploy_status = api.projects().operations().get(name=op_name).execute()\n    if deploy_status.get('done'):\n      break\n    if 'error' in deploy_status:\n      # The operation completed with an error.\n      tf.logging.error(deploy_status['error'])\n      raise RuntimeError(\n          'Failed to deploy model to CMLE for serving: {}'.format(\n              deploy_status['error']))\n\n    time.sleep(_POLLING_INTERVAL_IN_SECONDS)\n    tf.logging.info('Model still being deployed...')\n\n  tf.logging.info(\n      'Successfully deployed model {} with version {}, serving from {}'.format(\n          model_name, model_version, serving_path))\n", "path": "tfx/extensions/google_cloud_ai_platform/runner.py"}]}