How to use worker_starts method in Molotov

Best Python code snippet using molotov_python

dependencies.py

Source:dependencies.py

1import logging2import os3import platform4import time5from dataclasses import dataclass6from typing import Any, MutableMapping, Optional, Tuple, Union7from weakref import WeakKeyDictionary8try:9    from functools import singledispatchmethod10except ImportError:11    from singledispatchmethod import singledispatchmethod  # type: ignore12from nameko.containers import WorkerContext13from nameko.events import EventHandler14from nameko.extensions import DependencyProvider, Entrypoint15from nameko.rpc import Rpc16from nameko.web.handlers import HttpRequestHandler17from prometheus_client import Counter, Gauge, Histogram18try:19    from prometheus_client.exposition import choose_encoder20except ImportError:21    from prometheus_client.exposition import choose_formatter as choose_encoder22from prometheus_client.registry import REGISTRY, CollectorRegistry, RestrictedRegistry23from werkzeug.wrappers import Request, Response24logger = logging.getLogger(__name__)25START_TIME = time.time()26class MetricsServer:27    """28    Serves metrics in a format readable by Prometheus scraper.29    Call :meth:`~expose_metrics()` from a service method decorated with `@http`_30    entrypoint to present metrics to Prometheus over HTTP.31    .. _@http: https://nameko.readthedocs.io/en/stable/built_in_extensions.html#http32    """33    def expose_metrics(self, request: Request) -> Response:34        """35        Returns metrics as a HTTP response in Prometheus text format.36        """37        registry: Union[CollectorRegistry, RestrictedRegistry]38        if "name" not in request.args:39            logger.debug(40                "Registry name(s) not found in query string, using global registry"41            )42            registry = REGISTRY43        else:44            names = request.args.getlist("name")45            registry = REGISTRY.restricted_registry(names)46        encoder, content_type = choose_encoder(request.headers["Accept"])47        try:48            # note: registry may be an instance of RestrictedRegistry , but49            # the only thing encoder does, is call registry.collect(); perhaps50            # a Protocol-based type would be more suitable here?51            output = encoder(registry)  # type: ignore52            return Response(output, status=200, content_type=content_type)53        except Exception:54            message = "Failed to generate metrics"55            logger.exception(message)56            return Response(message, status=500)57@dataclass(frozen=True)58class WorkerSummary:59    """60    Represents the final result (or error) of a worker, including duration.61    This simplifies method signatures for singledispatch overrides.62    """63    duration: float64    result: Any65    exc_info: Optional[Tuple]66class PrometheusMetrics(DependencyProvider):67    """68    Dependency provider which measures RPC, event handler and HTTP endpoint69    latency.70    On service start, a few default metrics are declared. These are:71    - ``<prefix>_http_requests_total``72    - ``<prefix>_http_request_latency_seconds``73    - ``<prefix>_rpc_requests_total``74    - ``<prefix>_rpc_request_latency_seconds``75    - ``<prefix>_events_total``76    - ``<prefix>_events_latency_seconds``77    where ``prefix`` is either derived from ``name`` attribute of the service78    class, or :ref:`configured manually <configuration>`.79    """80    def __init__(self):81        self.worker_starts: MutableMapping[WorkerContext, float] = WeakKeyDictionary()82    def setup(self) -> None:83        """84        Configures the dependency provider and declares default metrics.85        """86        # read config from container, use service name as default prefix87        service_name = self.container.service_name88        config = self.container.config.get("PROMETHEUS", {})89        service_config = config.get(service_name, {})90        prefix = service_config.get("prefix", service_name)91        # read application version from an environment variable92        app_version_key = config.get("APP_VERSION_KEY", "APP_VERSION")93        self.app_version = os.environ.get(app_version_key, "unknown")94        self.python_version = platform.python_version()95        # initialize default metrics exposed for every service96        self.service_info = Gauge(97            f"{prefix}_service_info",98            "Always 1; see https://www.robustperception.io/exposing-the-software-version-to-prometheus",99            ["service_version", "python_version"],100        )101        self.service_uptime_seconds = Gauge(102            f"{prefix}_service_uptime_seconds",103            "Uptime of service in seconds",104        )105        self.service_max_workers = Gauge(106            f"{prefix}_service_max_workers",107            "Maximum number of available nameko workers",108        )109        self.service_running_workers = Gauge(110            f"{prefix}_service_running_workers",111            "Number of currently running nameko workers",112        )113        self.http_request_total_counter = Counter(114            f"{prefix}_http_requests_total",115            "Total number of HTTP requests",116            ["http_method", "endpoint", "status_code"],117        )118        self.http_request_latency_histogram = Histogram(119            f"{prefix}_http_request_latency_seconds",120            "HTTP request duration in seconds",121            ["http_method", "endpoint", "status_code"],122        )123        self.rpc_request_total_counter = Counter(124            f"{prefix}_rpc_requests_total",125            "Total number of RPC requests",126            ["method_name"],127        )128        self.rpc_request_latency_histogram = Histogram(129            f"{prefix}_rpc_request_latency_seconds",130            "RPC request duration in seconds",131            ["method_name"],132        )133        self.events_total_counter = Counter(134            f"{prefix}_events_total",135            "Total number of handled events",136            ["source_service", "event_type"],137        )138        self.events_latency_histogram = Histogram(139            f"{prefix}_events_latency_seconds",140            "Event handler duration in seconds",141            ["source_service", "event_type"],142        )143    def get_dependency(self, worker_ctx: WorkerContext) -> MetricsServer:144        """145        Returns an instance of146        :class:`~nameko_prometheus.dependencies.MetricsServer` to be injected147        into the worker.148        """149        return MetricsServer()150    def worker_setup(self, worker_ctx: WorkerContext) -> None:151        """152        Called before service worker starts.153        """154        self.worker_starts[worker_ctx] = time.perf_counter()155    def worker_result(156        self, worker_ctx: WorkerContext, result=None, exc_info=None157    ) -> None:158        """159        Called after service worker completes.160        At this point the default metrics such as worker latency are observed,161        regardless of whether the worker finished successfully or raised an162        exception.163        """164        try:165            start = self.worker_starts.pop(worker_ctx)166        except KeyError:167            logger.warning("No worker_ctx in request start dictionary")168            return169        worker_summary = WorkerSummary(170            duration=time.perf_counter() - start,171            result=result,172            exc_info=exc_info,173        )174        self.observe_entrypoint(worker_ctx.entrypoint, worker_summary)175        self.observe_state_metrics()176    @singledispatchmethod177    def observe_entrypoint(178        self, entrypoint: Entrypoint, worker_summary: WorkerSummary179    ) -> None:180        logger.warning(f"Entrypoint {entrypoint} is not traceable by nameko_prometheus")181    @observe_entrypoint.register(Rpc)182    def _observe_rpc(self, entrypoint: Rpc, worker_summary: WorkerSummary) -> None:183        logger.debug(f"Collect metrics from RPC entrypoint {entrypoint}")184        method_name = entrypoint.method_name185        self.rpc_request_total_counter.labels(method_name=method_name).inc()186        self.rpc_request_latency_histogram.labels(method_name=method_name).observe(187            worker_summary.duration188        )189    @observe_entrypoint.register(HttpRequestHandler)190    def _observe_http(191        self, entrypoint: HttpRequestHandler, worker_summary: WorkerSummary192    ) -> None:193        logger.debug(f"Collect metrics from HTTP entrypoint {entrypoint}")194        http_method = entrypoint.method195        url = entrypoint.url196        if worker_summary.exc_info:197            _, exc, _ = worker_summary.exc_info198            status_code = entrypoint.response_from_exception(exc).status_code199        else:200            status_code = entrypoint.response_from_result(201                worker_summary.result202            ).status_code203        logger.debug(f"Tracing HTTP request: {http_method} {url} {status_code}")204        self.http_request_total_counter.labels(205            http_method=http_method, endpoint=url, status_code=status_code206        ).inc()207        self.http_request_latency_histogram.labels(208            http_method=http_method, endpoint=url, status_code=status_code209        ).observe(worker_summary.duration)210    @observe_entrypoint.register(EventHandler)211    def _observe_event_handler(212        self, entrypoint: EventHandler, worker_summary: WorkerSummary213    ) -> None:214        logger.debug(f"Collect metrics from event handler entrypoint {entrypoint}")215        source_service = entrypoint.source_service216        event_type = entrypoint.event_type217        self.events_total_counter.labels(218            source_service=source_service, event_type=event_type219        ).inc()220        self.events_latency_histogram.labels(221            source_service=source_service, event_type=event_type222        ).observe(worker_summary.duration)223    def observe_state_metrics(self) -> None:224        self.service_info.labels(225            service_version=self.app_version, python_version=self.python_version226        ).set(1)227        self.service_uptime_seconds.set(time.time() - START_TIME)228        self.service_max_workers.set(self.container.max_workers)...

loadtest.py

Source:loadtest.py

...10    Notice that it's not a coroutine.11    """12    pass13@setup()14async def worker_starts(worker_id, args):15    """ This function is called once per worker.16    If it returns a mapping, it will be used with all requests.17    You can add things like Authorization headers for instance,18    by setting a "headers" key.19    """20    headers = {"SomeHeader": "1"}21    return {"headers": headers}22@teardown()23def worker_ends(worker_id):24    """ This functions is called when the worker is done.25    Notice that it's not a coroutine.26    """27    pass28@global_teardown()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.