Basic support for broadcast messages #66

cc-a · 2024-09-11T14:13:19Z

Description

Provides the simplest possible implementation for broadcast messages that I could come up with. A more sophisticated version will be possible after #7 is completed. Adds:

A web endpoint where messages can be deposited - currently saved into a global variable but will change to user sessions (i.e. databese) once we have users...
A script - kafka_consumer.py - that listens for messages send by Kafka and uploads to the web endpoint.
Basic display of messages on the index page.
Some supporting configuration changes.

Working with Kafka is a bit fiddly so let me know how you get on with the instructions added in the README.

No tests yet as I don't they're useful for something so hacky.

I'm not aiming to have this merged by the end of the day so take your time.

Fixes #11 and #10

Type of change

Documentation (non-breaking change that adds or improves the documentation)
New feature (non-breaking change which adds functionality)
Optimization (non-breaking, back-end change that speeds up the code)
Bug fix (non-breaking change which fixes an issue)
Breaking change (whatever its nature)

Key checklist

All tests pass (eg. python -m pytest)
The documentation builds and looks OK (eg. python -m sphinx -b html docs docs/build)
Pre-commit hooks run successfully (eg. pre-commit run --all-files)

Further checks

Code is commented, particularly in hard-to-understand areas
Tests added or an issue has been opened to tackle that in the future. (Indicate issue here: # (issue))

AdrianDAlessandro

I'm happy that everything looks sensible. But I was not able to run it:

$ poetry run drunc-unified-shell --log-level debug ./data/process-manager-pocket-kafka.json
[17:21:19] DEBUG    "unified_shell": Parent's PID: 36084 | This PID: 66041                                                                                                                                                 shell.py:41
Starting process manager with configuration file://./data/process-manager-pocket-kafka.json
Process Process-1:
Traceback (most recent call last):
  File "/Users/adalessa/.pyenv/versions/3.11.1/lib/python3.11/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/Users/adalessa/.pyenv/versions/3.11.1/lib/python3.11/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/Users/adalessa/Projects/QwikApp/QwikApp_web/.venv/lib/python3.11/site-packages/drunc/process_manager/interface/process_manager.py", line 15, in run_pm
    parent_death_pact() # If the parent dies (for example unified shell), we die too
    ^^^^^^^^^^^^^^^^^^^
  File "/Users/adalessa/Projects/QwikApp/QwikApp_web/.venv/lib/python3.11/site-packages/drunc/utils/utils.py", line 183, in parent_death_pact
    assert sys.platform == 'linux', \
           ^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: this fn only works on Linux right now

dune_processes/settings/settings.py

main/templates/main/index.html

AdrianDAlessandro · 2024-09-17T16:04:43Z

main/views.py

+# extreme hackiness suitable only for demonstration purposes
+# will replace this with per-user session storage - once we've added auth
+MESSAGES: list[str] = []
+"""Broadcast messages to display to the user."""


Commenting as a reminder to open this as an issue to address later

Add a TODO label

AdrianDAlessandro · 2024-09-17T16:07:54Z

main/views.py

+      A NO_CONTENT response.
+    """
+    MESSAGES.append(request.POST["message"])
+    return HttpResponse(status=HTTPStatus.NO_CONTENT)


I see how this is hacky. Is the plan to add a messages field to the user model?

Also, is there a desire to make this a REST API using the rest_framework?

Why would a REST API endpoint be best here?

Whilst we could store messages under the user (or other model) I think the ephemeral nature of the data makes storage in user sessions a better fit - https://docs.djangoproject.com/en/5.1/topics/http/sessions/. Ultimately this does store data in the database but it provides a need way of making sure we only collect messages when a user needs to see them.

No desire to use rest_framework as it's a bit of a sledge hammer. After thinking about it more I don't think we actually need an endpoint if we write the Kafka consumer as a custom django-admin command it can populate data into sessions directly.

Why is NO_CONTENT returned here? Is it the same in general for a successful POST?

data/process-manager-pocket-kafka.json

scripts/kafka_consumer.py

dalonsoa

Ok, so I manage to get it working in a VM with Ubuntu and Python 3.11.

All good but I had to invert steps 1 and 2 for it to work. Drunk refused to work with that configuration unless Kafka was up and running first.

I get a lot of scary looking, error messages in drunc, as well. See for example the following (but more nicely formatted). Not sure if that is normal:

[11:07:25] INFO "kafka.coordinator": Elected group leader -- performing partition base.py:521 assignments using range INFO "kafka.coordinator": Successfully joined group base.py:335 drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf with generation 3 INFO "kafka.consumer.subscription_state": Updated partition subscription_state.py:257 assignment: [TopicPartition(topic='control.dalonsoa.process_manage r', partition=0)] INFO "kafka.coordinator.consumer": Setting newly assigned partitions consumer.py:245 {TopicPartition(topic='control.dalonsoa.process_manager', partition=0)} for group drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf [11:08:14] WARNING "kafka.coordinator": Heartbeat session expired, marking coordinator base.py:985 dead WARNING "kafka.coordinator": Marking the coordinator dead (node base.py:715 coordinator-0) for group drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf: Heartbeat session expired. INFO "kafka.cluster": Group coordinator for cluster.py:371 drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf is BrokerMetadata(nodeId='coordinator-0', host='192.168.1.22', port=30092, rack=None) INFO "kafka.coordinator": Discovered coordinator coordinator-0 for group base.py:693 drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf WARNING "kafka.coordinator.consumer": Auto offset commit failed for consumer.py:794 group drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf : CommitFailedError: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max_poll_interval_ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the rebalance timeout with max_poll_interval_ms, or by reducing the maximum size of batches returned in poll() with max_poll_records.

       ERROR    "kafka.coordinator.consumer": Offset commit failed: This is      consumer.py:539
                likely to cause duplicate message delivery                                      
                Traceback (most recent call last):                                              
                  File                                                                          
                "/home/dalonsoa/.cache/pypoetry/virtualenvs/dune-processes-gWVRM                
                ber-py3.11/lib/python3.11/site-packages/kafka/coordinator/consum                
                er.py", line 528, in _maybe_auto_commit_offsets_sync                            
                    self.commit_offsets_sync(self._subscription.all_consumed_off                
                sets())                                                                         
                  File                                                                          
                "/home/dalonsoa/.cache/pypoetry/virtualenvs/dune-processes-gWVRM                
                ber-py3.11/lib/python3.11/site-packages/kafka/coordinator/consum                
                er.py", line 521, in commit_offsets_sync                                        
                    raise future.exception # pylint:                                            
                disable-msg=raising-bad-type                                                    
                    ^^^^^^^^^^^^^^^^^^^^^^                                                      
                kafka.errors.CommitFailedError: CommitFailedError: Commit cannot                
                be completed since the group has already                                        
                            rebalanced and assigned the partitions to another                   
                member.                                                                         
                            This means that the time between subsequent calls to                
                poll()                                                                          
                            was longer than the configured max_poll_interval_ms,                
                which                                                                           
                            typically implies that the poll loop is spending too                
                much                                                                            
                            time message processing. You can address this either                
                by                                                                              
                            increasing the rebalance timeout with                               
                max_poll_interval_ms,                                                           
                            or by reducing the maximum size of batches returned                 
                in poll()                                                                       
                            with max_poll_records.                                              
                                                                                                
       INFO     "kafka.coordinator.consumer": Revoking previously assigned       consumer.py:348
                partitions                                                                      
                {TopicPartition(topic='control.dalonsoa.process_manager',                       
                partition=0)} for group                                                         
                drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf                
       INFO     "kafka.coordinator": (Re-)joining group                              base.py:450
                drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf

[11:08:18] INFO "kafka.coordinator": Elected group leader -- performing partition base.py:521
assignments using range
INFO "kafka.coordinator": Successfully joined group base.py:335
drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf
with generation 5
INFO "kafka.consumer.subscription_state": Updated partition subscription_state.py:257
assignment:
[TopicPartition(topic='control.dalonsoa.process_manage
r', partition=0)]
INFO "kafka.coordinator.consumer": Setting newly assigned partitions consumer.py:245
{TopicPartition(topic='control.dalonsoa.process_manager',
partition=0)} for group
drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf
[11:10:31] WARNING "kafka.coordinator": Heartbeat session expired, marking coordinator base.py:985
dead
WARNING "kafka.coordinator": Marking the coordinator dead (node base.py:715
coordinator-0) for group
drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf:
Heartbeat session expired.
[11:10:34] INFO "kafka.cluster": Group coordinator for cluster.py:371
drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf
is BrokerMetadata(nodeId='coordinator-0', host='192.168.1.22',
port=30092, rack=None)
INFO "kafka.coordinator": Discovered coordinator coordinator-0 for group base.py:693
drunc-stdout-broadcasthandler-dalonsoa-2024-09-18-10-56-22-unssf

README.md

data/process-manager-pocket-kafka.json

docker-compose.yml

dalonsoa · 2024-09-18T09:57:25Z

main/views.py

+# extreme hackiness suitable only for demonstration purposes
+# will replace this with per-user session storage - once we've added auth
+MESSAGES: list[str] = []
+"""Broadcast messages to display to the user."""


Add a TODO label

dalonsoa · 2024-09-18T10:00:21Z

main/views.py

+@require_POST
+@csrf_exempt


What these two do?

require_POST makes it so that the view only accepts POST requests. csrf_exempt removes django's cross site request forgery protection which isn't relevant here.

dalonsoa · 2024-09-18T10:01:28Z

main/views.py

+      A NO_CONTENT response.
+    """
+    MESSAGES.append(request.POST["message"])
+    return HttpResponse(status=HTTPStatus.NO_CONTENT)


Why would a REST API endpoint be best here?

dalonsoa · 2024-09-18T10:06:08Z

scripts/kafka_consumer.py

So, is this all it is from the point of view of the consumer or it is just a minimal version to get going? Should this be running as some sort of daemon rather than a script?

As mentioned in a previous comment I suggest we turn this into a manage.py command. Could probably use a pass for robustness.

cc-a · 2024-09-18T16:17:58Z

Ok, so I manage to get it working in a VM with Ubuntu and Python 3.11.

All good but I had to invert steps 1 and 2 for it to work. Drunk refused to work with that configuration unless Kafka was up and running first.

I get a lot of scary looking, error messages in drunc, as well. See for example the following (but more nicely formatted). Not sure if that is normal:

Well done for persevering. So long as it worked I wouldn't worry about the messages.

jamesturner246

This all makes sense once we get Kafka running. LGTM 👍

Only minor comment is about default values for environment variables. But then I know nothing about best practice in wep apps. Cheers for the help earlier.

dune_processes/settings/settings.py

jamesturner246 · 2024-09-20T17:18:46Z

main/views.py

+      A NO_CONTENT response.
+    """
+    MESSAGES.append(request.POST["message"])
+    return HttpResponse(status=HTTPStatus.NO_CONTENT)


Why is NO_CONTENT returned here? Is it the same in general for a successful POST?

jamesturner246 · 2024-09-20T18:48:15Z

Did have one extra question actually: what's the difference between docker compose and kubernetes, and when/why do they prefer one over the other?

cc-a requested review from dalonsoa, jamesturner246, TinyMarsh and AdrianDAlessandro September 11, 2024 14:13

cc-a force-pushed the messages branch 3 times, most recently from ff49490 to 1ae36ec Compare September 11, 2024 19:35

cc-a added 3 commits September 11, 2024 20:39

Add ability to upload and display broadcast messages to users

51cd0cc

Make server locations configurable

2913cb4

Add instructions for developing with Kafka

8437dba

cc-a force-pushed the messages branch from 1ae36ec to 8437dba Compare September 11, 2024 19:39

cc-a linked an issue Sep 16, 2024 that may be closed by this pull request

API endpoint for Kafka messages #10

Open

AdrianDAlessandro reviewed Sep 17, 2024

View reviewed changes

dalonsoa approved these changes Sep 18, 2024

View reviewed changes

cc-a added 3 commits September 18, 2024 16:54

Add todo for message storage global

c0d66ab

Add readme to data directory

8987b95

Update instructions to start kafka first

10f05e9

jamesturner246 approved these changes Sep 20, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Basic support for broadcast messages #66

Basic support for broadcast messages #66

cc-a commented Sep 11, 2024

AdrianDAlessandro left a comment

AdrianDAlessandro Sep 17, 2024

dalonsoa Sep 18, 2024

AdrianDAlessandro Sep 17, 2024

dalonsoa Sep 18, 2024

cc-a Sep 18, 2024

jamesturner246 Sep 20, 2024

dalonsoa left a comment

dalonsoa Sep 18, 2024

dalonsoa Sep 18, 2024

cc-a Sep 18, 2024

dalonsoa Sep 18, 2024

dalonsoa Sep 18, 2024

cc-a Sep 18, 2024

cc-a commented Sep 18, 2024

jamesturner246 left a comment

jamesturner246 Sep 20, 2024

jamesturner246 commented Sep 20, 2024

		@require_POST
		@csrf_exempt

Basic support for broadcast messages #66

Are you sure you want to change the base?

Basic support for broadcast messages #66

Conversation

cc-a commented Sep 11, 2024

Description

Type of change

Key checklist

Further checks

AdrianDAlessandro left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dalonsoa left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cc-a commented Sep 18, 2024

jamesturner246 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jamesturner246 commented Sep 20, 2024