Merge pull request #4 from open-datastudio/tunnel

Move tunnel creation to staroid package

Author: Leemoonsoo

ods/ods.py

@@ -4,17 +4,6 @@
 import requests
 import os, stat, time
 from pathlib import Path
-from shutil import which
-import platform
-import subprocess
-import wget
-import atexit
-
-CHISEL_VERSION="1.6.0"
-CHISEL_ARCH_MAP={
-    "x86_64": "amd64",
-    "i386": "386"
-}
 
 class Ods:
     def __init__(self, staroid=None, ske=None, cache_dir=None, chisel_path=None):
@@ -48,40 +37,8 @@ def create_or_get_cache_dir(self, module = ""):
             os.makedirs(cache_dir)
         return cache_dir
 
-    def __check_cmd(self, cmd):
-        if which(cmd) == None:
-            raise Exception("'{}' command not found".format(cmd))
-
     def download_chisel_if_not_exists(self):
-        # check gunzip available
-        self.__check_cmd("gunzip")
-
-        if self.__chisel_path == None:
-            # download chisel binary for secure tunnel if not exists
-            uname = platform.uname()
-            uname.system.lower()
-            if uname.machine not in CHISEL_ARCH_MAP.keys():
-                raise Exception("Can not download chisel automatically. Please download manually from 'https://github.com/jpillora/chisel/releases/download/v{}' and set 'chisel_path' argument".format(CHISEL_VERSION))
-
-            download_url = "https://github.com/jpillora/chisel/releases/download/v{}/chisel_{}_{}_{}.gz".format(
-                CHISEL_VERSION, CHISEL_VERSION, uname.system.lower(), CHISEL_ARCH_MAP[uname.machine])
-            cache_bin = self.create_or_get_cache_dir("bin")
-            chisel_path = "{}/chisel".format(cache_bin)
-
-            if not os.path.exists(chisel_path):
-                # download
-                filename = wget.download(download_url, cache_bin)
-
-                # extract
-                subprocess.run(["gunzip", "-f", filename])
-
-                # rename
-                subprocess.run(["mv", filename.replace(".gz", ""), chisel_path])
-
-                # chmod
-                os.chmod(chisel_path, stat.S_IRWXU)
-
-            self.__chisel_path = chisel_path
+        self._staroid.get_chisel_path()
 
     def _start_instance_on_staroid(self, instance_name, commit_url):
         cluster = self._staroid.cluster().get(self.__ske)
@@ -118,53 +75,14 @@ def _start_tunnel(self, instance_name, tunnels):
         ns_api = self._staroid.namespace(cluster)
         ns = ns_api.get(instance_name)
         ns_api.shell_start(instance_name)
-        resources = ns_api.get_all_resources(instance_name)
-
-        shell_service = None
-        for s in resources["services"]["items"]:
-            if "labels" in s["metadata"]:
-                if "resource.staroid.com/system" in s["metadata"]["labels"]:
-                    if s["metadata"]["labels"]["resource.staroid.com/system"] == "shell":
-                        shell_service = s
-                        break                        
-
-        if shell_service == None:
-            raise Exception("Shell service not found")
-
-        tunnel_server = "https://p{}-{}--{}".format("57682", shell_service["metadata"]["name"], ns.url()[len("https://"):])
-        cmd = [
-            self.__chisel_path,
-            "client",
-            "--header",
-            "Authorization: token {}".format(self._staroid.get_access_token()),
-            "--keepalive",
-            "10s",
-            tunnel_server
-        ]
-        cmd.extend(tunnels)
-        self.__tunnel_processes[instance_name]=subprocess.Popen(cmd)
-        atexit.register(self.cleanup)
-
-    def cleanup(self):
-        timeout_sec = 5
-        for p in self.__tunnel_processes.values(): # list of your processes
-            p_sec = 0
-            for second in range(timeout_sec):
-                if p.poll() == None:
-                    time.sleep(1)
-                    p_sec += 1
-            if p_sec >= timeout_sec:
-                p.kill() # supported from python 2.6
+        ns_api.start_tunnel(instance_name, tunnels)
 
     def _stop_tunnel(self, instance_name):
-        if self._is_tunnel_running(instance_name):
-            self.__tunnel_processes[instance_name].kill()
-            del self.__tunnel_processes[instance_name]
-
         cluster = self._staroid.cluster().get(self.__ske)
         if cluster == None:
             raise Exception("Can't get ske cluster")
         ns_api = self._staroid.namespace(cluster)
+        ns_api.stop_tunnel(instance_name)
         ns_api.shell_stop(instance_name)
 
     def _stop_instance_on_staroid(self, instance_name):

ods/spark_cluster/spark_cluster.py

@@ -10,17 +10,12 @@
 SPARK_ARTIFACTS={
     "3.0.0": {
         "image": "opendatastudio/spark-py:v3.0.0-staroid-20200830-01",
-        "dist": "https://downloads.apache.org/spark/spark-3.0.0/spark-3.0.0-bin-hadoop3.2.tgz",
+        "dist": "https://archive.apache.org/dist/spark/spark-3.0.0/spark-3.0.0-bin-hadoop3.2.tgz",
         "commit_url": "GITHUB/open-datastudio/spark-serverless:master",
         "jars": {
             "hadoop-aws": "https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.2.0/hadoop-aws-3.2.0.jar",
             "aws-java-sdk": "https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.11.563/aws-java-sdk-bundle-1.11.563.jar"
         }
-    },
-    "2.4.6": {
-        "image": "opendatastudio/spark-py:v2.4.6-staroid",
-        "dist": "https://downloads.apache.org/spark/spark-2.4.6/spark-2.4.6-bin-hadoop2.7.tgz",
-        "commit_url": "GITHUB/open-datastudio/spark-serverless:master"
     }
 }
 

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="ods",
-    version="0.0.3",
+    version="0.0.4",
     license='MIT',
     author="Open Data Studio",
     author_email="moon@staroid.com",