Merge remote-tracking branch 'ar/migrate-to-postgres'

admin/sql/create_primary_keys.sql

@@ -1,5 +1,6 @@
 BEGIN;
 
 ALTER TABLE "user" ADD CONSTRAINT user_pkey PRIMARY KEY (id);
+ALTER TABLE listens ADD CONSTRAINT listens_pkey PRIMARY KEY (user_id, ts);
 
 COMMIT;

admin/sql/create_tables.sql

@@ -8,4 +8,14 @@ CREATE TABLE "user" (
 );
 ALTER TABLE "user" ADD CONSTRAINT user_musicbrainz_id_key UNIQUE (musicbrainz_id);
 
+CREATE TABLE listens (
+  id              SERIAL,
+  user_id         VARCHAR NOT NULL,
+  ts              TIMESTAMP WITH TIME ZONE NOT NULL,
+  artist_msid     UUID NOT NULL,
+  album_msid      UUID,
+  recording_msid  UUID NOT NULL,
+  raw_data        JSONB
+);
+
 COMMIT;

config.py.sample

@@ -7,6 +7,12 @@ SECRET_KEY = "CHANGE_ME"
 
 # Primary database
 SQLALCHEMY_DATABASE_URI = "postgresql://listenbrainz@/listenbrainz"
+# Oldest listens which can be stored in the database, in days.
+MAX_POSTGRES_LISTEN_HISTORY = "90"
+# Log Postgres queries if they execeed this time, in milliseconds.
+PG_QUERY_TIMEOUT = "3000"
+# Disable synchronous_commit to OFF for Postgres. Default: False
+PG_ASYNC_LISTEN_COMMIT = "False"
 
 MESSYBRAINZ_SQLALCHEMY_DATABASE_URI = "postgresql://messybrainz@/messybrainz"
 
@@ -26,12 +32,16 @@ KAFKA_RUN_CLASS_BINARY = "/opt/kafka/bin/kafka-run-class.sh"
 # CASSANDRA
 CASSANDRA_SERVER = "localhost"
 CASSANDRA_KEYSPACE = "listenbrainz"
+CASSANDRA_REPLICATION_FACTOR = "1"
 
 
 # MusicBrainz OAuth
 MUSICBRAINZ_CLIENT_ID = "CLIENT_ID"
 MUSICBRAINZ_CLIENT_SECRET = "CLIENT_SECRET"
 
+# Last.fm
+LASTFM_API_KEY = "USE_LASTFM_API_KEY"
+
 
 # LOGGING
 

listenstore/.gitignore

@@ -2,3 +2,4 @@
 build
 listenbrainz_store.egg-info
 dist
+listenstore.conf

listenstore/README.md

@@ -1,6 +1,5 @@
 
 ## Development
-
     $ PYTHONPATH=. bin/listenstore-test.py
 
 ## Pip, eggs, etc
@@ -13,12 +12,14 @@ then install the egg:
 
 bin scripts get installed to `/usr/local/bin`.
 
+## Configuration
+Listenstore uses same configuration file as used for ListenBrainz. If the
+location of the `config.py` is changed, the change them to appropriate values.
 
 ## Schema
 
-Cassandra table called "listens" stores everything. It is keyed on `(uid,
-idkey)`, where idkey is the first 3 numbers in the unixtimestamp of the
-listen.
+Postgres table called "listens" stores everything. It is keyed on `(user_id,
+ts)`, where ts is the timestamp (with timezone) of the listen.
 
 This effectively shards listens by a user over multiple row keys, since
 very wide rows are inefficient.

listenstore/bin/listenstore-kafka-to-cassandra-writer.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python
 import logging
 from listenstore.cli import Command
+from listenstore.listenstore import CassandraListenStore
+
 
 
 class KafkaToCassandra(Command):
@@ -11,7 +13,13 @@ def __init__(self):
         self.log = logging.getLogger(__name__)
 
     def run(self):
-        self.kafkaConsumer.start_listens(self.listenStore)
+        self.kafka_consumer.start_listens(self.listen_store)
+
+    @property
+    def listen_store(self):
+        self._listenStore = CassandraListenStore(self.config)
+        return self._listenStore
+
 
 
 if __name__ == '__main__':

listenstore/bin/listenstore-kafka-to-postgres-writer.py

@@ -0,0 +1,26 @@
+#!/usr/bin/env python
+import logging
+from listenstore.cli import Command
+from listenstore.listenstore import PostgresListenStore
+
+
+
+class KafkaToPostgres(Command):
+    desc = "Print listens fetched from kafka"
+
+    def __init__(self):
+        super(KafkaToPostgres, self).__init__()
+        self.log = logging.getLogger(__name__)
+
+    def run(self):
+        self.kafka_consumer.start_listens(self.listen_store)
+
+    @property
+    def listen_store(self):
+        self._listenStore = PostgresListenStore(self.config)
+        return self._listenStore
+
+
+
+if __name__ == '__main__':
+    KafkaToPostgres().start()

listenstore/bin/listenstore-test.py

@@ -9,9 +9,9 @@ def run(self):
         self.log.info("ListenPrinter starting..")
         #unixtime = int(calendar.timegm(time.gmtime()))
         #item = {'user_id':'rj','listened_at':unixtime,'body':{'foo':'bar','now':time.strftime("%c")}}
-        #self.listenStore.insert(item)
+        #self.listen_store.insert(item)
 
-        res = self.listenStore.fetch_listens(uid='rj')
+        res = self.listen_store.fetch_listens(uid='rj')
         for r in res:
             print repr(r)
 

listenstore/bin/listenstore-tool.py

@@ -13,7 +13,7 @@ def __init__(self):
     def run(self):
         uid = self.config['userid']
         self.log.info("ListenTool starting for uid %s.." % (uid))
-        res = self.listenStore.fetch_listens(uid=uid, limit=self.config['limit'])
+        res = self.listen_store.fetch_listens(uid=uid, limit=self.config['limit'])
         for r in res:
             print repr(r)
 

listenstore/listenstore/cli.py

@@ -3,7 +3,7 @@
 import logging
 
 from .utils import argparse_factory, parse_args_and_config
-from .listenstore import ListenStore
+from .listenstore import PostgresListenStore, CassandraListenStore
 from .kafkaconsumer import KafkaConsumer
 
 
@@ -24,7 +24,7 @@ def __init__(self):
 
     # NB: only sets level after our Command starts running
     def set_log_level(self):
-        l = self.config['loglevel']
+        l = self.config['LOGLEVEL']
         lev = logging.INFO
         if l == "DEBUG":
             lev = logging.DEBUG
@@ -61,13 +61,14 @@ def run():
         raise NotImplementedError()
 
     @property
-    def listenStore(self):
-        if self._listenStore is None:
-            self._listenStore = ListenStore(self.config)
-        return self._listenStore
+    def listen_store(self):
+        """ Override this method in bin scripts to support writing
+            to both Casandra and Postgres
+        """
+        pass
 
     @property
-    def kafkaConsumer(self):
+    def kafka_consumer(self):
         if self._kafkaConsumer is None:
             self._kafkaConsumer = KafkaConsumer(self.config)
         return self._kafkaConsumer

listenstore/listenstore/kafkaconsumer.py

@@ -4,6 +4,7 @@
 from .listen import Listen
 from time import time, sleep
 from cassandra.cluster import NoHostAvailable
+from .utils import wrapper, time_it
 
 KAFKA_READ_TIMEOUT = 5
 CASSANDRA_BATCH_SIZE = 1000
@@ -12,10 +13,11 @@
 class KafkaConsumer(object):
     def __init__(self, conf):
         self.log = logging.getLogger(__name__)
-        self.client = KafkaClient(conf["kafka_server"])
+        self.client = KafkaClient(conf["KAFKA_CONNECT"])
         self.total_inserts = 0
         self.inserts = 0
         self.listenstore = None
+        self.conf = conf
 
 
     def start_listens(self, listenstore):
@@ -51,7 +53,13 @@ def start(self, group_name, topic_name):
                 broken = True
                 while broken:
                     try:
-                        self.listenstore.insert_batch(listens)
+                        threshold = 10000       # Default threshold in ms
+                        if 'PG_QUERY_TIMEOUT' in self.conf:
+                            threshold = int(self.conf['PG_QUERY_TIMEOUT'])
+
+                        time_it(wrapper(self.listenstore.insert_postgresql, listens), \
+                                threshold, self.log, 'Threshold Postgres_Query_Time exceeded')
+
                         broken = False
                     except ValueError as e:
                         self.log.error("Cannot insert listens: %s" % unicode(e))