Remove etcd

Dockerfile

@@ -32,6 +32,6 @@ WORKDIR /app
 
 ENV PYTHONPATH "${PYTHONPATH}:/${WORKDIR}"
 
-EXPOSE 8899
+EXPOSE 50101
 
 CMD ["python", "./__init__.py"]

README.md

@@ -15,7 +15,7 @@ Install and run:
     * `python3 -m venv venv`
     * `source venv/bin/activate`
 * Run supporting infrastructure using Docker
-    * `./setup-docker.sh` 
-    * `docker-compose up -d`
+    * See the instructions in the [killrvideo-docker-common][https://github.com/KillrVideo/killrvideo-docker-common] repository for running the supporting infrastructure 
+    * The documentation describes an option to run the KillrVideo python services in a Docker container
 * Run the Python KillrVideo services
     * `python killrvideo/__init__.py`

VERSION

@@ -1 +1 @@
-DOCKER_BUILD_TAG=0.9.1
+DOCKER_BUILD_TAG=3.0.1-rc1

killrvideo/__init__.py

@@ -1,15 +1,14 @@
 from concurrent import futures
 import grpc
-import etcd
 import time
 import logging
 import json
 import os
 
-from dse.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT, EXEC_PROFILE_GRAPH_DEFAULT
-from dse_graph import DseGraph
-from dse.auth import PlainTextAuthProvider
 from dse import ConsistencyLevel
+from dse.auth import PlainTextAuthProvider
+from dse.cluster import Cluster, ExecutionProfile, EXEC_PROFILE_DEFAULT, EXEC_PROFILE_GRAPH_DEFAULT, NoHostAvailable
+from dse_graph import DseGraph
 from dse.policies import TokenAwarePolicy, DCAwareRoundRobinPolicy
 import dse.cqlengine.connection
 
@@ -37,28 +36,14 @@ def serve():
 
     dse_username = os.getenv('KILLRVIDEO_DSE_USERNAME')
     dse_password = os.getenv('KILLRVIDEO_DSE_PASSWORD')
+    dse_contact_points = os.getenv('KILLRVIDEO_DSE_CONTACT_POINTS', 'dse').split(',')
+    service_port = os.getenv('KILLRVIDEO_SERVICE_PORT', '50101')
 
     file = open('config.json', 'r')
     config = json.load(file)
 
-    service_port = config['SERVICE_PORT']
-    service_host = config['SERVICE_HOST']
-    etcd_port = config['ETCD_PORT']
-    contact_points = config['CONTACT_POINTS']
     default_consistency_level = config['DEFAULT_CONSISTENCY_LEVEL']
 
-    service_address = service_host + ":" + str(service_port)
-    etcd_client = etcd.Client(host=service_host, port=etcd_port)
-
-    # Wait for Cassandra (DSE) to be up, aka registered in etcd
-    while True:
-        try:
-            etcd_client.read('/killrvideo/services/cassandra')
-            break # if we get here, Cassandra is registered and should be available
-        except etcd.EtcdKeyNotFound:
-            logging.info('Waiting for Cassandra to be registered in etcd, sleeping 10s')
-            time.sleep(10)
-
     # Initialize Cassandra Driver and Mapper
     load_balancing_policy = TokenAwarePolicy(DCAwareRoundRobinPolicy())
     profile = ExecutionProfile(consistency_level=ConsistencyLevel.name_to_value[default_consistency_level],
@@ -69,11 +54,25 @@ def serve():
     if dse_username:
         auth_provider = PlainTextAuthProvider(username=dse_username, password=dse_password)
 
-    cluster = Cluster(contact_points=contact_points,
-                      execution_profiles={EXEC_PROFILE_DEFAULT: profile, EXEC_PROFILE_GRAPH_DEFAULT: graph_profile},
-                      auth_provider = auth_provider)
+    # Wait for Cassandra (DSE) to be up
+    session = None
+    while not session:
+        try:
+            session = Cluster(contact_points=dse_contact_points,
+                              execution_profiles={EXEC_PROFILE_DEFAULT: profile, EXEC_PROFILE_GRAPH_DEFAULT: graph_profile},
+                              auth_provider = auth_provider).connect("killrvideo")
+        except (NoHostAvailable):
+            logging.info('Waiting for Cassandra (DSE) to be available')
+            time.sleep(10)
+
+    # Additional retry loop to check if dummy keyspace exists
+    while True:
+        logging.info('Checking for schema to be created...')
+        result = session.execute('SELECT keyspace_name FROM system_schema.keyspaces WHERE keyspace_name=\'kv_init_done\'')
+        if result.one(): # any result indicates keyspace has been created
+            break
+        time.sleep(10)
 
-    session = cluster.connect("killrvideo")
     dse.cqlengine.connection.set_session(session)
 
     # Initialize GRPC Server
@@ -90,19 +89,9 @@ def serve():
     VideoCatalogServiceServicer(grpc_server, VideoCatalogService(session=session))
 
     # Start GRPC Server
-    grpc_server.add_insecure_port('[::]:' + str(service_port))
+    grpc_server.add_insecure_port('[::]:' + service_port)
     grpc_server.start()
 
-    # Register Services with etcd
-    etcd_client.write('/killrvideo/services/CommentsService/killrvideo-python', service_address)
-    etcd_client.write('/killrvideo/services/RatingsService/killrvideo-python', service_address)
-    etcd_client.write('/killrvideo/services/SearchService/killrvideo-python', service_address)
-    etcd_client.write('/killrvideo/services/StatisticsService/killrvideo-python', service_address)
-    etcd_client.write('/killrvideo/services/SuggestedVideoService/killrvideo-python', service_address)
-    #etcd_client.write('/killrvideo/services/UploadsService/killrvideo-python', service_address)
-    etcd_client.write('/killrvideo/services/UserManagementService/killrvideo-python', service_address)
-    etcd_client.write('/killrvideo/services/VideoCatalogService/killrvideo-python', service_address)
-
     # Keep application alive
     try:
         while True:

killrvideo/comments/comments_events_kafka.py

@@ -1,3 +1,4 @@
+import os
 from kafka import KafkaProducer
 from .comments_events_pb2 import UserCommentedOnVideo
 from common.common_types_conversions import UUID_to_grpc, datetime_to_Timestamp, TimeUUID_to_grpc
@@ -8,7 +9,7 @@ class CommentsPublisher(object):
     """Provides methods that publish events associated with the Comments Service."""
 
     def __init__(self):
-        self.producer = KafkaProducer(bootstrap_servers='10.0.75.1:9092',
+        self.producer = KafkaProducer(bootstrap_servers=os.getenv('KILLRVIDEO_KAFKA', 'kafka'),
                                       client_id='killrvideo-python:CommentsService')
 
 

killrvideo/config.json

@@ -1,7 +1,3 @@
 {
-  "SERVICE_PORT": 8899,
-  "SERVICE_HOST": "10.0.75.1",
-  "ETCD_PORT": 2379,
-  "CONTACT_POINTS": ["10.0.75.1"],
   "DEFAULT_CONSISTENCY_LEVEL": "LOCAL_QUORUM"
 }

killrvideo/ratings/ratings_events_kafka.py

@@ -1,4 +1,5 @@
 import logging
+import os
 from kafka import KafkaProducer
 from .ratings_events_pb2 import UserRatedVideo
 from common.common_types_conversions import UUID_to_grpc, grpc_to_UUID, datetime_to_Timestamp, Timestamp_to_datetime
@@ -9,7 +10,7 @@ class RatingPublisher(object):
     """Provides methods that publish events associated with the Ratings Service."""
 
     def __init__(self):
-        self.producer = KafkaProducer(bootstrap_servers='10.0.75.1:9092',
+        self.producer = KafkaProducer(bootstrap_servers=os.getenv('KILLRVIDEO_KAFKA', 'kafka'),
                                       client_id='killrvideo-python:RatingsService')
 
 

killrvideo/suggested_videos/suggested_videos_events_kafka.py

@@ -1,4 +1,5 @@
 import logging
+import os
 import threading
 from kafka import KafkaConsumer
 from user_management.user_management_events_pb2 import UserCreated
@@ -10,16 +11,13 @@
 USER_RATED_VIDEO_TOPIC = 'topic-kv-videoRating'
 YOUTUBE_VIDEO_ADDED_TOPIC = 'topic-kv-videoCreation'
 
-BOOTSTRAP_SERVERS = '10.0.75.1:9092'
-
-
 class SuggestedVideoKafkaConsumer(threading.Thread):
     def __init__(self, suggested_videos_consumer):
         threading.Thread.__init__(self)
         self.suggested_videos_consumer = suggested_videos_consumer
 
     def run(self):
-        consumer = KafkaConsumer(bootstrap_servers=BOOTSTRAP_SERVERS,
+        consumer = KafkaConsumer(bootstrap_servers=os.getenv('KILLRVIDEO_KAFKA', 'kafka'),
                                  client_id='killrvideo-python:SuggestedVideosService',
                                  fetch_max_wait_ms=10000) # poll every 10 seconds for new events
 

killrvideo/user_management/user_management_events_kafka.py

@@ -1,4 +1,5 @@
 import logging
+import os
 from kafka import KafkaProducer
 from .user_management_events_pb2 import UserCreated
 from common.common_types_conversions import UUID_to_grpc, datetime_to_Timestamp
@@ -9,7 +10,7 @@ class UserManagementPublisher(object):
     """Provides methods that publish events associated with the User Management Service."""
 
     def __init__(self):
-        self.producer = KafkaProducer(bootstrap_servers='10.0.75.1:9092',
+        self.producer = KafkaProducer(bootstrap_servers=os.getenv('KILLRVIDEO_KAFKA', 'kafka'),
                                       client_id='killrvideo-python:UserManagementService')
 
 

killrvideo/video_catalog/video_catalog_events_kafka.py

@@ -1,4 +1,5 @@
 import logging
+import os
 from kafka import KafkaProducer
 from .video_catalog_events_pb2 import YouTubeVideoAdded
 from common.common_types_conversions import UUID_to_grpc, datetime_to_Timestamp
@@ -9,7 +10,7 @@ class VideoCatalogPublisher(object):
     """Provides methods that publish events associated with the Video Catalog Service."""
 
     def __init__(self):
-        self.producer = KafkaProducer(bootstrap_servers='10.0.75.1:9092',
+        self.producer = KafkaProducer(bootstrap_servers=os.getenv('KILLRVIDEO_KAFKA', 'kafka'),
                                       client_id='killrvideo-python:VideoCatalogService')