refactor(python)!: use bidder indexing as input

locustfiles/_utils.py

@@ -25,6 +25,18 @@ class AdRequest(typing.TypedDict):
     """Device used by the user visiting the publisher's site"""
 
 
+class BidderFactory(polyfactory.factories.TypedDictFactory[Bidder]):
+    __set_as_default_factory_for_type__ = True
+    _names = [
+        polyfactory.factories.TypedDictFactory.__faker__.unique.company()
+        for _ in range(20)
+    ]
+
+    @classmethod
+    def name(cls) -> str:
+        return str(cls.__random__.choice(cls._names))
+
+
 class AdRequestFactory(polyfactory.factories.TypedDictFactory[AdRequest]):
     """Utility object to generate random AdRequests"""
 

locustfiles/root.py

@@ -10,8 +10,8 @@ class Root(locust.FastHttpUser):
     def on_start(self) -> None:
         ad_request = AdRequestFactory.build()
 
-        self.fabrics = [
-            {
+        self.fabrics = {
+            bidder["name"]: {
                 "featureMap": {
                     "bidder": bidder["name"],
                     "userSynced": bidder.get("user_id") is not None,
@@ -22,7 +22,7 @@ def on_start(self) -> None:
                 "groundTruth": {"hasResponse": True},
             }
             for bidder in ad_request["bidders"]
-        ]
+        }
         return super().on_start()
 
     @locust.task(100)

locustfiles/rtb.py

@@ -16,8 +16,8 @@ def handleAdRequest(self):
         ad_request = AdRequestFactory.build()
 
         # Build a list of features for each bidder in the ad request
-        feature_maps = [
-            {
+        fabrics = {
+            bidder["name"]: {
                 "featureMap": {
                     "bidder": bidder["name"],
                     "userSynced": bidder.get("user_id") is not None,
@@ -29,12 +29,14 @@ def handleAdRequest(self):
                 }
             }
             for bidder in ad_request["bidders"]
-        ]
+        }
         # Do a single call to the Greenbids Tailor service
-        fabrics = self.client.put("", json=feature_maps).json()
+        fabrics = self.client.put("", json=fabrics).json()
+        # Prepare a flag to know if you must make the POST call
+        is_exploration = False
 
         # Do your regular calls here to send a bid requests to the selected bidders
-        for fabric in fabrics:
+        for fabric in fabrics.values():
             if not fabric["prediction"]["shouldSend"]:
                 # Skip any bidder that as too few response probability
                 continue
@@ -49,7 +51,9 @@ def handleAdRequest(self):
             # Store the outcome in the fabric
             fabric["groundTruth"] = dict(hasResponse=hasResponse)
 
+            is_exploration = is_exploration or fabric["prediction"]["isExploration"]
+
         # For a sample of calls, report the outcomes to the Greenbids Tailor POST endpoint
-        if fabrics[0]["prediction"]["isExploration"]:
+        if is_exploration:
             # You may use a fire-and-forget mechanism
             self.client.post("", json=fabrics)

python/src/greenbids/tailor/core/app/main.py

@@ -47,6 +47,7 @@ def _setup_logging():
     tracer_provider=telemetry.tracer_provider,
     meter_provider=telemetry.meter_provider,
     excluded_urls=f"{healthz.router.prefix}/.*",
+    http_capture_headers_server_response=["greenbids-tailor-.*"],
 )
 
 app.include_router(root.router)

python/src/greenbids/tailor/core/app/routers/root.py

@@ -1,27 +1,53 @@
-from fastapi import APIRouter
-from greenbids.tailor.core import fabric
+from fastapi import APIRouter, Response
+from greenbids.tailor.core import fabric, telemetry
 from .. import resources
 
 router = APIRouter(tags=["Main"])
 
+_invokation_count = telemetry.meter.create_counter("greenbids_tailor_invokation_count")
+_fabrics_count = telemetry.meter.create_counter("greenbids_tailor_fabrics_count")
+
 
 @router.put("/")
 async def get_buyers_probabilities(
-    fabrics: list[fabric.Fabric],
-) -> list[fabric.Fabric]:
+    fabrics: dict[str, fabric.Fabric], response: Response
+) -> dict[str, fabric.Fabric]:
     """Compute the probability of the buyers to provide a bid.
 
     This must be called for each adcall.
     Only the feature map attribute of the fabrics needs to be present.
     The prediction attribute will be populated in the returned response.
     """
-    return resources.get_instance().gb_model.get_buyers_probabilities(fabrics)
+    inst = resources.get_instance()
+    results = inst.gb_model.get_buyers_probabilities(fabrics)
+    is_exploration = False
+    for k, f in results.items():
+        is_exploration = is_exploration or f.prediction.is_exploration
+        _fabrics_count.add(
+            1,
+            attributes={
+                "greenbids.tailor.model_name": inst.gb_model_name,
+                "greenbids.tailor.fabric_id": k,
+                "greenbids.tailor.should_send": f.prediction.should_send,
+            },
+        )
+    _invokation_count.add(
+        1,
+        {
+            "greenbids.tailor.model_name": inst.gb_model_name,
+            "greenbids.tailor.is_exploration": is_exploration,
+        },
+    )
+
+    response.headers["greenbids-tailor-is-exploration"] = str(is_exploration)
+    response.headers["greenbids-tailor-model-name"] = inst.gb_model_name
+    return results
 
 
 @router.post("/")
 async def report_buyers_status(
-    fabrics: list[fabric.Fabric],
-) -> list[fabric.Fabric]:
+    fabrics: dict[str, fabric.Fabric],
+) -> dict[str, fabric.Fabric]:
     """Train model according to actual outcome.
 
     This must NOT be called for each adcall, but only for exploration ones.

python/src/greenbids/tailor/core/models.py

@@ -17,15 +17,15 @@ class Model(ABC):
     @abstractmethod
     def get_buyers_probabilities(
         self,
-        fabrics: list[fabric.Fabric],
-    ) -> list[fabric.Fabric]:
+        fabrics: dict[str, fabric.Fabric],
+    ) -> dict[str, fabric.Fabric]:
         raise NotImplementedError
 
     @abstractmethod
     def report_buyers_status(
         self,
-        fabrics: list[fabric.Fabric],
-    ) -> list[fabric.Fabric]:
+        fabrics: dict[str, fabric.Fabric],
+    ) -> dict[str, fabric.Fabric]:
         raise NotImplementedError
 
     def dump(self, fp: typing.BinaryIO) -> None:
@@ -44,16 +44,23 @@ def __init__(self):
 
     def get_buyers_probabilities(
         self,
-        fabrics: list[fabric.Fabric],
-    ) -> list[fabric.Fabric]:
-        prediction = fabric.Prediction(exploration_rate=0.2)
-        return [f.model_copy(update=dict(prediction=prediction)) for f in fabrics]
+        fabrics: dict[str, fabric.Fabric],
+    ) -> dict[str, fabric.Fabric]:
+        prediction = fabric.Prediction(
+            exploration_rate=float(
+                os.environ.get("GREENBIDS_TAILOR_MODELS_NONE_EXPLORATION_RATE", 0.2)
+            )
+        )
+        return {
+            k: f.model_copy(update=dict(prediction=prediction))
+            for k, f in fabrics.items()
+        }
 
     def report_buyers_status(
         self,
-        fabrics: list[fabric.Fabric],
-    ) -> list[fabric.Fabric]:
-        self._logger.debug([f.feature_map.root for f in fabrics[:1]])
+        fabrics: dict[str, fabric.Fabric],
+    ) -> dict[str, fabric.Fabric]:
+        self._logger.debug(next(f.feature_map.root for f in fabrics.values()))
         return fabrics
 
 

python/src/greenbids/tailor/core/telemetry.py

@@ -22,14 +22,14 @@
 
 
 from .logging import RateLimitingFilter
-from greenbids.tailor.core import version
+from greenbids.tailor import core
 
 RESOURCE = resources.Resource.create(
     {
         resources.SERVICE_INSTANCE_ID: str(
             os.environ.get("GREENBIDS_TAILOR_API_USER", "Unknown")
         ),
-        resources.SERVICE_VERSION: version,
+        resources.SERVICE_VERSION: core.version,
     }
 )
 
@@ -41,10 +41,13 @@
     )
     metric_readers.append(PrometheusMetricReader())
 meter_provider = metrics.MeterProvider(metric_readers=metric_readers, resource=RESOURCE)
+_instrumentation_name = ".".join(core.__name__.split(".")[:-1])
+meter = meter_provider.get_meter(_instrumentation_name, core.version)
 
 _OTLP_TRACES_PROCESSOR = BatchSpanProcessor(OTLPSpanExporter())
 tracer_provider = trace.TracerProvider(resource=RESOURCE)
 tracer_provider.add_span_processor(_OTLP_TRACES_PROCESSOR)
+tracer = tracer_provider.get_tracer(_instrumentation_name, core.version)
 
 _OTLP_LOGS_PROCESSOR = BatchLogRecordProcessor(OTLPLogExporter())
 logger_provider = logs.LoggerProvider(resource=RESOURCE)