Refactors file pipeline to add queue

[cuducos]

Como discutido em #59, esse PR inicia uma tentativa de utilizar filas para reduzir o uso de memória – mas ainda é um rascunho, algumas coisas que sei que faltam:

• Flag para baixar os arquivos pelo pipeline ou pela fila
• Escolher o "modelo" no banco de dados a partir do tipo de item do objeto coletado no Scrapy
• Salvar checksum
• Testar se não quebrei nada no refactor do ExtractFileContentPipeline (tentei deixar o código mais legível, usando variáveis com nomes significativos para humanos e menos índices numéricos)
• Configurar RabbitMQ em produção
• Configurar RabbitMQ em desenvolvimento
• Testar se a chamada no send do ator a partir do Scrapy funciona (sendo que a tarefa depende do Django; creio que sim, pois é o worker quem executa mas preciso testar)
• Criar o worker com acesso ao ORM do Django (alguém já usou o django_dramatiq?)

E mais umas coisas que eu não sei ainda : )

Coisas importantes, agora:

• Abri esse PR para ser mergeado na branch em salva-diarios (e não em master), era essa a ideia?
• Claro que estou curioso para qualquer feedback até aqui, mas especialmente:
  • O que acham do # TODO que deixei no datasets/tasks.py (linha 16)?
  • O que acham do # TODO que deixei no scraper/pipelines.py (linha 33)?

[anapaulagomes]

Cuducos, obrigada pela sua primeira contribuição! 🎉

Abri esse PR para ser mergeado na branch em salva-diarios (e não em master), era essa a ideia?

Acredito que seria interessante ter uma feature flag aqui pra manter o funcionamento dos spiders que ainda não possuem dados sendo salvos no banco. Acho bom manter pra quem quiser rodar os spiders localmente fora do comando crawl e ainda assim ter o conteúdo dos arquivos. O que você acha?
Eu, por exemplo, ainda rodo os coletores locais pra fazer pesquisa nos arquivos e atualizar o Kaggle. Ainda é um passo manual que ainda está longe de ser automatizado.

De um jeito ou de outro, esse PR poderia ser mergeado na master mesmo. O salva-diarios é que não pode ser mergeado antes. 🙏

[cuducos]

Atualizando a lista de afazeres do post inicial com outros items:

UPDATE Movi para o post inicial.

[cuducos]

Pronto! Avencei um pouco e agora já não considero mais draft, yay!

Para testarmos em ambiente de staging ou produção no Heroku, precisamos que alguém com acesso ative o CloudAMQP. Deve funcionar normalmente, pois adicionei o worker no Procfile e usei a variável de ambiente padrão do Heroku no core/settings.py.

[anapaulagomes]

Caramba, essa é uma bela feature pra MQ! 🎉 Deixei alguns comentários.
Não pude deployar pra staging pq tua branch não pertence a esse repo mas adicionei o addon e te coloquei como colaborador. Fiquei com preguiça de configurar uma review app, hoje tá corrido (malz ae haha).

[anapaulagomes]

Boa, @cuducos! 🏆
Amanhã vou deployar para staging, assim podemos testar antes de mergear.

[anapaulagomes]

Depois de muitas lágrimas (risos), consegui identificar os motivos pelos quais não estava funcionando:

• Precisamos ter configurado o limite do broker (BROKER_POOL_LIMIT=1) e o número de processos do worker - assim respeitamos os limites do plano gratuito e evitamos o famigerado connection refused
• No método que salvava os diários estávamos aceitando o valor padrão dos itens (um conteúdo vazio) mas estávamos checando se ele é nulo ou não pra atualizar. Por isso nenhum diário tinha seu conteúdo atualizado.
• No pipeline não estávamos retornando o item de volta

Fora isso:

• Adicionei as atas ao rolê, já que o pipeline quebra sem suporte a elas

Commitei essas mudanças nessa branch de testes (baseada na tua). Lá dá pra ver as diferenças.

No momento estou rodando toda a coleta de diários e atas. O pico de mensagens na fila chegou a 11 mil. Três filas, 7 conexões abertas (máximo de 20 conexões). O consumo de memória no worker se manteve em 128MB - not bad.

Infelizmente uma parte ainda teremos que resolver: o arquivo não é encontrado (acredito que pelo fato do heroku não persisti-los), logo não é lido. :(

[anapaulagomes]

Na real, não está funcionando como esperado. Retorna um gerador e nada acontece feijoada. Primeiro pq temos que retornar o item e lá em cima retornamos None e embaixo um generator. Além disso, não dá pra retornar vários itens (o yield está dentro do loop). Consegui pôr pra funcionar assim:

    def item_completed(self, results, item, info):
        if not results:
            return item

        content_from_file_urls = []
        for result in results:
            ok, file_info = result
            if not ok:
                continue

            kwargs = {
                "item_name": item.__class__.__name__,
                "url": file_info["url"],
                "path": f"{FILES_STORE}{file_info['path']}",
                "checksum": file_info["checksum"],
                "save_to_db": ASYNC_FILE_PROCESSING,
                "keep_file": KEEP_FILES,
            }
            if ASYNC_FILE_PROCESSING:
                content_from_file.send(**kwargs)
            else:
                content_from_file_urls.append(content_from_file(**kwargs))
        item["file_content"] = content_from_file_urls
        return item

[cuducos]

Putz! Falha minha. Como results era plural, meu instinto foi criar um gerador, mas acho que (a) não faz sentido, e, com certeza, (b) justifica o erro.

[cuducos]

Infelizmente uma parte ainda teremos que resolver: o arquivo não é encontrado (acredito que pelo fato do heroku não persisti-los), logo não é lido. :(

Poder ser por isso, mas mesmo se persistisse: são processos diferentes, ou seja, Docker containers diferentes. Precisamos de um volume compartilhado.

Acho que se a gente usar um S3, fica barato, ou quase de graça. Vai ser só uma escrita, uma leitura por arquivo (depois apagamos). O que acha?

[anapaulagomes]

Fechando esse PR em favor do #101. Obrigada, @cuducos! 🏅