Add support for JSON_INPUT in gzipped configMap (#41)

Author: xhejtman

src/tesk_core/filer.py

@@ -11,6 +11,7 @@
 import logging
 import netrc
 import requests
+import gzip
 from tesk_core.exception import UnknownProtocol, FileProtocolDisabled
 import shutil
 from glob import glob
@@ -481,7 +482,11 @@ def main():
 
     logging.info('Starting %s filer...', args.transputtype)
 
-    data = json.loads(args.data)
+    if args.data.endswith('.gz'):
+        with gzip.open(args.data, 'rb') as fh:
+            data = json.loads(fh.read())
+    else:
+        data = json.loads(args.data)
 
     for afile in data[args.transputtype]:
         logging.debug('Processing file: %s', afile['path'])
@@ -494,4 +499,4 @@ def main():
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    sys.exit(main())

src/tesk_core/filer_class.py

@@ -14,8 +14,9 @@ def getEnv(self):               return self.getContainer(0)['env']
     def getImagePullPolicy(self):   return self.getContainer(0)['imagePullPolicy']
 
 
-    def __init__(self, name, data, filer_name='eu.gcr.io/tes-wes/filer', filer_version='v0.5', pullPolicyAlways = False):
+    def __init__(self, name, data, filer_name='eu.gcr.io/tes-wes/filer', filer_version='v0.5', pullPolicyAlways = False, json_pvc=None):
         self.name = name
+        self.json_pvc = json_pvc
         self.spec = {
             "kind": "Job",
             "apiVersion": "batch/v1",
@@ -41,11 +42,22 @@ def __init__(self, name, data, filer_name='eu.gcr.io/tes-wes/filer', filer_versi
         }
 
         env = self.getEnv()
-        env.append({"name": "JSON_INPUT", "value": json.dumps(data)})
+        if json_pvc is None:
+            env.append({"name": "JSON_INPUT", "value": json.dumps(data)})
         env.append({"name": "HOST_BASE_PATH", "value": path.HOST_BASE_PATH})
         env.append(
             {"name": "CONTAINER_BASE_PATH", "value": path.CONTAINER_BASE_PATH})
 
+        if json_pvc:
+            self.getVolumeMounts().append({
+                "name"        : 'jsoninput'
+              , 'mountPath'   : '/jsoninput'
+            })
+            self.getVolumes().append({
+                "name"        : 'jsoninput'
+              , "configMap"   : { 'name' : json_pvc }
+            })
+
         if fileEnabled():
             self.getVolumeMounts().append({
 
@@ -152,8 +164,12 @@ def add_netrc_mount(self, netrc_name='netrc'):
 
 
     def get_spec(self, mode, debug=False):
-        self.spec['spec']['template']['spec']['containers'][0]['args'] = [
-            mode, "$(JSON_INPUT)"]
+        if self.json_pvc is None:
+            self.spec['spec']['template']['spec']['containers'][0]['args'] = [
+                mode, "$(JSON_INPUT)"]
+        else:
+            self.spec['spec']['template']['spec']['containers'][0]['args'] = [
+                mode, "/jsoninput/JSON_INPUT.gz"]
 
         if debug:
             self.spec['spec']['template']['spec']['containers'][0][

src/tesk_core/taskmaster.py

@@ -6,6 +6,7 @@
 import re
 import sys
 import logging
+import gzip
 from kubernetes import client, config
 from tesk_core.job import Job
 from tesk_core.pvc import PVC
@@ -132,13 +133,18 @@ def init_pvc(data, filer):
     return pvc
 
 
-def run_task(data, filer_name, filer_version):
+def run_task(data, filer_name, filer_version, have_json_pvc):
     task_name = data['executors'][0]['metadata']['labels']['taskmaster-name']
     pvc = None
 
+    if have_json_pvc:
+        json_pvc = task_name
+    else:
+        json_pvc = None
+
     if data['volumes'] or data['inputs'] or data['outputs']:
 
-        filer = Filer(task_name + '-filer', data, filer_name, filer_version, args.pull_policy_always)
+        filer = Filer(task_name + '-filer', data, filer_name, filer_version, args.pull_policy_always, json_pvc)
 
         if os.environ.get('TESK_FTP_USERNAME') is not None:
             filer.set_ftp(
@@ -249,6 +255,7 @@ def newLogger(loglevel):
 
 
 def main():
+    have_json_pvc = False
 
     parser = newParser()
     global args
@@ -271,8 +278,13 @@ def main():
     elif args.file == '-':
         data = json.load(sys.stdin)
     else:
-        with open(args.file) as fh:
-            data = json.load(fh)
+        if args.file.endswith('.gz'):
+            with gzip.open(args.file, 'rb') as fh:
+                data = json.loads(fh.read())
+            have_json_pvc = True
+        else:
+            with open(args.file) as fh:
+                data = json.load(fh)
 
     # Load kubernetes config file
     if args.localKubeConfig:
@@ -287,7 +299,7 @@ def main():
     if check_cancelled():
         exit_cancelled('Cancelled during init')
 
-    run_task(data, args.filer_name, args.filer_version)
+    run_task(data, args.filer_name, args.filer_version, have_json_pvc)
 
 
 def clean_on_interrupt():