Examples

6 files changed, 508 insertions, 14 deletions

diff --git a/tools/gke/big_query_utils.py b/tools/gke/big_query_utils.py
new file mode 100644
index 0000000000..ebcf9d6ec3
--- /dev/null
+++ b/tools/gke/big_query_utils.py
@@ -0,0 +1,181 @@
+import argparse
+import json
+import uuid
+import httplib2
+
+from apiclient import discovery
+from apiclient.errors import HttpError
+from oauth2client.client import GoogleCredentials
+
+NUM_RETRIES = 3
+
+
+def create_bq():
+  """Authenticates with cloud platform and gets a BiqQuery service object
+  """
+  creds = GoogleCredentials.get_application_default()
+  return discovery.build('bigquery', 'v2', credentials=creds)
+
+
+def create_ds(biq_query, project_id, dataset_id):
+  is_success = True
+  body = {
+      'datasetReference': {
+          'projectId': project_id,
+          'datasetId': dataset_id
+      }
+  }
+  try:
+    dataset_req = biq_query.datasets().insert(projectId=project_id, body=body)
+    dataset_req.execute(num_retries=NUM_RETRIES)
+  except HttpError as http_error:
+    if http_error.resp.status == 409:
+      print 'Warning: The dataset %s already exists' % dataset_id
+    else:
+      # Note: For more debugging info, print "http_error.content"
+      print 'Error in creating dataset: %s. Err: %s' % (dataset_id, http_error)
+      is_success = False
+  return is_success
+
+
+def make_field(field_name, field_type, field_description):
+  return {
+      'name': field_name,
+      'type': field_type,
+      'description': field_description
+  }
+
+
+def create_table(big_query, project_id, dataset_id, table_id, fields_list,
+                 description):
+  is_success = True
+  body = {
+      'description': description,
+      'schema': {
+          'fields': fields_list
+      },
+      'tableReference': {
+          'datasetId': dataset_id,
+          'projectId': project_id,
+          'tableId': table_id
+      }
+  }
+  try:
+    table_req = big_query.tables().insert(projectId=project_id,
+                                          datasetId=dataset_id,
+                                          body=body)
+    res = table_req.execute(num_retries=NUM_RETRIES)
+    print 'Successfully created %s "%s"' % (res['kind'], res['id'])
+  except HttpError as http_error:
+    if http_error.resp.status == 409:
+      print 'Warning: Table %s already exists' % table_id
+    else:
+      print 'Error in creating table: %s. Err: %s' % (table_id, http_error)
+      is_success = False
+  return is_success
+
+
+def insert_rows(big_query, project_id, dataset_id, table_id, rows_list):
+  is_success = True
+  body = {'rows': rows_list}
+  try:
+    insert_req = big_query.tabledata().insertAll(projectId=project_id,
+                                                 datasetId=dataset_id,
+                                                 tableId=table_id,
+                                                 body=body)
+    print body
+    res = insert_req.execute(num_retries=NUM_RETRIES)
+    print res
+  except HttpError as http_error:
+    print 'Error in inserting rows in the table %s' % table_id
+    is_success = False
+  return is_success
+
+#####################
+
+
+def make_emp_row(emp_id, emp_name, emp_email):
+  return {
+      'insertId': str(emp_id),
+      'json': {
+          'emp_id': emp_id,
+          'emp_name': emp_name,
+          'emp_email_id': emp_email
+      }
+  }
+
+
+def get_emp_table_fields_list():
+  return [
+      make_field('emp_id', 'INTEGER', 'Employee id'),
+      make_field('emp_name', 'STRING', 'Employee name'),
+      make_field('emp_email_id', 'STRING', 'Employee email id')
+  ]
+
+
+def insert_emp_rows(big_query, project_id, dataset_id, table_id, start_idx,
+                    num_rows):
+  rows_list = [make_emp_row(i, 'sree_%d' % i, 'sreecha_%d@gmail.com' % i)
+               for i in range(start_idx, start_idx + num_rows)]
+  insert_rows(big_query, project_id, dataset_id, table_id, rows_list)
+
+
+def create_emp_table(big_query, project_id, dataset_id, table_id):
+  fields_list = get_emp_table_fields_list()
+  description = 'Test table created by sree'
+  create_table(big_query, project_id, dataset_id, table_id, fields_list,
+               description)
+
+
+def sync_query(big_query, project_id, query, timeout=5000):
+  query_data = {'query': query, 'timeoutMs': timeout}
+  query_job = None
+  try:
+    query_job = big_query.jobs().query(
+        projectId=project_id,
+        body=query_data).execute(num_retries=NUM_RETRIES)
+  except HttpError as http_error:
+    print 'Query execute job failed with error: %s' % http_error
+    print http_error.content
+  return query_job
+
+#[Start query_emp_records]
+def query_emp_records(big_query, project_id, dataset_id, table_id):
+  query = 'SELECT emp_id, emp_name FROM %s.%s ORDER BY emp_id;' % (dataset_id, table_id)
+  print query
+  query_job = sync_query(big_query, project_id, query, 5000)
+  job_id = query_job['jobReference']
+
+  print query_job
+  print '**Starting paging **'
+  #[Start Paging]
+  page_token = None
+  while True:
+    page = big_query.jobs().getQueryResults(
+        pageToken=page_token,
+        **query_job['jobReference']).execute(num_retries=NUM_RETRIES)
+    rows = page['rows']
+    for row in rows:
+      print row['f'][0]['v'], "---", row['f'][1]['v']
+    page_token = page.get('pageToken')
+    if not page_token:
+      break
+  #[End Paging]
+#[End query_emp_records]
+
+#########################
+DATASET_SEQ_NUM = 1
+TABLE_SEQ_NUM = 11
+
+PROJECT_ID = 'sree-gce'
+DATASET_ID = 'sree_test_dataset_%d' % DATASET_SEQ_NUM
+TABLE_ID = 'sree_test_table_%d' % TABLE_SEQ_NUM
+
+EMP_ROW_IDX = 10
+EMP_NUM_ROWS = 5
+
+bq = create_bq()
+create_ds(bq, PROJECT_ID, DATASET_ID)
+create_emp_table(bq, PROJECT_ID, DATASET_ID, TABLE_ID)
+insert_emp_rows(bq, PROJECT_ID, DATASET_ID, TABLE_ID, EMP_ROW_IDX, EMP_NUM_ROWS)
+query_emp_records(bq, PROJECT_ID, DATASET_ID, TABLE_ID)
diff --git a/tools/gke/create_client.py b/tools/gke/create_client.py
new file mode 100755
index 0000000000..bc56ef0ef1
--- /dev/null
+++ b/tools/gke/create_client.py
@@ -0,0 +1,108 @@
+#!/usr/bin/env python2.7
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import argparse
+
+import kubernetes_api
+
+argp = argparse.ArgumentParser(description='Launch Stress tests in GKE')
+
+argp.add_argument('-n',
+                  '--num_instances',
+                  required=True,
+                  type=int,
+                  help='The number of instances to launch in GKE')
+args = argp.parse_args()
+
+kubernetes_api_server="localhost"
+kubernetes_api_port=8001
+
+
+# Docker image
+image_name="gcr.io/sree-gce/grpc_stress_test_2"
+
+server_address = "stress-server.default.svc.cluster.local:8080"
+metrics_server_address = "localhost:8081"
+
+stress_test_arg_list=[
+    "--server_addresses=" + server_address,
+    "--test_cases=empty_unary:20,large_unary:20",
+    "--num_stubs_per_channel=10"
+]
+
+metrics_client_arg_list=[
+    "--metrics_server_address=" + metrics_server_address,
+    "--total_only=true"]
+
+env_dict={
+    "GPRC_ROOT": "/var/local/git/grpc",
+    "STRESS_TEST_IMAGE": "/var/local/git/grpc/bins/opt/stress_test",
+    "STRESS_TEST_ARGS_STR": ' '.join(stress_test_arg_list),
+    "METRICS_CLIENT_IMAGE": "/var/local/git/grpc/bins/opt/metrics_client",
+    "METRICS_CLIENT_ARGS_STR": ' '.join(metrics_client_arg_list)}
+
+cmd_list=["/var/local/git/grpc/bins/opt/stress_test"]
+arg_list=stress_test_arg_list # make this [] in future
+port_list=[8081]
+
+namespace = 'default'
+is_headless_service = False # Client is NOT headless service
+
+print('Creating %d instances of client..' % args.num_instances)
+
+for i in range(1, args.num_instances + 1):
+  service_name = 'stress-client-%d' % i
+  pod_name = service_name  # Use the same name for kubernetes Service and Pod
+  is_success = kubernetes_api.create_pod(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      pod_name,
+      image_name,
+      port_list,
+      cmd_list,
+      arg_list,
+      env_dict)
+  if not is_success:
+    print("Error in creating pod %s" % pod_name)
+  else:
+    is_success = kubernetes_api.create_service(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      service_name,
+      pod_name,
+      port_list,  # Service port list
+      port_list,  # Container port list (same as service port list)
+      is_headless_service)
+    if not is_success:
+      print("Error in creating service %s" % service_name)
+    else:
+      print("Created client %s" % pod_name)
diff --git a/tools/gke/create_server.py b/tools/gke/create_server.py
new file mode 100755
index 0000000000..23ab62c205
--- /dev/null
+++ b/tools/gke/create_server.py
@@ -0,0 +1,74 @@
+#!/usr/bin/env python2.7
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import argparse
+
+import kubernetes_api
+
+service_name = 'stress-server'
+pod_name = service_name  # Use the same name for kubernetes Service and Pod
+namespace = 'default'
+is_headless_service = True
+cmd_list=['/var/local/git/grpc/bins/opt/interop_server']
+arg_list=['--port=8080']
+port_list=[8080]
+image_name='gcr.io/sree-gce/grpc_stress_test_2'
+env_dict={}
+
+# Make sure you run kubectl proxy --port=8001
+kubernetes_api_server='localhost'
+kubernetes_api_port=8001
+
+is_success = kubernetes_api.create_pod(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      pod_name,
+      image_name,
+      port_list,
+      cmd_list,
+      arg_list,
+      env_dict)
+if not is_success:
+  print("Error in creating pod")
+else:
+  is_success = kubernetes_api.create_service(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      service_name,
+      pod_name,
+      port_list,  # Service port list
+      port_list,  # Container port list (same as service port list)
+      is_headless_service)
+  if not is_success:
+    print("Error in creating service")
+  else:
+    print("Successfully created the Server")
diff --git a/tools/gke/delete_client.py b/tools/gke/delete_client.py
new file mode 100755
index 0000000000..aa519f26b8
--- /dev/null
+++ b/tools/gke/delete_client.py
@@ -0,0 +1,66 @@
+#!/usr/bin/env python2.7
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import argparse
+
+import kubernetes_api
+
+argp = argparse.ArgumentParser(description='Delete Stress test clients in GKE')
+argp.add_argument('-n',
+                  '--num_instances',
+                  required=True,
+                  type=int,
+                  help='The number of instances currently running')
+
+args = argp.parse_args()
+for i in range(1, args.num_instances + 1):
+  service_name = 'stress-client-%d' % i
+  pod_name = service_name
+  namespace = 'default'
+  kubernetes_api_server="localhost"
+  kubernetes_api_port=8001
+
+  is_success=kubernetes_api.delete_pod(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      pod_name)
+  if not is_success:
+    print('Error in deleting Pod %s' % pod_name)
+  else:
+    is_success= kubernetes_api.delete_service(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      service_name)
+    if not is_success:
+      print('Error in deleting Service %s' % service_name)
+    else:
+      print('Deleted %s' % pod_name)
diff --git a/tools/gke/delete_server.py b/tools/gke/delete_server.py
new file mode 100755
index 0000000000..6e3fdcc33b
--- /dev/null
+++ b/tools/gke/delete_server.py
@@ -0,0 +1,58 @@
+#!/usr/bin/env python2.7
+# Copyright 2015, Google Inc.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import argparse
+
+import kubernetes_api
+
+service_name = 'stress-server'
+pod_name = service_name  # Use the same name for kubernetes Service and Pod
+namespace = 'default'
+is_headless_service = True
+kubernetes_api_server="localhost"
+kubernetes_api_port=8001
+
+is_success = kubernetes_api.delete_pod(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      pod_name)
+if not is_success:
+  print("Error in deleting Pod %s" % pod_name)
+else:
+  is_success = kubernetes_api.delete_service(
+      kubernetes_api_server,
+      kubernetes_api_port,
+      namespace,
+      service_name)
+  if not is_success:
+    print("Error in deleting Service %d" % service_name)
+  else:
+    print("Deleted server %s" % service_name)
diff --git a/tools/gke/kubernetes_api.py b/tools/gke/kubernetes_api.py
index 7dd3015365..14d724bd31 100755
--- a/tools/gke/kubernetes_api.py
+++ b/tools/gke/kubernetes_api.py
@@ -33,8 +33,9 @@ import json
 
 _REQUEST_TIMEOUT_SECS = 10
 
+
 def _make_pod_config(pod_name, image_name, container_port_list, cmd_list,
-                    arg_list):
+                     arg_list, env_dict):
   """Creates a string containing the Pod defintion as required by the Kubernetes API"""
   body = {
       'kind': 'Pod',
@@ -48,20 +49,21 @@ def _make_pod_config(pod_name, image_name, container_port_list, cmd_list,
               {
                   'name': pod_name,
                   'image': image_name,
-                  'ports': []
+                  'ports': [{'containerPort': port,
+                             'protocol': 'TCP'} for port in container_port_list]
               }
           ]
       }
   }
-  # Populate the 'ports' list
-  for port in container_port_list:
-    port_entry = {'containerPort': port, 'protocol': 'TCP'}
-    body['spec']['containers'][0]['ports'].append(port_entry)
+
+  env_list = [{'name': k, 'value': v} for (k, v) in env_dict.iteritems()]
+  if len(env_list) > 0:
+    body['spec']['containers'][0]['env'] = env_list
 
   # Add the 'Command' and 'Args' attributes if they are passed.
   # Note:
   #  - 'Command' overrides the ENTRYPOINT in the Docker Image
-  #  - 'Args' override the COMMAND in Docker image (yes, it is confusing!)
+  #  - 'Args' override the CMD in Docker image (yes, it is confusing!)
   if len(cmd_list) > 0:
     body['spec']['containers'][0]['command'] = cmd_list
   if len(arg_list) > 0:
@@ -70,7 +72,7 @@ def _make_pod_config(pod_name, image_name, container_port_list, cmd_list,
 
 
 def _make_service_config(service_name, pod_name, service_port_list,
-                        container_port_list, is_headless):
+                         container_port_list, is_headless):
   """Creates a string containing the Service definition as required by the Kubernetes API.
 
   NOTE:
@@ -124,6 +126,7 @@ def _print_connection_error(msg):
   print('ERROR: Connection failed. Did you remember to run Kubenetes proxy on '
         'localhost (i.e kubectl proxy --port=<proxy_port>) ?. Error: %s' % msg)
 
+
 def _do_post(post_url, api_name, request_body):
   """Helper to do HTTP POST.
 
@@ -135,7 +138,9 @@ def _do_post(post_url, api_name, request_body):
   """
   is_success = True
   try:
-    r = requests.post(post_url, data=request_body, timeout=_REQUEST_TIMEOUT_SECS)
+    r = requests.post(post_url,
+                      data=request_body,
+                      timeout=_REQUEST_TIMEOUT_SECS)
     if r.status_code == requests.codes.conflict:
       print('WARN: Looks like the resource already exists. Api: %s, url: %s' %
             (api_name, post_url))
@@ -143,7 +148,8 @@ def _do_post(post_url, api_name, request_body):
       print('ERROR: %s API returned error. HTTP response: (%d) %s' %
             (api_name, r.status_code, r.text))
       is_success = False
-  except(requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
+  except (requests.exceptions.Timeout,
+          requests.exceptions.ConnectionError) as e:
     is_success = False
     _print_connection_error(str(e))
   return is_success
@@ -165,7 +171,8 @@ def _do_delete(del_url, api_name):
       print('ERROR: %s API returned error. HTTP response: %s' %
             (api_name, r.text))
       is_success = False
-  except(requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
+  except (requests.exceptions.Timeout,
+          requests.exceptions.ConnectionError) as e:
     is_success = False
     _print_connection_error(str(e))
   return is_success
@@ -179,12 +186,12 @@ def create_service(kube_host, kube_port, namespace, service_name, pod_name,
   post_url = 'http://%s:%d/api/v1/namespaces/%s/services' % (
       kube_host, kube_port, namespace)
   request_body = _make_service_config(service_name, pod_name, service_port_list,
-                                     container_port_list, is_headless)
+                                      container_port_list, is_headless)
   return _do_post(post_url, 'Create Service', request_body)
 
 
 def create_pod(kube_host, kube_port, namespace, pod_name, image_name,
-               container_port_list, cmd_list, arg_list):
+               container_port_list, cmd_list, arg_list, env_dict):
   """Creates a Kubernetes Pod.
 
   Note that it is generally NOT considered a good practice to directly create
@@ -200,7 +207,7 @@ def create_pod(kube_host, kube_port, namespace, pod_name, image_name,
   post_url = 'http://%s:%d/api/v1/namespaces/%s/pods' % (kube_host, kube_port,
                                                          namespace)
   request_body = _make_pod_config(pod_name, image_name, container_port_list,
-                                 cmd_list, arg_list)
+                                  cmd_list, arg_list, env_dict)
   return _do_post(post_url, 'Create Pod', request_body)