var/opt/nydus/ops/primordial/log/analyticslogger.py000064400000011252147205541240016572 0ustar00# -*- coding: utf-8 -*- from logging import Logger, LoggerAdapter from typing import Any, Dict, MutableMapping, Optional, Tuple # pylint: disable=W0611 from primordial.log.extra import LogExtra from primordial.log.envelopedata import EnvelopeVersion, EnvelopeRole, EnvelopeDataSchema from primordial.log.envelope import Envelope class AnalyticsLogger(LoggerAdapter): """A LoggerAdapter which specifies an interface for log calls and adds default/initial log data into the record.""" def __init__( self, logger: Logger, envelope_version: Optional[str] = None, data_schema: Optional[str] = None) -> None: self.extra = {} # type: Dict[str, Any] super().__init__(logger, self.extra) # Set up any defaults if envelope_version is None: self._envelope_version = EnvelopeVersion() else: self._envelope_version = EnvelopeVersion.fromString(envelope_version) if data_schema is None: self._data_schema = EnvelopeDataSchema() else: self._data_schema = EnvelopeDataSchema.fromString(data_schema) self.setExtraAttr(Envelope.ENVELOPE_VERSION_FIELD, str(self._envelope_version)) self.setExtraAttr(Envelope.DATA_SCHEMA_FIELD, str(self._data_schema)) self.role = EnvelopeRole() self.setExtraAttr(EnvelopeRole.ROLE_FIELD, str(self.role)) self.validate_data_schema = False @property def performance(self) -> 'AnalyticsLogger': self._role(EnvelopeRole.ROLE_PERFORMANCE) return self @property def development(self) -> 'AnalyticsLogger': self._role(EnvelopeRole.ROLE_DEVELOPMENT) return self @property def business_analytics(self) -> 'AnalyticsLogger': self._role(EnvelopeRole.ROLE_BUSINESS_ANALYTICS) return self @property def validate(self) -> 'AnalyticsLogger': self.validate_data_schema = True return self def setExtraAttr(self, field: str, value: Any) -> None: self.extra[field] = value # Calls to info() etc will call this method by default def process(self, msg: str, kwargs: MutableMapping[str, Any]) -> Tuple[str, MutableMapping[str, Any]]: if 'envelope_version' in kwargs: self.v(kwargs['envelope_version']) if 'data_schema' in kwargs: self.s(kwargs['data_schema']) if 'role' in kwargs: self._role(kwargs['role']) self._validate_schema() logextra = LogExtra.getLogExtra(**self.extra) logextra.set(**kwargs) new_kwargs = {} new_kwargs['extra'] = logextra.get() self._reset() return msg, new_kwargs def envelope_version(self, envelope_version: str) -> 'AnalyticsLogger': """Permanently set envelope_version for the lifetime of this logger.""" self._envelope_version = EnvelopeVersion.fromString(envelope_version) self.setExtraAttr(Envelope.ENVELOPE_VERSION_FIELD, str(self._envelope_version)) return self def data_schema(self, data_schema: str) -> 'AnalyticsLogger': """Permanently set data_schema for the lifetime of this logger.""" self._data_schema = EnvelopeDataSchema.fromString(data_schema) self.setExtraAttr(Envelope.DATA_SCHEMA_FIELD, str(self._data_schema)) return self def v(self, envelope_version: str) -> 'AnalyticsLogger': """Temporary override envelope_version for one logging call.""" self.setExtraAttr(Envelope.ENVELOPE_VERSION_FIELD, envelope_version) return self def s(self, data_schema: str) -> 'AnalyticsLogger': """Temporary override data_schema for one logging call.""" self.setExtraAttr(Envelope.DATA_SCHEMA_FIELD, data_schema) return self def _role(self, role: str) -> None: self.role = EnvelopeRole().set(EnvelopeRole.ROLE_FIELD, role) self.setExtraAttr(EnvelopeRole.ROLE_FIELD, str(self.role)) # Implementers will need to override this class and function for data_schema validation def _validate_schema(self) -> bool: if self.validate_data_schema: # pylint: disable=no-else-return data_schema = self.extra['data_schema'] # pylint: disable=unused-variable # TODO - implement for envelope v2 return True else: return True def _reset(self) -> None: self.role = EnvelopeRole() self.validate_data_schema = False self.extra = {} self.setExtraAttr(EnvelopeRole.ROLE_FIELD, str(self.role)) self.setExtraAttr(Envelope.ENVELOPE_VERSION_FIELD, str(self._envelope_version)) self.setExtraAttr(Envelope.DATA_SCHEMA_FIELD, str(self._data_schema))