We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[FR]Allow different MLFLOW_TRACKING_URI inside Kubernetes
_As a Kubernetes workflow developer I want to submit jobs to kubernetes running in Docker Desktop that use the Tracker UI directly from my desktop command line so I can easily develop Kubernetes projects_ 

## Problem
Inside of the cluster, the URI for the tracking UI is `http://host.docker.internal:5000` outside of kubernetes the URI should be `http://localhost:5000`. When I submit a job, the MLflow CLI wants to send some updates to the tracker ui from my desktop and then once the job starts inside the cluster it wants to connect to the ui from inside.

## Proposed Solution
In a fork of MLFlow 1.2.0 I simply added a new environment variable, `K8S_MLFLOW_TRACKING_URI`. If this is set, it is the value that is passed into the job running inside the cluster. If it is not set then the usual behavior carries on where the same `MLFLOW_TRACKING_URI` is used inside and outside.

## Motivation
This seems like it would be a common development configuration. It's possible to get around this with some DNS hacking, or running the UI inside the cluster, but that is currently not as easy as it should be. This seems like a simple fix that doesn't add a lot of extra configuration complexity

## Next Steps
If this solution makes sense I'll package up my change and unit tests in a PR


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlflow/projects/kubernetes.py`
Content:
```
1 import logging
2 import docker
3 import time
4 from threading import RLock
5 from datetime import datetime
6 
7 import kubernetes
8 from kubernetes.config.config_exception import ConfigException
9 
10 from mlflow.exceptions import ExecutionException
11 from mlflow.projects.submitted_run import SubmittedRun
12 from mlflow.entities import RunStatus
13 
14 from shlex import split
15 from six.moves import shlex_quote as quote
16 
17 _logger = logging.getLogger(__name__)
18 
19 
20 def push_image_to_registry(image_tag):
21     client = docker.from_env()
22     _logger.info("=== Pushing docker image %s ===", image_tag)
23     for line in client.images.push(repository=image_tag, stream=True, decode=True):
24         if 'error' in line and line['error']:
25             raise ExecutionException("Error while pushing to docker registry: "
26                                      "{error}".format(error=line['error']))
27     return client.images.get_registry_data(image_tag).id
28 
29 
30 def _get_kubernetes_job_definition(project_name, image_tag, image_digest,
31                                    command, env_vars, job_template):
32     container_image = image_tag + '@' + image_digest
33     timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S-%f')
34     job_name = "{}-{}".format(project_name, timestamp)
35     _logger.info("=== Creating Job %s ===", job_name)
36     environment_variables = [{'name': k, 'value': v} for k, v in env_vars.items()]
37     job_template['metadata']['name'] = job_name
38     job_template['spec']['template']['spec']['containers'][0]['name'] = project_name
39     job_template['spec']['template']['spec']['containers'][0]['image'] = container_image
40     job_template['spec']['template']['spec']['containers'][0]['command'] = command
41     if 'env' not in job_template['spec']['template']['spec']['containers'][0].keys():
42         job_template['spec']['template']['spec']['containers'][0]['env'] = []
43     job_template['spec']['template']['spec']['containers'][0]['env'] += environment_variables
44     return job_template
45 
46 
47 def _get_run_command(entrypoint_command):
48     formatted_command = []
49     for cmd in entrypoint_command:
50         formatted_command.extend([quote(s) for s in split(cmd)])
51     return formatted_command
52 
53 
54 def _load_kube_context(context=None):
55     try:
56         # trying to load either the context passed as arg or, if None,
57         # the one provided as env var `KUBECONFIG` or in `~/.kube/config`
58         kubernetes.config.load_kube_config(context=context)
59     except (IOError, ConfigException) as e:
60         _logger.debug('Error loading kube context "%s": %s', context, e)
61         _logger.info('No valid kube config found, using in-cluster configuration')
62         kubernetes.config.load_incluster_config()
63 
64 
65 def run_kubernetes_job(project_name, active_run, image_tag, image_digest, command, env_vars,
66                        kube_context=None, job_template=None):
67     job_template = _get_kubernetes_job_definition(project_name,
68                                                   image_tag,
69                                                   image_digest,
70                                                   _get_run_command(command),
71                                                   env_vars,
72                                                   job_template)
73     job_name = job_template['metadata']['name']
74     job_namespace = job_template['metadata']['namespace']
75     _load_kube_context(context=kube_context)
76     api_instance = kubernetes.client.BatchV1Api()
77     api_instance.create_namespaced_job(namespace=job_namespace,
78                                        body=job_template, pretty=True)
79     return KubernetesSubmittedRun(active_run.info.run_id, job_name, job_namespace)
80 
81 
82 class KubernetesSubmittedRun(SubmittedRun):
83     """
84     Instance of SubmittedRun corresponding to a Kubernetes Job run launched to run an MLflow
85     project.
86     :param mlflow_run_id: ID of the MLflow project run.
87     :param job_name: Kubernetes job name.
88     :param job_namespace: Kubernetes job namespace.
89     """
90     # How often to poll run status when waiting on a run
91     POLL_STATUS_INTERVAL = 5
92 
93     def __init__(self, mlflow_run_id, job_name, job_namespace):
94         super(KubernetesSubmittedRun, self).__init__()
95         self._mlflow_run_id = mlflow_run_id
96         self._job_name = job_name
97         self._job_namespace = job_namespace
98         self._status = RunStatus.SCHEDULED
99         self._status_lock = RLock()
100 
101     @property
102     def run_id(self):
103         return self._mlflow_run_id
104 
105     def wait(self):
106         kube_api = kubernetes.client.BatchV1Api()
107         while not RunStatus.is_terminated(self._update_status(kube_api)):
108             time.sleep(self.POLL_STATUS_INTERVAL)
109 
110         return self._status == RunStatus.FINISHED
111 
112     def _update_status(self, kube_api=kubernetes.client.BatchV1Api()):
113         api_response = kube_api.read_namespaced_job_status(name=self._job_name,
114                                                            namespace=self._job_namespace,
115                                                            pretty=True)
116         status = api_response.status
117         with self._status_lock:
118             if RunStatus.is_terminated(self._status):
119                 return self._status
120             if self._status == RunStatus.SCHEDULED:
121                 if api_response.status.start_time is None:
122                     _logger.info("Waiting for Job to start")
123                 else:
124                     _logger.info("Job started.")
125                     self._status = RunStatus.RUNNING
126             if status.conditions is not None:
127                 for condition in status.conditions:
128                     if condition.status == "True":
129                         _logger.info(condition.message)
130                         if condition.type == "Failed":
131                             self._status = RunStatus.FAILED
132                         elif condition.type == "Complete":
133                             self._status = RunStatus.FINISHED
134         return self._status
135 
136     def get_status(self):
137         status = self._status
138         return status if RunStatus.is_terminated(status) else self._update_status()
139 
140     def cancel(self):
141         with self._status_lock:
142             if not RunStatus.is_terminated(self._status):
143                 _logger.info("Cancelling job.")
144                 kube_api = kubernetes.client.BatchV1Api()
145                 kube_api.delete_namespaced_job(name=self._job_name,
146                                                namespace=self._job_namespace,
147                                                body=kubernetes.client.V1DeleteOptions(),
148                                                pretty=True)
149                 self._status = RunStatus.KILLED
150                 _logger.info("Job cancelled.")
151             else:
152                 _logger.info("Attempting to cancel a job that is already terminated.")
153
```
--- 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/mlflow/projects/kubernetes.py b/mlflow/projects/kubernetes.py
--- a/mlflow/projects/kubernetes.py
+++ b/mlflow/projects/kubernetes.py
@@ -1,6 +1,7 @@
 import logging
 import docker
 import time
+import os
 from threading import RLock
 from datetime import datetime
 
@@ -33,6 +34,8 @@
     timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S-%f')
     job_name = "{}-{}".format(project_name, timestamp)
     _logger.info("=== Creating Job %s ===", job_name)
+    if os.environ.get('KUBE_MLFLOW_TRACKING_URI') is not None:
+        env_vars['MLFLOW_TRACKING_URI'] = os.environ['KUBE_MLFLOW_TRACKING_URI']
     environment_variables = [{'name': k, 'value': v} for k, v in env_vars.items()]
     job_template['metadata']['name'] = job_name
     job_template['spec']['template']['spec']['containers'][0]['name'] = project_name

{"golden_diff": "diff --git a/mlflow/projects/kubernetes.py b/mlflow/projects/kubernetes.py\n--- a/mlflow/projects/kubernetes.py\n+++ b/mlflow/projects/kubernetes.py\n@@ -1,6 +1,7 @@\n import logging\n import docker\n import time\n+import os\n from threading import RLock\n from datetime import datetime\n \n@@ -33,6 +34,8 @@\n     timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S-%f')\n     job_name = \"{}-{}\".format(project_name, timestamp)\n     _logger.info(\"=== Creating Job %s ===\", job_name)\n+    if os.environ.get('KUBE_MLFLOW_TRACKING_URI') is not None:\n+        env_vars['MLFLOW_TRACKING_URI'] = os.environ['KUBE_MLFLOW_TRACKING_URI']\n     environment_variables = [{'name': k, 'value': v} for k, v in env_vars.items()]\n     job_template['metadata']['name'] = job_name\n     job_template['spec']['template']['spec']['containers'][0]['name'] = project_name\n", "issue": "[FR]Allow different MLFLOW_TRACKING_URI inside Kubernetes\n_As a Kubernetes workflow developer I want to submit jobs to kubernetes running in Docker Desktop that use the Tracker UI directly from my desktop command line so I can easily develop Kubernetes projects_ \r\n\r\n## Problem\r\nInside of the cluster, the URI for the tracking UI is `http://host.docker.internal:5000` outside of kubernetes the URI should be `http://localhost:5000`. When I submit a job, the MLflow CLI wants to send some updates to the tracker ui from my desktop and then once the job starts inside the cluster it wants to connect to the ui from inside.\r\n\r\n## Proposed Solution\r\nIn a fork of MLFlow 1.2.0 I simply added a new environment variable, `K8S_MLFLOW_TRACKING_URI`. If this is set, it is the value that is passed into the job running inside the cluster. If it is not set then the usual behavior carries on where the same `MLFLOW_TRACKING_URI` is used inside and outside.\r\n\r\n## Motivation\r\nThis seems like it would be a common development configuration. It's possible to get around this with some DNS hacking, or running the UI inside the cluster, but that is currently not as easy as it should be. This seems like a simple fix that doesn't add a lot of extra configuration complexity\r\n\r\n## Next Steps\r\nIf this solution makes sense I'll package up my change and unit tests in a PR\r\n\n", "before_files": [{"content": "import logging\nimport docker\nimport time\nfrom threading import RLock\nfrom datetime import datetime\n\nimport kubernetes\nfrom kubernetes.config.config_exception import ConfigException\n\nfrom mlflow.exceptions import ExecutionException\nfrom mlflow.projects.submitted_run import SubmittedRun\nfrom mlflow.entities import RunStatus\n\nfrom shlex import split\nfrom six.moves import shlex_quote as quote\n\n_logger = logging.getLogger(__name__)\n\n\ndef push_image_to_registry(image_tag):\n    client = docker.from_env()\n    _logger.info(\"=== Pushing docker image %s ===\", image_tag)\n    for line in client.images.push(repository=image_tag, stream=True, decode=True):\n        if 'error' in line and line['error']:\n            raise ExecutionException(\"Error while pushing to docker registry: \"\n                                     \"{error}\".format(error=line['error']))\n    return client.images.get_registry_data(image_tag).id\n\n\ndef _get_kubernetes_job_definition(project_name, image_tag, image_digest,\n                                   command, env_vars, job_template):\n    container_image = image_tag + '@' + image_digest\n    timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S-%f')\n    job_name = \"{}-{}\".format(project_name, timestamp)\n    _logger.info(\"=== Creating Job %s ===\", job_name)\n    environment_variables = [{'name': k, 'value': v} for k, v in env_vars.items()]\n    job_template['metadata']['name'] = job_name\n    job_template['spec']['template']['spec']['containers'][0]['name'] = project_name\n    job_template['spec']['template']['spec']['containers'][0]['image'] = container_image\n    job_template['spec']['template']['spec']['containers'][0]['command'] = command\n    if 'env' not in job_template['spec']['template']['spec']['containers'][0].keys():\n        job_template['spec']['template']['spec']['containers'][0]['env'] = []\n    job_template['spec']['template']['spec']['containers'][0]['env'] += environment_variables\n    return job_template\n\n\ndef _get_run_command(entrypoint_command):\n    formatted_command = []\n    for cmd in entrypoint_command:\n        formatted_command.extend([quote(s) for s in split(cmd)])\n    return formatted_command\n\n\ndef _load_kube_context(context=None):\n    try:\n        # trying to load either the context passed as arg or, if None,\n        # the one provided as env var `KUBECONFIG` or in `~/.kube/config`\n        kubernetes.config.load_kube_config(context=context)\n    except (IOError, ConfigException) as e:\n        _logger.debug('Error loading kube context \"%s\": %s', context, e)\n        _logger.info('No valid kube config found, using in-cluster configuration')\n        kubernetes.config.load_incluster_config()\n\n\ndef run_kubernetes_job(project_name, active_run, image_tag, image_digest, command, env_vars,\n                       kube_context=None, job_template=None):\n    job_template = _get_kubernetes_job_definition(project_name,\n                                                  image_tag,\n                                                  image_digest,\n                                                  _get_run_command(command),\n                                                  env_vars,\n                                                  job_template)\n    job_name = job_template['metadata']['name']\n    job_namespace = job_template['metadata']['namespace']\n    _load_kube_context(context=kube_context)\n    api_instance = kubernetes.client.BatchV1Api()\n    api_instance.create_namespaced_job(namespace=job_namespace,\n                                       body=job_template, pretty=True)\n    return KubernetesSubmittedRun(active_run.info.run_id, job_name, job_namespace)\n\n\nclass KubernetesSubmittedRun(SubmittedRun):\n    \"\"\"\n    Instance of SubmittedRun corresponding to a Kubernetes Job run launched to run an MLflow\n    project.\n    :param mlflow_run_id: ID of the MLflow project run.\n    :param job_name: Kubernetes job name.\n    :param job_namespace: Kubernetes job namespace.\n    \"\"\"\n    # How often to poll run status when waiting on a run\n    POLL_STATUS_INTERVAL = 5\n\n    def __init__(self, mlflow_run_id, job_name, job_namespace):\n        super(KubernetesSubmittedRun, self).__init__()\n        self._mlflow_run_id = mlflow_run_id\n        self._job_name = job_name\n        self._job_namespace = job_namespace\n        self._status = RunStatus.SCHEDULED\n        self._status_lock = RLock()\n\n    @property\n    def run_id(self):\n        return self._mlflow_run_id\n\n    def wait(self):\n        kube_api = kubernetes.client.BatchV1Api()\n        while not RunStatus.is_terminated(self._update_status(kube_api)):\n            time.sleep(self.POLL_STATUS_INTERVAL)\n\n        return self._status == RunStatus.FINISHED\n\n    def _update_status(self, kube_api=kubernetes.client.BatchV1Api()):\n        api_response = kube_api.read_namespaced_job_status(name=self._job_name,\n                                                           namespace=self._job_namespace,\n                                                           pretty=True)\n        status = api_response.status\n        with self._status_lock:\n            if RunStatus.is_terminated(self._status):\n                return self._status\n            if self._status == RunStatus.SCHEDULED:\n                if api_response.status.start_time is None:\n                    _logger.info(\"Waiting for Job to start\")\n                else:\n                    _logger.info(\"Job started.\")\n                    self._status = RunStatus.RUNNING\n            if status.conditions is not None:\n                for condition in status.conditions:\n                    if condition.status == \"True\":\n                        _logger.info(condition.message)\n                        if condition.type == \"Failed\":\n                            self._status = RunStatus.FAILED\n                        elif condition.type == \"Complete\":\n                            self._status = RunStatus.FINISHED\n        return self._status\n\n    def get_status(self):\n        status = self._status\n        return status if RunStatus.is_terminated(status) else self._update_status()\n\n    def cancel(self):\n        with self._status_lock:\n            if not RunStatus.is_terminated(self._status):\n                _logger.info(\"Cancelling job.\")\n                kube_api = kubernetes.client.BatchV1Api()\n                kube_api.delete_namespaced_job(name=self._job_name,\n                                               namespace=self._job_namespace,\n                                               body=kubernetes.client.V1DeleteOptions(),\n                                               pretty=True)\n                self._status = RunStatus.KILLED\n                _logger.info(\"Job cancelled.\")\n            else:\n                _logger.info(\"Attempting to cancel a job that is already terminated.\")\n", "path": "mlflow/projects/kubernetes.py"}], "after_files": [{"content": "import logging\nimport docker\nimport time\nimport os\nfrom threading import RLock\nfrom datetime import datetime\n\nimport kubernetes\nfrom kubernetes.config.config_exception import ConfigException\n\nfrom mlflow.exceptions import ExecutionException\nfrom mlflow.projects.submitted_run import SubmittedRun\nfrom mlflow.entities import RunStatus\n\nfrom shlex import split\nfrom six.moves import shlex_quote as quote\n\n_logger = logging.getLogger(__name__)\n\n\ndef push_image_to_registry(image_tag):\n    client = docker.from_env()\n    _logger.info(\"=== Pushing docker image %s ===\", image_tag)\n    for line in client.images.push(repository=image_tag, stream=True, decode=True):\n        if 'error' in line and line['error']:\n            raise ExecutionException(\"Error while pushing to docker registry: \"\n                                     \"{error}\".format(error=line['error']))\n    return client.images.get_registry_data(image_tag).id\n\n\ndef _get_kubernetes_job_definition(project_name, image_tag, image_digest,\n                                   command, env_vars, job_template):\n    container_image = image_tag + '@' + image_digest\n    timestamp = datetime.now().strftime('%Y-%m-%d-%H-%M-%S-%f')\n    job_name = \"{}-{}\".format(project_name, timestamp)\n    _logger.info(\"=== Creating Job %s ===\", job_name)\n    if os.environ.get('KUBE_MLFLOW_TRACKING_URI') is not None:\n        env_vars['MLFLOW_TRACKING_URI'] = os.environ['KUBE_MLFLOW_TRACKING_URI']\n    environment_variables = [{'name': k, 'value': v} for k, v in env_vars.items()]\n    job_template['metadata']['name'] = job_name\n    job_template['spec']['template']['spec']['containers'][0]['name'] = project_name\n    job_template['spec']['template']['spec']['containers'][0]['image'] = container_image\n    job_template['spec']['template']['spec']['containers'][0]['command'] = command\n    if 'env' not in job_template['spec']['template']['spec']['containers'][0].keys():\n        job_template['spec']['template']['spec']['containers'][0]['env'] = []\n    job_template['spec']['template']['spec']['containers'][0]['env'] += environment_variables\n    return job_template\n\n\ndef _get_run_command(entrypoint_command):\n    formatted_command = []\n    for cmd in entrypoint_command:\n        formatted_command.extend([quote(s) for s in split(cmd)])\n    return formatted_command\n\n\ndef _load_kube_context(context=None):\n    try:\n        # trying to load either the context passed as arg or, if None,\n        # the one provided as env var `KUBECONFIG` or in `~/.kube/config`\n        kubernetes.config.load_kube_config(context=context)\n    except (IOError, ConfigException) as e:\n        _logger.debug('Error loading kube context \"%s\": %s', context, e)\n        _logger.info('No valid kube config found, using in-cluster configuration')\n        kubernetes.config.load_incluster_config()\n\n\ndef run_kubernetes_job(project_name, active_run, image_tag, image_digest, command, env_vars,\n                       kube_context=None, job_template=None):\n    job_template = _get_kubernetes_job_definition(project_name,\n                                                  image_tag,\n                                                  image_digest,\n                                                  _get_run_command(command),\n                                                  env_vars,\n                                                  job_template)\n    job_name = job_template['metadata']['name']\n    job_namespace = job_template['metadata']['namespace']\n    _load_kube_context(context=kube_context)\n    api_instance = kubernetes.client.BatchV1Api()\n    api_instance.create_namespaced_job(namespace=job_namespace,\n                                       body=job_template, pretty=True)\n    return KubernetesSubmittedRun(active_run.info.run_id, job_name, job_namespace)\n\n\nclass KubernetesSubmittedRun(SubmittedRun):\n    \"\"\"\n    Instance of SubmittedRun corresponding to a Kubernetes Job run launched to run an MLflow\n    project.\n    :param mlflow_run_id: ID of the MLflow project run.\n    :param job_name: Kubernetes job name.\n    :param job_namespace: Kubernetes job namespace.\n    \"\"\"\n    # How often to poll run status when waiting on a run\n    POLL_STATUS_INTERVAL = 5\n\n    def __init__(self, mlflow_run_id, job_name, job_namespace):\n        super(KubernetesSubmittedRun, self).__init__()\n        self._mlflow_run_id = mlflow_run_id\n        self._job_name = job_name\n        self._job_namespace = job_namespace\n        self._status = RunStatus.SCHEDULED\n        self._status_lock = RLock()\n\n    @property\n    def run_id(self):\n        return self._mlflow_run_id\n\n    def wait(self):\n        kube_api = kubernetes.client.BatchV1Api()\n        while not RunStatus.is_terminated(self._update_status(kube_api)):\n            time.sleep(self.POLL_STATUS_INTERVAL)\n\n        return self._status == RunStatus.FINISHED\n\n    def _update_status(self, kube_api=kubernetes.client.BatchV1Api()):\n        api_response = kube_api.read_namespaced_job_status(name=self._job_name,\n                                                           namespace=self._job_namespace,\n                                                           pretty=True)\n        status = api_response.status\n        with self._status_lock:\n            if RunStatus.is_terminated(self._status):\n                return self._status\n            if self._status == RunStatus.SCHEDULED:\n                if api_response.status.start_time is None:\n                    _logger.info(\"Waiting for Job to start\")\n                else:\n                    _logger.info(\"Job started.\")\n                    self._status = RunStatus.RUNNING\n            if status.conditions is not None:\n                for condition in status.conditions:\n                    if condition.status == \"True\":\n                        _logger.info(condition.message)\n                        if condition.type == \"Failed\":\n                            self._status = RunStatus.FAILED\n                        elif condition.type == \"Complete\":\n                            self._status = RunStatus.FINISHED\n        return self._status\n\n    def get_status(self):\n        status = self._status\n        return status if RunStatus.is_terminated(status) else self._update_status()\n\n    def cancel(self):\n        with self._status_lock:\n            if not RunStatus.is_terminated(self._status):\n                _logger.info(\"Cancelling job.\")\n                kube_api = kubernetes.client.BatchV1Api()\n                kube_api.delete_namespaced_job(name=self._job_name,\n                                               namespace=self._job_namespace,\n                                               body=kubernetes.client.V1DeleteOptions(),\n                                               pretty=True)\n                self._status = RunStatus.KILLED\n                _logger.info(\"Job cancelled.\")\n            else:\n                _logger.info(\"Attempting to cancel a job that is already terminated.\")\n", "path": "mlflow/projects/kubernetes.py"}]}