- try:
- srv_status = status[self._get_version_tag('applications')] \
- [service][self._get_version_tag('status')] \
- [self._get_version_tag('status')]
- self.log.debug("{}: Service {} status is {}".
- format(self, service, srv_status))
- return srv_status
-
- except KeyError as e:
- self.log.info("self: Did not find service {}, e={}".format(self, service, e))
- return 'NA'
-
- except Exception as e:
- msg = "{}: exception checking service status for {}, e {}". \
- format(self, service, e)
- self.log.error(msg)
- self.log.exception(e)
- raise JujuStatusError(msg)
-
-
- @asyncio.coroutine
- def get_service_status(self, service=None, status=None, env=None):
- ''' Get service status
-
- maintenance : The unit is not yet providing services, but is actively doing stuff.
- unknown : Service has finished an event but the charm has not called status-set yet.
- waiting : Service is unable to progress to an active state because of dependency.
- blocked : Service needs manual intervention to get back to the Running state.
- active : Service correctly offering all the services.
- NA : Service is not deployed
- '''
- pf = partial(self._get_service_status,
- service=service,
- status=status,
- env=env)
- srv_status = yield from self.loop.run_in_executor(
- None,
- pf,
- )
- return srv_status
-
- def _is_service_deployed(self, service=None, status=None, env=None):
- resp = self._get_service_status(service=service,
- status=status,
- env=env)
-
- if resp not in ['terminated', 'NA']:
- return True
-
- return False
-
- @asyncio.coroutine
- def is_service_deployed(self, service=None, status=None, env=None):
- '''Check if the service is deployed'''
- pf = partial(self._is_service_deployed,
- service=service,
- status=status,
- env=env)
- rc = yield from self.loop.run_in_executor(
- None,
- pf,
- )
- return rc
-
- def _is_service_error(self, service=None, status=None, env=None):
- resp = self._get_service_status(service=service,
- status=status,
- env=env)
-
- if resp in ['error']:
- return True
-
- return False
-
- @asyncio.coroutine
- def is_service_error(self, service=None, status=None, env=None):
- '''Check if the service is in error state'''
- pf = partial(self._is_service_error,
- service=service,
- status=status,
- env=env)
- rc = yield from self.loop.run_in_executor(
- None,
- pf,
- )
- return rc
-
- def _is_service_maint(self, service=None, status=None, env=None):
- resp = self._get_service_status(service=service,
- status=status,
- env=env)
-
- if resp in ['maintenance']:
- return True
-
- return False
-
- @asyncio.coroutine
- def is_service_maint(self, service=None, status=None, env=None):
- '''Check if the service is in error state'''
- pf = partial(self._is_service_maint,
- service=service,
- status=status,
- env=env)
- rc = yield from self.loop.run_in_executor(
- None,
- pf,
- )
- return rc
-
- def _is_service_active(self, service=None, status=None, env=None):
- resp = self._get_service_status(service=service,
- status=status,
- env=env)
-
- if resp in ['active']:
- return True
-
- return False
-
- @asyncio.coroutine
- def is_service_active(self, service=None, status=None, env=None):
- '''Check if the service is active'''
- pf = partial(self._is_service_active,
- service=service,
- status=status,
- env=env)
- rc = yield from self.loop.run_in_executor(
- None,
- pf,
- )
- return rc
-
- def _is_service_blocked(self, service=None, status=None, env=None):
- resp = self._get_service_status(service=service,
- status=status,
- env=env)
-
- if resp in ['blocked']:
- return True
-
- return False
-
- @asyncio.coroutine
- def is_service_blocked(self, service=None, status=None, env=None):
- '''Check if the service is blocked'''
- pf = partial(self._is_service_blocked,
- service=service,
- status=status,
- env=env)
- rc = yield from self.loop.run_in_executor(