Better error handling for the problems occuring inside docker images

beat/core/inputs.py

@@ -31,3 +31,4 @@ from beat.backend.python.inputs import Input
 from beat.backend.python.inputs import InputGroup
 from beat.backend.python.inputs import RemoteInput
 from beat.backend.python.inputs import RemoteInputGroup
+from beat.backend.python.inputs import RemoteException

beat/core/test/mocks.py

@@ -29,6 +29,7 @@
 from ..data import DataSource
 from ..data import DataSink
 
+
 class MockDataSource(DataSource):
 
   def __init__(self, data, indexes):
@@ -44,6 +45,10 @@ class MockDataSource(DataSource):
   def hasMoreData(self):
     return self.current < sum(1 for i in self.data)
 
+
+#----------------------------------------------------------
+
+
 class MockDataSink(DataSink):
 
   class WrittenData:
@@ -65,3 +70,16 @@ class MockDataSink(DataSink):
 
   def isConnected(self):
     return True
+
+
+#----------------------------------------------------------
+
+
+class MockDataSource_Crash(DataSource):
+
+  def next(self):
+    a = b
+
+  def hasMoreData(self):
+    a = b
+    return False

beat/core/test/prefix/algorithms/user/integers_crash/1.json

+{
+  "language": "python",
+  "splittable": true,
+  "groups": [
+    {
+      "name": "main",
+      "inputs": {
+        "in_data": {
+          "type": "user/single_integer/1"
+        }
+      },
+      "outputs": {
+        "out_data": {
+          "type": "user/single_integer/1"
+        }
+      }
+    }
+  ]
+}

beat/core/test/prefix/algorithms/user/integers_crash/1.py

+#!/usr/bin/env python
+# vim: set fileencoding=utf-8 :
+
+###############################################################################
+#                                                                             #
+# Copyright (c) 2016 Idiap Research Institute, http://www.idiap.ch/           #
+# Contact: beat.support@idiap.ch                                              #
+#                                                                             #
+# This file is part of the beat.core module of the BEAT platform.             #
+#                                                                             #
+# Commercial License Usage                                                    #
+# Licensees holding valid commercial BEAT licenses may use this file in       #
+# accordance with the terms contained in a written agreement between you      #
+# and Idiap. For further information contact tto@idiap.ch                     #
+#                                                                             #
+# Alternatively, this file may be used under the terms of the GNU Affero      #
+# Public License version 3 as published by the Free Software and appearing    #
+# in the file LICENSE.AGPL included in the packaging of this file.            #
+# The BEAT platform is distributed in the hope that it will be useful, but    #
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY  #
+# or FITNESS FOR A PARTICULAR PURPOSE.                                        #
+#                                                                             #
+# You should have received a copy of the GNU Affero Public License along      #
+# with the BEAT platform. If not, see http://www.gnu.org/licenses/.           #
+#                                                                             #
+###############################################################################
+
+class Algorithm:
+
+    def process(self, inputs, outputs):
+        a = b
+        return True

beat/core/test/prefix/databases/crash/1.py

@@ -35,7 +35,7 @@ class DoneCrashes:
 class NextCrashes:
 
     def setup(self, *args, **kwargs): return True
-    def done(self): return True
+    def done(self): return False
     def next(self):
         a = b
         return True

beat/core/test/prefix/experiments/user/user/single/1/single_crash.json

+{
+  "analyzers": {
+    "analysis": {
+      "algorithm": "user/integers_echo_analyzer/1",
+      "inputs": {
+        "in_data": "in"
+      }
+    }
+  },
+  "blocks": {
+    "echo": {
+      "algorithm": "user/integers_crash/1",
+      "inputs": {
+        "in_data": "in"
+      },
+      "outputs": {
+        "out_data": "out"
+      }
+    }
+  },
+  "datasets": {
+    "set": {
+      "database": "simple/1",
+      "protocol": "protocol",
+      "set": "set"
+    }
+  },
+  "globals": {
+    "queue": "queue",
+    "environment": {
+      "name": "environment",
+      "version": "1"
+    }
+  }
+}

beat/core/test/prefix/experiments/user/user/single/1/single_db_crash_done.json

+{
+  "analyzers": {
+    "analysis": {
+      "algorithm": "user/integers_echo_analyzer/1",
+      "inputs": {
+        "in_data": "in"
+      }
+    }
+  },
+  "blocks": {
+    "echo": {
+      "algorithm": "user/integers_crash/1",
+      "inputs": {
+        "in_data": "in"
+      },
+      "outputs": {
+        "out_data": "out"
+      }
+    }
+  },
+  "datasets": {
+    "set": {
+      "database": "crash/1",
+      "protocol": "protocol",
+      "set": "done_crashes"
+    }
+  },
+  "globals": {
+    "queue": "queue",
+    "environment": {
+      "name": "environment",
+      "version": "1"
+    }
+  }
+}

beat/core/test/prefix/experiments/user/user/single/1/single_db_crash_next.json

+{
+  "analyzers": {
+    "analysis": {
+      "algorithm": "user/integers_echo_analyzer/1",
+      "inputs": {
+        "in_data": "in"
+      }
+    }
+  },
+  "blocks": {
+    "echo": {
+      "algorithm": "user/integers_crash/1",
+      "inputs": {
+        "in_data": "in"
+      },
+      "outputs": {
+        "out_data": "out"
+      }
+    }
+  },
+  "datasets": {
+    "set": {
+      "database": "crash/1",
+      "protocol": "protocol",
+      "set": "next_crashes"
+    }
+  },
+  "globals": {
+    "queue": "queue",
+    "environment": {
+      "name": "environment",
+      "version": "1"
+    }
+  }
+}

beat/core/test/test_execution.py

@@ -135,7 +135,6 @@ class TestExecution(unittest.TestCase):
         assert 'status' in result
         assert 'stdout' in result
         assert 'stderr' in result
-        assert 'statistics' in result
         assert 'timed_out' in result
         assert 'system_error' in result
         assert 'user_error' in result
@@ -149,8 +148,9 @@ class TestExecution(unittest.TestCase):
           logger.warn("system errors: %s", result['system_error'])
           return result
         assert result['status'] == 0
-        assert isinstance(result['statistics'], dict)
-        assert result['statistics']
+
+        if result.has_key('statistics'):
+          assert isinstance(result['statistics'], dict)
 
       if executor.analysis:
         self.check_output(tmp_prefix, executor.data['result']['path'])
@@ -192,6 +192,30 @@ class TestExecution(unittest.TestCase):
     assert 'NameError' in result['user_error']
     self.assertEqual(result['system_error'], '')
 
+  def test_single_1_crash(self):
+    result = self.execute('user/user/single/1/single_crash', [None])
+    assert result
+    self.assertEqual(result['status'], 1)
+    assert result['user_error']
+    assert 'NameError' in result['user_error']
+    self.assertEqual(result['system_error'], '')
+
+  def test_single_1_db_crash_done(self):
+    result = self.execute('user/user/single/1/single_db_crash_done', [None])
+    assert result
+    self.assertTrue(result['status'] != 0)
+    assert result['user_error']
+    assert 'a = b' in result['user_error']
+    self.assertEqual(result['system_error'], '')
+
+  def test_single_1_db_crash_next(self):
+    result = self.execute('user/user/single/1/single_db_crash_next', [None])
+    assert result
+    self.assertTrue(result['status'] != 0)
+    assert result['user_error']
+    assert 'a = b' in result['user_error']
+    self.assertEqual(result['system_error'], '')
+
   def test_single_1_large(self):
     assert self.execute('user/user/single/1/single_large', [{'out_data': 2.0}]) is None
 

beat/core/test/test_message_handler.py

@@ -49,11 +49,13 @@ from ..agent import MessageHandler
 from ..dataformat import DataFormat
 from ..inputs import RemoteInput
 from ..inputs import RemoteInputGroup
+from ..inputs import RemoteException
 from ..inputs import Input
 from ..inputs import InputGroup
 from ..inputs import InputList
 
 from .mocks import MockDataSource
+from .mocks import MockDataSource_Crash
 
 from . import prefix
 
@@ -165,3 +167,62 @@ class TestMessageHandler(unittest.TestCase):
     nose.tools.eq_(self.remote_input_b.data.value, 200)
 
     assert not self.remote_group.hasMoreData()
+
+
+#----------------------------------------------------------
+
+
+class TestMessageHandlerErrorHandling(unittest.TestCase):
+
+  def setUp(self):
+    dataformat = DataFormat(prefix, 'user/single_integer/1')
+
+    data_source = MockDataSource_Crash()
+
+    input = Input('in', 'user/single_integer/1', data_source)
+
+    group = InputGroup('channel')
+    group.add(input)
+
+    self.input_list = InputList()
+    self.input_list.add(group)
+
+
+    self.server_context = zmq.Context()
+    server_socket = self.server_context.socket(zmq.PAIR)
+    address = 'tcp://127.0.0.1'
+    port = server_socket.bind_to_random_port(address)
+    address += ':%d' % port
+
+    self.message_handler = MessageHandler(self.input_list, self.server_context, server_socket)
+
+
+    self.client_context = zmq.Context()
+    client_socket = self.client_context.socket(zmq.PAIR)
+    client_socket.connect(address)
+
+    self.remote_input = RemoteInput('in', dataformat, client_socket)
+
+    self.remote_group = RemoteInputGroup('channel', False, client_socket)
+    self.remote_group.add(self.remote_input)
+
+    self.remote_input_list = InputList()
+    self.remote_input_list.add(self.remote_group)
+
+    self.message_handler.start()
+
+
+  def test_input_has_more_data(self):
+    self.assertRaises(RemoteException, self.remote_input.hasMoreData)
+
+
+  def test_input_next(self):
+    self.assertRaises(RemoteException, self.remote_input.next)
+
+
+  def test_group_has_more_data(self):
+    self.assertRaises(RemoteException, self.remote_group.hasMoreData)
+
+
+  def test_group_next(self):
+    self.assertRaises(RemoteException, self.remote_group.next)