ashokc

{ "type": "set_tempo", "tempo": 600000, "time": 0}, # tick = 0.025 seconds, time is #ticks
...
...
{ "type": "note_on", "time": 0, "note": 60, "velocity": 127, "channel": 0}, # note 60 (Key C4) pressed down                                                        
{ "type": "note_on", "time": 18, "note": 60, "velocity": 0, "channel": 0}, # Release C4 after 18*0.025 sec                                           
{ "type": "note_on", "time": 0, "note": 62, "velocity": 127, "channel": 0}, # Press D4 after 0 sec                                            
{ "type": "note_on", "time": 6, "note": 62, "velocity": 0, "channel": 0},                                            
{ "type": "note_on", "time": 0, "note": 64, "velocity": 127, "channel": 0},                                            
{ "type": "note_on", "time": 12, "note": 64, "velocity": 0, "channel": 0},                                                        
...

ashokc

PCollection<KV<String, RawVertex>> vertices = pipeline
  .apply(KafkaIO.<String, GenericRecord>read() // Read from a kafka topic
           
  ...
// key is the vertex label: "A" , "B" or "C"
  .withKeyDeserializer(StringDeserializer.class)
// value is an Avro object
  .withValueDeserializer(ConfluentSchemaRegistryDeserializerProvider.of(schema_registry,topic+"-value")
  .withCreateTime(Duration.ZERO) // event createtime is used by Beam windows & watermark
  .apply(ParDo.of(new EmitRawVertex())); // PCollection<KV<String, RawVertex>> 

ashokc

from locust import HttpLocust, TaskSet, task, between
import random, sys, json

# We have 7 ways of engaging the API via Http
url = "http://localhost/gunicorn/quotes" # for nginx + gunicorn
#url = "http://localhost/uwsgi/quotes" # for nginx + uwsgi
#url = "http://localhost/uwsgi-http/quotes" # for nginx + uwsgi-http
#url = "http://localhost/werkzeug/quotes" # for nginx + werkzeug
#url = "http://localhost:9999/quotes" # for direct gunicorn
#url = "http://localhost:9997/quotes" # for direct uwsgi-http

ashokc

#!/bin/bash
/usr/bin/cmonitor_collector --sampling-interval=3 --output-filename=./system_metrics.json # start gathering memory/cpu usage sats
sleep 10
pipenv run locust -f ./load_tests.py --host localhost --csv ./results -c 500 -r 10 -t 60m --no-web
sleep 10
pid=$(ps -ef | grep 'cmonitor_collector' | grep -v 'grep' | awk '{ print $2 }')
kill -15 $pid # stop the monitor

ashokc

    location /werkzeug/ {
      proxy_pass http://127.0.0.1:9996/;
    }
    location /uwsgi-http/ {
      proxy_pass http://127.0.0.1:9997/;
    }
    location /uwsgi/ {
      rewrite ^/uwsgi/(.*) /$1 break;
      include uwsgi_params;
      uwsgi_pass 127.0.0.1:9998;

ashokc

[program:werkzeug]
command=.../virtualenvs/.../bin/python .../quoteserver/quotes.py
directory=.../quoteserver
autostart=true
redirect_stderr=true
stdout_logfile=/var/log/supervisor/werkzeug.log

ashokc

# in quotes.py
if __name__ == "__main__":
    app.logger.disabled = True
    log = logging.getLogger('werkzeug')
    log.disabled = True
    app.run(host='localhost', port=9996, debug=False, threaded=True)

ashokc

[uwsgi] more wsgi.py 

from quotes import app as application

if __name__ == "__main__":
    application.run()

ashokc

[program:uwsgi-http]
command=.../virtualenvs/.../bin/uwsgi --chdir .../quoteserver --wsgi-file wsgi.py --http 127.0.0.1:9997 --processes 6 --threads 1 --master --disable-logging --log-4xx --log-5xx
directory=.../quoteserver
autostart=true
stdout_logfile=/var/log/supervisor/uwsgi-http.log
redirect_stderr=true

ashokc

workers = 6
threads = 1
worker_class = 'sync'
loglevel = 'ERROR'
error_log = '-' # means stdout
access_log = '-' # means stdout
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"'