fix fun local start could not receive logs bug (#166)

* fix fun local start could not receive logs bug * fix initializer when using http trigger * add force close express server * optimize force server close * process runtime exception when http invoke * add runtime error handler * check container exist before stopping container
alibaba · Mar 13, 2019 · 88e5e8c · 88e5e8c
1 parent df10ff8
commit 88e5e8c
Show file tree

Hide file tree

Showing 8 changed files with 149 additions and 40 deletions.
diff --git a/examples/local_http/nodejs8/index.js b/examples/local_http/nodejs8/index.js
@@ -1,7 +1,11 @@
 var getRawBody = require('raw-body')
 
-module.exports.handler = function (request, response, context) {
-
+module.exports.initializer = function(context, callback) {
+    console.log("initializer invoked");
+    callback(null, '');
+}
+
+module.exports.handler = function (request, response, context) {    
     // get request body
     getRawBody(request, function (err, body) {
         var respBody = {

diff --git a/examples/local_http/template.yml b/examples/local_http/template.yml
@@ -12,6 +12,7 @@ Resources:
         CodeUri: nodejs8/
         Description: 'http trigger demo with nodejs8!'
         Runtime: nodejs8
+        Initializer: index.initializer
       Events:
         http-test:
           Type: HTTP

diff --git a/lib/commands/local/start.js b/lib/commands/local/start.js
@@ -17,33 +17,54 @@ const definition = require('../../definition');
 const { getDebugPort, getDebugIde } = require('../../debug');
 
 const serverPort = 8000;
+const SERVER_CLOSE_TIMEOUT = 5000;
 
 function registerSigintForExpress(server) {
   var sockets = {}, nextSocketId = 0;
 
   // close express server 
   // https://stackoverflow.com/questions/14626636/how-do-i-shutdown-a-node-js-https-server-immediately/14636625#14636625
-  server.on('connection', function(socket) {
+  server.on('connection', function (socket) {
     let socketId = nextSocketId;
     sockets[socketId] = socket;
-    socket.on('close', function() {
+    socket.on('close', function () {
       delete sockets[socketId];
     });
   });
 
-  process.on('SIGINT', () => {
-    server.close();
+  process.once('SIGINT', () => {
+
+    console.log('begin to close server');
+
+    // force close if gracefully closing failed
+    // https://stackoverflow.com/a/36830072/6602338
+    const serverCloseTimeout = setTimeout(() => {
+      console.log('server close timeout, force to close server');
+
+      server.emit('close');
+
+      // if force close failed, exit directly
+      setTimeout(() => {
+        process.exit(-1);
+      }, SERVER_CLOSE_TIMEOUT);
+
+    }, SERVER_CLOSE_TIMEOUT);
+
+    // gracefully close server
+    server.close(() => {
+      clearTimeout(serverCloseTimeout);
+    });
 
     for (let socketId in sockets) {
-      if (!{}.hasOwnProperty.call(sockets, socketId)) {continue;}
+      if (!{}.hasOwnProperty.call(sockets, socketId)) { continue; }
 
       sockets[socketId].destroy();
     }
   });
 }
 
 function startExpress(app) {
-    
+
   const server = app.listen(serverPort, function () {
     console.log(`function compute app listening on port ${serverPort}!`);
     console.log();

diff --git a/lib/docker.js b/lib/docker.js
@@ -61,12 +61,22 @@ function waitingForContainerStopped() {
     const jobs = [];
 
     for (let container of containers) {
-      console.log(`stopping container ${container}`);
-      jobs.push(docker.getContainer(container).stop());
+      try {
+        const c = docker.getContainer(container);
+
+        await c.inspect();
+
+        console.log(`stopping container ${container}`);
+
+        jobs.push(c.stop());
+      } catch(error) {
+        debug('get container instance error, ignore container to stop, error is', error);
+      }
     }
 
     try {
       await Promise.all(jobs);
+      console.log('all containers stopped');
     } catch (error) {
       console.error(error);
       process.exit(-1);
@@ -176,7 +186,7 @@ async function imageExist(imageName) {
   return images.length > 0;
 }
 
-function generateDockerCmd(functionProps, httpMode) {
+function generateDockerCmd(functionProps, httpMode, invokeInitializer = true) {
   const cmd = ['-h', functionProps.Handler];
 
   // always pass event using stdin mode
@@ -188,7 +198,7 @@ function generateDockerCmd(functionProps, httpMode) {
 
   const initializer = functionProps.Initializer;
 
-  if (initializer) {
+  if (initializer && invokeInitializer) {
     cmd.push('-i', initializer);
   }
 
@@ -396,7 +406,9 @@ function resolveDockerUser(nasConfig) {
   return `${uid}:${gid}`;
 }
 
-async function startContainer(opts) {
+// outputStream, errorStream used for http invoke
+// because agent is started when container running and exec could not receive related logs
+async function startContainer(opts, outputStream, errorStream) {
 
   const container = await docker.createContainer(opts);
 
@@ -408,6 +420,27 @@ async function startContainer(opts) {
     console.error(err);
   }
 
+  const logs = outputStream || errorStream;
+
+  if (logs) {
+    if (!outputStream) {
+      outputStream = devnull();
+    }
+
+    if (!errorStream) {
+      errorStream = devnull();
+    }
+
+    // dockerode bugs on windows. attach could not receive output and error, must use logs
+    const logStream = await container.logs({
+      stdout: true,
+      stderr: true,
+      follow: true
+    });
+
+    container.modem.demuxStream(logStream, outputStream, errorStream);
+  }
+
   return {
     stop: async () => {
       await container.stop();

diff --git a/lib/local/http-invoke.js b/lib/local/http-invoke.js
@@ -24,6 +24,7 @@ class HttpInvoke extends Invoke {
 
     this.isAnonymous = authType === 'ANONYMOUS';
     this.endpointPrefix = endpointPrefix;
+    this._invokeInitializer = true;
   }
 
   async _disableRunner(evt, name) {
@@ -41,6 +42,7 @@ class HttpInvoke extends Invoke {
 
       this.runner = null;
       this.containerName = docker.generateRamdomContainerName();
+      this._invokeInitializer = true;
 
       console.log('reloading success, stop old container background...');
 
@@ -70,8 +72,6 @@ class HttpInvoke extends Invoke {
           if (!this.runner) {
             debug('acquire invoke lock success, ready to create runner');
 
-            await docker.pullImageIfNeed(this.imageName);
-
             if (!this.watcher) {
               watch(this.codeMount.Source, { recursive: true, persistent: false }, (evt, name) => {
                 if (this.runner) {
@@ -82,17 +82,7 @@ class HttpInvoke extends Invoke {
               });
             }
 
-            const envs = await docker.generateDockerEnvs(this.functionProps, this.debugPort, null);
-
-            const opts = await dockerOpts.generateLocalStartRunOpts(this.runtime,
-              this.containerName,
-              this.mounts,
-              ['--server'],
-              this.debugPort,
-              envs,
-              this.dockerUser);
-
-            this.runner = await startContainer(opts);
+            await this._generateRunner();            
           } else {
             debug('acquire invoke lock success, but runner already created, skipping...');
           }
@@ -101,6 +91,21 @@ class HttpInvoke extends Invoke {
     }
   }
 
+  async _generateRunner() {
+
+    const envs = await docker.generateDockerEnvs(this.functionProps, this.debugPort, null);
+
+    const opts = await dockerOpts.generateLocalStartRunOpts(this.runtime,
+      this.containerName,
+      this.mounts,
+      ['--server'],
+      this.debugPort,
+      envs,
+      this.dockerUser);
+
+    this.runner = await startContainer(opts, process.stdout, process.stderr);
+  }
+
   async doInvoke(req, res) {
     // only one invoke can be processed
     await lock.acquire('invoke', async () => {
@@ -137,7 +142,7 @@ class HttpInvoke extends Invoke {
         // reuse container
         debug('http doInvoke, acquire invoke lock');
 
-        const cmd = [dockerOpts.resolveMockScript(this.runtime), ...docker.generateDockerCmd(this.functionProps, true)];
+        const cmd = [dockerOpts.resolveMockScript(this.runtime), ...docker.generateDockerCmd(this.functionProps, true, this._invokeInitializer)];
 
         debug(`http doInvoke, cmd is : ${cmd}`);
 
@@ -146,14 +151,35 @@ class HttpInvoke extends Invoke {
           if (!await validateSignature(req, res, req.method)) { return; }
         }
 
-        await this.runner.exec(cmd, {
-          env: envs,
-          event,
-          outputStream,
-          errorStream,
-          verbose: true
-        });
-
+        try {
+          await this.runner.exec(cmd, {
+            env: envs,
+            event,
+            outputStream,
+            errorStream,
+            verbose: true
+          });
+
+          this._invokeInitializer = false;
+        } catch(error) {
+          // errors for runtime error
+          // for example, when using nodejs, use response.send(new Error('haha')) will lead to runtime error
+          // and container will auto exit, exec will receive no message
+          res.status(500);
+          res.setHeader('Content-Type', 'application/json');
+
+          res.send({
+            'errorMessage': `Process exited unexpectedly before completing request`
+          });
+
+          // for next invoke
+          this.runner = null;
+          this.containerName = docker.generateRamdomContainerName();
+
+          console.error(error);
+          return ;
+        }
+
         debug('http doInvoke exec end, begin to response');
       }
 

diff --git a/lib/package/ignore.js b/lib/package/ignore.js
@@ -5,7 +5,7 @@ const parser = require('git-ignore-parser'),
   fs = require('fs'),
   path = require('path');
 
-const ignoredFile = ['.git', '.svn', '.env'];
+const ignoredFile = ['.git', '.svn', '.env', '.fun/nas'];
 
 module.exports = function (baseDir) {
 

diff --git a/test/docker.test.js b/test/docker.test.js
@@ -63,6 +63,19 @@ describe('test generateDockerCmd', () => {
       '3'
     ]);
   });
+
+  it('test generate docker http cmd without initializer', () => {
+    const cmd = docker.generateDockerCmd(functionProps, true, false);
+
+    expect(cmd).to.eql([
+      '-h',
+      'index.handler',
+      '--stdin',
+      '--http',
+      '--initializationTimeout',
+      '3'
+    ]);
+  });
 });
 
 
@@ -255,7 +268,10 @@ describe('test docker run', async () => {
 
     sandbox.stub(DockerCli.prototype, 'pull').resolves({});
     sandbox.stub(DockerCli.prototype, 'run').resolves({});
-    sandbox.stub(DockerCli.prototype, 'getContainer').returns({ 'stop': sandbox.stub() });
+    sandbox.stub(DockerCli.prototype, 'getContainer').returns({ 
+      'stop': sandbox.stub(),
+      'inspect': sandbox.stub().resolves({})
+    });
 
     streamMock = {
       'write': sandbox.stub(),
@@ -384,6 +400,7 @@ describe('test docker run', async () => {
     await sleep(10);
 
     assert.calledWith(DockerCli.prototype.getContainer, 'containerId');
+    assert.calledOnce(DockerCli.prototype.getContainer().inspect);
     assert.calledOnce(DockerCli.prototype.getContainer().stop);
   });
 });

diff --git a/test/local/http-invoke.test.js b/test/local/http-invoke.test.js
@@ -101,6 +101,7 @@ describe('test http response', async () => {
   let app;
   let server;
   let restoreProcess;
+  let httpInvoke;
 
   beforeEach(async () => {
 
@@ -125,7 +126,7 @@ def handler(environ, start_response):
 `);
   });
 
-  afterEach(async function () {
+  afterEach(async () => {
     rimraf.sync(projectDir);
     restoreProcess();
 
@@ -142,7 +143,7 @@ def handler(environ, start_response):
     const endpointPrefix = `/2016-08-15/proxy/${serviceName}/${functionName}`;
     const endpoint = `${endpointPrefix}*`;
 
-    const httpInvoke = new HttpInvoke(serviceName, httptriggerServiceRes,
+    httpInvoke = new HttpInvoke(serviceName, httptriggerServiceRes,
       functionName, httpTriggerFunctionRes, null, null, ymlPath, 'ANONYMOUS', endpointPrefix);
 
     app.get(endpoint, async (req, res) => {
@@ -157,6 +158,8 @@ def handler(environ, start_response):
     const body = await httpx.read(resp, 'utf8');
 
     expect(body).to.contain('Hello world!');
+
+    await httpInvoke.runner.stop();
   });
 
   it('test http local invoke with authType function with invalid signature', async () => {
@@ -170,7 +173,7 @@ def handler(environ, start_response):
     const endpointPrefix = `/2016-08-15/proxy/${serviceName}/${functionName}`;
     const endpoint = `${endpointPrefix}*`;
 
-    const httpInvoke = new HttpInvoke(serviceName, httptriggerServiceRes,
+    httpInvoke = new HttpInvoke(serviceName, httptriggerServiceRes,
       functionName, httpTriggerFunctionRes, null, null, ymlPath, 'FUNCTION', endpointPrefix);
 
     app.get(endpoint, async (req, res) => {
@@ -185,6 +188,8 @@ def handler(environ, start_response):
     const body = await httpx.read(resp, 'utf8');
 
     expect(body).to.contain('Signature doesn\'t match, request signature is');
+
+    await httpInvoke.runner.stop();
   });
 
   it('test http local invoke with authType function with valid signature', async () => {
@@ -222,5 +227,7 @@ def handler(environ, start_response):
     const body = await httpx.read(resp, 'utf8');
 
     expect(body).to.contain('Hello world!');
+
+    await httpInvoke.runner.stop();
   });
 });