thelabnyc/django-logpipe

Automated mirror of https://gitlab.com/thelabnyc/django-logpipe

django-logpipe's Issues

How to trigger consumer after produce the message from the producer.

Hello,

I have setup the django logpipe in my project and I have created different services for different module so when I try to produce the message in any service then how to trigger or calling the consumer to consume the message in different services.

Thanks

Django project as a consumer

Is the project still active and stable?

Always getting "Apps aren't loaded yet" error

Hi,

I think I did everything according to the readme, but everytime I try to execute python manage.py run_kafka_consumer, I get the django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet. Error.

This is my apps.py file:

from django.apps import AppConfig
from logpipe import Consumer, register_consumer

from lptest.serializers import MissionSerializer


class LptestConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "lptest"


@register_consumer
def build_person_consumer():
    consumer = Consumer('missions')
    consumer.register(MissionSerializer)
    return consumer

this is my serializer:

class MissionSerializer(serializers.ModelSerializer):
    MESSAGE_TYPE = 'mission'
    VERSION = 1
    KEY_FIELD = 'name'

    class Meta:
        model = Mission
        fields = "__all__"

    @classmethod
    def lookup_instance(cls, name, **kwargs):
        try:
            return Mission.objects.get(name=name)
        except Mission.DoesNotExist:
            pass

I'm not really sure what I'm missing here. The app is listed in the INSTALLED_APPS as well.

Thank you in advance!

Infinite loop in case of ProvisionedThroughputExceededException from Kinesis

Hi, thanks for your amazing package!

We use it as a consumer for Kinesis.
We use the suggested configuration with multiconsumer:

multi = MultiConsumer(people_consumer, places_consumer)
multi.run()

At some point, our Kinesis service raises a ProvisionedThroughputExceededException, multiple times a row, and it ends up with the _get_records method of the kinesis based consumer returning None.

This further leads to a StopIteration but coupled with the indefinite consumer, it ends up in an infinite loop.

For the moment we overrode the _get_records to generate a specific exception and crash for good, but there must be a cleaner solution to avoid entering this infinite loop.

Thank you very much for your advice!

How do i handle errors?

After implementing logpipe on Django, everything worked perfectly, but it happened that a message arrived with all fields empty, this caused Kafka to freeze, which remained on that message and did not let the others pass. I have implemented error handling like this, but I would like to know how I could handle the error by sending messages containing errors to another queue.

    if type == 'add':
        if topic:
            try:
                topic = f"{topic}_posts"
                producer = Producer(topic, serializers.PostCreateKafkaSerializer)
                producer.send(data)
                logger.info("Try to Send to Kafka")
            except Exception as e:
                resp = requests.post(url, headers = {"Content-Type": "application/json", "Authorization": f"Token {token}"}, data=json.dumps(data))
                print(e)
                try:
                    resp.json()
                    logger.warning("Fallback send with POST, Kafka Problems!")
                except Exception:
                    traceback.print_exc()
                    raise NotExpectedException("Error Post")
        else:
            try:
                resp = requests.post(url, headers = {"Content-Type": "application/json", "Authorization": f"Token {token}"}, data=json.dumps(data))
                resp.json()
                logger.warning("No topic send by Post")
            except Exception as error:
                traceback.print_exc()
                logger.error("Error Post")
                raise NotExpectedException("Error Post")

Basically I do a fallback with a post in case I am unable to send it via Kafka, connection problems or Kafka down, what I would like to avoid is Kafka blocking when I send a message that is not in order. Not if it is possible to manage it directly from logpipe.

Add support for localstack?

We use localstack to test our integration w/ AWS infra locally, are you interested in adding a test_mode so that users can specify the local url for kinesis/kafka in test_mode?

logpipe.exceptions.MissingTopicError: Could not find topic test_topic. Does it exist?

Im using the logpipe to listen to my separate kafka server which is running locally on my machine. But it is not able to find it. I tried "localhost:9092" in the bootstrap server settings but its still not working. Appreciate your assistance.