.env

POSTGRES_URL=postgres://webapp:secret@10.1.0.5:5432/webapp
YUGABYTE_URL=postgres://yugabyte:yugabyte@10.1.0.3:5433/yugabyte

common.py

import os
import random
from urllib.parse import urlparse

from orator import DatabaseManager, Model
from orator_cache import DatabaseManager, Cache
from dotenv import load_dotenv

load_dotenv()

def init_db(key):
    result = urlparse(os.environ[key])
    config = {
        'pgsql': {
            'driver': 'pgsql',
            'host': result.hostname, # random.choice(['10.1.0.2','10.1.0.3','10.1.0.5'])
            'database': result.path[1:],
            'user': result.username,
            'port': result.port,
            'password': result.password,
            'prefix': ''
        }
    }

    db = DatabaseManager(config)
    Model.set_connection_resolver(db)
    db.update('set timezone = 0')
    return db

def safe_random_sample(x, n):
    if len(x) <= n: return x
    return random.sample(x, n)

create_data.py

import random
import sys
import time
from concurrent.futures import ProcessPoolExecutor

from common import init_db, safe_random_sample
from faker import Faker

from progress import Progress

fake = Faker()


if __name__ == '__main__':
    db = init_db(sys.argv[1])
    if '--init' in sys.argv:
        db.update('''
            create extension if not exists "uuid-ossp";
        
            create table channels (
              id uuid primary key default uuid_generate_v1(),
              name text
            );
            
            create table users (
                id uuid primary key default uuid_generate_v1(),
                is_admin boolean,
                name text,
                created_at timestamptz default current_timestamp
            );
            
            create table subscriptions (
                id uuid primary key default uuid_generate_v1(),
                user_id uuid not null references users(id),
                channel_id uuid not null references channels(id),
                unread_count int not null default 0 check ( unread_count >= 0 ),
                created_at timestamptz default current_timestamp,
                updated_at timestamptz default current_timestamp
            );
            
            create unique index on subscriptions (user_id, channel_id);
            create index on subscriptions (user_id, unread_count);     
            
            create table kv (k text primary key, v int); 
            insert into kv (k, v) values ('count', 0); 
        ''')
        exit()

    if '--delete' in sys.argv:
        db.update('truncate subscriptions cascade')
        db.update('truncate users cascade')
        db.update('truncate channels cascade')
        db.update("update kv set v = 0 where k = 'count'")
        exit()

    if '--drop' in sys.argv:
        db.update('drop table if exists subscriptions')
        db.update('drop table if exists users')
        db.update('drop table if exists channels')
        db.update('drop table if exists kv')
        exit()

    subs_count = [1, 16, 3, 12, 36, 40, 233, 184, 5, 1, 24, 98, 94, 1, 8, 1, 493, 8, 4, 2, 27, 249, 1, 56, 1, 5, 110, 1, 9, 1, 16, 1, 2, 1, 47, 2, 342, 273, 135, 3, 6, 2, 17, 2, 4, 497, 2, 5, 1, 28, 3, 103, 1, 53, 8, 19, 154, 7, 49, 3, 542, 5, 46, 1, 7, 16, 58, 4, 2, 7, 17, 161, 22, 3, 5, 5, 519, 5, 9, 1, 76, 72, 890, 3, 5, 1, 1, 2, 6, 1, 1, 1, 4, 178, 2, 8526, 18, 162, 4, 3, 4, 2287, 12, 1, 5, 11, 26, 1, 3, 1, 1, 155, 15, 15, 170, 9, 3, 7, 24, 35, 9, 1, 559, 1, 87, 6, 23, 4, 22, 45, 1, 20, 50, 74, 44, 2, 7, 1, 34, 19, 1, 9, 160, 7, 14, 217, 3, 5, 15, 15, 13, 1, 1, 4, 1, 3, 3, 1, 2, 5, 3, 9, 568, 4, 5, 4, 147, 1, 43, 1, 43, 2, 7, 10, 1, 10, 1, 55, 3, 318, 4, 19, 32, 1, 25, 4, 2, 35, 39, 1, 18, 1, 85, 47, 28, 1, 3, 15, 4, 125, 7, 6, 69, 1, 8, 8, 44, 940, 30, 2, 76, 33, 6, 9, 227, 1, 1, 1, 12, 7, 1, 10, 19, 4, 1, 10, 18, 4, 69, 26, 189, 2, 145, 6, 80, 15, 2, 4270, 6, 1052, 1, 171, 824, 21, 473, 16, 103, 2, 154, 7, 43, 2, 152, 1, 2, 6, 1, 2, 2, 2, 577, 1, 10, 49, 6, 5, 22, 1, 283, 406, 1, 1, 4, 9, 46, 242, 28, 9, 3, 1, 60, 3, 1, 6, 49, 21, 14, 1, 1, 408, 172, 108, 1, 9, 222, 8, 8, 45, 1, 1, 38, 14, 2, 274, 4, 13, 121, 54, 51, 3, 1, 2, 1, 18, 4, 30, 9, 9, 1, 1, 1, 38, 3, 357, 7, 36, 380, 2, 2, 387, 394, 9, 13, 233, 1, 1, 3, 9, 7, 4, 2, 25, 50, 2, 4, 2, 10, 4, 2, 6, 2, 9, 21, 1, 3, 1, 5, 6, 113, 24, 28, 5, 5, 19, 8, 363, 25, 425, 1, 1, 28, 7, 4, 3, 1, 1, 75, 19, 225, 5, 7, 1, 114, 5, 59, 44, 11, 1, 1, 19, 494, 125, 5, 17, 12, 1, 6, 4, 4, 3, 4, 88, 1, 19, 1, 68, 1, 23, 9, 9, 96, 14, 186, 1239, 100, 11, 184, 10, 6, 2, 6, 4, 21, 1, 9, 16, 911, 15, 10, 9, 6, 1, 5, 1, 4, 44, 39, 3, 24, 25, 2, 53, 1, 108, 35, 1, 2, 1, 12, 3, 80, 1, 325, 1, 256, 2, 30, 1, 1, 4, 214, 14, 210, 2, 213, 69, 3, 35, 52, 33, 2, 5, 43, 14, 14, 111, 3, 2, 1, 36, 13, 5, 1, 38, 34, 8, 1, 7, 8, 13, 1, 1, 1, 5, 17, 1, 218, 7, 13, 2, 10, 88, 3, 14, 22, 1, 21, 17, 293, 8, 1, 266, 5, 14, 23, 1, 742, 59, 428, 4, 3, 10, 1, 6, 1, 94, 37, 4, 1, 26, 10, 19, 3, 1, 29, 494, 2, 2, 2, 126, 286, 186, 22, 1, 15, 3, 27, 191, 3, 39, 41, 2, 19, 5, 16, 32, 1, 134, 220, 1, 3, 262, 10, 4, 32, 170, 3, 19, 189, 1, 6, 2, 201, 202, 1454, 2, 2, 16, 6, 4, 1, 20, 3, 7, 57, 1, 28, 10, 8, 4, 107, 2, 6, 24, 3, 322, 119, 20, 52, 7, 7, 1, 2, 129, 1, 3, 1, 113, 2, 3, 70, 7, 37, 6, 668, 2, 1, 1, 1, 263, 1, 2, 16, 12, 20, 1, 2, 8, 8, 10, 168, 2, 23, 2, 1, 17, 2, 8, 30, 2, 7, 115, 16, 3, 10, 16, 1, 181, 117, 1, 75, 1, 785, 8, 1, 3, 1, 11, 605, 7, 1, 6, 34, 3, 12, 118, 2, 2, 80, 56, 3, 5, 11, 9, 129, 5, 13, 47, 3, 7, 20, 3, 2, 10, 12, 78, 217, 6, 223, 77, 7, 44, 283, 2, 163, 118, 21, 439, 2, 2, 1, 4189, 4, 608, 12, 119, 76, 25, 1, 226, 49, 12, 1, 26, 5, 26, 63, 1, 36, 22, 6, 9, 1, 15, 4, 4, 32, 11, 6, 4, 43, 1, 1, 2, 67, 1, 2, 1, 28, 1, 2, 373, 643, 2, 3, 285, 23, 4, 5, 9, 6, 7, 24, 1, 4, 1, 2, 2, 235, 16, 20, 1, 108, 17, 1, 38, 1, 3, 31, 1, 48, 12, 4, 621, 2, 8, 39, 1, 143, 20, 470, 5, 25, 3, 101, 5, 8, 15, 16, 2, 3, 43, 4, 6, 219, 1, 62, 1, 1, 1, 94, 14, 33, 4, 1, 1, 140, 3, 1, 1, 16, 2, 18, 5, 7, 50, 10, 26, 2, 58, 52, 10, 8, 148, 92, 38, 1, 1, 9, 6, 15, 48, 6, 3, 2, 6, 67, 2, 1, 3, 2, 474, 34, 1, 10, 2, 21, 60, 3, 9, 1, 28, 7, 53, 224, 2, 306, 13, 1, 20, 1, 534, 5, 3, 3, 22, 1, 13, 31, 1, 1, 2, 18, 2, 4, 177, 1, 30, 29, 82, 7, 5, 50, 6, 1, 3, 1, 7, 59, 3, 96, 2, 265, 27, 4, 49, 2, 1, 23, 5, 25, 44, 1, 31, 3, 106, 2, 1, 12, 2, 4, 1, 1, 40, 147, 4, 2, 4, 36, 3, 3, 109, 2, 5, 26, 2, 2, 10, 126, 180, 27, 65, 2, 1, 3, 2, 19, 209, 138, 2, 66, 3, 2, 2, 271, 12, 16, 3, 2, 13, 95, 4, 5, 80, 1, 5, 1, 4, 17, 73, 2, 170, 535, 18, 4, 33, 1, 34, 3, 1, 2, 5, 28, 27, 50, 134, 1, 3, 7, 86, 133, 242, 22, 17, 1, 1, 81, 15, 1, 13, 20, 258, 5, 88, 7, 43, 1, 4, 192, 1]

    user_ids = []
    channel_ids = []
    print(f'{len(channel_ids)} channels total')
    print(f'{db.select("select count(*) from subscriptions")[0][0]} subscriptions total')

    def worker(i):
        p = Progress(10_000_000)

        db = init_db(sys.argv[1])
        total_subs = db.select("select v from kv where k = 'count'")[0][0]

        while True:
            p.set_gauge(total_subs)
            print(f'{total_subs} subscriptions total ({total_subs / 11263109. * 100.:.2f}%) {p.left_human()} {p.speed_per_sec()}/sec')
            if total_subs > 10_000_000: break

            t = time.time()
            n=10
            for _ in range(n):
                channel_ids.append(db.select('insert into channels (name) values (%s) returning id', [fake.sentence()[:-1]])[0][0])

                user_id = db.select('insert into users (name) values (%s) returning id', [fake.name()])[0][0]

                channels = safe_random_sample(channel_ids, random.choice(subs_count))
                for channel_id in channels:
                    db.select('insert into subscriptions (user_id, channel_id, unread_count) values (%s, %s, %s) returning id', [user_id, channel_id, random.uniform(0, 20)])

                total_subs = db.select("update kv set v=v+%s where k = 'count' returning v;", [len(channels)])[0][0]
            print(f'Took {time.time() - t}s')


    n_procs = int(sys.argv[2])
    print(f'Running in {n_procs} processes')
    with ProcessPoolExecutor(n_procs) as p:
        for _ in p.map(worker, range(n_procs)):
            print('Done')

progress.py

import time


class Progress:
    def __init__(self, target_value=100, max_values=10):
        self.target_value = target_value
        self.values_at_time = []
        self.cur_value = 0
        self.max_values = max_values

    def set_gauge(self, value):
        self.values_at_time.append((value, time.time()))
        self.values_at_time = self.values_at_time[-self.max_values:]
        self.cur_value = value

        # if decreasing, remove earlier estimates
        while len(self.values_at_time) > 1 and self.values_at_time[0][0] > value:
            self.values_at_time.pop(0)

    def incr(self, value = 1):
        self.cur_value += value
        self.set_gauge(self.cur_value)

    def left_human(self):
        sec = self.left_sec()
        if sec < 0:
            return f'?'
        if sec < 60:
            return f'{sec:.0f} sec'
        if sec < 3600:
            return f'{sec/60:.0f} min'
        if sec < 86400:
            return f'{sec/3600:.1f} hr'

        return f'{sec/86400:.1f} days'

    def left_sec(self):
        left_sec = 0.0
        if len(self.values_at_time) > 0:
            current_value = self.values_at_time[-1][0]
            try:
                left_sec = (self.target_value - current_value) / self.speed_per_sec()
            except ZeroDivisionError:
                return 0
            # print('left sec', left_sec)

        return left_sec

    def speed_per_sec(self):
        if len(self.values_at_time) < 2:
            return 0

        current_value, current_time = self.values_at_time[-1]
        first_value, first_time = self.values_at_time[0]
        # print(f'cur value: {current_value:.7f} last; first_value: {first_value:.7f}')
        time_passed = (current_time - first_time)
        if time_passed == 0:
            return 0
        # print('time_passed', time_passed)
        speed_per_sec = (current_value - first_value) / time_passed
        # print('speed per sec', speed_per_sec)
        return speed_per_sec

    def speed_per_hour(self):
        return self.speed_per_sec() * 3600


if __name__ == '__main__':
    p = Progress(100)
    p.set_gauge(7)
    time.sleep(3)
    p.set_gauge(10)
    print(p.left_sec())  # 90

rf1-master1.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-master \
  --master_addresses 10.1.0.2:7100 \
  --rpc_bind_addresses=10.1.0.2:7100 \
  --fs_data_dirs=/root/yugabyte_data/rf1/master1 \
  --webserver_interface=0.0.0.0 \
  --webserver_port=7000 \
  --replication_factor=1

rf1-tablet1.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-tserver \
  --tserver_master_addrs 10.1.0.2:7100 \
  --rpc_bind_addresses 10.1.0.2:9100 \
  --start_pgsql_proxy \
  --pgsql_proxy_bind_address 10.1.0.2:5433 \
  --fs_data_dirs=/root/yugabyte_data/rf1/tablet1 \
  --webserver_port=9001 \
  --redis_proxy_bind_address=10.1.0.2:6379 \
  --redis_proxy_webserver_port=11000 \
  --cql_proxy_bind_address=10.1.0.2:9042 \
  --cql_proxy_webserver_port=12000

rf1-tablet1a.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-tserver \
  --tserver_master_addrs 10.1.0.2:7100 \
  --rpc_bind_addresses 10.1.0.2:9101 \
  --start_pgsql_proxy \
  --pgsql_proxy_bind_address 10.1.0.2:5434 \
  --fs_data_dirs=/root/yugabyte_data/rf1/tablet1a \
  --webserver_port=9002 \
  --redis_proxy_bind_address=10.1.0.2:6380 \
  --redis_proxy_webserver_port=11001 \
  --cql_proxy_bind_address=10.1.0.2:9043 \
  --cql_proxy_webserver_port=12001

rf3-master1.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-master \
  --master_addresses 10.1.0.2:7100,10.1.0.3:7100,10.1.0.5:7100 \
  --rpc_bind_addresses=10.1.0.2:7100 \
  --fs_data_dirs=/root/yugabyte_data/rf3/master1 \
  --webserver_interface=0.0.0.0 \
  --webserver_port=7000 \
  --replication_factor=3

rf3-master2.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-master \
  --master_addresses 10.1.0.2:7100,10.1.0.3:7100,10.1.0.5:7100 \
  --rpc_bind_addresses=10.1.0.3:7100 \
  --fs_data_dirs=/root/yugabyte_data/rf3/master2 \
  --webserver_interface=0.0.0.0 \
  --webserver_port=7000 \
  --replication_factor=3

rf3-master3.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-master \
  --master_addresses 10.1.0.2:7100,10.1.0.3:7100,10.1.0.5:7100 \
  --rpc_bind_addresses=10.1.0.5:7100 \
  --fs_data_dirs=/root/yugabyte_data/rf3/master3 \
  --webserver_interface=0.0.0.0 \
  --webserver_port=7000 \
  --replication_factor=3

rf3-tablet1.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-tserver \
  --tserver_master_addrs 10.1.0.2:7100,10.1.0.3:7100,10.1.0.5:7100 \
  --rpc_bind_addresses 10.1.0.2:9100 \
  --start_pgsql_proxy \
  --pgsql_proxy_bind_address 10.1.0.2:5433 \
  --fs_data_dirs=/root/yugabyte_data/rf3/tablet1 \
  --webserver_port=9001 \
  --redis_proxy_bind_address=10.1.0.2:6379 \
  --redis_proxy_webserver_port=11000 \
  --cql_proxy_bind_address=10.1.0.2:9042 \
  --cql_proxy_webserver_port=12000

rf3-tablet2.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-tserver \
  --tserver_master_addrs 10.1.0.2:7100,10.1.0.3:7100,10.1.0.5:7100 \
  --rpc_bind_addresses 10.1.0.3:9100 \
  --start_pgsql_proxy \
  --pgsql_proxy_bind_address 10.1.0.3:5433 \
  --fs_data_dirs=/root/yugabyte_data/rf3/tablet2 \
  --webserver_port=9001 \
  --redis_proxy_bind_address=10.1.0.3:6379 \
  --redis_proxy_webserver_port=11000 \
  --cql_proxy_bind_address=10.1.0.3:9042 \
  --cql_proxy_webserver_port=12000

rf3-tablet3.sh

#!/usr/bin/env bash

/root/yugabyte-2.9.0.0/bin/yb-tserver \
  --tserver_master_addrs 10.1.0.2:7100,10.1.0.3:7100,10.1.0.5:7100 \
  --rpc_bind_addresses 10.1.0.5:9100 \
  --start_pgsql_proxy \
  --pgsql_proxy_bind_address 10.1.0.5:5433 \
  --fs_data_dirs=/root/yugabyte_data/rf3/tablet3 \
  --webserver_port=9001 \
  --redis_proxy_bind_address=10.1.0.5:6379 \
  --redis_proxy_webserver_port=11000 \
  --cql_proxy_bind_address=10.1.0.5:9042 \
  --cql_proxy_webserver_port=12000