Feat boot restore (SYN-4163)

synapse/lib/cell.py

@@ -13,6 +13,7 @@
 import contextlib
 import multiprocessing
 
+import aiohttp
 import tornado.web as t_web
 
 import synapse.exc as s_exc
@@ -1647,11 +1648,6 @@ async def getBackupInfo(self):
         '''
         running = int(time.monotonic() - self.backmonostart * 1000) if self.backmonostart else None
 
-        def epochmillis(dtornone):
-            if dtornone is None:
-                return None
-            return int(dtornone.timestamp() * 1000)
-
         retn = {
             'currduration': running,
             'laststart': int(self.backstartdt.timestamp() * 1000) if self.backstartdt else None,
@@ -2587,6 +2583,106 @@ async def _initCellBoot(self):
         async with s_telepath.loadTeleCell(self.dirn):
             await self._bootCellMirror(provconf)
 
+    @classmethod
+    async def _initBootRestore(cls, dirn):
+
+        env = 'SYN_RESTORE_HTTPS_URL'
+        rurl = os.getenv(env, None)
+
+        if rurl is None:
+            return
+
+        dirn = s_common.gendir(dirn)
+
+        # restore.done - Allow an existing URL to be left in the configuration
+        # for a service without issues.
+        doneiden = None
+
+        donepath = s_common.genpath(dirn, 'restore.done')
+        if os.path.isfile(donepath):
+            with s_common.genfile(donepath) as fd:
+                doneiden = fd.read().decode().strip()
+
+        rurliden = s_common.guid(rurl)
+
+        if doneiden == rurliden:
+            return
+
+        # If there is a mismatch vs restore.done, we hulk smash in the restored contents.
+        # There are no seatbelts here to check for existing files / etc.
+
+        clean_url = s_urlhelp.sanitizeUrl(rurl).rsplit('?', 1)[0]
+        logger.warning(f'Restoring {cls.getCellType()} from {env}={clean_url}')
+
+        # Setup get args
+        insecure_marker = 'https+insecure://'
+        kwargs = {}
+        if rurl.startswith(insecure_marker):
+            logger.warning(f'Disabling SSL verification for restore request.')
+            kwargs['ssl'] = False
+            rurl = 'https://' + rurl[len(insecure_marker):]
+
+        tmppath = s_common.gendir(dirn, 'tmp')
+        tarpath = s_common.genpath(tmppath, f'restore_{rurliden}.tgz')
+
+        try:
+
+            with s_common.genfile(tarpath) as fd:
+                async with aiohttp.client.ClientSession() as sess:
+                    async with sess.get(rurl, **kwargs) as resp:
+                        resp.raise_for_status()
+
+                        content_length = int(resp.headers.get('content-length', 0))
+                        if content_length > 100:
+                            logger.warning(f'Downloading {content_length/s_const.megabyte:.3f} MB of data.')
+                            pvals = [int((content_length * 0.01) * i) for i in range(1, 100)]
+                        else:  # pragma: no cover
+                            logger.warning(f'Odd content-length encountered: {content_length}')
+                            pvals = []
+
+                        csize = s_const.kibibyte * 64  # default read chunksize for ClientSession
+                        tsize = 0
+
+                        async for chunk in resp.content.iter_chunked(csize):
+                            fd.write(chunk)
+
+                            tsize = tsize + len(chunk)
+                            if pvals and tsize > pvals[0]:
+                                pvals.pop(0)
+                                percentage = (tsize / content_length) * 100
+                                logger.warning(f'Downloaded {tsize/s_const.megabyte:.3f} MB, {percentage:.3f}%')
+
+            logger.warning(f'Extracting {tarpath} to {dirn}')
+
+            with tarfile.open(tarpath) as tgz:
+                for memb in tgz.getmembers():
+                    if memb.name.find('/') == -1:
+                        continue
+                    memb.name = memb.name.split('/', 1)[1]
+                    logger.warning(f'Extracting {memb.name}')
+                    tgz.extract(memb, dirn)
+
+            # and record the rurliden
+            with s_common.genfile(donepath) as fd:
+                fd.truncate(0)
+                fd.seek(0)
+                fd.write(rurliden.encode())
+
+        except asyncio.CancelledError:
+            raise
+
+        except Exception:
+            logger.exception('Failed to restore cell from URL.')
+            raise
+
+        else:
+            logger.warning('Restored service from URL')
+            return
+
+        finally:
+            if os.path.isfile(tarpath):
+                os.unlink(tarpath)
+
     async def _bootCellProv(self):
 
         provurl = self.conf.get('aha:provision')
@@ -2780,6 +2876,9 @@ async def initFromArgv(cls, argv, outp=None):
 
         logconf = s_common.setlogging(logger, defval=opts.log_level,
                                       structlog=opts.structured_logging)
+
+        await cls._initBootRestore(opts.dirn)
+
         try:
             conf.setdefault('_log_conf', logconf)
             conf.setConfFromOpts(opts)

synapse/tests/test_lib_cell.py

@@ -10,6 +10,7 @@
 
 import synapse.exc as s_exc
 import synapse.common as s_common
+import synapse.cortex as s_cortex
 import synapse.daemon as s_daemon
 import synapse.telepath as s_telepath
 
@@ -1565,3 +1566,92 @@ async def test_mirror_badiden(self):
                     async with await s_cell.Cell.anit(conf=conf01, dirn=path01) as cell01:
                         await stream.wait(timeout=2)
                         self.true(await cell01.waitfini(6))
+
+    async def test_backup_restore(self):
+
+        async with self.getTestAxon(conf={'auth:passwd': 'root'}) as axon:
+            addr, port = await axon.addHttpsPort(0)
+            url = f'https+insecure://root:root@localhost:{port}/api/v1/axon/files/by/sha256/'
+
+            # Make our first backup
+            async with self.getTestCore() as core:
+                self.len(1, await core.nodes('[inet:ipv4=1.2.3.4]'))
+
+                # Punch in a value to the cell.yaml to ensure it persists
+                core.conf['storm:log'] = True
+                core.conf.reqConfValid()
+                s_common.yamlmod({'storm:log': True}, core.dirn, 'cell.yaml')
+
+                async with await axon.upload() as upfd:
+
+                    async with core.getLocalProxy() as prox:
+
+                        async for chunk in prox.iterNewBackupArchive():
+                            await upfd.write(chunk)
+
+                        size, sha256 = await upfd.save()
+                        await asyncio.sleep(0)
+
+            furl = f'{url}{s_common.ehex(sha256)}'
+
+            # Happy test for URL based restore.
+            with self.setTstEnvars(SYN_RESTORE_HTTPS_URL=furl):
+                with self.getTestDir() as cdir:
+                    # Restore works
+                    with self.getAsyncLoggerStream('synapse.lib.cell',
+                                                   'Restoring cortex from SYN_RESTORE_HTTPS_URL') as stream:
+                        async with await s_cortex.Cortex.initFromArgv([cdir]) as core:
+                            self.true(await stream.wait(6))
+                            self.len(1, await core.nodes('inet:ipv4=1.2.3.4'))
+                            self.true(core.conf.get('storm:log'))
+
+                    # Turning the service back on with the restore URL is fine too.
+                    with self.getAsyncLoggerStream('synapse.lib.cell') as stream:
+                        async with await s_cortex.Cortex.initFromArgv([cdir]) as core:
+                            self.len(1, await core.nodes('inet:ipv4=1.2.3.4'))
+
+                            # Take a backup of the cell with the restore.done file in place
+                            async with await axon.upload() as upfd:
+                                async with core.getLocalProxy() as prox:
+                                    async for chunk in prox.iterNewBackupArchive():
+                                        await upfd.write(chunk)
+
+                                    size, sha256r = await upfd.save()
+                                    await asyncio.sleep(0)
+
+                    stream.seek(0)
+                    logs = stream.read()
+                    self.notin('Restoring from url', logs)
+
+                    # grab the restore iden for later use
+                    rpath = s_common.genpath(cdir, 'restore.done')
+                    with s_common.genfile(rpath) as fd:
+                        doneiden = fd.read().decode().strip()
+                    self.true(s_common.isguid(doneiden))
+
+                    # FIXME - DISCUSS THIS. I REMOVED THE SAFETIES THAT WERE HERE, BUT THEY
+                    # WERE PRETTY FRAGILE. SHOULD BE ALL OR NOTHING.
+                    #
+                    # # Restoring into a directory that has been used to boot a cell not okay.
+                    # # Remove the restore.done file forces the restore from happening again.
+                    # os.unlink(rpath)
+                    # with self.raises(s_exc.BadConfValu) as cm:
+                    #     async with self.getTestCore(dirn=cdir) as core:
+                    #         pass
+
+            # Restore a backup which has an existing restore.done file in it - that marker file will get overwritten
+            furl2 = f'{url}{s_common.ehex(sha256r)}'
+            with self.setTstEnvars(SYN_RESTORE_HTTPS_URL=furl2):
+                with self.getTestDir() as cdir:
+                    # Restore works
+                    with self.getAsyncLoggerStream('synapse.lib.cell',
+                                                   'Restoring cortex from SYN_RESTORE_HTTPS_URL') as stream:
+                        async with await s_cortex.Cortex.initFromArgv([cdir]) as core:
+                            self.true(await stream.wait(6))
+                            self.len(1, await core.nodes('inet:ipv4=1.2.3.4'))
+
+                    rpath = s_common.genpath(cdir, 'restore.done')
+                    with s_common.genfile(rpath) as fd:
+                        second_doneiden = fd.read().decode().strip()
+                        self.true(s_common.isguid(second_doneiden))
+                    self.ne(doneiden, second_doneiden)