Merge pull request #4707 from golemfactory/task_api_envs

Update to Task API v0.16.1

golem/database/database.py

@@ -61,7 +61,7 @@ def execute_sql(self, sql, params=None, require_commit=True):
 
 
 class Database:
-    SCHEMA_VERSION = 36
+    SCHEMA_VERSION = 37
 
     def __init__(self,  # noqa pylint: disable=too-many-arguments
                  db: peewee.Database,

golem/database/schemas/037_requested_task_env_id.py

@@ -0,0 +1,17 @@
+# pylint: disable=no-member
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+import peewee as pw
+
+SCHEMA_VERSION = 37
+
+
+def migrate(migrator, database, fake=False, **kwargs):
+    migrator.add_fields(
+        'requestedtask',
+        env_id=pw.CharField(max_length=255, null=True))
+
+
+def rollback(migrator, database, fake=False, **kwargs):
+    migrator.remove_fields('requestedtask', 'env_id')

golem/envs/docker/cpu.py

@@ -10,6 +10,7 @@
 
 from dataclasses import dataclass, field, asdict
 from docker.errors import APIError
+from golem_task_api.envs import DOCKER_CPU_ENV_ID
 from twisted.internet.defer import Deferred, inlineCallbacks, succeed
 from twisted.internet.threads import deferToThread
 from urllib3.contrib.pyopenssl import WrappedSocket
@@ -47,7 +48,6 @@
 mem = CONSTRAINT_KEYS['mem']
 cpu = CONSTRAINT_KEYS['cpu']
 
-DOCKER_CPU_ENV_ID = 'docker_cpu'
 DOCKER_CPU_METADATA = EnvMetadata(
     id=DOCKER_CPU_ENV_ID,
     description='Docker environment using CPU'

golem/envs/docker/gpu.py

@@ -3,6 +3,7 @@
 from typing import Any, Dict, List, Optional
 
 from dataclasses import dataclass, field
+from golem_task_api.envs import DOCKER_GPU_ENV_ID
 
 from golem.core.common import update_dict
 from golem.envs import (
@@ -22,7 +23,6 @@
 logger = getLogger(__name__)
 
 
-DOCKER_GPU_ENV_ID = 'docker_gpu'
 DOCKER_GPU_METADATA = EnvMetadata(
     id=DOCKER_GPU_ENV_ID,
     description='Docker environment using GPU'

golem/model.py

@@ -684,6 +684,7 @@ class RequestedTask(BaseModel):
     name = CharField(null=True)
     status = StringEnumField(enum_type=taskstate.TaskStatus, null=False)
 
+    env_id = CharField(null=True)
     prerequisites = JsonField(null=False, default='{}')
 
     task_timeout = IntegerField(null=False)  # milliseconds

golem/task/requestedtaskmanager.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 from datetime import timedelta
 import logging
 from pathlib import Path
@@ -122,7 +123,6 @@ def create_task(
             app_id=golem_params.app_id,
             name=golem_params.name,
             status=TaskStatus.creating,
-            # prerequisites='{}',
             task_timeout=golem_params.task_timeout,
             subtask_timeout=golem_params.subtask_timeout,
             start_time=default_now(),
@@ -177,11 +177,14 @@ async def init_task(self, task_id: TaskId) -> None:
 
         app_client = await self._get_app_client(task.app_id)
         logger.debug('init_task(task_id=%r) - creating task', task_id)
-        await app_client.create_task(
+        reply = await app_client.create_task(
             task.task_id,
             task.max_subtasks,
             task.app_params,
         )
+        task.env_id = reply.env_id
+        task.prerequisites = json.loads(reply.prerequisites_json)
+        task.save()
         logger.debug('init_task(task_id=%r) after', task_id)
 
     @staticmethod

golem/task/taskcomputer.py

@@ -11,6 +11,7 @@
 from dataclasses import dataclass
 from golem_messages.message.tasks import ComputeTaskDef, TaskHeader
 from golem_task_api import ProviderAppClient, constants as task_api_constants
+from golem_task_api.envs import DOCKER_CPU_ENV_ID, DOCKER_GPU_ENV_ID
 from pydispatch import dispatcher
 from twisted.internet import defer
 
@@ -22,8 +23,8 @@
 from golem.docker.manager import DockerManager
 from golem.docker.task_thread import DockerTaskThread
 from golem.envs import EnvId
-from golem.envs.docker.cpu import DockerCPUConfig, DOCKER_CPU_ENV_ID
-from golem.envs.docker.gpu import DockerGPUConfig, DOCKER_GPU_ENV_ID
+from golem.envs.docker.cpu import DockerCPUConfig
+from golem.envs.docker.gpu import DockerGPUConfig
 from golem.hardware import scale_memory, MemSize
 from golem.manager.nodestatesnapshot import ComputingSubtaskStateSnapshot
 from golem.resource.dirmanager import DirManager

requirements.txt

@@ -41,7 +41,7 @@ eventlet==0.24.1
 fs==2.4.4
 Golem-Messages==3.11.0
 Golem-Smart-Contracts-Interface==1.10.2
-golem_task_api==0.15.1
+golem_task_api==0.16.1
 greenlet==0.4.15
 h2==3.0.1
 hpack==3.0.0

requirements_to-freeze.txt

@@ -20,7 +20,7 @@ eth-utils==1.0.3
 ethereum==1.6.1
 Golem-Messages==3.11.0
 Golem-Smart-Contracts-Interface==1.10.2
-golem_task_api==0.15.1
+golem_task_api==0.16.1
 html2text==2018.1.9
 humanize==0.5.1
 incremental==17.5.0

tests/golem/task/test_newtaskcomputer.py

@@ -6,16 +6,14 @@
 
 from golem_messages.message import ComputeTaskDef
 from golem_task_api import ProviderAppClient, TaskApiService
+from golem_task_api.envs import DOCKER_CPU_ENV_ID
 from twisted.internet import defer
 
 from golem.clientconfigdescriptor import ClientConfigDescriptor
 from golem.core.deferred import deferred_from_future
 from golem.core.statskeeper import IntStatsKeeper
 from golem.envs import Runtime
-from golem.envs.docker.cpu import (
-    DOCKER_CPU_ENV_ID,
-    DockerCPUConfig
-)
+from golem.envs.docker.cpu import DockerCPUConfig
 from golem.task.envmanager import EnvironmentManager
 from golem.task.taskcomputer import NewTaskComputer
 from golem.testutils import TempDirFixture

tests/golem/task/test_requestedtaskmanager.py

@@ -2,6 +2,7 @@
 # ^^ Pytest fixtures in the same file require the same name
 
 import asyncio
+import json
 from pathlib import Path
 
 from freezegun import freeze_time
@@ -26,14 +27,19 @@
 
 @pytest.fixture
 def mock_client(monkeypatch):
-    _mock_client = AsyncMock(spec=RequestorAppClient)
+    client_mock = AsyncMock(spec=RequestorAppClient)
 
     @asyncio.coroutine
     def mock_create(*_args, **_kwargs):
-        return _mock_client
+        return client_mock
 
     monkeypatch.setattr(RequestorAppClient, 'create', mock_create)
-    return _mock_client
+
+    client_mock.create_task.return_value = Mock(
+        env_id='env_id',
+        prerequisites_json='null'
+    )
+    return client_mock
 
 
 @pytest.mark.usefixtures('pytest_database_fixture')
@@ -74,21 +80,28 @@ def test_create_task(self):
     async def test_init_task(self, mock_client):
         # given
         task_id = self._create_task()
+        env_id = 'test_env'
+        prerequisites = {'key': 'value'}
+        mock_client.create_task.return_value = Mock(
+            env_id=env_id,
+            prerequisites_json=json.dumps(prerequisites)
+        )
         # when
         await self.rtm.init_task(task_id)
         row = RequestedTask.get(RequestedTask.task_id == task_id)
         # then
         assert row.status == TaskStatus.creating
+        assert row.env_id == env_id
+        assert row.prerequisites == prerequisites
         mock_client.create_task.assert_called_once_with(
             row.task_id,
             row.max_subtasks,
             row.app_params
         )
         self.app_manager.enabled.assert_called_once_with(row.app_id)
 
-    @pytest.mark.usefixtures('mock_client')
     @pytest.mark.asyncio
-    async def test_init_task_wrong_status(self):
+    async def test_init_task_wrong_status(self, mock_client):
         # given
         task_id = self._create_task()
         # when
@@ -99,9 +112,8 @@ async def test_init_task_wrong_status(self):
         with pytest.raises(RuntimeError):
             await self.rtm.init_task(task_id)
 
-    @pytest.mark.usefixtures('mock_client')
     @pytest.mark.asyncio
-    async def test_start_task(self):
+    async def test_start_task(self, mock_client):
         # given
         task_id = self._create_task()
         await self.rtm.init_task(task_id)

tests/golem/task/test_taskserver.py

@@ -21,6 +21,7 @@
 from golem_messages.factories.datastructures import p2p as dt_p2p_factory
 from golem_messages.message import ComputeTaskDef
 from golem_messages.utils import encode_hex as encode_key_id, pubkey_to_address
+from golem_task_api.envs import DOCKER_CPU_ENV_ID
 from requests import HTTPError
 
 from golem import testutils
@@ -34,7 +35,6 @@
     UnsupportReason,
 )
 from golem.envs import Environment as NewEnv
-from golem.envs.docker.cpu import DOCKER_CPU_ENV_ID
 from golem.network.hyperdrive.client import HyperdriveClientOptions, \
     HyperdriveClient, to_hyperg_peer
 from golem.resource import resourcemanager