Added example for Linux Application using CCNx
[nepi.git] / src / neco / resources / linux / application.py
index b7b3f4e..953c651 100644 (file)
@@ -110,7 +110,7 @@ class LinuxApplication(ResourceManager):
 
     @property
     def app_home(self):
-        return os.path.join(self.node.exp_dir, self._home)
+        return os.path.join(self.node.exp_home, self._home)
 
     @property
     def src_dir(self):
@@ -133,8 +133,8 @@ class LinuxApplication(ResourceManager):
 
         path = os.path.join(self.app_home, name)
         
-        cmd = "(test -f %s && echo 'success') || echo 'error'" % path
-        (out, err), proc = self.node.execute(cmd)
+        command = "(test -f %s && echo 'success') || echo 'error'" % path
+        (out, err), proc = self.node.execute(command)
 
         if (err and proc.poll()) or out.find("error") != -1:
             msg = " Couldn't find trace %s " % name
@@ -181,6 +181,9 @@ class LinuxApplication(ResourceManager):
         # upload code
         self.upload_code()
 
+        # upload stdin
+        self.upload_stdin()
+
         # install dependencies
         self.install_dependencies()
 
@@ -190,14 +193,23 @@ class LinuxApplication(ResourceManager):
         # Install
         self.install()
 
-        command = self.replace_paths(self.get("command"))
-        x11 = self.get("forwardX11") or False
-        if not x11:
+        command = self.get("command")
+        x11 = self.get("forwardX11")
+        if not x11 and command:
             self.info("Uploading command '%s'" % command)
-            
+
+            # Export environment
+            environ = ""
+            env = self.get("env") or ""
+            for var in env.split(" "):
+                environ += 'export %s\n' % var
+
+            command = environ + command
+
             # If the command runs asynchronous, pre upload the command 
             # to the app.sh file in the remote host
             dst = os.path.join(self.app_home, "app.sh")
+            command = self.replace_paths(command)
             self.node.upload(command, dst, text = True)
 
         super(LinuxApplication, self).provision()
@@ -220,14 +232,26 @@ class LinuxApplication(ResourceManager):
                     sources.remove(source)
 
             # Download http sources
-            for source in http_sources:
-                dst = os.path.join(self.src_dir, source.split("/")[-1])
-                # TODO: Check if the tar.gz is already downloaded using a hash
-                # and don't download twice !!
-                command = "wget -o %s %s" % (dst, source)
-                self.node.execute(command)
-
-            self.node.upload(sources, self.src_dir)
+            if http_sources:
+                cmd = " wget -c --directory-prefix=${SOURCES} "
+                verif = ""
+
+                for source in http_sources:
+                    cmd += " %s " % (source)
+                    verif += " ls ${SOURCES}/%s ;" % os.path.basename(source)
+                
+                # Wget output goes to stderr :S
+                cmd += " 2> /dev/null ; "
+
+                # Add verification
+                cmd += " %s " % verif
+
+                # Upload the command to a file, and execute asynchronously
+                self.upload_and_run(cmd, 
+                        "http_sources.sh", "http_sources_pid", 
+                        "http_sources_out", "http_sources_err")
+            if sources:
+                self.node.upload(sources, self.src_dir)
 
     def upload_code(self):
         code = self.get("code")
@@ -240,6 +264,15 @@ class LinuxApplication(ResourceManager):
             dst = os.path.join(self.src_dir, "code")
             self.node.upload(sources, dst, text = True)
 
+    def upload_stdin(self):
+        stdin = self.get("stdin")
+        if stdin:
+            # create dir for sources
+            self.info(" Uploading stdin ")
+
+            dst = os.path.join(self.app_home, "stdin")
+            self.node.upload(stdin, dst, text = True)
+
     def install_dependencies(self):
         depends = self.get("depends")
         if depends:
@@ -254,32 +287,22 @@ class LinuxApplication(ResourceManager):
             # create dir for build
             self.node.mkdir(self.build_dir)
 
-            cmd = self.replace_paths(build)
-
-            (out, err), proc = self.run_and_wait(cmd, self.app_home,
-                pidfile = "build_pid",
-                stdout = "build_out", 
-                stderr = "build_err", 
-                raise_on_error = True)
+            # Upload the command to a file, and execute asynchronously
+            self.upload_and_run(build, 
+                    "build.sh", "build_pid", 
+                    "build_out", "build_err")
  
     def install(self):
         install = self.get("install")
         if install:
             self.info(" Installing sources ")
 
-            cmd = self.replace_paths(install)
-
-            (out, err), proc = self.run_and_wait(cmd, self.app_home, 
-                pidfile = "install_pid",
-                stdout = "install_out", 
-                stderr = "install_err", 
-                raise_on_error = True)
+            # Upload the command to a file, and execute asynchronously
+            self.upload_and_run(install, 
+                    "install.sh", "install_pid", 
+                    "install_out", "install_err")
 
     def deploy(self):
-        command = self.replace_paths(self.get("command"))
-        
-        self.info(" Deploying command '%s' " % command)
-
         # Wait until node is associated and deployed
         node = self.node
         if not node or node.state < ResourceState.READY:
@@ -287,6 +310,8 @@ class LinuxApplication(ResourceManager):
             self.ec.schedule(reschedule_delay, self.deploy)
         else:
             try:
+                command = self.get("command") or ""
+                self.info(" Deploying command '%s' " % command)
                 self.discover()
                 self.provision()
             except:
@@ -296,33 +321,53 @@ class LinuxApplication(ResourceManager):
             super(LinuxApplication, self).deploy()
 
     def start(self):
-        command = self.replace_paths(self.get("command"))
-        env = self.get("env")
-        stdin = 'stdin' if self.get("stdin") else None
+        command = self.get('command')
+        env = self.get('env')
+        stdin = 'stdin' if self.get('stdin') else None
+        stdout = 'stdout' if self.get('stdout') else 'stdout'
+        stderr = 'stderr' if self.get('stderr') else 'stderr'
         sudo = self.get('sudo') or False
-        x11 = self.get("forwardX11") or False
+        x11 = self.get('forwardX11') or False
         failed = False
 
         super(LinuxApplication, self).start()
 
+        if not command:
+            self.info("No command to start ")
+            self._state = ResourceState.FINISHED
+            return 
+    
         self.info("Starting command '%s'" % command)
 
         if x11:
+            if env:
+                # Export environment
+                environ = ""
+                for var in env.split(" "):
+                    environ += ' %s ' % var
+
+                command = "(" + environ + " ; " + command + ")"
+                command = self.replace_paths(command)
+
+            # If the command requires X11 forwarding, we
+            # can't run it asynchronously
             (out, err), proc = self.node.execute(command,
                     sudo = sudo,
                     stdin = stdin,
-                    stdout = 'stdout',
-                    stderr = 'stderr',
-                    env = env,
                     forward_x11 = x11)
 
+            self._state = ResourceState.FINISHED
+
             if proc.poll() and err:
                 failed = True
         else:
-            # Run the command asynchronously
-            command = "bash ./app.sh"
-            (out, err), proc = self.node.run(command, self.app_home, 
+            # Command was  previously uploaded, now run the remote
+            # bash file asynchronously
+            cmd = "bash ./app.sh"
+            (out, err), proc = self.node.run(cmd, self.app_home, 
                 stdin = stdin, 
+                stdout = stdout,
+                stderr = stderr,
                 sudo = sudo)
 
             if proc.poll() and err:
@@ -406,44 +451,38 @@ class LinuxApplication(ResourceManager):
 
         return self._state
 
-    def valid_connection(self, guid):
-        # TODO: Validate!
-        return True
-        # XXX: What if it is connected to more than one node?
-        resources = self.find_resources(exact_tags = [tags.NODE])
-        self._node = resources[0] if len(resources) == 1 else None
-        return self._node
-
-    def hash_app(self):
-        """ Generates a hash representing univokely the application.
-        Is used to determine whether the home directory should be cleaned
-        or not.
+    def upload_and_run(self, cmd, fname, pidfile, outfile, errfile):
+        dst = os.path.join(self.app_home, fname)
+        cmd = self.replace_paths(cmd)
+        self.node.upload(cmd, dst, text = True)
 
-        """
-        command = self.get("command")
-        forwards_x11 = self.get("forwardX11")
-        env = self.get("env")
-        sudo = self.get("sudo")
-        depends = self.get("depends")
-        sources = self.get("sources")
-        cls._register_attribute(sources)
-        cls._register_attribute(build)
-        cls._register_attribute(install)
-        cls._register_attribute(stdin)
-        cls._register_attribute(stdout)
-        cls._register_attribute(stderr)
-        cls._register_attribute(tear_down)
-        skey = "".join(map(str, args))
-        return hashlib.md5(skey).hexdigest()
+        cmd = "bash ./%s" % fname
+        (out, err), proc = self.node.run_and_wait(cmd, self.app_home,
+            pidfile = pidfile,
+            stdout = outfile, 
+            stderr = errfile, 
+            raise_on_error = True)
 
     def replace_paths(self, command):
         """
         Replace all special path tags with shell-escaped actual paths.
         """
-        return ( command
-            .replace("${SOURCES}", self.src_dir)
-            .replace("${BUILD}", self.build_dir) 
-            .replace("${APPHOME}", self.app_home) 
-            .replace("${NODEHOME}", self.node.home) )
+        def absolute_dir(d):
+            return d if d.startswith("/") else os.path.join("${HOME}", d)
 
+        return ( command
+            .replace("${SOURCES}", absolute_dir(self.src_dir))
+            .replace("${BUILD}", absolute_dir(self.build_dir))
+            .replace("${APP_HOME}", absolute_dir(self.app_home))
+            .replace("${NODE_HOME}", absolute_dir(self.node.node_home))
+            .replace("${EXP_HOME}", absolute_dir(self.node.exp_home) )
+            )
+        
+    def valid_connection(self, guid):
+        # TODO: Validate!
+        return True
+        # XXX: What if it is connected to more than one node?
+        resources = self.find_resources(exact_tags = [tags.NODE])
+        self._node = resources[0] if len(resources) == 1 else None
+        return self._node