Chose producer_cls globally and for each producer

[Gerleff]

I'd like to use something like this:

def make_func_push(func: Callable) -> Callable:
    """Decores func or method, making it push metrics into Prometeus."""
    return prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)(metrics.calculate(func))


class PushingProducer(AIOKafkaProducer):
    """Pushing to Prometeus Kafka producer."""

    send = make_func_push(AIOKafkaProducer.send)
    send_batch = make_func_push(AIOKafkaProducer.send_batch)

as producer for whole application by default or with chosen producers.

For now, monkeypatch is an option, but I'd thank much for more flexible solution.

[sternakt]

Hi,

This is how i understood the issue, please tell if I misinterpreted.
So, you want to pass a function to metrics.calculate() which will call it, I guess, when the metrics are calculated to push them to Kafka.

Wouldn't this solve your problem?

app = FastKafka(...)

@app.produces()
async def to_prometheus_metrics(metrics: Metrics) -> MetricsDataFormat:
    return MetricsDataFormat(metrics=metrics)

prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)(
    metrics.calculate(to_prometheus_metrics)
)

[Gerleff]

My goal is to decorate AIOKafkaProducer.send & AIOKafkaProducer.send_batch methods, so

class MyProd(AIOKafkaProducer):
    @prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)
    @metrics.calculate
    async def send(...):
        ...
        
    @prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)
    @metrics.calculate
    async def send_batch(...):
        ...

With that producer all "send" calls will push metrics.

Can you please explain a little deeper, how provided above solution will solve my problem?

[sternakt]

My mistake, I think I understand now,

So, the prometheus_pusher and metrics decorators collect and push metrics from the producing calls.

Is there a particular reason why you don't decorate the producing functions instead of send and send_batch? Something like this:

@app.produces()
@prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)
@metrics.calculate
async def to_some_topic(...) -> ...:
    ...

And about the global producer:

I guess that you want to have a global producer to collect the metrics for all the messages being sent, no matter the topic?
I'm afraid this is not easily done at the moment, the app internally creates a producer for each topic individually.

When we started with the implementation, we had an option to pass the producer object to the produces decorator, so you can pass your own custom producer (or a global one if that would be the case), we could plan for the reintroduction of that feature in the next sprints if that would solve your problem more elegantly

[Gerleff]

Sorry for late answer(

We are collecting metrics, including Summary. Summary measures time, required to execute function.

@prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)
@metrics.calculate
@app.produces()
async def to_some_topic(x):
    return x

mb if decorators are placed that way, we'll summarize time to execute producer's functionality instead of lambda x: x? =)
If so, it's gonna be solution, which can be developed into overriding produces method in FastKafka's child class.