Add support for stream writer Synapse workers

As stream writer workers are also powered by the `generic_worker`
Synapse app, this necessitated that we provide means for distinguishing
between them and regular `generic_workers`.

I've also taken the time to optimize nginx configuration generation
(more Jinja2 macro usage, less duplication).

Worker names have also changed.
Workers are now named sequentially like this:
- `matrix-synapse-worker-0-generic`
- `matrix-synapse-worker-1-stream-writer-typing`
- `matrix-synapse-worker-2-pusher`

instead of `matrix-synapse-worker_generic_worker-18111` (indexed with a
port number).

People who modify `matrix_synapse_workers_enabled_list` directly will
need to adjust their configuration.

roles/matrix-synapse/templates/synapse/homeserver.yaml.j2

@@ -352,13 +352,13 @@ worker_app: synapse.app.homeserver
 
 # thx https://oznetnerd.com/2017/04/18/jinja2-selectattr-filter/
 # reduce the main worker's offerings to core homeserver business
-{% if matrix_synapse_workers_enabled_list | selectattr('type', 'equalto', 'federation_sender') | list  %}
+{% if matrix_synapse_workers_enabled_list | selectattr('type', 'equalto', 'federation_sender') | list | length > 0  %}
 send_federation: false
 {% endif %}
-{% if matrix_synapse_workers_enabled_list | selectattr('type', 'equalto', 'media_repository') | list  %}
+{% if matrix_synapse_workers_enabled_list | selectattr('type', 'equalto', 'media_repository') | list | length > 0  %}
 enable_media_repo: false
 {% endif %}
-{% if matrix_synapse_workers_enabled_list | selectattr('type', 'equalto', 'pusher') | list  %}
+{% if matrix_synapse_workers_enabled_list | selectattr('type', 'equalto', 'pusher') | list | length > 0  %}
 start_pushers: false
 {% endif %}
 
@@ -2870,6 +2870,7 @@ opentracing:
 #  worker1:
 #    host: localhost
 #    port: 8034
+instance_map: {{ matrix_synapse_instance_map | to_json }}
 
 # Experimental: When using workers you can define which workers should
 # handle event persistence and typing notifications. Any worker
@@ -2878,6 +2879,7 @@ opentracing:
 #stream_writers:
 #  events: worker1
 #  typing: worker1
+stream_writers: {{ matrix_synapse_stream_writers | to_json }}
 
 # The worker that is used to run background tasks (e.g. cleaning up expired
 # data). If not provided this defaults to the main process.

roles/matrix-synapse/templates/synapse/prometheus/external_prometheus.yml.example.j2

@@ -24,8 +24,8 @@ scrape_configs:
           job: "master"
           index: "0"
 {% for worker in matrix_synapse_workers_enabled_list %}
-  - job_name: 'synapse-{{ worker.type }}-{{ worker.instanceId }}'
-    metrics_path: /metrics/synapse/worker/{{ worker.type }}-{{ worker.instanceId }}
+  - job_name: '{{ worker.name }}'
+    metrics_path: /metrics/synapse/worker/{{ worker.id }}
     scheme: {{ 'https' if matrix_nginx_proxy_https_enabled|default(true) else 'http' }}
 {% if matrix_nginx_proxy_proxy_matrix_metrics_basic_auth_enabled|default(true) %}
     basic_auth:
@@ -35,6 +35,7 @@ scrape_configs:
     static_configs:
       - targets: ['{{ matrix_server_fqn_matrix }}:{{ matrix_nginx_proxy_container_https_host_bind_port|default(443) if matrix_nginx_proxy_https_enabled|default(true) else matrix_nginx_proxy_container_http_host_bind_port|default(80) }}']
         labels:
+          worker_id: {{ worker.id }}
           job: "{{ worker.type }}"
-          index: "{{ worker.instanceId }}"
+          app: {{ worker.app }}
 {% endfor %}

roles/matrix-synapse/templates/synapse/systemd/matrix-synapse-worker.service.j2

@@ -43,7 +43,7 @@ ExecStart={{ matrix_host_command_docker }} run --rm --name {{ matrix_synapse_wor
 			{{ arg }} \
 			{% endfor %}
 			{{ matrix_synapse_docker_image }} \
-			run -m synapse.app.{{ matrix_synapse_worker_details.type }} -c /data/homeserver.yaml -c /data/{{ matrix_synapse_worker_config_file_name }}
+			run -m synapse.app.{{ matrix_synapse_worker_details.app }} -c /data/homeserver.yaml -c /data/{{ matrix_synapse_worker_config_file_name }}
 
 
 ExecStop=-{{ matrix_host_command_sh }} -c '{{ matrix_host_command_docker }} kill {{ matrix_synapse_worker_container_name }} 2>/dev/null || true'

roles/matrix-synapse/templates/synapse/systemd/matrix-synapse.service.j2

@@ -12,7 +12,7 @@ Wants={{ service }}
 
 {% if matrix_synapse_workers_enabled %}
 {% for matrix_synapse_worker_details in matrix_synapse_workers_enabled_list %}
-Wants=matrix-synapse-worker-{{ matrix_synapse_worker_details.type }}-{{ matrix_synapse_worker_details.port }}.service
+Wants={{ matrix_synapse_worker_details.name }}.service
 {% endfor %}
 {% endif %}
 

roles/matrix-synapse/templates/synapse/worker.yaml.j2

@@ -1,27 +1,38 @@
 #jinja2: lstrip_blocks: "True"
-worker_app: synapse.app.{{ matrix_synapse_worker_details.type }}
-worker_name: {{ matrix_synapse_worker_details.type ~ ':' ~ matrix_synapse_worker_details.port }}
+worker_app: synapse.app.{{ matrix_synapse_worker_details.app }}
+worker_name: {{ matrix_synapse_worker_details.name }}
 
 {% if matrix_synapse_replication_listener_enabled %}
 worker_replication_host: matrix-synapse
 worker_replication_http_port: {{ matrix_synapse_replication_http_port }}
 {% endif %}
 
-{% set has_listeners = (matrix_synapse_worker_details.type not in [ 'appservice', 'federation_sender', 'pusher' ] or matrix_synapse_metrics_enabled) %}
-
 {% set http_resources = [] %}
 
-{% if matrix_synapse_worker_details.type in ['generic_worker', 'user_dir'] %}
+{% if matrix_synapse_worker_details.type == 'user_dir' %}
   {% set http_resources = http_resources + ['client'] %}
 {% endif %}
-{% if matrix_synapse_worker_details.type in ['generic_worker'] %}
-  {% set http_resources = http_resources+ ['federation'] %}
+{% if matrix_synapse_worker_details.type == 'generic_worker' %}
+  {% set http_resources = http_resources + ['client', 'federation'] %}
 {% endif %}
-{% if matrix_synapse_worker_details.type in ['media_repository'] %}
+{#
+  None of the background workers need to handle federation traffic.
+  Only some of the stream writers need to handle client traffic.
+#}
+{% if matrix_synapse_worker_details.type == 'stream_writer' and matrix_synapse_worker_details.webserving %}
+  {% set http_resources = http_resources + ['client'] %}
+{% endif %}
+{% if matrix_synapse_worker_details.type == 'media_repository' %}
   {% set http_resources = http_resources + ['media'] %}
 {% endif %}
 
-{% if http_resources|length > 0 or matrix_synapse_metrics_enabled %}
+{% set replication_http_resources = [] %}
+{% if matrix_synapse_worker_details.type == 'stream_writer' %}
+  {# All background workers need to handle replication traffic. #}
+  {% set replication_http_resources = replication_http_resources + ['replication'] %}
+{% endif %}
+
+{% if http_resources|length > 0 or matrix_synapse_metrics_enabled or replication_http_resources|length > 0 %}
 worker_listeners:
 {% if http_resources|length > 0 %}
   - type: http
@@ -36,6 +47,13 @@ worker_listeners:
     bind_addresses: ['0.0.0.0']
     port: {{ matrix_synapse_worker_details.metrics_port }}
 {% endif %}
+{% if replication_http_resources|length > 0 %}
+  - type: http
+    bind_addresses: ['::']
+    port: {{ matrix_synapse_worker_details.replication_port }}
+    resources:
+      - names: {{ replication_http_resources|to_json }}
+{% endif %}
 {% endif %}
 
 {% if matrix_synapse_worker_details.type == 'generic_worker' %}