rwlaunchpad/plugins/rwstagingmgr/rift/tasklets/rwstagingmgr/store/file_store.py

   1 """
   2 #
   3 #   Copyright 2016 RIFT.IO Inc
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17
  18 @file file_store.py
  19 @author Varun Prasad (varun.prasad@riftio.com)
  20 @date 28-Sep-2016
  21
  22 """
  23
  24 import logging
  25 import os
  26 import shutil
  27 import tempfile
  28 import time
  29 import uuid
  30 import yaml
  31
  32 import gi
  33 gi.require_version("RwStagingMgmtYang", "1.0")
  34 from gi.repository import RwStagingMgmtYang
  35 import rift.mano.dts as mano_dts
  36 from rift.mano.utils.project import DEFAULT_PROJECT
  37
  38 from .. import model
  39 from ..protocol import StagingStorePublisherProtocol
  40
  41
  42 class StagingAreaExists(Exception):
  43     pass
  44
  45 class InvalidStagingArea(Exception):
  46     pass
  47
  48 class StagingStructureError(Exception):
  49     pass
  50
  51 class StagingFileStore(StagingStorePublisherProtocol):
  52     """File based store for creating and managing staging areas.
  53     """
  54     META_YAML = "meta.yaml"
  55     DEFAULT_EXPIRY = 60 * 60
  56
  57     def __init__(self, tasklet, root_dir=None):
  58         default_path = os.path.join(
  59             os.getenv('RIFT_VAR_ROOT'),
  60             "launchpad/staging")
  61
  62         self.root_dir = root_dir or default_path
  63
  64         if not os.path.isdir(self.root_dir):
  65             os.makedirs(self.root_dir)
  66
  67         self.log = tasklet.log
  68         self.tmp_dir = tempfile.mkdtemp(dir=self.root_dir)
  69
  70         self._cache = {}
  71         self.tasklet = tasklet
  72
  73     def on_recovery(self, staging_areas):
  74         for area in staging_areas:
  75             staging_area = model.StagingArea(area)
  76             self._cache[area.area_id] = staging_area
  77
  78
  79     def get_staging_area(self, area_id):
  80         if area_id not in self._cache:
  81             raise InvalidStagingArea
  82
  83         return self._cache[area_id]
  84
  85
  86     def get_delegate(self, project_name):
  87         if not project_name:
  88             project_name = DEFAULT_PROJECT
  89
  90         try:
  91             proj = self.tasklet.projects[project_name]
  92         except Exception as e:
  93             err = "Project or project name not found {}: {}". \
  94                   format(msg.as_dict(), e)
  95             self.log.error (err)
  96             raise Exception (err)
  97
  98         return proj.publisher
  99
 100     def create_staging_area(self, staging_area_config):
 101         """Create the staging area
 102         Args:
 103             staging_area_config (YangInput_RwStagingMgmt_CreateStagingArea): Rpc input
 104
 105         Returns:
 106             model.StagingArea
 107
 108         Raises:
 109             StagingAreaExists: if the staging area already exists
 110         """
 111         delegate = self.get_delegate(staging_area_config.project_name)
 112
 113         area_id = str(uuid.uuid4())
 114
 115         container_path = os.path.join(self.root_dir, str(area_id))
 116         meta_path = os.path.join(container_path, self.META_YAML)
 117
 118         if os.path.exists(container_path):
 119             raise StagingAreaExists
 120
 121         # Create the dir
 122         os.makedirs(container_path)
 123
 124         config_dict = staging_area_config.as_dict()
 125         config_dict.update({
 126             "area_id": area_id,
 127             "created_time": time.time(),
 128             "status": "LIVE",
 129             "path": container_path
 130             })
 131
 132         staging_area = RwStagingMgmtYang.YangData_RwProject_Project_StagingAreas_StagingArea.from_dict(config_dict)
 133         staging_area = model.StagingArea(staging_area)
 134
 135         self._cache[area_id] = staging_area
 136
 137         try:
 138             if delegate:
 139                 delegate.on_staging_area_create(staging_area.model)
 140         except Exception as e:
 141             self.log.exception(e)
 142
 143         return staging_area
 144
 145     def remove_staging_area(self, staging_area):
 146         """Delete the staging area
 147         Args:
 148             staging_area (str or model.StagingArea): Staging ID or the
 149                 StagingArea object
 150         """
 151         if type(staging_area) is str:
 152             staging_area = self.get_staging_area(staging_area)
 153
 154         delegate = self.get_delegate(staging_area.project_name)
 155
 156         if os.path.isdir(staging_area.model.path):
 157             shutil.rmtree(staging_area.model.path)
 158
 159         staging_area.model.status = "EXPIRED"
 160
 161         try:
 162             if delegate:
 163                 delegate.on_staging_area_delete(staging_area.model)
 164         except Exception as e:
 165             self.log.exception(e)