diff --git a/paasta_tools/api/api.py b/paasta_tools/api/api.py
index f01a48dac1..a0f20d89c8 100644
--- a/paasta_tools/api/api.py
+++ b/paasta_tools/api/api.py
@@ -148,6 +148,10 @@ def make_app(global_config=None):
         "service.instance.tasks.task",
         "/v1/services/{service}/{instance}/tasks/{task_id}",
     )
+    config.add_route(
+        "service.instance.remote_run",
+        "/v1/services/{service}/{instance}/remote_run",
+    )
     config.add_route("service.list", "/v1/services/{service}")
     config.add_route("services", "/v1/services")
     config.add_route(
diff --git a/paasta_tools/api/api_docs/oapi.yaml b/paasta_tools/api/api_docs/oapi.yaml
index 1518b6b65d..a5b4310684 100644
--- a/paasta_tools/api/api_docs/oapi.yaml
+++ b/paasta_tools/api/api_docs/oapi.yaml
@@ -1627,6 +1627,42 @@ paths:
       summary: Get mesos task of service_name.instance_name by task_id
       tags:
       - service
+  /services/{service}/{instance}/remote_run:
+    post:
+      operationId: remote_run
+      parameters:
+      - description: Service name
+        in: path
+        name: service
+        required: true
+        schema:
+          type: string
+      - description: Instance name
+        in: path
+        name: instance
+        required: true
+        schema:
+          type: string
+      - description: Username
+        in: query
+        name: user
+        required: true
+        schema:
+          type: string
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                type: string
+          description: The service is delayed for these possible reasons
+        "404":
+          description: Deployment key not found
+        "500":
+          description: Failure
+      summary: Launch a remote-run pod
+      tags:
+      - service
   /version:
     get:
       operationId: showVersion
diff --git a/paasta_tools/api/api_docs/swagger.json b/paasta_tools/api/api_docs/swagger.json
index 9caa3c8cac..7ee0a58740 100644
--- a/paasta_tools/api/api_docs/swagger.json
+++ b/paasta_tools/api/api_docs/swagger.json
@@ -846,6 +846,52 @@
                     }
                 ]
             }
+        },
+        "/services/{service}/{instance}/remote_run": {
+            "post": {
+                "responses": {
+                    "200": {
+                        "description": "It worked!",
+                        "schema": {
+                            "type": "string"
+                        }
+                    },
+                    "404": {
+                        "description": "Deployment key not found"
+                    },
+                    "500": {
+                        "description": "Failure"
+                    }
+                },
+                "summary": "Do a remote run",
+                "operationId": "remote_run",
+                "tags": [
+                    "service"
+                ],
+                "parameters": [
+                    {
+                        "in": "path",
+                        "description": "Service name",
+                        "name": "service",
+                        "required": true,
+                        "type": "string"
+                    },
+                    {
+                        "in": "path",
+                        "description": "Instance name",
+                        "name": "instance",
+                        "required": true,
+                        "type": "string"
+                    },
+                    {
+                        "in": "query",
+                        "description": "Username",
+                        "name": "user",
+                        "required": true,
+                        "type": "string"
+                    }
+                ]
+            }
         }
     },
     "definitions": {
diff --git a/paasta_tools/api/views/instance.py b/paasta_tools/api/views/instance.py
index 945a1d5187..274fe782af 100644
--- a/paasta_tools/api/views/instance.py
+++ b/paasta_tools/api/views/instance.py
@@ -31,6 +31,7 @@
 
 import paasta_tools.mesos.exceptions as mesos_exceptions
 from paasta_tools import paasta_remote_run
+from paasta_tools import paasta_remote_run_2
 from paasta_tools import tron_tools
 from paasta_tools.api import settings
 from paasta_tools.api.views.exception import ApiFailure
@@ -385,3 +386,18 @@ def instance_mesh_status(request):
         raise ApiFailure(error_message, 500)
 
     return instance_mesh
+
+
+@view_config(
+    route_name="service.instance.remote_run", request_method="POST", renderer="json"
+)
+def remote_run(request):
+    service = request.swagger_data.get("service")
+    instance = request.swagger_data.get("instance")
+    user = request.swagger_data.get("user")
+
+    try:
+        paasta_remote_run_2.remote_run_start(service, instance, user, settings.cluster)
+    except Exception:
+        error_message = traceback.format_exc()
+        raise ApiFailure(error_message, 500)
diff --git a/paasta_tools/cli/cli.py b/paasta_tools/cli/cli.py
index 92f9dd222b..22af8a138d 100755
--- a/paasta_tools/cli/cli.py
+++ b/paasta_tools/cli/cli.py
@@ -118,6 +118,7 @@ def add_subparser(command, subparsers):
     "pause_service_autoscaler": "pause_service_autoscaler",
     "push-to-registry": "push_to_registry",
     "remote-run": "remote_run",
+    "remote-run-2": "remote_run_2",
     "rollback": "rollback",
     "secret": "secret",
     "security-check": "security_check",
diff --git a/paasta_tools/cli/cmds/remote_run_2.py b/paasta_tools/cli/cmds/remote_run_2.py
new file mode 100644
index 0000000000..dd380ccaf1
--- /dev/null
+++ b/paasta_tools/cli/cmds/remote_run_2.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python
+# Copyright 2015-2016 Yelp Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from paasta_tools.api.client import get_paasta_oapi_client
+from paasta_tools.cli.utils import get_paasta_oapi_api_clustername
+from paasta_tools.cli.utils import lazy_choices_completer
+from paasta_tools.utils import list_services
+from paasta_tools.utils import load_system_paasta_config
+from paasta_tools.utils import PaastaColors
+from paasta_tools.utils import SystemPaastaConfig
+
+
+def add_common_args_to_parser(parser):
+    parser.add_argument(
+        "-s",
+        "--service",
+        help="The name of the service you wish to inspect. Required.",
+        required=True,
+    ).completer = lazy_choices_completer(list_services)
+    parser.add_argument(
+        "-i",
+        "--instance",
+        help=(
+            "Simulate a docker run for a particular instance of the "
+            "service, like 'main' or 'canary'. Required."
+        ),
+        required=True,
+    )
+    parser.add_argument(
+        "-c",
+        "--cluster",
+        help=(
+            "The name of the cluster you wish to run your task on. "
+            "If omitted, uses the default cluster defined in the paasta "
+            f"remote-run configs."
+        ),
+    )
+
+
+def add_subparser(
+    subparsers,
+) -> None:
+    remote_run_parser = subparsers.add_parser(
+        "remote-run-2",
+        help="Run stuff remotely.",
+        description=("'paasta remote-run' runs stuff remotely "),
+    )
+    add_common_args_to_parser(remote_run_parser)
+    remote_run_parser.set_defaults(command=remote_run)
+
+
+def paasta_remote_run(
+    cluster: str,
+    service: str,
+    instance: str,
+    system_paasta_config: SystemPaastaConfig,
+    verbose: int,
+    is_eks: bool = False,
+) -> int:
+    output = []
+    ret_code = 0
+    client = get_paasta_oapi_client(
+        cluster=get_paasta_oapi_api_clustername(cluster=cluster, is_eks=is_eks),
+        system_paasta_config=system_paasta_config,
+    )
+    if not client:
+        print("Cannot get a paasta-api client")
+        exit(1)
+
+    try:
+        response = client.service.remote_run(
+            service=service, instance=instance, user="qlo"
+        )
+        print(response)
+    except client.api_error as exc:
+        output.append(PaastaColors.red(exc.reason))
+        ret_code = exc.status
+    except (client.connection_error, client.timeout_error) as exc:
+        output.append(
+            PaastaColors.red(f"Could not connect to API: {exc.__class__.__name__}")
+        )
+        ret_code = 1
+    except Exception as e:
+        output.append(PaastaColors.red(f"Exception when talking to the API:"))
+        output.append(str(e))
+        ret_code = 1
+
+    print("\n".join(output))
+
+    return ret_code
+
+
+def remote_run(args) -> int:
+    """Run stuff, but remotely!"""
+    system_paasta_config = load_system_paasta_config()
+    paasta_remote_run(
+        args.cluster, args.service, args.instance, system_paasta_config, 1, False
+    )
diff --git a/paasta_tools/paasta_remote_run_2.py b/paasta_tools/paasta_remote_run_2.py
new file mode 100755
index 0000000000..ed60a73aa7
--- /dev/null
+++ b/paasta_tools/paasta_remote_run_2.py
@@ -0,0 +1,51 @@
+from time import sleep
+
+from paasta_tools.kubernetes.application.controller_wrappers import (
+    get_application_wrapper,
+)
+from paasta_tools.kubernetes_tools import KubeClient
+from paasta_tools.kubernetes_tools import load_kubernetes_service_config_no_cache
+from paasta_tools.utils import DEFAULT_SOA_DIR
+
+
+def remote_run_start(service, instance, user, cluster):
+    kube_client = KubeClient()
+    is_eks = False
+    deployment = load_kubernetes_service_config_no_cache(
+        service, instance, cluster, DEFAULT_SOA_DIR
+    )
+    namespace = deployment.get_namespace()
+
+    formatted_application = deployment.format_kubernetes_app()
+    formatted_application.metadata.name += f"-remote-run-{user}"
+    pod_name = formatted_application.metadata.name
+    app_wrapper = get_application_wrapper(formatted_application)
+    app_wrapper.load_local_config(DEFAULT_SOA_DIR, cluster, is_eks)
+    app_wrapper.create(kube_client)
+
+    # Get pod status and name
+    for retry in range(5):
+        pod_list = kube_client.core.list_namespaced_pod(namespace)
+        matching_pod = None
+        for pod in pod_list.items:
+            if pod.metadata.name.startswith(pod_name):
+                matching_pod = pod
+                break
+
+        if not matching_pod:
+            sleep(1)
+            continue
+
+        if pod.status.phase == "Running":
+            break
+        elif pod.status.phase != "Initializing":
+            raise Exception(f"Pod state is {pod.status.phase}")
+
+    if not matching_pod:
+        raise Exception("No matching pod")
+
+    return {"Status": "Success!", "pod_name": pod.metadata.name, "namespace": namespace}
+
+
+def remote_run_stop():
+    pass
diff --git a/paasta_tools/paastaapi/api/service_api.py b/paasta_tools/paastaapi/api/service_api.py
index 63901f7cee..7db7aec9fe 100644
--- a/paasta_tools/paastaapi/api/service_api.py
+++ b/paasta_tools/paastaapi/api/service_api.py
@@ -1299,6 +1299,142 @@ def __mesh_instance(
             callable=__mesh_instance
         )
 
+        def __remote_run(
+            self,
+            service,
+            instance,
+            user,
+            **kwargs
+        ):
+            """Launch a remote-run pod  # noqa: E501
+
+            This method makes a synchronous HTTP request by default. To make an
+            asynchronous HTTP request, please pass async_req=True
+
+            >>> thread = api.remote_run(service, instance, user, async_req=True)
+            >>> result = thread.get()
+
+            Args:
+                service (str): Service name
+                instance (str): Instance name
+                user (str): Username
+
+            Keyword Args:
+                _return_http_data_only (bool): response data without head status
+                    code and headers. Default is True.
+                _preload_content (bool): if False, the urllib3.HTTPResponse object
+                    will be returned without reading/decoding response data.
+                    Default is True.
+                _request_timeout (float/tuple): timeout setting for this request. If one
+                    number provided, it will be total request timeout. It can also
+                    be a pair (tuple) of (connection, read) timeouts.
+                    Default is None.
+                _check_input_type (bool): specifies if type checking
+                    should be done one the data sent to the server.
+                    Default is True.
+                _check_return_type (bool): specifies if type checking
+                    should be done one the data received from the server.
+                    Default is True.
+                _host_index (int/None): specifies the index of the server
+                    that we want to use.
+                    Default is read from the configuration.
+                async_req (bool): execute request asynchronously
+
+            Returns:
+                str
+                    If the method is called asynchronously, returns the request
+                    thread.
+            """
+            kwargs['async_req'] = kwargs.get(
+                'async_req', False
+            )
+            kwargs['_return_http_data_only'] = kwargs.get(
+                '_return_http_data_only', True
+            )
+            kwargs['_preload_content'] = kwargs.get(
+                '_preload_content', True
+            )
+            kwargs['_request_timeout'] = kwargs.get(
+                '_request_timeout', None
+            )
+            kwargs['_check_input_type'] = kwargs.get(
+                '_check_input_type', True
+            )
+            kwargs['_check_return_type'] = kwargs.get(
+                '_check_return_type', True
+            )
+            kwargs['_host_index'] = kwargs.get('_host_index')
+            kwargs['service'] = \
+                service
+            kwargs['instance'] = \
+                instance
+            kwargs['user'] = \
+                user
+            return self.call_with_http_info(**kwargs)
+
+        self.remote_run = Endpoint(
+            settings={
+                'response_type': (str,),
+                'auth': [],
+                'endpoint_path': '/services/{service}/{instance}/remote_run',
+                'operation_id': 'remote_run',
+                'http_method': 'POST',
+                'servers': None,
+            },
+            params_map={
+                'all': [
+                    'service',
+                    'instance',
+                    'user',
+                ],
+                'required': [
+                    'service',
+                    'instance',
+                    'user',
+                ],
+                'nullable': [
+                ],
+                'enum': [
+                ],
+                'validation': [
+                ]
+            },
+            root_map={
+                'validations': {
+                },
+                'allowed_values': {
+                },
+                'openapi_types': {
+                    'service':
+                        (str,),
+                    'instance':
+                        (str,),
+                    'user':
+                        (str,),
+                },
+                'attribute_map': {
+                    'service': 'service',
+                    'instance': 'instance',
+                    'user': 'user',
+                },
+                'location_map': {
+                    'service': 'path',
+                    'instance': 'path',
+                    'user': 'query',
+                },
+                'collection_format_map': {
+                }
+            },
+            headers_map={
+                'accept': [
+                    'application/json'
+                ],
+                'content_type': [],
+            },
+            api_client=api_client,
+            callable=__remote_run
+        )
+
         def __status_instance(
             self,
             service,